fix a bunch of deprecated things

pkg/mover/exec.go

@@ -2,6 +2,7 @@ package mover
 
 import (
 	"bytes"
+	"context"
 	"io"
 	"os"
 
@@ -33,13 +34,16 @@ func (m *MoverJob) Exec(pod v1.Pod, config *rest.Config, cmd []string, input io.
 	done := false
 	go func() {
 		for {
-			io.Copy(os.Stdout, prefixReader)
+			_, err := io.Copy(os.Stdout, prefixReader)
+			if err != nil {
+				m.log.WithError(err).Warning("failed to copy")
+			}
 			if done {
 				return
 			}
 		}
 	}()
-	err = exec.Stream(remotecommand.StreamOptions{
+	err = exec.StreamWithContext(context.Background(), remotecommand.StreamOptions{
 		Stdin:  input,
 		Stdout: output,
 		Stderr: os.Stdout,

pkg/mover/mover.go

@@ -11,7 +11,6 @@ import (
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -107,7 +106,7 @@ func (m *MoverJob) Start() *MoverJob {
 						{
 							Name:            ContainerName,
 							Image:           config.ContainerImage,
-							ImagePullPolicy: v1.PullAlways,
+							ImagePullPolicy: corev1.PullAlways,
 							Args:            []string{string(m.mode)},
 							VolumeMounts:    mounts,
 							TTY:             true,
@@ -127,15 +126,15 @@ func (m *MoverJob) Start() *MoverJob {
 		}
 	}
 
-	j, err := m.kClient.BatchV1().Jobs(m.Namespace).Create(context.TODO(), job, metav1.CreateOptions{})
+	j, err := m.kClient.BatchV1().Jobs(m.Namespace).Create(context.Background(), job, metav1.CreateOptions{})
 	if err != nil {
 		panic(err)
 	}
 	m.kJob = j
 	return m
 }
 
-func (m *MoverJob) followLogs(pod v1.Pod) {
+func (m *MoverJob) followLogs(pod corev1.Pod) {
 	req := m.kClient.CoreV1().Pods(m.Namespace).GetLogs(pod.Name, &corev1.PodLogOptions{
 		Follow:    true,
 		Container: ContainerName,
@@ -149,7 +148,10 @@ func (m *MoverJob) followLogs(pod v1.Pod) {
 	prefixReader := prefixer.New(podLogs, "[mover logs]: ")
 
 	for {
-		io.Copy(os.Stdout, prefixReader)
+		_, err := io.Copy(os.Stdout, prefixReader)
+		if err != nil {
+			m.log.WithError(err).Warning("failed to copy")
+		}
 	}
 }
 
@@ -161,14 +163,17 @@ func (m *MoverJob) getDeleteOptions() metav1.DeleteOptions {
 }
 
 func (m *MoverJob) Cleanup() error {
-	err := m.kClient.BatchV1().Jobs(m.Namespace).Delete(context.TODO(), m.Name, m.getDeleteOptions())
+	err := m.kClient.BatchV1().Jobs(m.Namespace).Delete(context.Background(), m.Name, m.getDeleteOptions())
 	if err != nil {
-		m.log.WithError(err).Debug("Failed to delete job")
+		m.log.WithError(err).WithField("name", m.Name).Debug("Failed to delete job")
 		return err
 	}
-	pods := m.getPods()
+	pods := m.getPods(context.Background())
 	for _, pod := range pods {
-		m.kClient.CoreV1().Pods(m.Namespace).Delete(context.TODO(), pod.Name, m.getDeleteOptions())
+		err := m.kClient.CoreV1().Pods(m.Namespace).Delete(context.Background(), pod.Name, m.getDeleteOptions())
+		if err != nil {
+			m.log.WithError(err).WithField("name", pod.Name).Warning("failed to delete pod")
+		}
 	}
 	return nil
 }

pkg/strategies/copyTwiceName.go

@@ -102,7 +102,11 @@ func (c *CopyTwiceNameStrategy) Do(sourcePVC *v1.PersistentVolumeClaim, destTemp
 		c.log.WithError(err).Warning("Failed to delete source pvc")
 		return c.Cleanup()
 	}
-	c.waitForPVCDeletion(sourcePVC)
+	err = c.waitForPVCDeletion(sourcePVC)
+	if err != nil {
+		c.log.WithError(err).Warning("failed to delete source pvc")
+		return c.Cleanup()
+	}
 
 	c.log.WithField("stage", 4).Debug("creating final destination PVC")
 	destInst, err := c.kClient.CoreV1().PersistentVolumeClaims(destTemplate.ObjectMeta.Namespace).Create(context.TODO(), destTemplate, metav1.CreateOptions{})
@@ -128,10 +132,14 @@ func (c *CopyTwiceNameStrategy) Do(sourcePVC *v1.PersistentVolumeClaim, destTemp
 	c.log.WithField("stage", 6).Debug("deleting temporary PVC")
 	err = c.kClient.CoreV1().PersistentVolumeClaims(destTemplate.ObjectMeta.Namespace).Delete(context.TODO(), c.TempDestPVC.Name, c.getDeleteOptions())
 	if err != nil {
-		c.log.WithError(err).Warning("Failed to delete temporary destination pvc")
+		c.log.WithError(err).Warning("failed to delete temporary destination pvc")
+		return c.Cleanup()
+	}
+	err = c.waitForPVCDeletion(c.TempDestPVC)
+	if err != nil {
+		c.log.WithError(err).Warning("failed to delete temporary destination pvc")
 		return c.Cleanup()
 	}
-	c.waitForPVCDeletion(c.TempDestPVC)
 
 	c.log.Info("And we're done")
 
@@ -157,8 +165,8 @@ func (c *CopyTwiceNameStrategy) setTimeout(pvc *v1.PersistentVolumeClaim) {
 }
 
 func (c *CopyTwiceNameStrategy) waitForPVCDeletion(pvc *v1.PersistentVolumeClaim) error {
-	return wait.Poll(2*time.Second, c.timeout, func() (bool, error) {
-		_, err := c.kClient.CoreV1().PersistentVolumeClaims(pvc.ObjectMeta.Namespace).Get(context.TODO(), pvc.ObjectMeta.Name, metav1.GetOptions{})
+	return wait.PollUntilContextTimeout(context.Background(), 2*time.Second, c.timeout, true, func(ctx context.Context) (bool, error) {
+		_, err := c.kClient.CoreV1().PersistentVolumeClaims(pvc.ObjectMeta.Namespace).Get(ctx, pvc.ObjectMeta.Name, metav1.GetOptions{})
 		if errors.IsNotFound(err) {
 			return true, nil
 		}
@@ -168,8 +176,8 @@ func (c *CopyTwiceNameStrategy) waitForPVCDeletion(pvc *v1.PersistentVolumeClaim
 }
 
 func (c *CopyTwiceNameStrategy) waitForBound(pvc *v1.PersistentVolumeClaim) error {
-	return wait.Poll(2*time.Second, c.timeout, func() (bool, error) {
-		pvc, err := c.kClient.CoreV1().PersistentVolumeClaims(pvc.ObjectMeta.Namespace).Get(context.TODO(), pvc.ObjectMeta.Name, metav1.GetOptions{})
+	return wait.PollUntilContextTimeout(context.Background(), 2*time.Second, c.timeout, true, func(ctx context.Context) (bool, error) {
+		pvc, err := c.kClient.CoreV1().PersistentVolumeClaims(pvc.ObjectMeta.Namespace).Get(ctx, pvc.ObjectMeta.Name, metav1.GetOptions{})
 		if err != nil {
 			return false, err
 		}

pkg/strategies/export.go

@@ -97,7 +97,7 @@ func (c *ExportStrategy) CopyOut(pod v1.Pod, config *rest.Config, name string) (
 func (c *ExportStrategy) Cleanup() error {
 	c.log.Info("Cleaning up...")
 	if c.tempMover != nil {
-		c.tempMover.Cleanup()
+		return c.tempMover.Cleanup()
 	}
 	return nil
 }

pkg/strategies/import.go

@@ -92,7 +92,7 @@ func (c *ImportStrategy) CopyInto(pod v1.Pod, config *rest.Config, localPath str
 func (c *ImportStrategy) Cleanup() error {
 	c.log.Info("Cleaning up...")
 	if c.tempMover != nil {
-		c.tempMover.Cleanup()
+		return c.tempMover.Cleanup()
 	}
 	return nil
 }