Merge branch 'main' into fix/reintroduce-streamingbatchwriterfix

Author: disq

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-  ".": "4.64.1"
+  ".": "4.65.0"
 }

CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [4.65.0](https://github.com/cloudquery/plugin-sdk/compare/v4.64.1...v4.65.0) (2024-10-04)
+
+
+### Features
+
+* Implement RandomQueue scheduler strategy ([#1914](https://github.com/cloudquery/plugin-sdk/issues/1914)) ([af8ac87](https://github.com/cloudquery/plugin-sdk/commit/af8ac87178cc318d2f31cd17efc7c921d6d52e6b))
+
+
+### Bug Fixes
+
+* Revert "fix: Error handling in StreamingBatchWriter" ([#1918](https://github.com/cloudquery/plugin-sdk/issues/1918)) ([38b4bfd](https://github.com/cloudquery/plugin-sdk/commit/38b4bfd20e17a00d5a2c83e1d48b8b16270592ba))
+* **tests:** WriterTestSuite.handleNulls should not overwrite columns ([#1920](https://github.com/cloudquery/plugin-sdk/issues/1920)) ([08e18e2](https://github.com/cloudquery/plugin-sdk/commit/08e18e265dfb7e6e77c32244f56acd0f63bf4ead))
+
 ## [4.64.1](https://github.com/cloudquery/plugin-sdk/compare/v4.64.0...v4.64.1) (2024-10-02)
 
 

examples/simple_plugin/go.mod

@@ -4,7 +4,7 @@ go 1.22.0
 
 require (
 	github.com/apache/arrow/go/v17 v17.0.0
-	github.com/cloudquery/plugin-sdk/v4 v4.64.1
+	github.com/cloudquery/plugin-sdk/v4 v4.65.0
 	github.com/rs/zerolog v1.33.0
 )
 

scheduler/metrics.go scheduler/metrics/metrics.goscheduler/metrics.go renamed to scheduler/metrics/metrics.go

@@ -1,4 +1,4 @@
-package scheduler
+package metrics
 
 import (
 	"context"
@@ -12,6 +12,10 @@ import (
 	"go.opentelemetry.io/otel/metric"
 )
 
+const (
+	OtelName = "io.cloudquery"
+)
+
 // Metrics is deprecated as we move toward open telemetry for tracing and metrics
 type Metrics struct {
 	TableClient map[string]map[string]*TableClientMetrics
@@ -82,39 +86,39 @@ func (s *Metrics) Equal(other *Metrics) bool {
 }
 
 func getOtelMeters(tableName string, clientID string) *OtelMeters {
-	resources, err := otel.Meter(otelName).Int64Counter("sync.table.resources",
+	resources, err := otel.Meter(OtelName).Int64Counter("sync.table.resources",
 		metric.WithDescription("Number of resources synced for a table"),
 		metric.WithUnit("/{tot}"),
 	)
 	if err != nil {
 		return nil
 	}
 
-	errors, err := otel.Meter(otelName).Int64Counter("sync.table.errors",
+	errors, err := otel.Meter(OtelName).Int64Counter("sync.table.errors",
 		metric.WithDescription("Number of errors encountered while syncing a table"),
 		metric.WithUnit("/{tot}"),
 	)
 	if err != nil {
 		return nil
 	}
 
-	panics, err := otel.Meter(otelName).Int64Counter("sync.table.panics",
+	panics, err := otel.Meter(OtelName).Int64Counter("sync.table.panics",
 		metric.WithDescription("Number of panics encountered while syncing a table"),
 		metric.WithUnit("/{tot}"),
 	)
 	if err != nil {
 		return nil
 	}
 
-	startTime, err := otel.Meter(otelName).Int64Counter("sync.table.start_time",
+	startTime, err := otel.Meter(OtelName).Int64Counter("sync.table.start_time",
 		metric.WithDescription("Start time of syncing a table"),
 		metric.WithUnit("ns"),
 	)
 	if err != nil {
 		return nil
 	}
 
-	endTime, err := otel.Meter(otelName).Int64Counter("sync.table.end_time",
+	endTime, err := otel.Meter(OtelName).Int64Counter("sync.table.end_time",
 		metric.WithDescription("End time of syncing a table"),
 		metric.WithUnit("ns"),
 	)
@@ -136,7 +140,7 @@ func getOtelMeters(tableName string, clientID string) *OtelMeters {
 	}
 }
 
-func (s *Metrics) initWithClients(table *schema.Table, clients []schema.ClientMeta) {
+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
@@ -146,7 +150,7 @@ func (s *Metrics) initWithClients(table *schema.Table, clients []schema.ClientMe
 		}
 	}
 	for _, relation := range table.Relations {
-		s.initWithClients(relation, clients)
+		s.InitWithClients(relation, clients)
 	}
 }
 

scheduler/metrics_test.go scheduler/metrics/metrics_test.goscheduler/metrics_test.go renamed to scheduler/metrics/metrics_test.go

@@ -1,4 +1,4 @@
-package scheduler
+package metrics
 
 import "testing"
 

scheduler/queue/active_work_signal.go

@@ -0,0 +1,78 @@
+package queue
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// activeWorkSignal is a thread-safe coordinator for awaiting a worker pool
+// that relies on a queue that might be temporarily empty.
+//
+// If queue is empty and workers idle, done!
+//
+// If the queue is empty but a worker is working on a task, we must wait and check
+// if it's empty after that worker finishes. That's why we need this.
+//
+// Use it like this:
+//
+// - When a worker picks up a task, call `Add()` (like a WaitGroup)
+// - When a worker finishes a task, call `Done()` (like a WaitGroup)
+//
+// - If the queue is empty, check `IsIdle()` to check if no workers are active.
+// - If workers are still active, call `Wait()` to block until state changes.
+type activeWorkSignal struct {
+	countChangeSignal   *sync.Cond
+	activeWorkUnitCount *atomic.Int32
+	isStarted           *atomic.Bool
+}
+
+func newActiveWorkSignal() *activeWorkSignal {
+	return &activeWorkSignal{
+		countChangeSignal:   sync.NewCond(&sync.Mutex{}),
+		activeWorkUnitCount: &atomic.Int32{},
+		isStarted:           &atomic.Bool{},
+	}
+}
+
+// Add means a worker has started working on a task.
+//
+// Wake up the work queuing goroutine.
+func (s *activeWorkSignal) Add() {
+	s.activeWorkUnitCount.Add(1)
+	s.isStarted.Store(true)
+	s.countChangeSignal.Signal()
+}
+
+// Done means a worker has finished working on a task.
+//
+// If the count became zero, wake up the work queuing goroutine (might have finished).
+func (s *activeWorkSignal) Done() {
+	s.activeWorkUnitCount.Add(-1)
+	s.countChangeSignal.Signal()
+}
+
+// IsIdle returns true if no workers are active. If queue is empty and workers idle, done!
+func (s *activeWorkSignal) IsIdle() bool {
+	return s.isStarted.Load() && s.activeWorkUnitCount.Load() <= 0
+}
+
+// Wait blocks until the count of active workers changes.
+func (s *activeWorkSignal) Wait() {
+	// A race condition is possible when the last active table asynchronously
+	// queues a relation. The table finishes (calling `.Done()`) a moment
+	// before the queue receives the `.Push()`. At this point, the queue is
+	// empty and there are no active workers.
+	//
+	// A moment later, the queue receives the `.Push()` and queues a new task.
+	//
+	// This is a very infrequent case according to tests, but it happens.
+	time.Sleep(10 * time.Millisecond)
+
+	if s.activeWorkUnitCount.Load() <= 0 {
+		return
+	}
+	s.countChangeSignal.L.Lock()
+	defer s.countChangeSignal.L.Unlock()
+	s.countChangeSignal.Wait()
+}

scheduler/queue/concurrent_random_queue.go

@@ -0,0 +1,41 @@
+package queue
+
+import (
+	"math/rand"
+	"sync"
+)
+
+// ConcurrentRandomQueue is a generic, thread-safe queue
+// that pops random elements in O(1) time.
+type ConcurrentRandomQueue[T any] struct {
+	mu     sync.Mutex
+	queue  []T
+	random *rand.Rand
+}
+
+func NewConcurrentRandomQueue[T any](seed int64, capacityHint int) *ConcurrentRandomQueue[T] {
+	return &ConcurrentRandomQueue[T]{queue: make([]T, 0, capacityHint), random: rand.New(rand.NewSource(seed))}
+}
+
+func (q *ConcurrentRandomQueue[T]) Push(item T) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.queue = append(q.queue, item)
+}
+
+func (q *ConcurrentRandomQueue[T]) Pop() *T {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if len(q.queue) == 0 {
+		return nil
+	}
+	idx := q.random.Intn(len(q.queue))
+	lastIdx := len(q.queue) - 1
+	q.queue[idx], q.queue[lastIdx] = q.queue[lastIdx], q.queue[idx]
+	item := q.queue[lastIdx]
+	q.queue = q.queue[:lastIdx]
+
+	return &item
+}

scheduler/queue/scheduler.go

@@ -0,0 +1,138 @@
+package queue
+
+import (
+	"context"
+
+	"github.com/cloudquery/plugin-sdk/v4/caser"
+	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
+	"github.com/cloudquery/plugin-sdk/v4/schema"
+	"github.com/google/uuid"
+	"github.com/rs/zerolog"
+	"golang.org/x/sync/errgroup"
+)
+
+const DefaultWorkerCount = 1000
+
+// WorkUnit is an atomic unit of work that the scheduler syncs.
+//
+// It is one table resolver (same as all other scheduler strategies).
+//
+// But if it is a non-top-level table, it is bound to a single parent resource.
+type WorkUnit struct {
+	Table  *schema.Table
+	Client schema.ClientMeta
+	Parent *schema.Resource
+}
+
+type Scheduler struct {
+	workerCount       int
+	logger            zerolog.Logger
+	caser             *caser.Caser
+	deterministicCQID bool
+	metrics           *metrics.Metrics
+	invocationID      string
+	seed              int64
+}
+
+type Option func(*Scheduler)
+
+func WithWorkerCount(workerCount int) Option {
+	return func(d *Scheduler) {
+		d.workerCount = workerCount
+	}
+}
+
+func WithCaser(c *caser.Caser) Option {
+	return func(d *Scheduler) {
+		d.caser = c
+	}
+}
+
+func WithDeterministicCQID(deterministicCQID bool) Option {
+	return func(d *Scheduler) {
+		d.deterministicCQID = deterministicCQID
+	}
+}
+
+func WithInvocationID(invocationID string) Option {
+	return func(d *Scheduler) {
+		d.invocationID = invocationID
+	}
+}
+
+func NewShuffleQueueScheduler(logger zerolog.Logger, m *metrics.Metrics, seed int64, opts ...Option) *Scheduler {
+	scheduler := &Scheduler{
+		logger:       logger,
+		metrics:      m,
+		workerCount:  DefaultWorkerCount,
+		caser:        caser.New(),
+		invocationID: uuid.New().String(),
+		seed:         seed,
+	}
+
+	for _, opt := range opts {
+		opt(scheduler)
+	}
+
+	return scheduler
+}
+
+func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedResources chan<- *schema.Resource) {
+	if len(tableClients) == 0 {
+		return
+	}
+	queue := NewConcurrentRandomQueue[WorkUnit](d.seed, len(tableClients))
+	for _, tc := range tableClients {
+		queue.Push(tc)
+	}
+
+	jobs := make(chan *WorkUnit)
+	activeWorkSignal := newActiveWorkSignal()
+
+	// Worker pool
+	workerPool, _ := errgroup.WithContext(ctx)
+	for w := 0; w < d.workerCount; w++ {
+		workerPool.Go(func() error {
+			newWorker(
+				jobs,
+				queue,
+				resolvedResources,
+				d.logger,
+				d.caser,
+				d.invocationID,
+				d.deterministicCQID,
+				d.metrics,
+			).work(ctx, activeWorkSignal)
+			return nil
+		})
+	}
+
+	// Work distribution
+	go func() {
+		defer close(jobs)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				item := queue.Pop()
+
+				// There is work to do
+				if item != nil {
+					jobs <- item
+					continue
+				}
+
+				// Queue is empty and no active work, done!
+				if activeWorkSignal.IsIdle() {
+					return
+				}
+
+				// Queue is empty and there is active work, wait for changes
+				activeWorkSignal.Wait()
+			}
+		}
+	}()
+
+	_ = workerPool.Wait()
+}