feat: add synced_tables metric

flow/activities/flowable.go

@@ -11,6 +11,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	lua "github.com/yuin/gopher-lua"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/log"
@@ -767,6 +768,20 @@ func (a *FlowableActivity) RecordSlotSizes(ctx context.Context) error {
 					return
 				}
 				slotMetricGauges.IntervalSinceLastNormalizeGauge = intervalSinceLastNormalizeGauge
+
+				syncedTablesGauge, err := a.OtelManager.GetOrInitInt64Gauge(
+					otel_metrics.BuildMetricName(otel_metrics.SyncedTablesGaugeName),
+					metric.WithDescription("Number of tables synced"),
+				)
+				if err != nil {
+					logger.Error("Failed to get synced tables gauge", slog.Any("error", err))
+					return
+				}
+				syncedTablesGauge.Record(ctx, int64(len(config.TableMappings)), metric.WithAttributeSet(attribute.NewSet(
+					attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),
+					attribute.String(otel_metrics.PeerNameKey, peerName),
+				)))
+
 			}
 
 			if err := srcConn.HandleSlotInfo(ctx, a.Alerter, a.CatalogPool, &alerting.AlertKeys{

flow/activities/flowable_core.go

@@ -332,7 +332,6 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 				attribute.String(otel_metrics.FlowNameKey, flowName),
 			)))
 		}
-
 	}
 
 	syncState.Store(shared.Ptr("updating schema"))

flow/otel_metrics/otel_manager.go

@@ -31,6 +31,10 @@ const (
 	ErrorsEmittedCounterName = "errors_emitted"
 	// SyncRecordsSyncedGaugeName is the gauge name for the number of records synced for every Sync batch
 	SyncRecordsSyncedGaugeName = "sync_records_synced"
+	// SyncedTablesGaugeName is the gauge name for the number of tables being synced for a mirror
+	SyncedTablesGaugeName = "synced_tables"
+	// InstanceStatusGaugeName  used for notifying the status of the instance, like if it is healthy/under maintenance etc.
+	InstanceStatusGaugeName = "instance_status"
 )
 
 type SlotMetricGauges struct {
@@ -40,6 +44,7 @@ type SlotMetricGauges struct {
 	OpenConnectionsGauge            metric.Int64Gauge
 	OpenReplicationConnectionsGauge metric.Int64Gauge
 	IntervalSinceLastNormalizeGauge metric.Float64Gauge
+	InstanceStatusGauge             metric.Int64Gauge
 }
 
 func BuildMetricName(baseName string) string {