Fix an issue where we were trying to log metrics for full table (#473)

For full table partitions skip logging metrics.
PeerDB-io · Oct 3, 2023 · af126b5 · af126b5
1 parent 8a434aa
commit af126b5
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/flow/cmd/worker.go b/flow/cmd/worker.go
@@ -137,7 +137,7 @@ func WorkerMain(opts *WorkerOptions) error {
 	w.RegisterWorkflow(peerflow.DropFlowWorkflow)
 	w.RegisterActivity(&activities.FlowableActivity{
 		EnableMetrics:        opts.EnableMetrics,
-		CatalogMirrorMonitor: &catalogMirrorMonitor,
+		CatalogMirrorMonitor: catalogMirrorMonitor,
 	})
 
 	err = w.Run(worker.InterruptCh())

diff --git a/flow/connectors/utils/monitoring/monitoring.go b/flow/connectors/utils/monitoring/monitoring.go
@@ -25,8 +25,8 @@ type CDCBatchInfo struct {
 	StartTime     time.Time
 }
 
-func NewCatalogMirrorMonitor(catalogConn *pgxpool.Pool) CatalogMirrorMonitor {
-	return CatalogMirrorMonitor{
+func NewCatalogMirrorMonitor(catalogConn *pgxpool.Pool) *CatalogMirrorMonitor {
+	return &CatalogMirrorMonitor{
 		catalogConn: catalogConn,
 	}
 }
@@ -189,6 +189,11 @@ func (c *CatalogMirrorMonitor) AddPartitionToQRepRun(ctx context.Context, flowJo
 		return nil
 	}
 
+	if partition.Range == nil && partition.FullTablePartition {
+		log.Infof("partition %s is a full table partition. Metrics logging is skipped.", partition.PartitionId)
+		return nil
+	}
+
 	var rangeStart, rangeEnd string
 	switch x := partition.Range.Range.(type) {
 	case *protos.PartitionRange_IntRange: