test: Fix tests

scheduler/metrics/metrics.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/cloudquery/plugin-sdk/v4/schema"
-	"github.com/rs/zerolog"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -270,17 +269,3 @@ func (m *TableClientMetrics) OtelEndTime(ctx context.Context, end time.Time) {
 	}
 	m.otelMeters.previousEndTime = val
 }
-
-func LogTablesMetrics(logger zerolog.Logger, m *Metrics, tables schema.Tables, client schema.ClientMeta) {
-	clientName := client.ID()
-	for _, table := range tables {
-		tableMetrics := m.TableClient[table.Name][clientName]
-		duration := tableMetrics.Duration.Load()
-		if duration == nil {
-			// This can happen for a relation when there are no resources to resolve from the parent
-			duration = new(time.Duration)
-		}
-		logger.Info().Str("table", table.Name).Str("client", clientName).Uint64("resources", tableMetrics.Resources).Dur("duration_ms", *duration).Uint64("errors", tableMetrics.Errors).Msg("table sync finished")
-		LogTablesMetrics(logger, m, table.Relations, client)
-	}
-}

scheduler/queue/dispatcher.go

@@ -69,7 +69,6 @@ func (w *worker) resolveTable(ctx context.Context, table *schema.Table, client s
 	tableMetrics.OtelEndTime(ctx, endTime)
 	if parent == nil {
 		logger.Info().Uint64("resources", tableMetrics.Resources).Uint64("errors", tableMetrics.Errors).Dur("duration_ms", duration).Msg("table sync finished")
-		metrics.LogTablesMetrics(w.logger, w.metrics, table.Relations, client)
 	}
 }
 
@@ -281,6 +280,7 @@ func (d *Dispatcher) Dispatch(ctx context.Context, tableClients []TableClientPai
 			item := queue.Pop()
 			if item == nil {
 				if !workStarted.Load() || activeWorkers.Load() != 0 {
+					time.Sleep(10 * time.Millisecond)
 					continue
 				}
 				break