Set invocation ID in newResource

Author: disq

scheduler/metrics/metrics.go

@@ -85,7 +85,7 @@ func (s *Metrics) Equal(other *Metrics) bool {
 	return true
 }
 
-func getOtelMeters(tableName, clientID, invocationID string) *OtelMeters {
+func getOtelMeters(tableName, clientID string) *OtelMeters {
 	resources, err := otel.Meter(OtelName).Int64Counter("sync.table.resources",
 		metric.WithDescription("Number of resources synced for a table"),
 		metric.WithUnit("/{tot}"),
@@ -136,22 +136,21 @@ func getOtelMeters(tableName, clientID, invocationID 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, invocationID string) {
+func (s *Metrics) InitWithClients(table *schema.Table, clients []schema.ClientMeta) {
 	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, invocationID),
+			otelMeters: getOtelMeters(tableName, clientID),
 		}
 	}
 	for _, relation := range table.Relations {
-		s.InitWithClients(relation, clients, invocationID)
+		s.InitWithClients(relation, clients)
 	}
 }
 

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}, scheduler.invocationID)
+		m.InitWithClients(tc.Table, []schema.ClientMeta{tc.Client})
 	}
 
 	resolvedResources := make(chan *schema.Resource)

scheduler/queue/worker.go

@@ -70,7 +70,6 @@ func (w *worker) resolveTable(ctx context.Context, table *schema.Table, client s
 		"sync.table."+table.Name,
 		trace.WithAttributes(
 			attribute.Key("sync.client.id").String(clientName),
-			attribute.Key("sync.invocation.id").String(w.invocationID),
 		),
 	)
 	defer span.End()

scheduler/scheduler.go

@@ -14,8 +14,6 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/samber/lo"
 	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/semaphore"
 )
 
@@ -196,10 +194,7 @@ func (s *Scheduler) SyncAll(ctx context.Context, client schema.ClientMeta, table
 }
 
 func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables schema.Tables, res chan<- message.SyncMessage, opts ...SyncOption) error {
-	ctx, span := otel.Tracer(metrics.OtelName).Start(ctx,
-		"sync",
-		trace.WithAttributes(attribute.Key("sync.invocation.id").String(s.invocationID)),
-	)
+	ctx, span := otel.Tracer(metrics.OtelName).Start(ctx, "sync")
 	defer span.End()
 	if len(tables) == 0 {
 		return ErrNoTables

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.invocationID)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	// 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.invocationID)
+		s.metrics.InitWithClients(table, clients)
 	}
 
 	tableClients := make([]tableClient, 0)
@@ -81,7 +81,6 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c
 		"sync.table."+table.Name,
 		trace.WithAttributes(
 			attribute.Key("sync.client.id").String(clientName),
-			attribute.Key("sync.invocation.id").String(s.invocationID),
 		),
 	)
 	defer span.End()

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

serve/opentelemetry.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/cloudquery/plugin-sdk/v4/plugin"
 	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/attribute"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
@@ -32,6 +33,7 @@ func newResource(p *plugin.Plugin) *resource.Resource {
 			semconv.SchemaURL,
 			semconv.ServiceName("cloudquery-"+p.Name()),
 			semconv.ServiceVersion(p.Version()),
+			attribute.Key("sync.invocation.id").String(p.InvocationID()),
 		),
 	)
 	if err != nil {