Add more information logs for UI logs table

flow/activities/flowable.go

@@ -219,6 +219,7 @@
 ) (*protos.SetupNormalizedTableBatchOutput, error) {
 	logger := activity.GetLogger(ctx)
 	ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowName)
+	a.Alerter.LogFlowInfo(ctx, config.FlowName, "Setting up destination tables")
 	conn, err := connectors.GetByNameAs[connectors.NormalizedTablesConnector](ctx, config.Env, a.CatalogPool, config.PeerName)
 	if err != nil {
 		if errors.Is(err, errors.ErrUnsupported) {
@@ -247,6 +248,7 @@
 	})
 	defer shutdown()
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowName, "Setting up destination tables")
 	tableExistsMapping := make(map[string]bool, len(tableNameSchemaMapping))
 	for tableIdentifier, tableSchema := range tableNameSchemaMapping {
 		existing, err := conn.SetupNormalizedTable(
@@ -265,15 +267,19 @@
 		numTablesSetup.Add(1)
 		if !existing {
 			logger.Info("created table " + tableIdentifier)
+			a.Alerter.LogFlowInfo(ctx, config.FlowName, "created table "+tableIdentifier+" in destination")
 		} else {
 			logger.Info("table already exists " + tableIdentifier)
 		}
+
 	}
 
 	if err := conn.FinishSetupNormalizedTables(ctx, tx); err != nil {
 		return nil, fmt.Errorf("failed to commit normalized tables tx: %w", err)
 	}
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowName, "All destination tables have been setup")
+
 	return &protos.SetupNormalizedTableBatchOutput{
 		TableExistsMapping: tableExistsMapping,
 	}, nil
@@ -509,6 +515,8 @@
 		}
 	}
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, fmt.Sprintf("obtained partitions for table %s", config.WatermarkTable))
+
 	return &protos.QRepParitionResult{
 		Partitions: partitions,
 	}, nil
@@ -576,6 +584,7 @@
 		}
 	}
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "replicated all rows to destination for table "+config.DestinationTableIdentifier)
 	return nil
 }
 
@@ -637,6 +646,8 @@
 		return pullCleanupErr
 	}
 
+	a.Alerter.LogFlowInfo(ctx, req.FlowJobName, "Cleaned up source peer replication objects. Any replication slot or publication created by PeerDB has been removed.")
+
 	return nil
 }
 
@@ -656,6 +667,8 @@
 		return syncFlowCleanupErr
 	}
 
+	a.Alerter.LogFlowInfo(ctx, req.FlowJobName, "Cleaned up destination peer replication objects. Any PeerDB metadata storage has been dropped.")
+
 	return nil
 }
 
@@ -900,6 +913,8 @@
 		}
 	}
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Resync completed for all tables")
+
 	return renameOutput, tx.Commit(ctx)
 }
 
@@ -973,6 +988,9 @@
 	if err != nil {
 		a.Alerter.LogFlowError(ctx, cfg.FlowJobName, err)
 	}
+
+	a.Alerter.LogFlowInfo(ctx, cfg.FlowJobName, fmt.Sprintf("ensured %d tables exist in publication %s",
+		len(additionalTableMappings), cfg.PublicationName))
 	return err
 }
 
@@ -996,6 +1014,9 @@
 	if err != nil {
 		a.Alerter.LogFlowError(ctx, cfg.FlowJobName, err)
 	}
+
+	a.Alerter.LogFlowInfo(ctx, cfg.FlowJobName, fmt.Sprintf("removed %d tables from publication %s",
+		len(removedTablesMapping), cfg.PublicationName))
 	return err
 }
 

flow/activities/flowable_core.go

@@ -228,6 +228,11 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 			return nil, fmt.Errorf("failed to sync schema: %w", err)
 		}
 
+		numberOfSchemaChanges := len(recordBatchSync.SchemaDeltas)
+		if numberOfSchemaChanges > 0 {
+			a.Alerter.LogFlowInfo(ctx, flowName, fmt.Sprintf("synced %d schema changes from source to destination", numberOfSchemaChanges))
+		}
+
 		return &model.SyncCompositeResponse{
 			SyncResponse: &model.SyncResponse{
 				CurrentSyncBatchID: -1,
@@ -321,7 +326,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		}
 	}
 
-	pushedRecordsWithCount := fmt.Sprintf("pushed %d records", numRecords)
+	pushedRecordsWithCount := fmt.Sprintf("pushed %d records into intermediate storage", numRecords)
 	activity.RecordHeartbeat(ctx, pushedRecordsWithCount)
 	a.Alerter.LogFlowInfo(ctx, flowName, pushedRecordsWithCount)
 

flow/activities/snapshot_activity.go

@@ -58,7 +58,7 @@ func (a *SnapshotActivity) SetupReplication(
 ) (*protos.SetupReplicationOutput, error) {
 	ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName)
 	logger := activity.GetLogger(ctx)
-
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Setting up replication slot and publication")
 	a.Alerter.LogFlowEvent(ctx, config.FlowJobName, "Started Snapshot Flow Job")
 
 	conn, err := connectors.GetByNameAs[*connpostgres.PostgresConnector](ctx, nil, a.CatalogPool, config.PeerName)
@@ -98,6 +98,8 @@ func (a *SnapshotActivity) SetupReplication(
 		connector:    conn,
 	}
 
+	a.Alerter.LogFlowInfo(ctx, config.FlowJobName, "Replication slot and publication setup complete")
+
 	return &protos.SetupReplicationOutput{
 		SlotName:         slotInfo.SlotName,
 		SnapshotName:     slotInfo.SnapshotName,