virtctl imageupload should create datavolumes instead of PVCs

Signed-off-by: Michael Henriksen <[email protected]>

.gitignore

@@ -37,3 +37,4 @@ manifests/**/*.tmp
 /bazel-*
 /.bazelrc
 _ci-configs/
+.history

BUILD.bazel

@@ -33,6 +33,10 @@ load("@com_github_atlassian_bazel_tools//goimports:def.bzl", "goimports")
 goimports(
     name = "goimports",
     display_diffs = True,
+    exclude_paths = [
+        "./vendor/*",
+        "./.history/*",
+    ],
     local = ["kubevirt.io"],
     prefix = "kubevirt.io/kubevirt",
     write = True,

go.mod

@@ -74,7 +74,7 @@ require (
 	k8s.io/kube-aggregator v0.0.0-20190228175259-3e0149950b0e
 	k8s.io/utils v0.0.0-20190607212802-c55fbcfc754a
 	kubevirt.io/client-go v0.0.0-00010101000000-000000000000
-	kubevirt.io/containerized-data-importer v1.10.6
+	kubevirt.io/containerized-data-importer v1.10.9
 	kubevirt.io/qe-tools v0.1.3-0.20190512140058-934db0579e0c
 	sigs.k8s.io/controller-runtime v0.1.9 // indirect
 )

pkg/virtctl/imageupload/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/tools/clientcmd:go_default_library",
+        "//vendor/kubevirt.io/containerized-data-importer/pkg/apis/core/v1alpha1:go_default_library",
         "//vendor/kubevirt.io/containerized-data-importer/pkg/apis/upload/v1alpha1:go_default_library",
         "//vendor/kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned:go_default_library",
     ],

pkg/virtctl/imageupload/imageupload.go

@@ -26,6 +26,7 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 	"time"
 
@@ -40,6 +41,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 
 	"kubevirt.io/client-go/kubecli"
+	cdiv1 "kubevirt.io/containerized-data-importer/pkg/apis/core/v1alpha1"
 	uploadcdiv1 "kubevirt.io/containerized-data-importer/pkg/apis/upload/v1alpha1"
 	cdiClientset "kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned"
 	"kubevirt.io/kubevirt/pkg/virtctl/templates"
@@ -49,9 +51,12 @@ const (
 	// PodPhaseAnnotation is the annotation on a PVC containing the upload pod phase
 	PodPhaseAnnotation = "cdi.kubevirt.io/storage.pod.phase"
 
+	// PodReadyAnnotation tells whether the uploadserver pod is ready
+	PodReadyAnnotation = "cdi.kubevirt.io/storage.pod.ready"
+
 	uploadRequestAnnotation = "cdi.kubevirt.io/storage.upload.target"
 
-	uploadPodWaitInterval = 2 * time.Second
+	uploadReadyWaitInterval = 2 * time.Second
 
 	//UploadProxyURI is a URI of the upoad proxy
 	UploadProxyURI = "/v1alpha1/upload"
@@ -107,15 +112,15 @@ func NewImageUploadCommand(clientConfig clientcmd.ClientConfig) *cobra.Command {
 	}
 	cmd.Flags().BoolVar(&insecure, "insecure", insecure, "Allow insecure server connections when using HTTPS.")
 	cmd.Flags().StringVar(&uploadProxyURL, "uploadproxy-url", "", "The URL of the cdi-upload proxy service.")
-	cmd.Flags().StringVar(&pvcName, "pvc-name", "", "The destination PVC.")
+	cmd.Flags().StringVar(&pvcName, "pvc-name", "", "The destination DataVolume/PVC name.")
 	cmd.MarkFlagRequired("pvc-name")
 	cmd.Flags().StringVar(&pvcSize, "pvc-size", "", "The size of the PVC to create (ex. 10Gi, 500Mi).")
 	cmd.Flags().StringVar(&storageClass, "storage-class", "", "The storage class for the PVC.")
 	cmd.Flags().StringVar(&accessMode, "access-mode", accessMode, "The access mode for the PVC.")
 	cmd.Flags().BoolVar(&blockVolume, "block-volume", blockVolume, "Create a PVC with VolumeMode=Block (default Filesystem).")
 	cmd.Flags().StringVar(&imagePath, "image-path", "", "Path to the local VM image.")
 	cmd.MarkFlagRequired("image-path")
-	cmd.Flags().BoolVar(&noCreate, "no-create", noCreate, "Don't attempt to create a new PVC.")
+	cmd.Flags().BoolVar(&noCreate, "no-create", noCreate, "Don't attempt to create a new DataVolume/PVC.")
 	cmd.Flags().UintVar(&uploadPodWaitSecs, "wait-secs", uploadPodWaitSecs, "Seconds to wait for upload pod to start.")
 	cmd.SetUsageTemplate(templates.UsageTemplate())
 	return cmd
@@ -153,16 +158,17 @@ func (c *command) run(cmd *cobra.Command, args []string) error {
 		if !(k8serrors.IsNotFound(err) && !noCreate) {
 			return err
 		}
+
 		if !noCreate && len(pvcSize) == 0 {
 			return fmt.Errorf("When creating PVC, the size must be specified")
 		}
 
-		pvc, err = createUploadPVC(virtClient, namespace, pvcName, pvcSize, storageClass, accessMode, blockVolume)
+		dv, err := createUploadDataVolume(virtClient, namespace, pvcName, pvcSize, storageClass, accessMode, blockVolume)
 		if err != nil {
 			return err
 		}
 
-		fmt.Printf("PVC %s/%s created\n", namespace, pvc.Name)
+		fmt.Printf("DataVolume %s/%s created\n", dv.Namespace, dv.Name)
 	} else {
 		pvc, err = ensurePVCSupportsUpload(virtClient, pvc)
 		if err != nil {
@@ -172,15 +178,11 @@ func (c *command) run(cmd *cobra.Command, args []string) error {
 		fmt.Printf("Using existing PVC %s/%s\n", namespace, pvc.Name)
 	}
 
-	err = waitUploadPodRunning(virtClient, namespace, pvcName, uploadPodWaitInterval, time.Duration(uploadPodWaitSecs)*time.Second)
+	err = waitUploadServerReady(virtClient, namespace, pvcName, uploadReadyWaitInterval, time.Duration(uploadPodWaitSecs)*time.Second)
 	if err != nil {
 		return err
 	}
 
-	token, err := getUploadToken(virtClient.CdiClient(), namespace, pvcName)
-	if err != nil {
-		return err
-	}
 	if uploadProxyURL == "" {
 		uploadProxyURL, err = getUploadProxyURL(virtClient.CdiClient())
 		if err != nil {
@@ -190,14 +192,23 @@ func (c *command) run(cmd *cobra.Command, args []string) error {
 			return fmt.Errorf("Upload Proxy URL not found")
 		}
 	}
+
 	u, err := url.Parse(uploadProxyURL)
 	if err != nil {
 		return err
-	} else if u.Scheme == "" {
+	}
+
+	if u.Scheme == "" {
 		uploadProxyURL = fmt.Sprintf("https://%s", uploadProxyURL)
 	}
+
 	fmt.Printf("Uploading data to %s\n", uploadProxyURL)
 
+	token, err := getUploadToken(virtClient.CdiClient(), namespace, pvcName)
+	if err != nil {
+		return err
+	}
+
 	err = uploadData(uploadProxyURL, token, file, insecure)
 	if err != nil {
 		return err
@@ -236,7 +247,6 @@ func ConstructUploadProxyPath(uploadProxyURL string) (string, error) {
 }
 
 func uploadData(uploadProxyURL, token string, file *os.File, insecure bool) error {
-
 	url, err := ConstructUploadProxyPath(uploadProxyURL)
 	if err != nil {
 		return err
@@ -294,52 +304,31 @@ func getUploadToken(client cdiClientset.Interface, namespace, name string) (stri
 	return response.Status.Token, nil
 }
 
-func waitUploadPodRunning(client kubernetes.Interface, namespace, name string, interval, timeout time.Duration) error {
-	serviceName := "cdi-upload-" + name
+func waitUploadServerReady(client kubernetes.Interface, namespace, name string, interval, timeout time.Duration) error {
 	loggedStatus := false
 
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		pvc, err := client.CoreV1().PersistentVolumeClaims(namespace).Get(name, metav1.GetOptions{})
 		if err != nil {
-			return false, err
-		}
-
-		endpoints, err := client.CoreV1().Endpoints(namespace).Get(serviceName, metav1.GetOptions{})
-		if err != nil {
+			// DataVolume controller may not have created the PVC yet
 			if k8serrors.IsNotFound(err) {
 				return false, nil
 			}
-			return false, err
-		}
 
-		podPhase, _ := pvc.Annotations[PodPhaseAnnotation]
-
-		done := false
-		availableEndpoint := false
-		for _, subset := range endpoints.Subsets {
-			if len(subset.Addresses) > 0 {
-				// we're looking to make sure the service endpoint has
-				// the upload pod marked as being available, which means
-				// that it is ready to accept connections
-				availableEndpoint = true
-				break
-			}
+			return false, err
 		}
-		running := (podPhase == string(v1.PodRunning))
 
-		if running && availableEndpoint {
-			done = true
-		}
+		// upload controler sets this to true when uploadserver pod is ready to receive data
+		podReady, _ := pvc.Annotations[PodReadyAnnotation]
+		done, _ := strconv.ParseBool(podReady)
 
 		if !done && !loggedStatus {
-			fmt.Printf("Waiting for PVC %s upload pod to be running...\n", name)
+			fmt.Printf("Waiting for PVC %s upload pod to be ready...\n", name)
 			loggedStatus = true
 		}
 
 		if done && loggedStatus {
-			// be really sure
-			time.Sleep(interval)
-			fmt.Printf("Pod now running\n")
+			fmt.Printf("Pod now ready\n")
 		}
 
 		return done, nil
@@ -348,48 +337,50 @@ func waitUploadPodRunning(client kubernetes.Interface, namespace, name string, i
 	return err
 }
 
-func createUploadPVC(client kubernetes.Interface, namespace, name, size, storageClass, accessMode string, blockVolume bool) (*v1.PersistentVolumeClaim, error) {
+func createUploadDataVolume(client kubecli.KubevirtClient, namespace, name, size, storageClass, accessMode string, blockVolume bool) (*cdiv1.DataVolume, error) {
 	quantity, err := resource.ParseQuantity(size)
 	if err != nil {
 		return nil, fmt.Errorf("Validation failed for size=%s: %s", size, err)
 	}
 
-	pvc := &v1.PersistentVolumeClaim{
+	dv := &cdiv1.DataVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
-			Annotations: map[string]string{
-				uploadRequestAnnotation: "",
-			},
 		},
-		Spec: v1.PersistentVolumeClaimSpec{
-			Resources: v1.ResourceRequirements{
-				Requests: v1.ResourceList{
-					v1.ResourceStorage: quantity,
+		Spec: cdiv1.DataVolumeSpec{
+			Source: cdiv1.DataVolumeSource{
+				Upload: &cdiv1.DataVolumeSourceUpload{},
+			},
+			PVC: &v1.PersistentVolumeClaimSpec{
+				Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceStorage: quantity,
+					},
 				},
 			},
 		},
 	}
 
 	if storageClass != "" {
-		pvc.Spec.StorageClassName = &storageClass
+		dv.Spec.PVC.StorageClassName = &storageClass
 	}
 
 	if accessMode != "" {
-		pvc.Spec.AccessModes = []v1.PersistentVolumeAccessMode{v1.PersistentVolumeAccessMode(accessMode)}
+		dv.Spec.PVC.AccessModes = []v1.PersistentVolumeAccessMode{v1.PersistentVolumeAccessMode(accessMode)}
 	}
 
 	if blockVolume {
 		volMode := v1.PersistentVolumeBlock
-		pvc.Spec.VolumeMode = &volMode
+		dv.Spec.PVC.VolumeMode = &volMode
 	}
 
-	pvc, err = client.CoreV1().PersistentVolumeClaims(namespace).Create(pvc)
+	dv, err = client.CdiClient().CdiV1alpha1().DataVolumes(namespace).Create(dv)
 	if err != nil {
 		return nil, err
 	}
 
-	return pvc, nil
+	return dv, nil
 }
 
 func ensurePVCSupportsUpload(client kubernetes.Interface, pvc *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {