chore: lint

Author: Amogh-Bharadwaj

flow/connectors/postgres/pgdump_schema.go

@@ -40,8 +40,8 @@ func pipeCommand(
 	psqlCmd.Env = append(os.Environ(), "PGPASSWORD="+dstConfig.Password)
 
 	// handle TLS env vars
-	appendTLSEnv(srcCmd, srcConfig)
-	appendTLSEnv(psqlCmd, dstConfig)
+	appendTLSEnv(ctx, srcCmd, srcConfig)
+	appendTLSEnv(ctx, psqlCmd, dstConfig)
 
 	// pipe source command stdout -> psql stdin
 	pipe, err := srcCmd.StdoutPipe()
@@ -118,19 +118,19 @@ func buildPsqlArgs(config *protos.PostgresConfig) []string {
 	return args
 }
 
-func appendTLSEnv(cmd *exec.Cmd, config *protos.PostgresConfig) {
+func appendTLSEnv(ctx context.Context, cmd *exec.Cmd, config *protos.PostgresConfig) {
 	if config.RequireTls {
 		cmd.Env = append(cmd.Env, "PGSSLMODE=require")
 
 		if config.RootCa != nil && *config.RootCa != "" {
 			// write root CA to a temp file
 			tmpFile, err := os.CreateTemp("", "peerdb-root-ca-*.pem")
 			if err != nil {
-				slog.Warn("failed to create temp file for root CA, skipping sslrootcert", slog.Any("error", err))
+				slog.WarnContext(ctx, "failed to create temp file for root CA, skipping sslrootcert", slog.Any("error", err))
 				return
 			}
 			if _, err := tmpFile.WriteString(*config.RootCa); err != nil {
-				slog.Warn("failed to write root CA to temp file", slog.Any("error", err))
+				slog.WarnContext(ctx, "failed to write root CA to temp file", slog.Any("error", err))
 				tmpFile.Close()
 				os.Remove(tmpFile.Name())
 				return

flow/e2e/pg_schema_dump_test.go

@@ -220,13 +220,13 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_And_CDC() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load parent", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", dstParent)).Scan(&count)
+			"SELECT COUNT(*) FROM " + dstParent).Scan(&count)
 		return err == nil && count == 5
 	})
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load child", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", dstChild)).Scan(&count)
+			"SELECT COUNT(*) FROM " + dstChild).Scan(&count)
 		return err == nil && count == 10
 	})
 
@@ -306,29 +306,29 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_And_CDC() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc parent rows", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", dstParent)).Scan(&count)
+			"SELECT COUNT(*) FROM " + dstParent).Scan(&count)
 		return err == nil && count == 8
 	})
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc child rows", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", dstChild)).Scan(&count)
+			"SELECT COUNT(*) FROM " + dstChild).Scan(&count)
 		return err == nil && count == 15
 	})
 
 	// verify data integrity: compare actual row content
 	// query source and destination and compare
 	var srcParentCount, dstParentCount int64
-	err = srcConn.QueryRow(s.t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", srcParent)).Scan(&srcParentCount)
+	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + srcParent).Scan(&srcParentCount)
 	require.NoError(s.t, err)
-	err = dstConn.QueryRow(s.t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstParent)).Scan(&dstParentCount)
+	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + dstParent).Scan(&dstParentCount)
 	require.NoError(s.t, err)
 	require.Equal(s.t, srcParentCount, dstParentCount, "parent table row counts should match")
 
 	var srcChildCount, dstChildCount int64
-	err = srcConn.QueryRow(s.t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", srcChild)).Scan(&srcChildCount)
+	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + srcChild).Scan(&srcChildCount)
 	require.NoError(s.t, err)
-	err = dstConn.QueryRow(s.t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstChild)).Scan(&dstChildCount)
+	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + dstChild).Scan(&dstChildCount)
 	require.NoError(s.t, err)
 	require.Equal(s.t, srcChildCount, dstChildCount, "child table row counts should match")
 
@@ -358,7 +358,8 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_No_Owner_No_Privileges() {
 	dstConnStr := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s",
 		dstCfg.Host, dstCfg.Port, dstCfg.User, dstCfg.Password, dstCfg.Database)
 	dstConn, err := pgx.Connect(s.t.Context(), dstConnStr)
-	require.NoError(s.t, err, "failed to connect to secondary postgres on %s:%d (is the postgres2 tilt resource running?)", dstCfg.Host, dstCfg.Port)
+	require.NoError(s.t, err, "failed to connect to secondary postgres on %s:%d (postgres2 tilt resource running?)",
+		dstCfg.Host, dstCfg.Port)
 	s.t.Cleanup(func() { dstConn.Close(s.t.Context()) })
 
 	// sanity: role must not exist on destination
@@ -436,7 +437,7 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_No_Owner_No_Privileges() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load owned_tbl", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", qualified)).Scan(&count)
+			"SELECT COUNT(*) FROM " + qualified).Scan(&count)
 		return err == nil && count == 5
 	})
 
@@ -451,7 +452,7 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_No_Owner_No_Privileges() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc owned_tbl", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			fmt.Sprintf("SELECT COUNT(*) FROM %s", qualified)).Scan(&count)
+			"SELECT COUNT(*) FROM " + qualified).Scan(&count)
 		return err == nil && count == 10
 	})