fix: Remove invocationID from metric point attributes

Author: murarustefaan

scheduler/metrics/metrics.go

@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -20,42 +21,38 @@ const (
 	durationMetricName  = "sync.table.duration"
 )
 
-func NewMetrics(invocationID string) *Metrics {
-	resources, err := otel.Meter(ResourceName).Int64Counter(resourcesMetricName,
-		metric.WithDescription("Number of resources synced for a table"),
-		metric.WithUnit("/{tot}"),
-	)
-	if err != nil {
-		return nil
-	}
-
-	errors, err := otel.Meter(ResourceName).Int64Counter(errorsMetricName,
-		metric.WithDescription("Number of errors encountered while syncing a table"),
-		metric.WithUnit("/{tot}"),
-	)
-	if err != nil {
-		return nil
-	}
-
-	panics, err := otel.Meter(ResourceName).Int64Counter(panicsMetricName,
-		metric.WithDescription("Number of panics encountered while syncing a table"),
-		metric.WithUnit("/{tot}"),
-	)
-	if err != nil {
-		return nil
-	}
+var (
+	resources metric.Int64Counter
+	errors    metric.Int64Counter
+	panics    metric.Int64Counter
+	duration  metric.Int64Counter
+	once      sync.Once
+)
 
-	duration, err := otel.Meter(ResourceName).Int64Counter(durationMetricName,
-		metric.WithDescription("Duration of syncing a table"),
-		metric.WithUnit("ms"),
-	)
-	if err != nil {
-		return nil
-	}
+func NewMetrics(invocationID string) *Metrics {
+	once.Do(func() {
+		resources, _ = otel.Meter(ResourceName).Int64Counter(resourcesMetricName,
+			metric.WithDescription("Number of resources synced for a table"),
+			metric.WithUnit("/{tot}"),
+		)
+
+		errors, _ = otel.Meter(ResourceName).Int64Counter(errorsMetricName,
+			metric.WithDescription("Number of errors encountered while syncing a table"),
+			metric.WithUnit("/{tot}"),
+		)
+
+		panics, _ = otel.Meter(ResourceName).Int64Counter(panicsMetricName,
+			metric.WithDescription("Number of panics encountered while syncing a table"),
+			metric.WithUnit("/{tot}"),
+		)
+
+		duration, _ = otel.Meter(ResourceName).Int64Counter(durationMetricName,
+			metric.WithDescription("Duration of syncing a table"),
+			metric.WithUnit("ms"),
+		)
+	})
 
 	return &Metrics{
-		invocationID: invocationID,
-
 		resources: resources,
 		errors:    errors,
 		panics:    panics,
@@ -66,8 +63,6 @@ func NewMetrics(invocationID string) *Metrics {
 }
 
 type Metrics struct {
-	invocationID string
-
 	resources metric.Int64Counter
 	errors    metric.Int64Counter
 	panics    metric.Int64Counter
@@ -88,59 +83,23 @@ type measurement struct {
 	duration  *durationMeasurement
 }
 
-func (m *measurement) Equal(other *measurement) bool {
-	return m.resources == other.resources && m.errors == other.errors && m.panics == other.panics && m.duration == other.duration
-}
-
-// Equal compares to stats. Mostly useful in testing
-func (m *Metrics) Equal(other *Metrics) bool {
-	for table, clientStats := range m.measurements {
-		for client, stats := range clientStats.clients {
-			if _, ok := other.measurements[table]; !ok {
-				return false
-			}
-			if _, ok := other.measurements[table].clients[client]; !ok {
-				return false
-			}
-			if !stats.Equal(other.measurements[table].clients[client]) {
-				return false
-			}
-		}
-	}
-	for table, clientStats := range other.measurements {
-		for client, stats := range clientStats.clients {
-			if _, ok := m.measurements[table]; !ok {
-				return false
-			}
-			if _, ok := m.measurements[table].clients[client]; !ok {
-				return false
-			}
-			if !stats.Equal(m.measurements[table].clients[client]) {
-				return false
-			}
-		}
-	}
-	return true
-}
-
 func (m *Metrics) NewSelector(clientID, tableName string) Selector {
 	return Selector{
 		Set: attribute.NewSet(
-			attribute.Key("sync.invocation.id").String(m.invocationID),
 			attribute.Key("sync.table.name").String(tableName),
 		),
 		clientID:  clientID,
 		tableName: tableName,
 	}
 }
 
-func (m *Metrics) InitWithClients(invocationID string, table *schema.Table, clients []schema.ClientMeta) {
+func (m *Metrics) InitWithClients(table *schema.Table, clients []schema.ClientMeta) {
 	m.measurements[table.Name] = tableMeasurements{clients: make(map[string]*measurement), duration: &durationMeasurement{}}
 	for _, client := range clients {
 		m.measurements[table.Name].clients[client.ID()] = &measurement{duration: &durationMeasurement{}}
 	}
 	for _, relation := range table.Relations {
-		m.InitWithClients(invocationID, relation, clients)
+		m.InitWithClients(relation, clients)
 	}
 }
 

scheduler/queue/scheduler_test.go

@@ -82,7 +82,7 @@ func TestScheduler(t *testing.T) {
 	}
 
 	for _, tc := range tableClients {
-		m.InitWithClients(scheduler.invocationID, tc.Table, []schema.ClientMeta{tc.Client})
+		m.InitWithClients(tc.Table, []schema.ClientMeta{tc.Client})
 	}
 
 	resolvedResources := make(chan *schema.Resource)

scheduler/scheduler_debug.go

@@ -50,7 +50,7 @@ func (s *syncClient) syncTest(ctx context.Context, syncMultiplier int, resolvedR
 		preInitialisedClients[i] = clients
 		// we do this here to avoid locks so we initialize the metrics structure once in the main goroutine
 		// and then we can just read from it in the other goroutines concurrently given we are not writing to it.
-		s.metrics.InitWithClients(s.invocationID, table, clients)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	// First interleave the tables like in round-robin

scheduler/scheduler_dfs.go

@@ -40,7 +40,7 @@ func (s *syncClient) syncDfs(ctx context.Context, resolvedResources chan<- *sche
 		preInitialisedClients[i] = clients
 		// we do this here to avoid locks so we initial the metrics structure once in the main goroutines
 		// and then we can just read from it in the other goroutines concurrently given we are not writing to it.
-		s.metrics.InitWithClients(s.invocationID, table, clients)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	tableClients := make([]tableClient, 0)

scheduler/scheduler_round_robin.go

@@ -33,7 +33,7 @@ func (s *syncClient) syncRoundRobin(ctx context.Context, resolvedResources chan<
 		preInitialisedClients[i] = clients
 		// we do this here to avoid locks so we initial the metrics structure once in the main goroutines
 		// and then we can just read from it in the other goroutines concurrently given we are not writing to it.
-		s.metrics.InitWithClients(s.invocationID, table, clients)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	tableClients := roundRobinInterleave(s.tables, preInitialisedClients)

scheduler/scheduler_shuffle.go

@@ -33,7 +33,7 @@ func (s *syncClient) syncShuffle(ctx context.Context, resolvedResources chan<- *
 		preInitialisedClients[i] = clients
 		// we do this here to avoid locks so we initial the metrics structure once in the main goroutines
 		// and then we can just read from it in the other goroutines concurrently given we are not writing to it.
-		s.metrics.InitWithClients(s.invocationID, table, clients)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	// First interleave the tables like in round-robin

scheduler/scheduler_shuffle_queue.go

@@ -21,7 +21,7 @@ func (s *syncClient) syncShuffleQueue(ctx context.Context, resolvedResources cha
 		preInitialisedClients[i] = clients
 		// we do this here to avoid locks so we initial the metrics structure once in the main goroutines
 		// and then we can just read from it in the other goroutines concurrently given we are not writing to it.
-		s.metrics.InitWithClients(s.invocationID, table, clients)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	tableClients := roundRobinInterleave(s.tables, preInitialisedClients)