feat: allow collecting node-level metrics for Couchbase buckets

plugins/inputs/couchbase/README.md

@@ -29,6 +29,14 @@ This plugin gets metrics for each Couchbase node, as well as detailed metrics fo
   ## Use TLS but skip chain & host verification (defaults to false)
   ## If set to false, tls_cert and tls_key are required
   # insecure_skip_verify = false
+
+  ## Whether to collect cluster-wide bucket statistics
+  ## It is recommended to disable this in favor of node_stats
+  ## to get a better view of the cluster.
+  cluster_bucket_stats = true
+
+  ## Whether to collect bucket stats for each individual node
+  node_bucket_stats = false
 ```
 
 ## Measurements
@@ -45,12 +53,13 @@ Fields:
 - memory_free (unit: bytes, example: 23181365248.0)
 - memory_total (unit: bytes, example: 64424656896.0)
 
-### couchbase_bucket
+### couchbase_bucket and couchbase_node_bucket
 
 Tags:
 
 - cluster: whatever you called it in `servers` in the configuration, e.g.: `http://couchbase-0.example.com/`)
 - bucket: the name of the couchbase bucket, e.g., `blastro-df`
+- hostname: the hostname of the node the bucket metrics were collected from, e.g., `172.16.10.187:8091` (only present in `couchbase_node_bucket`)
 
 Default bucket fields:
 

plugins/inputs/couchbase/couchbase.go

@@ -20,6 +20,9 @@ type Couchbase struct {
 
 	BucketStatsIncluded []string `toml:"bucket_stats_included"`
 
+	ClusterBucketStats bool `toml:"cluster_bucket_stats"`
+	NodeBucketStats    bool `toml:"node_bucket_stats"`
+
 	bucketInclude filter.Filter
 	client        *http.Client
 
@@ -75,32 +78,55 @@ func (cb *Couchbase) gatherServer(acc telegraf.Accumulator, addr string) error {
 		acc.AddFields("couchbase_node", fields, tags)
 	}
 
-	for bucketName := range pool.BucketMap {
-		tags := map[string]string{"cluster": escapedAddr, "bucket": bucketName}
-		bs := pool.BucketMap[bucketName].BasicStats
-		fields := make(map[string]interface{})
-		cb.addBucketField(fields, "quota_percent_used", bs["quotaPercentUsed"])
-		cb.addBucketField(fields, "ops_per_sec", bs["opsPerSec"])
-		cb.addBucketField(fields, "disk_fetches", bs["diskFetches"])
-		cb.addBucketField(fields, "item_count", bs["itemCount"])
-		cb.addBucketField(fields, "disk_used", bs["diskUsed"])
-		cb.addBucketField(fields, "data_used", bs["dataUsed"])
-		cb.addBucketField(fields, "mem_used", bs["memUsed"])
-
-		err := cb.gatherDetailedBucketStats(addr, bucketName, fields)
-		if err != nil {
-			return err
+	for name, bucket := range pool.BucketMap {
+		cluster := regexpURI.ReplaceAllString(addr, "${1}")
+
+		if cb.ClusterBucketStats {
+			fields := cb.basicBucketStats(bucket.BasicStats)
+			tags := map[string]string{"cluster": cluster, "bucket": name}
+
+			err := cb.gatherDetailedBucketStats(addr, name, nil, fields)
+			if err != nil {
+				return err
+			}
+
+			acc.AddFields("couchbase_bucket", fields, tags)
 		}
 
-		acc.AddFields("couchbase_bucket", fields, tags)
+		if cb.NodeBucketStats {
+			for _, node := range bucket.Nodes() {
+				fields := cb.basicBucketStats(bucket.BasicStats)
+				tags := map[string]string{"cluster": cluster, "bucket": name, "hostname": node.Hostname}
+
+				err := cb.gatherDetailedBucketStats(addr, name, &node.Hostname, fields)
+				if err != nil {
+					return err
+				}
+
+				acc.AddFields("couchbase_node_bucket", fields, tags)
+			}
+		}
 	}
 
 	return nil
 }
 
-func (cb *Couchbase) gatherDetailedBucketStats(server, bucket string, fields map[string]interface{}) error {
+// basicBucketStats gets the basic bucket statistics
+func (cb *Couchbase) basicBucketStats(basicStats map[string]interface{}) map[string]interface{} {
+	fields := make(map[string]interface{})
+	cb.addBucketField(fields, "quota_percent_used", basicStats["quotaPercentUsed"])
+	cb.addBucketField(fields, "ops_per_sec", basicStats["opsPerSec"])
+	cb.addBucketField(fields, "disk_fetches", basicStats["diskFetches"])
+	cb.addBucketField(fields, "item_count", basicStats["itemCount"])
+	cb.addBucketField(fields, "disk_used", basicStats["diskUsed"])
+	cb.addBucketField(fields, "data_used", basicStats["dataUsed"])
+	cb.addBucketField(fields, "mem_used", basicStats["memUsed"])
+	return fields
+}
+
+func (cb *Couchbase) gatherDetailedBucketStats(server, bucket string, nodeHostname *string, fields map[string]interface{}) error {
 	extendedBucketStats := &BucketStats{}
-	err := cb.queryDetailedBucketStats(server, bucket, extendedBucketStats)
+	err := cb.queryDetailedBucketStats(server, bucket, nodeHostname, extendedBucketStats)
 	if err != nil {
 		return err
 	}
@@ -341,9 +367,15 @@ func (cb *Couchbase) addBucketFieldChecked(fields map[string]interface{}, fieldK
 	cb.addBucketField(fields, fieldKey, values[index])
 }
 
-func (cb *Couchbase) queryDetailedBucketStats(server, bucket string, bucketStats *BucketStats) error {
+func (cb *Couchbase) queryDetailedBucketStats(server, bucket string, nodeHostname *string, bucketStats *BucketStats) error {
+	url := server + "/pools/default/buckets/" + bucket
+	if nodeHostname != nil {
+		url += "/nodes/" + *nodeHostname
+	}
+	url += "/stats?"
+
 	// Set up an HTTP request to get the complete set of bucket stats.
-	req, err := http.NewRequest("GET", server+"/pools/default/buckets/"+bucket+"/stats?", nil)
+	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
 		return err
 	}
@@ -391,6 +423,7 @@ func init() {
 	inputs.Add("couchbase", func() telegraf.Input {
 		return &Couchbase{
 			BucketStatsIncluded: []string{"quota_percent_used", "ops_per_sec", "disk_fetches", "item_count", "disk_used", "data_used", "mem_used"},
+			ClusterBucketStats:  true,
 		}
 	})
 }