Stop using deprecated Executor.Stream in k8s.io/client-go/tools/remotecommand package. Start using StreamWithContext.

dsessler7 · dsessler7 · commit c7a885d862fa · 2024-06-20T16:14:15.000-07:00
diff --git a/internal/controller/postgrescluster/controller.go b/internal/controller/postgrescluster/controller.go
@@ -65,7 +65,7 @@ type Reconciler struct {
 	IsOpenShift bool
 	Owner       client.FieldOwner
 	PodExec     func(
-		namespace, pod, container string,
+		ctx context.Context, namespace, pod, container string,
 		stdin io.Reader, stdout, stderr io.Writer, command ...string,
 	) error
 	Recorder     record.EventRecorder
diff --git a/internal/controller/postgrescluster/instance.go b/internal/controller/postgrescluster/instance.go
@@ -792,7 +792,7 @@ func (r *Reconciler) rolloutInstance(
 
 	pod := instance.Pods[0]
 	exec := func(_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
-		return r.PodExec(pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
+		return r.PodExec(ctx, pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
 	}
 
 	primary, known := instance.IsPrimary()
diff --git a/internal/controller/postgrescluster/instance_rollout_test.go b/internal/controller/postgrescluster/instance_rollout_test.go
@@ -75,7 +75,7 @@ func TestReconcilerRolloutInstance(t *testing.T) {
 
 		execCalls := 0
 		reconciler.PodExec = func(
-			namespace, pod, container string, stdin io.Reader, _, _ io.Writer, command ...string,
+			ctx context.Context, namespace, pod, container string, stdin io.Reader, _, _ io.Writer, command ...string,
 		) error {
 			execCalls++
 
@@ -134,7 +134,7 @@ func TestReconcilerRolloutInstance(t *testing.T) {
 			reconciler := &Reconciler{}
 			reconciler.Tracer = otel.Tracer(t.Name())
 			reconciler.PodExec = func(
-				namespace, pod, container string, _ io.Reader, stdout, _ io.Writer, command ...string,
+				ctx context.Context, namespace, pod, container string, _ io.Reader, stdout, _ io.Writer, command ...string,
 			) error {
 				execCalls++
 
@@ -162,7 +162,7 @@ func TestReconcilerRolloutInstance(t *testing.T) {
 			reconciler := &Reconciler{}
 			reconciler.Tracer = otel.Tracer(t.Name())
 			reconciler.PodExec = func(
-				_, _, _ string, _ io.Reader, _, _ io.Writer, _ ...string,
+				ctx context.Context, _, _, _ string, _ io.Reader, _, _ io.Writer, _ ...string,
 			) error {
 				// Nothing useful in stdout.
 				return nil
diff --git a/internal/controller/postgrescluster/patroni.go b/internal/controller/postgrescluster/patroni.go
@@ -103,7 +103,7 @@ func (r *Reconciler) handlePatroniRestarts(
 			ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 		) error {
 			pod := primaryNeedsRestart.Pods[0]
-			return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+			return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 		})
 
 		return errors.WithStack(exec.RestartPendingMembers(ctx, "master", naming.PatroniScope(cluster)))
@@ -128,7 +128,7 @@ func (r *Reconciler) handlePatroniRestarts(
 			ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 		) error {
 			pod := replicaNeedsRestart.Pods[0]
-			return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+			return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 		})
 
 		return errors.WithStack(exec.RestartPendingMembers(ctx, "replica", naming.PatroniScope(cluster)))
@@ -212,8 +212,8 @@ func (r *Reconciler) reconcilePatroniDynamicConfiguration(
 	// NOTE(cbandy): Despite the guards above, calling PodExec may still fail
 	// due to a missing or stopped container.
 
-	exec := func(_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
-		return r.PodExec(pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
+	exec := func(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
+		return r.PodExec(ctx, pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
 	}
 
 	var configuration map[string]any
@@ -535,7 +535,7 @@ func (r *Reconciler) reconcilePatroniSwitchover(ctx context.Context,
 	}
 	exec := func(_ context.Context, stdin io.Reader, stdout, stderr io.Writer,
 		command ...string) error {
-		return r.PodExec(runningPod.Namespace, runningPod.Name, naming.ContainerDatabase, stdin,
+		return r.PodExec(ctx, runningPod.Namespace, runningPod.Name, naming.ContainerDatabase, stdin,
 			stdout, stderr, command...)
 	}
 
diff --git a/internal/controller/postgrescluster/patroni_test.go b/internal/controller/postgrescluster/patroni_test.go
@@ -544,7 +544,7 @@ func TestReconcilePatroniSwitchover(t *testing.T) {
 	var timelineCallNoLeader, timelineCall bool
 	r := Reconciler{
 		Client: client,
-		PodExec: func(namespace, pod, container string,
+		PodExec: func(ctx context.Context, namespace, pod, container string,
 			stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
 			called = true
 			switch {
diff --git a/internal/controller/postgrescluster/pgadmin.go b/internal/controller/postgrescluster/pgadmin.go
@@ -454,9 +454,9 @@ func (r *Reconciler) reconcilePGAdminUsers(
 		ctx = logging.NewContext(ctx, logging.FromContext(ctx).WithValues("pod", pod.Name))
 
 		podExecutor = func(
-			_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
+			ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 		) error {
-			return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+			return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 		}
 	}
 	if podExecutor == nil {
diff --git a/internal/controller/postgrescluster/pgadmin_test.go b/internal/controller/postgrescluster/pgadmin_test.go
@@ -785,7 +785,7 @@ func TestReconcilePGAdminUsers(t *testing.T) {
 
 		calls := 0
 		r.PodExec = func(
-			namespace, pod, container string,
+			ctx context.Context, namespace, pod, container string,
 			stdin io.Reader, stdout, stderr io.Writer, command ...string,
 		) error {
 			calls++
diff --git a/internal/controller/postgrescluster/pgbackrest.go b/internal/controller/postgrescluster/pgbackrest.go
@@ -2634,7 +2634,7 @@ func (r *Reconciler) reconcileStanzaCreate(ctx context.Context,
 	// create a pgBackRest executor and attempt stanza creation
 	exec := func(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer,
 		command ...string) error {
-		return r.PodExec(postgresCluster.GetNamespace(), writableInstanceName,
+		return r.PodExec(ctx, postgresCluster.GetNamespace(), writableInstanceName,
 			naming.ContainerDatabase, stdin, stdout, stderr, command...)
 	}
 
diff --git a/internal/controller/postgrescluster/pgbackrest_test.go b/internal/controller/postgrescluster/pgbackrest_test.go
@@ -700,13 +700,13 @@ func TestReconcileStanzaCreate(t *testing.T) {
 		},
 	}})
 
-	stanzaCreateFail := func(namespace, pod, container string, stdin io.Reader, stdout,
-		stderr io.Writer, command ...string) error {
+	stanzaCreateFail := func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+		stdout, stderr io.Writer, command ...string) error {
 		return errors.New("fake stanza create failed")
 	}
 
-	stanzaCreateSuccess := func(namespace, pod, container string, stdin io.Reader, stdout,
-		stderr io.Writer, command ...string) error {
+	stanzaCreateSuccess := func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+		stdout, stderr io.Writer, command ...string) error {
 		return nil
 	}
 
diff --git a/internal/controller/postgrescluster/pgbouncer.go b/internal/controller/postgrescluster/pgbouncer.go
@@ -181,8 +181,8 @@ func (r *Reconciler) reconcilePGBouncerInPostgreSQL(
 
 	if err == nil {
 		ctx := logging.NewContext(ctx, logging.FromContext(ctx).WithValues("revision", revision))
-		err = action(ctx, func(_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
-			return r.PodExec(pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
+		err = action(ctx, func(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {
+			return r.PodExec(ctx, pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
 		})
 	}
 	if err == nil {
diff --git a/internal/controller/postgrescluster/pgmonitor.go b/internal/controller/postgrescluster/pgmonitor.go
@@ -144,9 +144,9 @@ func (r *Reconciler) reconcilePGMonitorExporter(ctx context.Context,
 
 		// Apply the necessary SQL and record its hash in cluster.Status
 		if err == nil {
-			err = action(ctx, func(_ context.Context, stdin io.Reader,
+			err = action(ctx, func(ctx context.Context, stdin io.Reader,
 				stdout, stderr io.Writer, command ...string) error {
-				return r.PodExec(writablePod.Namespace, writablePod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
+				return r.PodExec(ctx, writablePod.Namespace, writablePod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
 			})
 		}
 		if err == nil {
diff --git a/internal/controller/postgrescluster/pgmonitor_test.go b/internal/controller/postgrescluster/pgmonitor_test.go
@@ -506,8 +506,8 @@ func TestReconcilePGMonitorExporterSetupErrors(t *testing.T) {
 			ctx := context.Background()
 			var called bool
 			reconciler := &Reconciler{
-				PodExec: func(namespace, pod, container string, stdin io.Reader, stdout,
-					stderr io.Writer, command ...string) error {
+				PodExec: func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+					stdout, stderr io.Writer, command ...string) error {
 					called = true
 					return nil
 				},
@@ -530,8 +530,8 @@ func TestReconcilePGMonitorExporter(t *testing.T) {
 	ctx := context.Background()
 	var called bool
 	reconciler := &Reconciler{
-		PodExec: func(namespace, pod, container string, stdin io.Reader, stdout,
-			stderr io.Writer, command ...string) error {
+		PodExec: func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+			stdout, stderr io.Writer, command ...string) error {
 			called = true
 			return nil
 		},
@@ -624,8 +624,8 @@ func TestReconcilePGMonitorExporterStatus(t *testing.T) {
 
 			// Create reconciler with mock PodExec function
 			reconciler := &Reconciler{
-				PodExec: func(namespace, pod, container string, stdin io.Reader, stdout,
-					stderr io.Writer, command ...string) error {
+				PodExec: func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+					stdout, stderr io.Writer, command ...string) error {
 					called = true
 					return nil
 				},
diff --git a/internal/controller/postgrescluster/postgres.go b/internal/controller/postgrescluster/postgres.go
@@ -199,9 +199,9 @@ func (r *Reconciler) reconcilePostgresDatabases(
 
 	ctx = logging.NewContext(ctx, logging.FromContext(ctx).WithValues("pod", pod.Name))
 	podExecutor = func(
-		_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
+		ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 	) error {
-		return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+		return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 	}
 
 	// Gather the list of database that should exist in PostgreSQL.
@@ -515,9 +515,9 @@ func (r *Reconciler) reconcilePostgresUsersInPostgreSQL(
 			ctx = logging.NewContext(ctx, logging.FromContext(ctx).WithValues("pod", pod.Name))
 
 			podExecutor = func(
-				_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
+				ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 			) error {
-				return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+				return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 			}
 			break
 		}
@@ -840,7 +840,7 @@ func (r *Reconciler) reconcilePostgresWALVolume(
 				// This assumes that $PGDATA matches the configured PostgreSQL "data_directory".
 				var stdout bytes.Buffer
 				err = errors.WithStack(r.PodExec(
-					observed.Pods[0].Namespace, observed.Pods[0].Name, naming.ContainerDatabase,
+					ctx, observed.Pods[0].Namespace, observed.Pods[0].Name, naming.ContainerDatabase,
 					nil, &stdout, nil, "bash", "-ceu", "--", `exec realpath "${PGDATA}/pg_wal"`))
 
 				walDirectory = strings.TrimRight(stdout.String(), "\n")
@@ -944,9 +944,9 @@ func (r *Reconciler) reconcileDatabaseInitSQL(ctx context.Context,
 	}
 
 	podExecutor = func(
-		_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
+		ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 	) error {
-		return r.PodExec(pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
+		return r.PodExec(ctx, pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)
 	}
 
 	// A writable pod executor has been found and we have the sql provided by
diff --git a/internal/controller/postgrescluster/postgres_test.go b/internal/controller/postgrescluster/postgres_test.go
@@ -359,7 +359,7 @@ volumeMode: Filesystem
 
 					expected := errors.New("flop")
 					reconciler.PodExec = func(
-						namespace, pod, container string,
+						ctx context.Context, namespace, pod, container string,
 						_ io.Reader, _, _ io.Writer, command ...string,
 					) error {
 						assert.Equal(t, namespace, "pod-ns")
@@ -376,7 +376,7 @@ volumeMode: Filesystem
 
 					// Files are in the wrong place; expect no changes to the PVC.
 					reconciler.PodExec = func(
-						_, _, _ string, _ io.Reader, stdout, _ io.Writer, _ ...string,
+						ctx context.Context, _, _, _ string, _ io.Reader, stdout, _ io.Writer, _ ...string,
 					) error {
 						assert.Assert(t, stdout != nil)
 						_, err := stdout.Write([]byte("some-place\n"))
@@ -399,7 +399,7 @@ volumeMode: Filesystem
 						new(corev1.ContainerStateRunning)
 
 					reconciler.PodExec = func(
-						_, _, _ string, _ io.Reader, stdout, _ io.Writer, _ ...string,
+						ctx context.Context, _, _, _ string, _ io.Reader, stdout, _ io.Writer, _ ...string,
 					) error {
 						assert.Assert(t, stdout != nil)
 						_, err := stdout.Write([]byte(postgres.WALDirectory(cluster, spec) + "\n"))
@@ -751,8 +751,8 @@ func TestReconcileDatabaseInitSQL(t *testing.T) {
 
 		// Overwrite the PodExec function with a check to ensure the exec
 		// call would have been made
-		PodExec: func(namespace, pod, container string, stdin io.Reader, stdout,
-			stderr io.Writer, command ...string) error {
+		PodExec: func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+			stdout, stderr io.Writer, command ...string) error {
 			called = true
 			return nil
 		},
@@ -875,8 +875,8 @@ func TestReconcileDatabaseInitSQLConfigMap(t *testing.T) {
 
 		// Overwrite the PodExec function with a check to ensure the exec
 		// call would have been made
-		PodExec: func(namespace, pod, container string, stdin io.Reader, stdout,
-			stderr io.Writer, command ...string) error {
+		PodExec: func(ctx context.Context, namespace, pod, container string, stdin io.Reader,
+			stdout, stderr io.Writer, command ...string) error {
 			called = true
 			return nil
 		},
diff --git a/internal/controller/runtime/pod_client.go b/internal/controller/runtime/pod_client.go
@@ -16,6 +16,7 @@
 package runtime
 
 import (
+	"context"
 	"io"
 
 	corev1 "k8s.io/api/core/v1"
@@ -29,7 +30,7 @@ import (
 // podExecutor runs command on container in pod in namespace. Non-nil streams
 // (stdin, stdout, and stderr) are attached the to the remote process.
 type podExecutor func(
-	namespace, pod, container string,
+	ctx context.Context, namespace, pod, container string,
 	stdin io.Reader, stdout, stderr io.Writer, command ...string,
 ) error
 
@@ -49,7 +50,7 @@ func NewPodExecutor(config *rest.Config) (podExecutor, error) {
 	client, err := newPodClient(config)
 
 	return func(
-		namespace, pod, container string,
+		ctx context.Context, namespace, pod, container string,
 		stdin io.Reader, stdout, stderr io.Writer, command ...string,
 	) error {
 		request := client.Post().
@@ -66,7 +67,7 @@ func NewPodExecutor(config *rest.Config) (podExecutor, error) {
 		exec, err := remotecommand.NewSPDYExecutor(config, "POST", request.URL())
 
 		if err == nil {
-			err = exec.Stream(remotecommand.StreamOptions{
+			err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
 				Stdin:  stdin,
 				Stdout: stdout,
 				Stderr: stderr,
diff --git a/internal/controller/standalone_pgadmin/controller.go b/internal/controller/standalone_pgadmin/controller.go
@@ -37,7 +37,7 @@ type PGAdminReconciler struct {
 	client.Client
 	Owner   client.FieldOwner
 	PodExec func(
-		namespace, pod, container string,
+		ctx context.Context, namespace, pod, container string,
 		stdin io.Reader, stdout, stderr io.Writer, command ...string,
 	) error
 	Recorder    record.EventRecorder
diff --git a/internal/controller/standalone_pgadmin/users.go b/internal/controller/standalone_pgadmin/users.go
@@ -80,9 +80,9 @@ func (r *PGAdminReconciler) reconcilePGAdminUsers(ctx context.Context, pgadmin *
 		ctx = logging.NewContext(ctx, logging.FromContext(ctx).WithValues("pod", pod.Name))
 
 		podExecutor = func(
-			_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
+			ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,
 		) error {
-			return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
+			return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)
 		}
 	}
 	if podExecutor == nil {
diff --git a/internal/controller/standalone_pgadmin/users_test.go b/internal/controller/standalone_pgadmin/users_test.go

Original file line number	Diff line number	Diff line change
`@@ -792,7 +792,7 @@ func (r *Reconciler) rolloutInstance(`
`792`	`792`
`793`	`793`	`pod := instance.Pods[0]`
`794`	`794`	`exec := func(_ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string) error {`
`795`		`- return r.PodExec(pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)`
	`795`	`+ return r.PodExec(ctx, pod.Namespace, pod.Name, naming.ContainerDatabase, stdin, stdout, stderr, command...)`
`796`	`796`	`}`
`797`	`797`
`798`	`798`	`primary, known := instance.IsPrimary()`
Original file line number	Diff line number	Diff line change
`@@ -454,9 +454,9 @@ func (r *Reconciler) reconcilePGAdminUsers(`
`454`	`454`	`ctx = logging.NewContext(ctx, logging.FromContext(ctx).WithValues("pod", pod.Name))`
`455`	`455`
`456`	`456`	`podExecutor = func(`
`457`		`- _ context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,`
	`457`	`+ ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, command ...string,`
`458`	`458`	`) error {`
`459`		`- return r.PodExec(pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)`
	`459`	`+ return r.PodExec(ctx, pod.Namespace, pod.Name, container, stdin, stdout, stderr, command...)`
`460`	`460`	`}`
`461`	`461`	`}`
`462`	`462`	`if podExecutor == nil {`
Original file line number	Diff line number	Diff line change
`@@ -2634,7 +2634,7 @@ func (r *Reconciler) reconcileStanzaCreate(ctx context.Context,`
`2634`	`2634`	`// create a pgBackRest executor and attempt stanza creation`
`2635`	`2635`	`exec := func(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer,`
`2636`	`2636`	`command ...string) error {`
`2637`		`- return r.PodExec(postgresCluster.GetNamespace(), writableInstanceName,`
	`2637`	`+ return r.PodExec(ctx, postgresCluster.GetNamespace(), writableInstanceName,`
`2638`	`2638`	`naming.ContainerDatabase, stdin, stdout, stderr, command...)`
`2639`	`2639`	`}`
`2640`	`2640`