feat: Add error handling and per-table metrics (#20909)

Przemysław Stępień · web-flow · commit a2d94730cfec · 2025-07-03T12:14:53.000Z
#### Summary ⚠️ **If you're contributing to a plugin please read this section of the [contribution guidelines](https://github.com/cloudquery/cloudquery/blob/main/CONTRIBUTING.md#open-core-vs-open-source) 🧑‍🎓 before submitting this PR** ⚠️ This PR adds handling of error messages from source plugins and add new statistics for platform calls. This solves two things: 1. Having proper error statistics when using plugins in development mode 2. Adding ability to show detailed per table metrics in platform without need to rely on otel_traces
diff --git a/cli/cmd/sync_v3.go b/cli/cmd/sync_v3.go
@@ -149,6 +149,7 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 		syncTimeTook   time.Duration
 		totalResources = int64(0)
 		totals         = sourceClient.Metrics()
+		statsPerTable  = cloudquery_api.SyncRunTableProgress{}
 	)
 	defer func() {
 		analytics.TrackSyncCompleted(ctx, invocationUUID.UUID, analytics.SyncFinishedEvent{
@@ -316,6 +317,7 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 		SkipTables:          sourceSpec.SkipTables,
 		SkipDependentTables: *sourceSpec.SkipDependentTables,
 		DeterministicCqId:   sourceSpec.DeterministicCQID,
+		WithErrorMessages:   true,
 	}
 	if sourceSpec.BackendOptions != nil {
 		syncReq.Backend = &plugin.Sync_BackendOptions{
@@ -351,6 +353,22 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 	// Read from the sync stream and write to all destinations.
 	isComplete := int64(0)
 
+	sourceTables, err := getTables(ctx, sourcePbClient, &plugin.GetTables_Request{
+		Tables:              sourceSpec.Tables,
+		SkipTables:          sourceSpec.SkipTables,
+		SkipDependentTables: *sourceSpec.SkipDependentTables})
+
+	if err != nil {
+		return err
+	}
+	// Pre init stats per table
+	for _, table := range sourceTables {
+		statsPerTable[table.Name] = cloudquery_api.SyncRunTableProgressValue{
+			Rows:   0,
+			Errors: 0,
+		}
+	}
+
 	var remoteProgressReporter *godebouncer.Debouncer
 	if progressAPIClient != nil {
 		teamName, syncName, syncRunId := os.Getenv("_CQ_TEAM_NAME"), os.Getenv("_CQ_SYNC_NAME"), os.Getenv("_CQ_SYNC_RUN_ID")
@@ -373,10 +391,11 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 				status = cloudquery_api.SyncRunStatusCompleted
 			}
 			obj := cloudquery_api.CreateSyncRunProgressJSONRequestBody{
-				Rows:     atomic.LoadInt64(&totalResources),
-				Errors:   int64(totals.Errors),
-				Warnings: int64(totals.Warnings),
-				Status:   &status,
+				Rows:          atomic.LoadInt64(&totalResources),
+				Errors:        int64(totals.Errors),
+				Warnings:      int64(totals.Warnings),
+				Status:        &status,
+				TableProgress: &statsPerTable,
 			}
 			if shard != nil {
 				obj.ShardNum = &shard.num
@@ -487,6 +506,10 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 
 				atomic.AddInt64(&newResources, record.NumRows())
 				atomic.AddInt64(&totalResources, record.NumRows())
+				tableName, _ := record.Schema().Metadata().GetValue(schema.MetadataTableName)
+				stats := statsPerTable[tableName]
+				stats.Rows += record.NumRows()
+				statsPerTable[tableName] = stats
 				if remoteProgressReporter != nil {
 					remoteProgressReporter.SendSignal()
 				}
@@ -598,6 +621,11 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 						return handleSendError(err, writeClients[i], "migrate")
 					}
 				}
+			case *plugin.Sync_Response_Error:
+				log.Error().Str("table", m.Error.TableName).Msg(m.Error.Error)
+				stats := statsPerTable[m.Error.TableName]
+				stats.Errors++
+				statsPerTable[m.Error.TableName] = stats
 			default:
 				return fmt.Errorf("unknown message type: %T", m)
 			}
@@ -608,27 +636,19 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 		return err
 	}
 
-	syncSummaryEnabled := summaryLocation != "" || lo.SomeBy(destinationSpecs, func(d specs.Destination) bool { return d.SyncSummary })
-	var sourceTables schema.Tables
-	if syncSummaryEnabled {
-		sourceTables, err = getTables(ctx, sourcePbClient, &plugin.GetTables_Request{
-			Tables:              sourceSpec.Tables,
-			SkipTables:          sourceSpec.SkipTables,
-			SkipDependentTables: *sourceSpec.SkipDependentTables,
-		})
-
-		if err != nil {
-			return err
-		}
-	}
-
 	err = syncClient.CloseSend()
 	if err != nil {
 		return err
 	}
 	totals = sourceClient.Metrics()
 	sourceWarnings := totals.Warnings
-	sourceErrors := totals.Errors
+	var sourceErrors uint64
+	for _, val := range statsPerTable {
+		sourceErrors += uint64(val.Errors)
+	}
+	if totals.Errors > sourceErrors {
+		sourceErrors = totals.Errors
+	}
 	var metadataDataErrors error
 	for i := range destinationsClients {
 		m := destinationsClients[i].Metrics()
@@ -697,13 +717,13 @@ func syncConnectionV3(ctx context.Context, syncOptions syncV3Options) (syncErr e
 	exitReason = ExitReasonCompleted
 
 	msg := "Sync completed successfully"
-	if totals.Errors > 0 {
+	if sourceErrors > 0 {
 		msg = "Sync completed with errors, see logs for details"
 	}
-	fmt.Printf("%s. Resources: %d, Errors: %d, Warnings: %d, Time: %s\n", msg, totalResources, totals.Errors, totals.Warnings, syncTimeTook.Truncate(time.Second).String())
+	fmt.Printf("%s. Resources: %d, Errors: %d, Warnings: %d, Time: %s\n", msg, totalResources, sourceErrors, totals.Warnings, syncTimeTook.Truncate(time.Second).String())
 	log.Info().
 		Int64("resources", totalResources).
-		Uint64("errors", totals.Errors).
+		Uint64("errors", sourceErrors).
 		Uint64("warnings", totals.Warnings).
 		Str("duration", syncTimeTook.Truncate(time.Second).String()).
 		Str("result", msg).
