chore: consolidation time metric (#35)

* chore: add ConsolidationRunDuration metric

* refactor: add env attribute to the provider instead of every metric

* go mod tidy

* tidy again

Author: volmedo

cmd/etracker/start.go

@@ -33,8 +33,6 @@ import (
 	"github.com/storacha/etracker/internal/presets"
 	"github.com/storacha/etracker/internal/server"
 	"github.com/storacha/etracker/internal/service"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/metric"
 )
 
 var startCmd = &cobra.Command{
@@ -179,8 +177,6 @@ func startService(cmd *cobra.Command, args []string) error {
 		}
 	}
 
-	env := cfg.MetricsEnvironment
-
 	// Create DynamoDB client
 	dynamoClient := dynamodb.NewFromConfig(cfg.AWSConfig)
 
@@ -203,7 +199,7 @@ func startService(cmd *cobra.Command, args []string) error {
 
 	// Initialize metrics if metrics are configured
 	if cfg.MetricsAuthToken != "" {
-		if err := metrics.Init(); err != nil {
+		if err := metrics.Init(cfg.MetricsEnvironment); err != nil {
 			return fmt.Errorf("initializing metrics: %w", err)
 		}
 
@@ -212,16 +208,14 @@ func startService(cmd *cobra.Command, args []string) error {
 		if err != nil {
 			log.Warnf("failed to reconcile unprocessed batches count: %v", err)
 		} else {
-			envAttr := attribute.String("env", env)
-			metrics.UnprocessedBatches.Add(ctx, count, metric.WithAttributeSet(attribute.NewSet(envAttr)))
+			metrics.UnprocessedBatches.Add(ctx, count)
 			log.Infof("reconciled unprocessed batches count: %d", count)
 		}
 	}
 
 	// Create service
 	svc, err := service.New(
 		id,
-		env,
 		egressTable,
 		consolidatedTable,
 		storageProviderTable,
@@ -272,7 +266,6 @@ func startService(cmd *cobra.Command, args []string) error {
 
 	cons, err := consolidator.New(
 		id,
-		env,
 		egressTable,
 		consolidatedTable,
 		spaceStatsTable,

go.mod

@@ -20,6 +20,7 @@ require (
 	go.opentelemetry.io/otel v1.38.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.60.0
 	go.opentelemetry.io/otel/metric v1.38.0
+	go.opentelemetry.io/otel/sdk v1.38.0
 	go.opentelemetry.io/otel/sdk/metric v1.38.0
 )
 
@@ -116,7 +117,6 @@ require (
 	github.com/ucan-wg/go-ucan v0.0.0-20240916120445-37f52863156c // indirect
 	github.com/whyrusleeping/cbor-gen v0.3.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
 	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/mock v0.5.2 // indirect

internal/consolidator/consolidator.go

@@ -45,7 +45,6 @@ var ErrNotFound = consolidated.ErrNotFound
 
 type Consolidator struct {
 	id                    principal.Signer
-	environment           string
 	egressTable           egress.EgressTable
 	consolidatedTable     consolidated.ConsolidatedTable
 	spaceStatsTable       spacestats.SpaceStatsTable
@@ -61,7 +60,6 @@ type Consolidator struct {
 
 func New(
 	id principal.Signer,
-	environment string,
 	egressTable egress.EgressTable,
 	consolidatedTable consolidated.ConsolidatedTable,
 	spaceStatsTable spacestats.SpaceStatsTable,
@@ -91,7 +89,6 @@ func New(
 
 	c := &Consolidator{
 		id:                    id,
-		environment:           environment,
 		egressTable:           egressTable,
 		consolidatedTable:     consolidatedTable,
 		spaceStatsTable:       spaceStatsTable,
@@ -145,8 +142,12 @@ func (c *Consolidator) Stop() {
 func (c *Consolidator) Consolidate(ctx context.Context) error {
 	log.Info("Starting consolidation cycle")
 
-	// Environment attribute for metrics
-	envAttr := attribute.String("env", c.environment)
+	// Track consolidation run duration
+	startTime := time.Now()
+	defer func() {
+		durationMs := time.Since(startTime).Milliseconds()
+		metrics.ConsolidationRunDuration.Record(ctx, durationMs)
+	}()
 
 	// Get unprocessed records
 	records, err := c.egressTable.GetUnprocessed(ctx, c.batchSize)
@@ -226,7 +227,7 @@ func (c *Consolidator) Consolidate(ctx context.Context) error {
 
 		// Increment consolidated bytes counter for this node
 		nodeAttr := attribute.String("node", record.Node.String())
-		metrics.ConsolidatedBytesPerNode.Add(ctx, int64(totalEgress), metric.WithAttributeSet(attribute.NewSet(nodeAttr, envAttr)))
+		metrics.ConsolidatedBytesPerNode.Add(ctx, int64(totalEgress), metric.WithAttributeSet(attribute.NewSet(nodeAttr)))
 
 		bLog.Infof("Consolidated %d bytes", totalEgress)
 	}
@@ -236,7 +237,7 @@ func (c *Consolidator) Consolidate(ctx context.Context) error {
 		return fmt.Errorf("marking records as processed: %w", err)
 	}
 
-	metrics.UnprocessedBatches.Add(ctx, int64(-len(records)), metric.WithAttributeSet(attribute.NewSet(envAttr)))
+	metrics.UnprocessedBatches.Add(ctx, int64(-len(records)))
 
 	log.Infof("Consolidation cycle completed. Processed %d records", len(records))
 

internal/consolidator/consolidator_test.go

@@ -83,7 +83,6 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 	// Create a consolidator instance to test the validation context it creates works as expected
 	c, err := New(
 		consolidatorID,
-		"test",
 		nil,
 		nil,
 		nil,

internal/metrics/metrics.go

@@ -5,9 +5,11 @@ import (
 
 	logging "github.com/ipfs/go-log/v2"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/metric"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
 )
 
 var log = logging.Logger("metrics")
@@ -21,17 +23,26 @@ var (
 
 	// UnprocessedBatches keeps track of the total number of batches pending consolidation
 	UnprocessedBatches metric.Int64UpDownCounter
+
+	// ConsolidationRunDuration tracks the time (in milliseconds) each consolidation run takes to process all batches
+	ConsolidationRunDuration metric.Int64Histogram
 )
 
 // Init initializes the OpenTelemetry metrics with Prometheus exporter
-func Init() error {
+func Init(environment string) error {
 	exporter, err := prometheus.New()
 	if err != nil {
 		return fmt.Errorf("failed to create prometheus exporter: %w", err)
 	}
 
-	// Create a MeterProvider with the Prometheus exporter
-	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(exporter))
+	// Create a resource with the environment attribute
+	res := resource.NewSchemaless(attribute.String("env", environment))
+
+	// Create a MeterProvider with the Prometheus exporter and resource
+	provider := sdkmetric.NewMeterProvider(
+		sdkmetric.WithReader(exporter),
+		sdkmetric.WithResource(res),
+	)
 
 	// Set the global MeterProvider
 	otel.SetMeterProvider(provider)
@@ -64,6 +75,14 @@ func Init() error {
 		return fmt.Errorf("failed to create UnprocessedBatches counter: %w", err)
 	}
 
+	ConsolidationRunDuration, err = meter.Int64Histogram(
+		"etracker_consolidation_run_duration_ms",
+		metric.WithDescription("Time in milliseconds for each consolidation run to process all batches"),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create ConsolidationRunDuration histogram: %w", err)
+	}
+
 	log.Info("OpenTelemetry metrics initialized with Prometheus exporter")
 	return nil
 }

internal/service/service.go

@@ -107,7 +107,6 @@ type Service interface {
 
 type service struct {
 	id                   principal.Signer
-	environment          string
 	egressTable          egress.EgressTable
 	consolidatedTable    consolidated.ConsolidatedTable
 	storageProviderTable storageproviders.StorageProviderTable
@@ -118,7 +117,6 @@ type service struct {
 
 func New(
 	id principal.Signer,
-	environment string,
 	egressTable egress.EgressTable,
 	consolidatedTable consolidated.ConsolidatedTable,
 	storageProviderTable storageproviders.StorageProviderTable,
@@ -128,7 +126,6 @@ func New(
 ) (*service, error) {
 	return &service{
 		id:                   id,
-		environment:          environment,
 		egressTable:          egressTable,
 		consolidatedTable:    consolidatedTable,
 		storageProviderTable: storageProviderTable,
@@ -144,9 +141,8 @@ func (s *service) Record(ctx context.Context, node did.DID, receipts ucan.Link,
 	}
 
 	nodeAttr := attribute.String("node", node.String())
-	envAttr := attribute.String("env", s.environment)
-	metrics.TrackedBatchesPerNode.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(nodeAttr, envAttr)))
-	metrics.UnprocessedBatches.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(envAttr)))
+	metrics.TrackedBatchesPerNode.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(nodeAttr)))
+	metrics.UnprocessedBatches.Add(ctx, 1)
 
 	return nil
 }