feat: Expose invocation ID in OpenTelemetry metrics

Author: disq

scheduler/metrics/metrics.go

@@ -85,7 +85,7 @@ func (s *Metrics) Equal(other *Metrics) bool {
 	return true
 }
 
-func getOtelMeters(tableName string, clientID string) *OtelMeters {
+func getOtelMeters(tableName, clientID, invocationID string) *OtelMeters {
 	resources, err := otel.Meter(OtelName).Int64Counter("sync.table.resources",
 		metric.WithDescription("Number of resources synced for a table"),
 		metric.WithUnit("/{tot}"),
@@ -136,21 +136,22 @@ func getOtelMeters(tableName string, clientID string) *OtelMeters {
 		attributes: []attribute.KeyValue{
 			attribute.Key("sync.client.id").String(clientID),
 			attribute.Key("sync.table.name").String(tableName),
+			attribute.Key("sync.invocation.id").String(invocationID),
 		},
 	}
 }
 
-func (s *Metrics) InitWithClients(table *schema.Table, clients []schema.ClientMeta) {
+func (s *Metrics) InitWithClients(table *schema.Table, clients []schema.ClientMeta, invocationID string) {
 	s.TableClient[table.Name] = make(map[string]*TableClientMetrics, len(clients))
 	for _, client := range clients {
 		tableName := table.Name
 		clientID := client.ID()
 		s.TableClient[tableName][clientID] = &TableClientMetrics{
-			otelMeters: getOtelMeters(tableName, clientID),
+			otelMeters: getOtelMeters(tableName, clientID, invocationID),
 		}
 	}
 	for _, relation := range table.Relations {
-		s.InitWithClients(relation, clients)
+		s.InitWithClients(relation, clients, invocationID)
 	}
 }
 

scheduler/queue/scheduler_test.go

@@ -79,7 +79,7 @@ func TestScheduler(t *testing.T) {
 	}
 
 	for _, tc := range tableClients {
-		m.InitWithClients(tc.Table, []schema.ClientMeta{tc.Client})
+		m.InitWithClients(tc.Table, []schema.ClientMeta{tc.Client}, scheduler.invocationID)
 	}
 
 	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(table, clients)
+		s.metrics.InitWithClients(table, clients, s.invocationID)
 	}
 
 	// First interleave the tables like in round-robin

scheduler/scheduler_dfs.go

@@ -39,7 +39,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(table, clients)
+		s.metrics.InitWithClients(table, clients, s.invocationID)
 	}
 
 	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(table, clients)
+		s.metrics.InitWithClients(table, clients, s.invocationID)
 	}
 
 	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(table, clients)
+		s.metrics.InitWithClients(table, clients, s.invocationID)
 	}
 
 	// 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(table, clients)
+		s.metrics.InitWithClients(table, clients, s.invocationID)
 	}
 
 	tableClients := roundRobinInterleave(s.tables, preInitialisedClients)