more logging, align connector close for source peer types (#3555)

Author: heavycrystal

flow/connectors/mongo/mongo.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"log/slog"
 	"net"
 	"sync/atomic"
 	"time"
@@ -81,18 +82,22 @@ func NewMongoConnector(ctx context.Context, config *protos.MongoConfig) (*MongoC
 }
 
 func (c *MongoConnector) Close() error {
+	var errs []error
 	if c != nil && c.client != nil {
 		// Use a timeout to ensure the disconnect operation does not hang indefinitely
 		timeout, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
-		return c.client.Disconnect(timeout)
-	}
-	if c.ssh != nil && c.ssh.Client != nil {
-		if err := c.ssh.Close(); err != nil {
-			return fmt.Errorf("failed to close SSH tunnel: %w", err)
+		if err := c.client.Disconnect(timeout); err != nil {
+			c.logger.Error("failed to disconnect MongoDB client", slog.Any("error", err))
+			errs = append(errs, fmt.Errorf("failed to disconnect MongoDB client: %w", err))
 		}
 	}
-	return nil
+
+	if err := c.ssh.Close(); err != nil {
+		c.logger.Error("[mongo] failed to close SSH tunnel", slog.Any("error", err))
+		errs = append(errs, fmt.Errorf("[mongo] failed to close SSH tunnel: %w", err))
+	}
+	return errors.Join(errs...)
 }
 
 func (c *MongoConnector) ConnectionActive(ctx context.Context) error {

flow/connectors/mysql/mysql.go

@@ -131,13 +131,13 @@ func (c *MySqlConnector) Close() error {
 	}
 	if conn := c.conn.Swap(nil); conn != nil {
 		if err := conn.Close(); err != nil {
-			c.logger.Error("Failed to close MySQL connection", slog.Any("error", err))
-			errs = append(errs, err)
+			c.logger.Error("failed to close MySQL connection", slog.Any("error", err))
+			errs = append(errs, fmt.Errorf("failed to close MySQL connection: %w", err))
 		}
 	}
 	if err := c.ssh.Close(); err != nil {
-		c.logger.Error("Failed to close SSH tunnel", slog.Any("error", err))
-		errs = append(errs, err)
+		c.logger.Error("[mysql] failed to close SSH tunnel", slog.Any("error", err))
+		errs = append(errs, fmt.Errorf("[mysql] failed to close SSH tunnel: %w", err))
 	}
 	return errors.Join(errs...)
 }

flow/connectors/postgres/postgres.go

@@ -269,21 +269,30 @@ func (c *PostgresConnector) replicationOptions(publicationName string, pgVersion
 
 // Close closes all connections.
 func (c *PostgresConnector) Close() error {
-	var connerr, replerr error
+	var errs []error
 	if c != nil {
 		timeout, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
-		connerr = c.conn.Close(timeout)
+		if err := c.conn.Close(timeout); err != nil {
+			c.logger.Error("failed to close Postgres connection", slog.Any("error", err))
+			errs = append(errs, fmt.Errorf("failed to close Postgres connection: %w", err))
+		}
 
 		if c.replConn != nil {
 			timeout, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 			defer cancel()
-			replerr = c.replConn.Close(timeout)
+			if err := c.replConn.Close(timeout); err != nil {
+				c.logger.Error("failed to close Postgres replication connection", slog.Any("error", err))
+				errs = append(errs, fmt.Errorf("failed to close Postgres replication connection: %w", err))
+			}
 		}
 
-		c.ssh.Close()
+		if err := c.ssh.Close(); err != nil {
+			c.logger.Error("[postgres] failed to close SSH tunnel", slog.Any("error", err))
+			errs = append(errs, fmt.Errorf("[postgres] failed to close SSH tunnel: %w", err))
+		}
 	}
-	return errors.Join(connerr, replerr)
+	return errors.Join(errs...)
 }
 
 func (c *PostgresConnector) Conn() *pgx.Conn {

flow/workflows/qrep_flow.go

@@ -617,8 +617,8 @@ func QRepFlowWorkflow(
 	}
 
 	q.logger.Info("Continuing as new workflow",
-		slog.Any("Last Partition", state.LastPartition),
-		slog.Uint64("Number of Partitions Processed", state.NumPartitionsProcessed))
+		slog.Any("lastPartition", state.LastPartition),
+		slog.Uint64("numPartitionsProcessed", state.NumPartitionsProcessed))
 
 	if q.activeSignal == model.PauseSignal {
 		updateStatus(ctx, q.logger, state, protos.FlowStatus_STATUS_PAUSED)

flow/workflows/xmin_flow.go

@@ -118,8 +118,8 @@ func XminFlowWorkflow(
 	}
 
 	logger.Info("Continuing as new workflow",
-		slog.Any("Last Partition", state.LastPartition),
-		slog.Uint64("Number of Partitions Processed", state.NumPartitionsProcessed))
+		slog.Any("lastPartition", state.LastPartition),
+		slog.Uint64("numPartitionsProcessed", state.NumPartitionsProcessed))
 
 	if q.activeSignal == model.PauseSignal {
 		updateStatus(ctx, q.logger, state, protos.FlowStatus_STATUS_PAUSED)