feat: Add handling of error messages to sdk

Author: Przemyslaw Stepien

internal/servers/plugin/v3/plugin.go

@@ -259,6 +259,17 @@ func (s *Server) Sync(req *pb.Sync_Request, stream pb.Plugin_SyncServer) error {
 					WhereClause:    whereClause,
 				},
 			}
+		case *message.SyncError:
+			if req.WithErrorMessages {
+				pbMsg.Message = &pb.Sync_Response_Error{
+					Error: &pb.Sync_MessageError{
+						TableName: m.TableName,
+						Error:     m.Error,
+					},
+				}
+			} else {
+				continue
+			}
 		default:
 			return status.Errorf(codes.Internal, "unknown message type: %T", msg)
 		}

message/sync_message.go

@@ -121,3 +121,13 @@ type SyncDeleteRecord struct {
 func (m SyncDeleteRecord) GetTable() *schema.Table {
 	return &schema.Table{Name: m.TableName}
 }
+
+type SyncError struct {
+	syncBaseMessage
+	TableName string
+	Error     string
+}
+
+func (e SyncError) GetTable() *schema.Table {
+	return &schema.Table{Name: e.TableName}
+}

scheduler/queue/scheduler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/cloudquery/plugin-sdk/v4/caser"
+	"github.com/cloudquery/plugin-sdk/v4/message"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/google/uuid"
@@ -32,6 +33,8 @@ type Scheduler struct {
 	metrics           *metrics.Metrics
 	invocationID      string
 	seed              int64
+	// message channel for sending SyncError messages
+	msgChan chan<- message.SyncMessage
 }
 
 type Option func(*Scheduler)
@@ -77,7 +80,7 @@ func NewShuffleQueueScheduler(logger zerolog.Logger, m *metrics.Metrics, seed in
 	return scheduler
 }
 
-func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedResources chan<- *schema.Resource) {
+func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedResources chan<- *schema.Resource, msgChan chan<- message.SyncMessage) {
 	if len(tableClients) == 0 {
 		return
 	}
@@ -102,6 +105,7 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR
 				d.invocationID,
 				d.deterministicCQID,
 				d.metrics,
+				msgChan,
 			).work(ctx, activeWorkSignal)
 			return nil
 		})

scheduler/queue/worker.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/cloudquery/plugin-sdk/v4/caser"
 	"github.com/cloudquery/plugin-sdk/v4/helpers"
+	"github.com/cloudquery/plugin-sdk/v4/message"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/resolvers"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
@@ -29,6 +30,8 @@ type worker struct {
 	invocationID      string
 	deterministicCQID bool
 	metrics           *metrics.Metrics
+	// message channel for sending SyncError messages
+	msgChan chan<- message.SyncMessage
 }
 
 func (w *worker) work(ctx context.Context, activeWorkSignal *activeWorkSignal) {
@@ -51,6 +54,7 @@ func newWorker(
 	invocationID string,
 	deterministicCQID bool,
 	m *metrics.Metrics,
+	msgChan chan<- message.SyncMessage,
 ) *worker {
 	return &worker{
 		jobs:              jobs,
@@ -61,6 +65,7 @@ func newWorker(
 		deterministicCQID: deterministicCQID,
 		invocationID:      invocationID,
 		metrics:           m,
+		msgChan:           msgChan,
 	}
 }
 
@@ -105,6 +110,12 @@ func (w *worker) resolveTable(ctx context.Context, table *schema.Table, client s
 			logger.Error().Err(err).Msg("table resolver finished with error")
 			tableMetrics.OtelErrorsAdd(ctx, 1)
 			atomic.AddUint64(&tableMetrics.Errors, 1)
+			// Send SyncError message
+			syncErrorMsg := &message.SyncError{
+				TableName: table.Name,
+				Error:     err.Error(),
+			}
+			w.msgChan <- syncErrorMsg
 			return
 		}
 	}()

scheduler/scheduler.go

@@ -139,6 +139,8 @@ type syncClient struct {
 	metrics      *metrics.Metrics
 	logger       zerolog.Logger
 	invocationID string
+	// message channel for sending SyncError messages
+	msgChan chan<- message.SyncMessage
 
 	shard *shard
 }
@@ -213,6 +215,7 @@ func (s *Scheduler) Sync(ctx context.Context, client schema.ClientMeta, tables s
 		scheduler:    s,
 		logger:       s.logger,
 		invocationID: s.invocationID,
+		msgChan:      res,
 	}
 	for _, opt := range opts {
 		opt(syncClient)

scheduler/scheduler_dfs.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/cloudquery/plugin-sdk/v4/helpers"
+	"github.com/cloudquery/plugin-sdk/v4/message"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/batchsender"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/resolvers"
@@ -119,6 +120,12 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c
 			logger.Error().Err(err).Msg("table resolver finished with error")
 			tableMetrics.OtelErrorsAdd(ctx, 1)
 			atomic.AddUint64(&tableMetrics.Errors, 1)
+			// Send SyncError message
+			syncErrorMsg := &message.SyncError{
+				TableName: table.Name,
+				Error:     err.Error(),
+			}
+			s.msgChan <- syncErrorMsg
 			return
 		}
 	}()

scheduler/scheduler_shuffle_queue.go

@@ -45,5 +45,5 @@ func (s *syncClient) syncShuffleQueue(ctx context.Context, resolvedResources cha
 			Client: tc.client,
 		})
 	}
-	scheduler.Sync(ctx, queueClients, resolvedResources)
+	scheduler.Sync(ctx, queueClients, resolvedResources, s.msgChan)
 }