feat: Add handling of error messages to sdk (#2195)

#### Summary

<!-- 🎉 Thank you for making CloudQuery awesome by submitting a PR 🎉 -->


<!--
Explain what problem this PR addresses
-->


This PR requires changes to plugin-pb 


---

Use the following steps to ensure your PR is ready to be reviewed

- [ ] Read the [contribution guidelines](../blob/main/CONTRIBUTING.md)
🧑‍🎓
- [ ] Run `go fmt` to format your code 🖊
- [ ] Lint your changes via `golangci-lint run` 🚨 (install golangci-lint
[here](https://golangci-lint.run/usage/install/#local-installation))
- [ ] Update or add tests 🧪
- [ ] Ensure the status checks below are successful ✅

Author: Przemysław Stępień

internal/servers/plugin/v3/plugin.go

@@ -259,6 +259,16 @@ func (s *Server) Sync(req *pb.Sync_Request, stream pb.Plugin_SyncServer) error {
 					WhereClause:    whereClause,
 				},
 			}
+		case *message.SyncError:
+			if !req.WithErrorMessages {
+				continue
+			}
+			pbMsg.Message = &pb.Sync_Response_Error{
+				Error: &pb.Sync_MessageError{
+					TableName: m.TableName,
+					Error:     m.Error,
+				},
+			}
 		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"
@@ -77,7 +78,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 +103,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/scheduler_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/cloudquery/plugin-sdk/v4/message"
 	"github.com/cloudquery/plugin-sdk/v4/scheduler/metrics"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/cloudquery/plugin-sdk/v4/transformers"
@@ -83,9 +84,10 @@ func TestScheduler(t *testing.T) {
 	}
 
 	resolvedResources := make(chan *schema.Resource)
+	msgs := make(chan message.SyncMessage, 10)
 	go func() {
 		defer close(resolvedResources)
-		scheduler.Sync(context.Background(), tableClients, resolvedResources)
+		scheduler.Sync(context.Background(), tableClients, resolvedResources, msgs)
 	}()
 
 	gotResources := make([]*schema.Resource, 0)

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)
 }