refactor(kube): execStream requires a context.Context (#2878)

pkg/app/cassandra.go

@@ -228,5 +228,5 @@ func (cas *CassandraInstance) execCommand(ctx context.Context, command []string)
 	if err != nil || podname == "" {
 		return "", "", errors.Wrapf(err, "Error getting the pod and container name %s.", cas.name)
 	}
-	return kube.Exec(cas.cli, cas.namespace, podname, containername, command, nil)
+	return kube.Exec(ctx, cas.cli, cas.namespace, podname, containername, command, nil)
 }

pkg/app/cockroachdb.go

@@ -295,7 +295,7 @@ func (c *CockroachDB) execCommand(ctx context.Context, command []string) (string
 	if err != nil || podName == "" {
 		return "", "", errors.Wrapf(err, "Error  getting pod and container name %s.", c.name)
 	}
-	return kube.Exec(c.cli, c.namespace, podName, containerName, command, nil)
+	return kube.Exec(ctx, c.cli, c.namespace, podName, containerName, command, nil)
 }
 
 func (c *CockroachDB) waitForGC(ctx context.Context) error {

pkg/app/couchbase.go

@@ -262,7 +262,7 @@ func (cb CouchbaseDB) execCommand(ctx context.Context, command []string) (string
 	if err != nil || len(container) == 0 {
 		return "", "", err
 	}
-	return kube.Exec(cb.cli, cb.namespace, podName, container[0].Name, command, nil)
+	return kube.Exec(ctx, cb.cli, cb.namespace, podName, container[0].Name, command, nil)
 }
 
 // getRunningCBPod name of running couchbase cluster pod if its in ready state

pkg/app/csi-snapshot.go

@@ -169,7 +169,7 @@ func (tlc *TimeLogCSI) execCommand(ctx context.Context, command []string) (strin
 	if err != nil || podname == "" {
 		return "", errors.Wrapf(err, "Error getting pod and containername %s.", tlc.name)
 	}
-	_, stderr, err := kube.Exec(tlc.cli, tlc.namespace, podname, containername, command, nil)
+	_, stderr, err := kube.Exec(ctx, tlc.cli, tlc.namespace, podname, containername, command, nil)
 	return stderr, err
 }
 

pkg/app/elasticsearch.go

@@ -230,7 +230,7 @@ func (esi *ElasticsearchInstance) execCommand(ctx context.Context, command []str
 	if err != nil || podname == "" {
 		return "", "", errors.Wrapf(err, "Error getting the pod and container name %s.", esi.name)
 	}
-	return kube.Exec(esi.cli, esi.namespace, podname, containername, command, nil)
+	return kube.Exec(ctx, esi.cli, esi.namespace, podname, containername, command, nil)
 }
 
 func (esi *ElasticsearchInstance) curlCommand(method, path string) string {

pkg/app/foundationdb.go

@@ -245,5 +245,5 @@ func (fdb *FoundationDB) execCommand(ctx context.Context, command []string) (str
 		return "", "", err
 	}
 
-	return kube.Exec(fdb.cli, fdb.namespace, podName, containers[0].Name, command, nil)
+	return kube.Exec(ctx, fdb.cli, fdb.namespace, podName, containers[0].Name, command, nil)
 }

pkg/app/mariadb.go

@@ -220,7 +220,7 @@ func (m *MariaDB) execCommand(ctx context.Context, command []string) (string, st
 	if err != nil || podname == "" {
 		return "", "", errors.Wrapf(err, "Error  getting pod and containername %s.", m.name)
 	}
-	return kube.Exec(m.cli, m.namespace, podname, containername, command, nil)
+	return kube.Exec(ctx, m.cli, m.namespace, podname, containername, command, nil)
 }
 
 func mariaDBSTSName(release string) string {

pkg/app/mongodb-deploymentconfig.go

@@ -183,7 +183,7 @@ func (mongo *MongoDBDepConfig) execCommand(ctx context.Context, command []string
 	if err != nil {
 		return "", "", err
 	}
-	stdout, stderr, err := kube.Exec(mongo.cli, mongo.namespace, podName, containerName, command, nil)
+	stdout, stderr, err := kube.Exec(ctx, mongo.cli, mongo.namespace, podName, containerName, command, nil)
 	log.Print("Executing the command in pod and container", field.M{"pod": podName, "container": containerName, "cmd": command})
 
 	return stdout, stderr, errors.Wrapf(err, "Error executing command in the pod")

pkg/app/mongodb.go

@@ -229,5 +229,5 @@ func (mongo *MongoDB) execCommand(ctx context.Context, command []string) (string
 	if err != nil || podName == "" {
 		return "", "", err
 	}
-	return kube.Exec(mongo.cli, mongo.namespace, podName, containerName, command, nil)
+	return kube.Exec(ctx, mongo.cli, mongo.namespace, podName, containerName, command, nil)
 }

pkg/app/mssql.go

@@ -245,7 +245,7 @@ func (m MssqlDB) execCommand(ctx context.Context, command []string) (string, str
 	if err != nil || podName == "" {
 		return "", "", errors.Wrapf(err, "Error getting pod and container name for app %s.", m.name)
 	}
-	return kube.Exec(m.cli, m.namespace, podName, containerName, command, nil)
+	return kube.Exec(ctx, m.cli, m.namespace, podName, containerName, command, nil)
 }
 
 func (m *MssqlDB) getDeploymentObj() (*appsv1.Deployment, error) {

pkg/app/mysql-deploymentconfig.go

@@ -194,7 +194,7 @@ func (mdep *MysqlDepConfig) execCommand(ctx context.Context, command []string) (
 	if err != nil {
 		return "", "", err
 	}
-	stdout, stderr, err := kube.Exec(mdep.cli, mdep.namespace, podName, containerName, command, nil)
+	stdout, stderr, err := kube.Exec(ctx, mdep.cli, mdep.namespace, podName, containerName, command, nil)
 	if err != nil {
 		return stdout, stderr, errors.Wrapf(err, "Error executing command in the pod.")
 	}

pkg/app/mysql.go

@@ -244,5 +244,5 @@ func (mdb *MysqlDB) execCommand(ctx context.Context, command []string) (string,
 	if err != nil || podname == "" {
 		return "", "", errors.Wrapf(err, "Error  getting pod and containername %s.", mdb.name)
 	}
-	return kube.Exec(mdb.cli, mdb.namespace, podname, containername, command, nil)
+	return kube.Exec(ctx, mdb.cli, mdb.namespace, podname, containername, command, nil)
 }

pkg/app/postgresql-deploymentconfig.go

@@ -223,5 +223,5 @@ func (pgres *PostgreSQLDepConfig) execCommand(ctx context.Context, command []str
 	if err != nil {
 		return "", "", err
 	}
-	return kube.Exec(pgres.cli, pgres.namespace, pod, container, command, nil)
+	return kube.Exec(ctx, pgres.cli, pgres.namespace, pod, container, command, nil)
 }

pkg/app/postgresql.go

@@ -225,5 +225,5 @@ func (pdb PostgresDB) execCommand(ctx context.Context, command []string) (string
 	if err != nil {
 		return "", "", err
 	}
-	return kube.Exec(pdb.cli, pdb.namespace, pod, container, command, nil)
+	return kube.Exec(ctx, pdb.cli, pdb.namespace, pod, container, command, nil)
 }

pkg/app/rds_aurora_mysql.go

@@ -30,7 +30,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
-	aws "github.com/kanisterio/kanister/pkg/aws"
+	"github.com/kanisterio/kanister/pkg/aws"
 	"github.com/kanisterio/kanister/pkg/aws/ec2"
 	"github.com/kanisterio/kanister/pkg/aws/rds"
 	"github.com/kanisterio/kanister/pkg/field"
@@ -396,5 +396,5 @@ func (a RDSAuroraMySQLDB) execCommand(ctx context.Context, command []string) (st
 	if err != nil || podName == "" {
 		return "", "", err
 	}
-	return kube.Exec(a.cli, a.namespace, podName, containerName, command, nil)
+	return kube.Exec(ctx, a.cli, a.namespace, podName, containerName, command, nil)
 }

pkg/app/rds_postgres.go

@@ -32,7 +32,7 @@ import (
 	"sigs.k8s.io/yaml"
 
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
-	aws "github.com/kanisterio/kanister/pkg/aws"
+	"github.com/kanisterio/kanister/pkg/aws"
 	"github.com/kanisterio/kanister/pkg/aws/ec2"
 	"github.com/kanisterio/kanister/pkg/aws/rds"
 	"github.com/kanisterio/kanister/pkg/field"
@@ -477,5 +477,5 @@ func (pdb RDSPostgresDB) execCommand(ctx context.Context, command []string) (str
 	if err != nil || podName == "" {
 		return "", "", err
 	}
-	return kube.Exec(pdb.cli, pdb.namespace, podName, containerName, command, nil)
+	return kube.Exec(ctx, pdb.cli, pdb.namespace, podName, containerName, command, nil)
 }

pkg/controllers/repositoryserver/repository.go

@@ -15,13 +15,15 @@
 package repositoryserver
 
 import (
+	"context"
+
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
 	"github.com/kanisterio/kanister/pkg/kopia/command"
 	"github.com/kanisterio/kanister/pkg/kopia/repository"
 	reposerver "github.com/kanisterio/kanister/pkg/secrets/repositoryserver"
 )
 
-func (h *RepoServerHandler) connectToKopiaRepository() error {
+func (h *RepoServerHandler) connectToKopiaRepository(ctx context.Context) error {
 	repoConfiguration := h.getRepositoryConfiguration()
 	cacheSizeSettings := h.getRepositoryCacheSettings()
 	args := command.RepositoryCommandArgs{
@@ -43,6 +45,7 @@ func (h *RepoServerHandler) connectToKopiaRepository() error {
 	}
 
 	return repository.ConnectToKopiaRepository(
+		ctx,
 		h.KubeCli,
 		h.RepositoryServer.Namespace,
 		h.RepositoryServer.Status.ServerInfo.PodName,

pkg/controllers/repositoryserver/repositoryserver_controller.go

@@ -105,7 +105,7 @@ func (r *RepositoryServerReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	}
 
 	logger.Info("Connect to Kopia Repository")
-	if err := repoServerHandler.connectToKopiaRepository(); err != nil {
+	if err := repoServerHandler.connectToKopiaRepository(ctx); err != nil {
 		condition := getCondition(metav1.ConditionFalse, conditionReasonRepositoryConnectedErr, err.Error(), crv1alpha1.RepositoryConnected)
 		if uerr := repoServerHandler.setCondition(ctx, condition, crv1alpha1.Failed); uerr != nil {
 			return ctrl.Result{}, uerr

pkg/controllers/repositoryserver/repositoryserver_controller_test.go

@@ -273,7 +273,7 @@ func (s *RepoServerControllerSuite) TestRepositoryServerStatusIsServerReady(c *C
 	err = kube.WaitForPodReady(ctx, s.kubeCli, s.repoServerControllerNamespace, repoServerCRCreated.Status.ServerInfo.PodName)
 	c.Assert(err, IsNil)
 
-	err = testutil.CreateTestKopiaRepository(s.kubeCli, repoServerCRCreated, testutil.GetDefaultS3CompliantStorageLocation())
+	err = testutil.CreateTestKopiaRepository(ctx, s.kubeCli, repoServerCRCreated, testutil.GetDefaultS3CompliantStorageLocation())
 	c.Assert(err, IsNil)
 
 	_, err = s.waitOnRepositoryServerState(c, repoServerCRCreated.Name)

pkg/controllers/repositoryserver/server.go

@@ -64,7 +64,7 @@ func (h *RepoServerHandler) startRepoProxyServer(ctx context.Context) (err error
 			Background:       true,
 		},
 	)
-	stdout, stderr, err := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 	if err != nil {
@@ -99,7 +99,7 @@ func (h *RepoServerHandler) checkServerStatus(ctx context.Context, serverAddress
 	if err != nil {
 		return errors.Wrap(err, "Failed to extract fingerprint from Kopia API server certificate secret data")
 	}
-	stdout, stderr, exErr := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+	stdout, stderr, exErr := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 	return exErr
@@ -127,7 +127,7 @@ func (h *RepoServerHandler) createOrUpdateClientUsers(ctx context.Context) error
 				LogDirectory:   command.DefaultLogDirectory,
 			},
 		})
-	stdout, stderr, err := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 	if err != nil {
@@ -164,7 +164,7 @@ func (h *RepoServerHandler) createOrUpdateClientUsers(ctx context.Context) error
 					NewUsername:  serverUsername,
 					UserPassword: string(password),
 				})
-			stdout, stderr, err := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+			stdout, stderr, err := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 			format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 			format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 			if err != nil {
@@ -182,7 +182,7 @@ func (h *RepoServerHandler) createOrUpdateClientUsers(ctx context.Context) error
 				NewUsername:  serverUsername,
 				UserPassword: string(password),
 			})
-		stdout, stderr, err := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+		stdout, stderr, err := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 		format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 		format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 		if err != nil {
@@ -215,7 +215,7 @@ func (h *RepoServerHandler) refreshServer(ctx context.Context) error {
 			ServerPassword: password,
 			Fingerprint:    fingerprint,
 		})
-	stdout, stderr, err := kube.Exec(h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, h.KubeCli, h.RepositoryServer.Namespace, h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, cmd, nil)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stdout)
 	format.Log(h.RepositoryServer.Status.ServerInfo.PodName, repoServerPodContainerName, stderr)
 	if err != nil {

pkg/controllers/repositoryserver/utils.go

@@ -247,8 +247,8 @@ func WaitTillCommandSucceed(ctx context.Context, cli kubernetes.Interface, cmd [
 		Jitter: false,
 		Min:    100 * time.Millisecond,
 		Max:    180 * time.Second,
-	}, func(context.Context) (bool, error) {
-		stdout, stderr, exErr := kube.Exec(cli, namespace, podName, container, cmd, nil)
+	}, func(localCtx context.Context) (bool, error) {
+		stdout, stderr, exErr := kube.Exec(localCtx, cli, namespace, podName, container, cmd, nil)
 		format.Log(podName, container, stdout)
 		format.Log(podName, container, stderr)
 		if exErr != nil {

pkg/function/backup_data.go

@@ -171,7 +171,7 @@ func backupData(ctx context.Context, cli kubernetes.Interface, namespace, pod, c
 		return backupDataParsedOutput{}, err
 	}
 	defer CleanUpCredsFile(ctx, pw, namespace, pod, container)
-	if err = restic.GetOrCreateRepository(cli, namespace, pod, container, backupArtifactPrefix, encryptionKey, insecureTLS, tp.Profile); err != nil {
+	if err = restic.GetOrCreateRepository(ctx, cli, namespace, pod, container, backupArtifactPrefix, encryptionKey, insecureTLS, tp.Profile); err != nil {
 		return backupDataParsedOutput{}, err
 	}
 
@@ -181,7 +181,7 @@ func backupData(ctx context.Context, cli kubernetes.Interface, namespace, pod, c
 	if err != nil {
 		return backupDataParsedOutput{}, err
 	}
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, cli, namespace, pod, container, cmd, nil)
 	format.LogWithCtx(ctx, pod, container, stdout)
 	format.LogWithCtx(ctx, pod, container, stderr)
 	if err != nil {

pkg/function/backup_data_using_kopia_server.go

@@ -142,6 +142,7 @@ func (b *backupDataUsingKopiaServerFunc) Exec(ctx context.Context, tp param.Temp
 	}
 
 	snapInfo, err := backupDataUsingKopiaServer(
+		ctx,
 		cli,
 		container,
 		hostname,
@@ -182,6 +183,7 @@ func (b *backupDataUsingKopiaServerFunc) ExecutionProgress() (crv1alpha1.PhasePr
 }
 
 func backupDataUsingKopiaServer(
+	ctx context.Context,
 	cli kubernetes.Interface,
 	container,
 	hostname,
@@ -212,7 +214,7 @@ func backupDataUsingKopiaServer(
 		},
 	})
 
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	if err != nil {
@@ -234,7 +236,7 @@ func backupDataUsingKopiaServer(
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to construct snapshot create command")
 	}
-	stdout, stderr, err = kube.Exec(cli, namespace, pod, container, cmd, nil)
+	stdout, stderr, err = kube.Exec(ctx, cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 

pkg/function/checkRepository.go

@@ -92,6 +92,7 @@ func CheckRepositoryPodFunc(
 		defer remover.Remove(context.Background()) //nolint:errcheck
 
 		err = restic.CheckIfRepoIsReachable(
+			ctx,
 			tp.Profile,
 			targetPath,
 			encryptionKey,

pkg/function/copy_volume_data.go

@@ -134,6 +134,7 @@ func copyVolumeDataPodFunc(
 		pod := pc.Pod()
 		// Get restic repository
 		if err := restic.GetOrCreateRepository(
+			ctx,
 			cli,
 			pod.Namespace,
 			pod.Name,

pkg/function/kube_exec.go

@@ -99,7 +99,7 @@ func (kef *kubeExecFunc) Exec(ctx context.Context, tp param.TemplateParams, args
 		bufStdout  = &bytes.Buffer{}
 		outWriters = io.MultiWriter(os.Stdout, bufStdout)
 	)
-	if err := kube.ExecOutput(cli, namespace, pod, container, cmd, nil, outWriters, os.Stderr); err != nil {
+	if err := kube.ExecOutput(ctx, cli, namespace, pod, container, cmd, nil, outWriters, os.Stderr); err != nil {
 		return nil, err
 	}
 

pkg/function/kube_exec_all.go

@@ -118,7 +118,7 @@ func execAll(ctx context.Context, cli kubernetes.Interface, namespace string, ps
 	for _, p := range ps {
 		for _, c := range cs {
 			go func(p string, c string) {
-				stdout, stderr, err := kube.Exec(cli, namespace, p, c, cmd, nil)
+				stdout, stderr, err := kube.Exec(ctx, cli, namespace, p, c, cmd, nil)
 				format.LogWithCtx(ctx, p, c, stdout)
 				format.LogWithCtx(ctx, p, c, stderr)
 				errChan <- err

pkg/kopia/repository/connect.go

@@ -1,6 +1,7 @@
 package repository
 
 import (
+	"context"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -14,6 +15,7 @@ import (
 
 // ConnectToKopiaRepository connects to an already existing kopia repository
 func ConnectToKopiaRepository(
+	ctx context.Context,
 	cli kubernetes.Interface,
 	namespace,
 	pod,
@@ -25,7 +27,7 @@ func ConnectToKopiaRepository(
 		return errors.Wrap(err, "Failed to generate repository connect command")
 	}
 
-	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd, nil)
+	stdout, stderr, err := kube.Exec(ctx, cli, namespace, pod, container, cmd, nil)
 	format.Log(pod, container, stdout)
 	format.Log(pod, container, stderr)
 	if err == nil {