feat(observability): attribute dashboard panels per cluster and server

core/internal/cnpgi/backup.go

@@ -117,16 +117,16 @@ func (b backupServiceImplementation) Backup(
 	)
 
 	backupStart := time.Now()
-	recordBackupStart(ctx)
-	defer recordBackupFinished(ctx)
+	recordBackupStart(ctx, cluster.Name)
+	defer recordBackupFinished(ctx, cluster.Name)
 
 	metadata, err := b.runBackup(
 		ctx,
 		backupName,
 		isPrimary,
 	)
 	if err != nil {
-		recordBackupFailure(ctx, time.Since(backupStart), err)
+		recordBackupFailure(ctx, cluster.Name, time.Since(backupStart), err)
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "backup failed")
 
@@ -138,14 +138,14 @@ func (b backupServiceImplementation) Backup(
 	// verification is folded into the successful-backup recording below.
 	corruption, verifyErr := b.runVerify(ctx, backupName)
 	if corruption {
-		recordBackupFailure(ctx, time.Since(backupStart), verifyErr)
+		recordBackupFailure(ctx, cluster.Name, time.Since(backupStart), verifyErr)
 		span.RecordError(verifyErr)
 		span.SetStatus(codes.Error, "verification detected corruption")
 
 		return nil, verifyErr
 	}
 
-	recordBackupSuccess(ctx, time.Since(backupStart))
+	recordBackupSuccess(ctx, cluster.Name, time.Since(backupStart))
 
 	return &backup.BackupResult{
 		BackupName:        backupName,

core/internal/cnpgi/metrics.go

@@ -23,44 +23,63 @@ import (
 	"context"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 
 	"github.com/cloudnative-pg/klio/core/internal/opentelemetry"
 )
 
+// clusterAttr returns the `cluster_name` attribute every plugin backup metric
+// carries, so panels can attribute backup activity to a specific PostgreSQL
+// cluster even when several clusters share one namespace.
+func clusterAttr(clusterName string) attribute.KeyValue {
+	return opentelemetry.AttributeKeyClusterName.Of(clusterName)
+}
+
 // recordBackupStart records that a backup has started. Callers must pair this
 // with a deferred recordBackupFinished so the in-progress counter decrements
 // on every exit path, including panics.
-func recordBackupStart(ctx context.Context) {
-	opentelemetry.PluginBackup.LatestStartTime.Record(ctx, time.Now().Unix())
-	opentelemetry.PluginBackup.InProgress.Add(ctx, 1)
+func recordBackupStart(ctx context.Context, clusterName string) {
+	cluster := clusterAttr(clusterName)
+	opentelemetry.PluginBackup.LatestStartTime.Record(ctx, time.Now().Unix(),
+		metric.WithAttributes(cluster))
+	opentelemetry.PluginBackup.InProgress.Add(ctx, 1, metric.WithAttributes(cluster))
 }
 
 // recordBackupFinished decrements the in-progress counter. Always invoke via
 // defer immediately after recordBackupStart so concurrent backup accounting
-// stays correct even when a backup panics or returns early.
-func recordBackupFinished(ctx context.Context) {
-	opentelemetry.PluginBackup.InProgress.Add(ctx, -1)
+// stays correct even when a backup panics or returns early. It must pass the
+// same clusterName as recordBackupStart so the up/down counter cancels out per
+// cluster.
+func recordBackupFinished(ctx context.Context, clusterName string) {
+	opentelemetry.PluginBackup.InProgress.Add(ctx, -1,
+		metric.WithAttributes(clusterAttr(clusterName)))
 }
 
 // recordBackupSuccess records a successful backup completion.
-func recordBackupSuccess(ctx context.Context, duration time.Duration) {
-	opentelemetry.PluginBackup.LatestCompletionTime.Record(ctx, time.Now().Unix())
-	opentelemetry.PluginBackup.LatestDuration.Record(ctx, duration.Seconds())
+func recordBackupSuccess(ctx context.Context, clusterName string, duration time.Duration) {
+	cluster := clusterAttr(clusterName)
+	opentelemetry.PluginBackup.LatestCompletionTime.Record(ctx, time.Now().Unix(),
+		metric.WithAttributes(cluster))
+	opentelemetry.PluginBackup.LatestDuration.Record(ctx, duration.Seconds(),
+		metric.WithAttributes(cluster))
 	opentelemetry.PluginBackup.Duration.Record(ctx, duration.Seconds(),
-		metric.WithAttributes(opentelemetry.OutcomeSuccess.Attribute()))
+		metric.WithAttributes(cluster, opentelemetry.OutcomeSuccess.Attribute()))
 	opentelemetry.PluginBackup.Runs.Add(ctx, 1,
-		metric.WithAttributes(opentelemetry.OutcomeSuccess.Attribute()))
+		metric.WithAttributes(cluster, opentelemetry.OutcomeSuccess.Attribute()))
 }
 
 // recordBackupFailure records a failed backup.
-func recordBackupFailure(ctx context.Context, duration time.Duration, err error) {
+func recordBackupFailure(ctx context.Context, clusterName string, duration time.Duration, err error) {
+	cluster := clusterAttr(clusterName)
 	category := classifyRunBackupError(ctx, err)
-	opentelemetry.PluginBackup.LatestFailureTime.Record(ctx, time.Now().Unix())
+	opentelemetry.PluginBackup.LatestFailureTime.Record(ctx, time.Now().Unix(),
+		metric.WithAttributes(cluster))
 	opentelemetry.PluginBackup.Duration.Record(ctx, duration.Seconds(),
-		metric.WithAttributes(opentelemetry.OutcomeFailure.Attribute()))
+		metric.WithAttributes(cluster, opentelemetry.OutcomeFailure.Attribute()))
 	opentelemetry.PluginBackup.Runs.Add(ctx, 1,
 		metric.WithAttributes(
+			cluster,
 			opentelemetry.OutcomeFailure.Attribute(),
 			opentelemetry.AttributeKeyFailureCategory.Of(category.Name),
 		))

core/internal/cnpgi/metrics_test.go

@@ -29,13 +29,18 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 
 	"github.com/cloudnative-pg/klio/core/internal/backupfailure"
 	"github.com/cloudnative-pg/klio/core/internal/opentelemetry"
 )
 
+// testClusterName is the PostgreSQL cluster name every record* call in these
+// tests tags its metrics with.
+const testClusterName = "cluster-test"
+
 // setupTestMeter installs a test MeterProvider with a ManualReader,
 // re-creates all instruments against it, and returns the reader.
 func setupTestMeter(t *testing.T) *sdkmetric.ManualReader {
@@ -197,7 +202,7 @@ func TestRecordBackupStart(t *testing.T) {
 	reader := setupTestMeter(t)
 
 	before := time.Now().Unix()
-	recordBackupStart(context.Background())
+	recordBackupStart(context.Background(), testClusterName)
 
 	rm := collectOTelMetrics(t, reader)
 
@@ -213,11 +218,11 @@ func TestRecordBackupStart(t *testing.T) {
 func TestRecordBackupSuccess(t *testing.T) {
 	reader := setupTestMeter(t)
 
-	recordBackupStart(context.Background())
+	recordBackupStart(context.Background(), testClusterName)
 	duration := 42 * time.Second
 	before := time.Now().Unix()
-	recordBackupSuccess(context.Background(), duration)
-	recordBackupFinished(context.Background())
+	recordBackupSuccess(context.Background(), testClusterName, duration)
+	recordBackupFinished(context.Background(), testClusterName)
 
 	rm := collectOTelMetrics(t, reader)
 
@@ -252,15 +257,15 @@ func TestRecordBackupSuccess(t *testing.T) {
 func TestRecordBackupFailure(t *testing.T) {
 	reader := setupTestMeter(t)
 
-	recordBackupStart(t.Context())
+	recordBackupStart(t.Context(), testClusterName)
 	before := time.Now().Unix()
 
 	//nolint:gosec // hardcoded test input
 	exitErr := exec.CommandContext(t.Context(), "sh",
 		"-c", fmt.Sprintf("exit %d", backupfailure.RepositoryError.ExitCode)).Run()
 
-	recordBackupFailure(t.Context(), 7*time.Second, exitErr)
-	recordBackupFinished(t.Context())
+	recordBackupFailure(t.Context(), testClusterName, 7*time.Second, exitErr)
+	recordBackupFinished(t.Context(), testClusterName)
 
 	rm := collectOTelMetrics(t, reader)
 
@@ -294,16 +299,16 @@ func TestRecordBackupFailureCategoriesAreSeparateSeries(t *testing.T) {
 	expiredCtx, cancelTimeout := context.WithTimeout(t.Context(), 0)
 	defer cancelTimeout()
 
-	recordBackupFailure(expiredCtx, time.Second, nil)
+	recordBackupFailure(expiredCtx, testClusterName, time.Second, nil)
 	canceledCtx, cancelCanceled := context.WithCancel(t.Context())
 	cancelCanceled()
-	recordBackupFailure(canceledCtx, time.Second, nil)
-	recordBackupFailure(canceledCtx, time.Second, nil)
+	recordBackupFailure(canceledCtx, testClusterName, time.Second, nil)
+	recordBackupFailure(canceledCtx, testClusterName, time.Second, nil)
 
 	//nolint:gosec // hardcoded test input
 	exitErr := exec.CommandContext(t.Context(), "sh",
 		"-c", fmt.Sprintf("exit %d", backupfailure.Verification.ExitCode)).Run()
-	recordBackupFailure(t.Context(), time.Second, exitErr)
+	recordBackupFailure(t.Context(), testClusterName, time.Second, exitErr)
 
 	rm := collectOTelMetrics(t, reader)
 
@@ -327,7 +332,7 @@ func TestRecordBackupFailureCategoriesAreSeparateSeries(t *testing.T) {
 func TestRecordBackupFailureNilErrorDefaultsToUnknown(t *testing.T) {
 	reader := setupTestMeter(t)
 
-	recordBackupFailure(context.Background(), time.Second, nil)
+	recordBackupFailure(context.Background(), testClusterName, time.Second, nil)
 
 	rm := collectOTelMetrics(t, reader)
 
@@ -337,23 +342,113 @@ func TestRecordBackupFailureNilErrorDefaultsToUnknown(t *testing.T) {
 	assert.Equal(t, int64(1), unknown)
 }
 
+// metricAttributeSets returns the attribute set of every data point on a
+// metric, across the aggregation types the plugin backup instruments use
+// (Int64/Float64 gauges, the Int64 up/down counter and runs counter, and the
+// Float64 duration histogram).
+func metricAttributeSets(a metricdata.Aggregation) []attribute.Set {
+	var sets []attribute.Set
+	switch d := a.(type) {
+	case metricdata.Gauge[int64]:
+		for _, dp := range d.DataPoints {
+			sets = append(sets, dp.Attributes)
+		}
+	case metricdata.Gauge[float64]:
+		for _, dp := range d.DataPoints {
+			sets = append(sets, dp.Attributes)
+		}
+	case metricdata.Sum[int64]:
+		for _, dp := range d.DataPoints {
+			sets = append(sets, dp.Attributes)
+		}
+	case metricdata.Histogram[float64]:
+		for _, dp := range d.DataPoints {
+			sets = append(sets, dp.Attributes)
+		}
+	}
+
+	return sets
+}
+
+// dataPointClusterNames collects the distinct `cluster_name` attribute values
+// seen across every plugin backup instrument (the empty string keys the data
+// points that carry no cluster_name), so a test can assert every series is
+// attributed to a cluster.
+func dataPointClusterNames(rm metricdata.ResourceMetrics) map[string]int {
+	names := map[string]int{}
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			for _, set := range metricAttributeSets(m.Data) {
+				if v, ok := set.Value("cluster_name"); ok {
+					names[v.AsString()]++
+				} else {
+					names[""]++
+				}
+			}
+		}
+	}
+
+	return names
+}
+
+func TestBackupMetricsCarryClusterName(t *testing.T) {
+	reader := setupTestMeter(t)
+
+	recordBackupStart(context.Background(), testClusterName)
+	recordBackupSuccess(context.Background(), testClusterName, 5*time.Second)
+	recordBackupFinished(context.Background(), testClusterName)
+	recordBackupFailure(context.Background(), testClusterName, time.Second, assert.AnError)
+
+	rm := collectOTelMetrics(t, reader)
+
+	names := dataPointClusterNames(rm)
+	require.NotEmpty(t, names)
+	assert.Equal(t, 0, names[""], "every plugin backup data point must carry a cluster_name attribute")
+	assert.Contains(t, names, testClusterName)
+}
+
+func TestBackupMetricsClustersAreSeparateSeries(t *testing.T) {
+	reader := setupTestMeter(t)
+
+	recordBackupStart(context.Background(), "cluster-a")
+	recordBackupSuccess(context.Background(), "cluster-a", 5*time.Second)
+	recordBackupFinished(context.Background(), "cluster-a")
+
+	recordBackupStart(context.Background(), "cluster-b")
+	recordBackupSuccess(context.Background(), "cluster-b", 9*time.Second)
+	recordBackupFinished(context.Background(), "cluster-b")
+
+	rm := collectOTelMetrics(t, reader)
+
+	// The runs counter must expose one data point per (cluster_name, outcome),
+	// not a single folded series.
+	perCluster := map[string]int64{}
+	for _, dp := range findInt64SumDataPoints(rm, opentelemetry.PluginBackupRunsMetric) {
+		v, ok := dp.Attributes.Value("cluster_name")
+		require.True(t, ok, "runs data point missing cluster_name")
+		perCluster[v.AsString()] += dp.Value
+	}
+	assert.Equal(t, int64(1), perCluster["cluster-a"])
+	assert.Equal(t, int64(1), perCluster["cluster-b"])
+}
+
 func TestBackupMetricsMultipleRuns(t *testing.T) {
 	reader := setupTestMeter(t)
 
 	// First backup: success.
-	recordBackupStart(context.Background())
-	recordBackupSuccess(context.Background(), 10*time.Second)
-	recordBackupFinished(context.Background())
+	recordBackupStart(context.Background(), testClusterName)
+	recordBackupSuccess(context.Background(), testClusterName, 10*time.Second)
+	recordBackupFinished(context.Background(), testClusterName)
 
 	// Second backup: failure.
-	recordBackupStart(context.Background())
-	recordBackupFailure(context.Background(), 3*time.Second, assert.AnError)
-	recordBackupFinished(context.Background())
+	recordBackupStart(context.Background(), testClusterName)
+	recordBackupFailure(context.Background(), testClusterName, 3*time.Second, assert.AnError)
+	recordBackupFinished(context.Background(), testClusterName)
 
 	// Third backup: success.
-	recordBackupStart(context.Background())
-	recordBackupSuccess(context.Background(), 30*time.Second)
-	recordBackupFinished(context.Background())
+	recordBackupStart(context.Background(), testClusterName)
+	recordBackupSuccess(context.Background(), testClusterName, 30*time.Second)
+	recordBackupFinished(context.Background(), testClusterName)
 
 	rm := collectOTelMetrics(t, reader)