| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package couchdb |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "maps" |
| 12 | "math" |
| 13 | "net/http" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | |
| 17 | "github.com/netdata/netdata/go/plugins/pkg/stm" |
| 18 | "github.com/netdata/netdata/go/plugins/pkg/web" |
| 19 | ) |
| 20 | |
| 21 | const ( |
| 22 | urlPathActiveTasks = "/_active_tasks" |
| 23 | urlPathOverviewStats = "/_node/%s/_stats" |
| 24 | urlPathSystemStats = "/_node/%s/_system" |
| 25 | urlPathDatabases = "/_dbs_info" |
| 26 | |
| 27 | httpStatusCodePrefix = "couchdb_httpd_status_codes_" |
| 28 | httpStatusCodePrefixLen = len(httpStatusCodePrefix) |
| 29 | ) |
| 30 | |
| 31 | func (c *Collector) collect() (map[string]int64, error) { |
| 32 | ms := c.scrapeCouchDB() |
| 33 | if ms.empty() { |
| 34 | return nil, nil |
| 35 | } |
| 36 | |
| 37 | collected := make(map[string]int64) |
| 38 | c.collectNodeStats(collected, ms) |
| 39 | c.collectSystemStats(collected, ms) |
| 40 | c.collectActiveTasks(collected, ms) |
| 41 | c.collectDBStats(collected, ms) |
| 42 | |
| 43 | return collected, nil |
| 44 | } |
| 45 | |
| 46 | func (c *Collector) collectNodeStats(collected map[string]int64, ms *cdbMetrics) { |
| 47 | if !ms.hasNodeStats() { |
| 48 | return |
| 49 | } |
| 50 | |
| 51 | for metric, value := range stm.ToMap(ms.NodeStats) { |
| 52 | collected[metric] = value |
| 53 | if strings.HasPrefix(metric, httpStatusCodePrefix) { |
| 54 | code := metric[httpStatusCodePrefixLen:] |
| 55 | collected["couchdb_httpd_status_codes_"+string(code[0])+"xx"] += value |
| 56 | } |
| 57 | } |
| 58 | } |
| 59 | |
| 60 | func (c *Collector) collectSystemStats(collected map[string]int64, ms *cdbMetrics) { |
| 61 | if !ms.hasNodeSystem() { |
| 62 | return |
| 63 | } |
| 64 | |
| 65 | maps.Copy(collected, stm.ToMap(ms.NodeSystem)) |
| 66 | |
| 67 | collected["peak_msg_queue"] = findMaxMQSize(ms.NodeSystem.MessageQueues) |
| 68 | } |
| 69 | |
| 70 | func (c *Collector) collectActiveTasks(collected map[string]int64, ms *cdbMetrics) { |
| 71 | collected["active_tasks_indexer"] = 0 |
| 72 | collected["active_tasks_database_compaction"] = 0 |
| 73 | collected["active_tasks_replication"] = 0 |
| 74 | collected["active_tasks_view_compaction"] = 0 |
| 75 | |
| 76 | if !ms.hasActiveTasks() { |
| 77 | return |
| 78 | } |
| 79 | |
| 80 | for _, task := range ms.ActiveTasks { |
| 81 | collected["active_tasks_"+task.Type]++ |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | func (c *Collector) collectDBStats(collected map[string]int64, ms *cdbMetrics) { |
| 86 | if !ms.hasDBStats() { |
| 87 | return |
| 88 | } |
| 89 | |
| 90 | for _, dbStats := range ms.DBStats { |
| 91 | if dbStats.Error != "" { |
| 92 | c.Warning("database '", dbStats.Key, "' doesn't exist") |
| 93 | continue |
| 94 | } |
| 95 | merge(collected, stm.ToMap(dbStats.Info), "db_"+dbStats.Key) |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | func (c *Collector) scrapeCouchDB() *cdbMetrics { |
| 100 | ms := &cdbMetrics{} |
| 101 | wg := &sync.WaitGroup{} |
| 102 | |
| 103 | wg.Go(func() { c.scrapeNodeStats(ms) }) |
| 104 | |
| 105 | wg.Go(func() { c.scrapeSystemStats(ms) }) |
| 106 | |
| 107 | wg.Go(func() { c.scrapeActiveTasks(ms) }) |
| 108 | |
| 109 | if len(c.databases) > 0 { |
| 110 | wg.Go(func() { c.scrapeDBStats(ms) }) |
| 111 | } |
| 112 | |
| 113 | wg.Wait() |
| 114 | return ms |
| 115 | } |
| 116 | |
| 117 | func (c *Collector) scrapeNodeStats(ms *cdbMetrics) { |
| 118 | req, _ := web.NewHTTPRequestWithPath(c.RequestConfig, fmt.Sprintf(urlPathOverviewStats, c.Config.Node)) |
| 119 | |
| 120 | var stats cdbNodeStats |
| 121 | |
| 122 | if err := c.client().RequestJSON(req, &stats); err != nil { |
| 123 | c.Warning(err) |
| 124 | return |
| 125 | } |
| 126 | |
| 127 | ms.NodeStats = &stats |
| 128 | } |
| 129 | |
| 130 | func (c *Collector) scrapeSystemStats(ms *cdbMetrics) { |
| 131 | req, _ := web.NewHTTPRequestWithPath(c.RequestConfig, fmt.Sprintf(urlPathSystemStats, c.Config.Node)) |
| 132 | |
| 133 | var stats cdbNodeSystem |
| 134 | |
| 135 | if err := c.client().RequestJSON(req, &stats); err != nil { |
| 136 | c.Warning(err) |
| 137 | return |
| 138 | } |
| 139 | |
| 140 | ms.NodeSystem = &stats |
| 141 | } |
| 142 | |
| 143 | func (c *Collector) scrapeActiveTasks(ms *cdbMetrics) { |
| 144 | req, _ := web.NewHTTPRequestWithPath(c.RequestConfig, urlPathActiveTasks) |
| 145 | |
| 146 | var stats []cdbActiveTask |
| 147 | |
| 148 | if err := c.client().RequestJSON(req, &stats); err != nil { |
| 149 | c.Warning(err) |
| 150 | return |
| 151 | } |
| 152 | |
| 153 | ms.ActiveTasks = stats |
| 154 | } |
| 155 | |
| 156 | func (c *Collector) scrapeDBStats(ms *cdbMetrics) { |
| 157 | req, _ := web.NewHTTPRequestWithPath(c.RequestConfig, urlPathDatabases) |
| 158 | req.Method = http.MethodPost |
| 159 | req.Header.Add("Accept", "application/json") |
| 160 | req.Header.Add("Content-Type", "application/json") |
| 161 | |
| 162 | var q struct { |
| 163 | Keys []string `json:"keys"` |
| 164 | } |
| 165 | q.Keys = c.databases |
| 166 | body, err := json.Marshal(q) |
| 167 | if err != nil { |
| 168 | c.Error(err) |
| 169 | return |
| 170 | } |
| 171 | req.Body = io.NopCloser(bytes.NewReader(body)) |
| 172 | |
| 173 | var stats []cdbDBStats |
| 174 | |
| 175 | if err := c.client().RequestJSON(req, &stats); err != nil { |
| 176 | c.Warning(err) |
| 177 | return |
| 178 | } |
| 179 | |
| 180 | ms.DBStats = stats |
| 181 | } |
| 182 | |
| 183 | func findMaxMQSize(MessageQueues map[string]any) int64 { |
| 184 | var maxSize float64 |
| 185 | for _, mq := range MessageQueues { |
| 186 | switch mqSize := mq.(type) { |
| 187 | case float64: |
| 188 | maxSize = math.Max(maxSize, mqSize) |
| 189 | case map[string]any: |
| 190 | if v, ok := mqSize["count"].(float64); ok { |
| 191 | maxSize = math.Max(maxSize, v) |
| 192 | } |
| 193 | } |
| 194 | } |
| 195 | return int64(maxSize) |
| 196 | } |
| 197 | |
| 198 | func (c *Collector) pingCouchDB() error { |
| 199 | req, err := web.NewHTTPRequest(c.RequestConfig) |
| 200 | if err != nil { |
| 201 | return err |
| 202 | } |
| 203 | |
| 204 | var info struct{ Couchdb string } |
| 205 | |
| 206 | if err := c.client().RequestJSON(req, &info); err != nil { |
| 207 | return err |
| 208 | } |
| 209 | |
| 210 | if info.Couchdb != "Welcome" { |
| 211 | return errors.New("not a CouchDB endpoint") |
| 212 | } |
| 213 | |
| 214 | return nil |
| 215 | } |
| 216 | |
| 217 | func (c *Collector) client() *web.Client { |
| 218 | return web.DoHTTP(c.httpClient).OnNokCode(func(resp *http.Response) (bool, error) { |
| 219 | var msg struct { |
| 220 | Error string `json:"error"` |
| 221 | Reason string `json:"reason"` |
| 222 | } |
| 223 | if err := json.NewDecoder(resp.Body).Decode(&msg); err == nil && msg.Error != "" { |
| 224 | return false, fmt.Errorf("error '%s', reason '%s'", msg.Error, msg.Reason) |
| 225 | } |
| 226 | return false, nil |
| 227 | }) |
| 228 | } |
| 229 | |
| 230 | func merge(dst, src map[string]int64, prefix string) { |
| 231 | for k, v := range src { |
| 232 | dst[prefix+"_"+k] = v |
| 233 | } |
| 234 | } |