feat(metrics): tables synced per batch (#3524)

Author: iamKunalGupta

flow/connectors/utils/monitoring/monitoring.go

@@ -136,11 +136,15 @@ func AddCDCBatchTablesForFlow(
 	if enabled, err := internal.PeerDBMetricsRecordAggregatesEnabled(ctx, nil); err == nil && enabled {
 		recordAggregateMetrics = true
 	}
+	var syncedTablesCount int64
 	tableNameOperations := make(map[string][3]opWithValue, len(tableNameRowsMapping))
 	for destinationTableName, rowCounts := range tableNameRowsMapping {
 		inserts := rowCounts.InsertCount.Load()
 		updates := rowCounts.UpdateCount.Load()
 		deletes := rowCounts.DeleteCount.Load()
+		if inserts+updates+deletes > 0 {
+			syncedTablesCount++
+		}
 		if recordAggregateMetrics {
 			tableNameOperations[destinationTableName] = [3]opWithValue{
 				{op: otel_metrics.RecordOperationTypeInsert, count: inserts},
@@ -199,6 +203,9 @@ func AddCDCBatchTablesForFlow(
 		return fmt.Errorf("error while committing transaction for inserting and updating statistics: %w", err)
 	}
 
+	otelManager.Metrics.SyncedTablesPerBatchGauge.Record(ctx, syncedTablesCount, metric.WithAttributeSet(attribute.NewSet(
+		attribute.Int64(otel_metrics.BatchIdKey, batchID))))
+
 	for destinationTableName, operations := range tableNameOperations {
 		for _, opAndCount := range operations {
 			otelManager.Metrics.RecordsSyncedPerTableCounter.Add(ctx, int64(opAndCount.count), metric.WithAttributeSet(attribute.NewSet(

flow/otel_metrics/otel_manager.go

@@ -48,6 +48,7 @@ const (
 	RecordsSyncedPerTableGaugeName       = "records_synced_per_table"
 	RecordsSyncedPerTableCounterName     = "records_synced_per_table_counter"
 	SyncedTablesGaugeName                = "synced_tables"
+	SyncedTablesPerBatchGaugeName        = "synced_tables_per_batch"
 	InstanceStatusGaugeName              = "instance_status"
 	MaintenanceStatusGaugeName           = "maintenance_status"
 	FlowStatusGaugeName                  = "flow_status"
@@ -85,6 +86,7 @@ type Metrics struct {
 	RecordsSyncedPerTableGauge       metric.Int64Gauge
 	RecordsSyncedPerTableCounter     metric.Int64Counter
 	SyncedTablesGauge                metric.Int64Gauge
+	SyncedTablesPerBatchGauge        metric.Int64Gauge
 	InstanceStatusGauge              metric.Int64Gauge
 	MaintenanceStatusGauge           metric.Int64Gauge
 	FlowStatusGauge                  metric.Int64Gauge
@@ -335,6 +337,12 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error {
 		return err
 	}
 
+	if om.Metrics.SyncedTablesPerBatchGauge, err = om.GetOrInitInt64Gauge(BuildMetricName(SyncedTablesPerBatchGaugeName),
+		metric.WithDescription("Number of tables synced for every Sync batch"),
+	); err != nil {
+		return err
+	}
+
 	if om.Metrics.InstanceStatusGauge, err = om.GetOrInitInt64Gauge(BuildMetricName(InstanceStatusGaugeName),
 		metric.WithDescription("Status of the instance, always emits a 1 metric with different attributes for different statuses"),
 	); err != nil {