Clone source program calls tar instead of getting piped input. This e…

…nsures we trap tar errors. (#1530)

Fix ready/complet check in clone controller

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

cmd/cdi-cloner/clone-source.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"os/exec"
 
 	"github.com/golang/snappy"
 	"github.com/prometheus/client_golang/prometheus"
@@ -21,12 +22,35 @@ import (
 
 var (
 	contentType string
+	mountPoint  string
 	uploadBytes uint64
 )
 
+type execReader struct {
+	cmd    *exec.Cmd
+	stdout io.ReadCloser
+	stderr io.ReadCloser
+}
+
+func (er *execReader) Read(p []byte) (n int, err error) {
+	n, err = er.stdout.Read(p)
+	if err == io.EOF {
+		if err2 := er.cmd.Wait(); err2 != nil {
+			errBytes, _ := ioutil.ReadAll(er.stderr)
+			klog.Fatalf("Subprocess did not execute successfully, result is: %q\n%s", er.cmd.ProcessState.ExitCode(), string(errBytes))
+		}
+	}
+	return
+}
+
+func (er *execReader) Close() error {
+	return er.stdout.Close()
+}
+
 func init() {
-	flag.StringVar(&contentType, "content_type", "", "archive|kubevirt|filesystem-clone|blockdevice-clone")
-	flag.Uint64Var(&uploadBytes, "upload_bytes", 0, "approx number of bytes in input")
+	flag.StringVar(&contentType, "content-type", "", "filesystem-clone|blockdevice-clone")
+	flag.StringVar(&mountPoint, "mount", "", "pvc mount point")
+	flag.Uint64Var(&uploadBytes, "upload-bytes", 0, "approx number of bytes in input")
 	klog.InitFlags(nil)
 }
 
@@ -105,12 +129,68 @@ func pipeToSnappy(reader io.ReadCloser) io.ReadCloser {
 	return pr
 }
 
+func validateContentType() {
+	switch contentType {
+	case "filesystem-clone", "blockdevice-clone":
+	default:
+		klog.Fatalf("Invalid content-type %q", contentType)
+	}
+}
+
+func validateMount() {
+	if mountPoint == "" {
+		klog.Fatalf("Invalid mount %q", mountPoint)
+	}
+}
+
+func newTarReader() (io.ReadCloser, error) {
+	cmd := exec.Command("/usr/bin/tar", "Scv", ".")
+	cmd.Dir = mountPoint
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	var stderr bytes.Buffer
+	cmd.Stderr = &stderr
+
+	if err = cmd.Start(); err != nil {
+		return nil, err
+	}
+
+	return &execReader{cmd: cmd, stdout: stdout, stderr: ioutil.NopCloser(&stderr)}, nil
+}
+
+func getInputStream() (rc io.ReadCloser) {
+	var err error
+	switch contentType {
+	case "filesystem-clone":
+		rc, err = newTarReader()
+		if err != nil {
+			klog.Fatalf("Error creating tar reader for %q: %+v", mountPoint, err)
+		}
+	case "blockdevice-clone":
+		rc, err = os.Open(mountPoint)
+		if err != nil {
+			klog.Fatalf("Error opening block device %q: %+v", mountPoint, err)
+		}
+	default:
+		klog.Fatalf("Invalid content-type %q", contentType)
+	}
+	return
+}
+
 func main() {
 	flag.Parse()
 	defer klog.Flush()
 
-	klog.Infof("content_type is %q\n", contentType)
-	klog.Infof("upload_bytes is %d", uploadBytes)
+	klog.Infof("content-type is %q\n", contentType)
+	klog.Infof("mount is %q\n", mountPoint)
+	klog.Infof("upload-bytes is %d", uploadBytes)
+
+	validateContentType()
+	validateMount()
 
 	ownerUID := getEnvVarOrDie(common.OwnerUID)
 
@@ -122,7 +202,7 @@ func main() {
 
 	klog.V(1).Infoln("Starting cloner target")
 
-	reader := pipeToSnappy(createProgressReader(os.Stdin, ownerUID, uploadBytes))
+	reader := pipeToSnappy(createProgressReader(getInputStream(), ownerUID, uploadBytes))
 
 	startPrometheus()
 

cmd/cdi-cloner/cloner_startup.sh

@@ -33,13 +33,13 @@ if [ "$VOLUME_MODE" == "block" ]; then
     UPLOAD_BYTES=$(blockdev --getsize64 $MOUNT_POINT)
     echo "UPLOAD_BYTES=$UPLOAD_BYTES"
 
-    /usr/bin/cdi-cloner -v=3 -alsologtostderr -content_type blockdevice-clone -upload_bytes $UPLOAD_BYTES <$MOUNT_POINT
+    /usr/bin/cdi-cloner -v=3 -alsologtostderr -content-type blockdevice-clone -upload-bytes $UPLOAD_BYTES -mount $MOUNT_POINT
 else
     pushd $MOUNT_POINT
     UPLOAD_BYTES=$(du -sb . | cut -f1)
     echo "UPLOAD_BYTES=$UPLOAD_BYTES"
 
-    tar Scv . | /usr/bin/cdi-cloner -v=3 -alsologtostderr -content_type filesystem-clone -upload_bytes $UPLOAD_BYTES
+    /usr/bin/cdi-cloner -v=3 -alsologtostderr -content-type filesystem-clone -upload-bytes $UPLOAD_BYTES -mount $MOUNT_POINT
 
     popd
 fi

pkg/controller/clone-controller.go

@@ -131,6 +131,28 @@ func addCloneControllerWatches(mgr manager.Manager, cloneController controller.C
 	}); err != nil {
 		return err
 	}
+	if err := cloneController.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestsFromMapFunc{
+		ToRequests: handler.ToRequestsFunc(func(obj handler.MapObject) []reconcile.Request {
+			target, ok := obj.Meta.GetAnnotations()[AnnOwnerRef]
+			if !ok {
+				return nil
+			}
+			namespace, name, err := cache.SplitMetaNamespaceKey(target)
+			if err != nil {
+				return nil
+			}
+			return []reconcile.Request{
+				{
+					NamespacedName: types.NamespacedName{
+						Namespace: namespace,
+						Name:      name,
+					},
+				},
+			}
+		}),
+	}); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -306,12 +328,7 @@ func (r *CloneReconciler) waitTargetPodRunningOrSucceeded(pvc *corev1.Persistent
 		return false, errors.Wrapf(err, "error parsing %s annotation", AnnPodReady)
 	}
 
-	if !ready {
-		log.V(3).Info("clone target pod not ready")
-		return podSucceededFromPVC(pvc), nil
-	}
-
-	return true, nil
+	return ready || podSucceededFromPVC(pvc), nil
 }
 
 func (r *CloneReconciler) findCloneSourcePod(pvc *corev1.PersistentVolumeClaim) (*corev1.Pod, error) {