func()

in apps/couchbase.go [46:167]
113 lines of code
2 McCabe index (conditional complexity)

func (r MetricsReceiverCouchbase) Pipelines(_ context.Context) ([]otel.ReceiverPipeline, error) {
	targets := []string{r.Endpoint}
	if r.Endpoint == "" {
		targets = []string{defaultCouchbaseEndpoint}
	}

	config := map[string]interface{}{
		"config": map[string]interface{}{
			"scrape_configs": []map[string]interface{}{
				{
					"job_name":        r.Type(),
					"scrape_interval": r.CollectionIntervalString(),
					"basic_auth": map[string]interface{}{
						"username": r.Username,
						"password": r.Password.SecretValue(),
					},
					"metric_relabel_configs": []map[string]interface{}{
						{
							"source_labels": []string{"__name__"},
							"regex":         "(kv_ops)|(kv_vb_curr_items)|(kv_num_vbuckets)|(kv_total_memory_used_bytes)|(kv_ep_num_value_ejects)|(kv_ep_mem_high_wat)|(kv_ep_mem_low_wat)|(kv_ep_oom_errors)",
							"action":        "keep",
						},
					},
					"static_configs": []map[string]interface{}{
						{
							"targets": targets,
						},
					},
				},
			},
		},
	}
	return []otel.ReceiverPipeline{{
		Receiver: otel.Component{
			Type:   "prometheus",
			Config: config,
		},
		Processors: map[string][]otel.Component{"metrics": {
			otel.NormalizeSums(),
			// remove prometheus scraping meta-metrics
			otel.MetricsFilter("exclude", "strict",
				"scrape_samples_post_metric_relabeling",
				"scrape_series_added",
				"scrape_duration_seconds",
				"scrape_samples_scraped",
				"up",
			),
			otel.MetricsTransform(
				// renaming from prometheus style to otel style, order is important before workload prefix
				otel.RenameMetric(
					"kv_ops",
					"couchbase.bucket.operation.count",
					otel.ToggleScalarDataType,
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_vb_curr_items",
					"couchbase.bucket.item.count",
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_num_vbuckets",
					"couchbase.bucket.vbucket.count",
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_total_memory_used_bytes",
					"couchbase.bucket.memory.usage",
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_ep_num_value_ejects",
					"couchbase.bucket.item.ejection.count",
					otel.ToggleScalarDataType,
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_ep_mem_high_wat",
					"couchbase.bucket.memory.high_water_mark.limit",
					otel.RenameLabel("bucket", "bucket_name")),
				otel.RenameMetric(
					"kv_ep_mem_low_wat",
					"couchbase.bucket.memory.low_water_mark.limit",
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_ep_tmp_oom_errors",
					"couchbase.bucket.error.oom.count.recoverable",
					otel.ToggleScalarDataType,
					otel.RenameLabel("bucket", "bucket_name"),
				),
				otel.RenameMetric(
					"kv_ep_oom_errors",
					"couchbase.bucket.error.oom.count.unrecoverable",
					otel.ToggleScalarDataType,
					otel.RenameLabel("bucket", "bucket_name"),
				),

				// combine OOM metrics
				otel.CombineMetrics(
					`^couchbase\.bucket\.error\.oom\.count\.(?P<error_type>unrecoverable|recoverable)$$`,
					"couchbase.bucket.error.oom.count",
				),

				// group by bucket and op
				otel.UpdateMetric(
					`couchbase.bucket.operation.count`,
					map[string]interface{}{
						"action":           "aggregate_labels",
						"label_set":        []string{"bucket_name", "op"},
						"aggregation_type": "sum",
					},
				),

				otel.AddPrefix("workload.googleapis.com"),
			),
			// Using the transform processor for metrics
			otel.TransformationMetrics(r.transformMetrics()...),
			otel.ModifyInstrumentationScope(r.Type(), "1.0"),
		}},
	}}, nil
}