send flow name with batch id metrics

PeerDB-io · Jan 16, 2025 · f52382e · f52382e
1 parent b910baa
commit f52382e
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go
@@ -11,6 +11,8 @@ import (
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/log"
 	"go.temporal.io/sdk/temporal"
@@ -315,7 +317,9 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		if err != nil {
 			logger.Error("Failed to get current batch id gauge", slog.Any("error", err))
 		} else {
-			currentBatchID.Record(ctx, res.CurrentSyncBatchID)
+			currentBatchID.Record(ctx, res.CurrentSyncBatchID, metric.WithAttributeSet(attribute.NewSet(
+				attribute.String(otel_metrics.FlowNameKey, flowName),
+			)))
 		}
 	}
 
@@ -670,7 +674,9 @@ func (a *FlowableActivity) normalizeLoop(
 					if err != nil {
 						logger.Error("Failed to get normalized batch id gauge", slog.Any("error", err))
 					} else {
-						lastNormalizedBatchID.Record(ctx, req.BatchID)
+						lastNormalizedBatchID.Record(ctx, req.BatchID, metric.WithAttributeSet(attribute.NewSet(
+							attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
+						)))
 					}
 				}
 				break