Merge pull request #485 from zvikorn/upload-dv

Orchestrate upload using a DataVolume

Author: awels

manifests/example/upload-datavolume.yaml

@@ -0,0 +1,16 @@
+apiVersion: cdi.kubevirt.io/v1alpha1
+kind: DataVolume
+metadata:
+  name: upload-datavolume
+spec:
+  source:
+      upload:
+        target: {}
+  pvc:
+    #the storageClassName is optional, if missing we use the default storage class.
+    #storageClassName: "hostpath"
+    accessModes:
+      - ReadWriteOnce
+    resources:
+      requests:
+        storage: 500Mi

manifests/example/upload-dv-token.yaml

@@ -0,0 +1,7 @@
+apiVersion: upload.cdi.kubevirt.io/v1alpha1
+kind: UploadTokenRequest
+metadata:
+  name: upload-datavolume
+  namespace: default
+spec:
+  pvcName: upload-datavolume

pkg/apis/datavolumecontroller/v1alpha1/types.go

@@ -42,9 +42,10 @@ type DataVolumeSpec struct {
 
 // DataVolumeSource represents the source for our Data Volume, this can be HTTP, S3 or an existing PVC
 type DataVolumeSource struct {
-	HTTP *DataVolumeSourceHTTP `json:"http,omitempty"`
-	S3   *DataVolumeSourceS3   `json:"s3,omitempty"`
-	PVC  *DataVolumeSourcePVC  `json:"pvc,omitempty"`
+	HTTP   *DataVolumeSourceHTTP   `json:"http,omitempty"`
+	S3     *DataVolumeSourceS3     `json:"s3,omitempty"`
+	PVC    *DataVolumeSourcePVC    `json:"pvc,omitempty"`
+	UPLOAD *DataVolumeSourceUpload `json:"upload,omitempty"`
 }
 
 // DataVolumeSourcePVC provides the parameters to create a Data Volume from an existing PVC
@@ -53,6 +54,11 @@ type DataVolumeSourcePVC struct {
 	Name      string `json:"name,omitempty"`
 }
 
+// DataVolumeSourceUpload provides the parameters to create a Data Volume by uploading the source
+type DataVolumeSourceUpload struct {
+	//Target string `json:"shouldUpload,omitempty"`
+}
+
 // DataVolumeSourceS3 provides the parameters to create a Data Volume from an S3 source
 type DataVolumeSourceS3 struct {
 	//URL is the url of the S3 source
@@ -109,6 +115,12 @@ const (
 	// CloneInProgress represents a data volume with a current phase of CloneInProgress
 	CloneInProgress DataVolumePhase = "CloneInProgress"
 
+	// UploadScheduled represents a data volume with a current phase of UploadScheduled
+	UploadScheduled DataVolumePhase = "UploadScheduled"
+
+	// UploadReady represents a data volume with a current phase of UploadReady
+	UploadReady DataVolumePhase = "UploadReady"
+
 	// Succeeded represents a DataVolumePhase of Succeeded
 	Succeeded DataVolumePhase = "Succeeded"
 	// Failed represents a DataVolumePhase of Failed

pkg/apis/datavolumecontroller/v1alpha1/zz_generated.deepcopy.go

@@ -334,6 +334,24 @@ func (c *DataVolumeController) updateCloneStatusPhase(pvc *corev1.PersistentVolu
 	}
 }
 
+func (c *DataVolumeController) updateUploadStatusPhase(pvc *corev1.PersistentVolumeClaim, dataVolumeCopy *cdiv1.DataVolume) {
+	phase, ok := pvc.Annotations[AnnPodPhase]
+	if ok {
+		switch phase {
+		case string(corev1.PodPending):
+			// TODO: Use a more generic Scheduled, like maybe TransferScheduled.
+			dataVolumeCopy.Status.Phase = cdiv1.UploadScheduled
+		case string(corev1.PodRunning):
+			// TODO: Use a more generic In Progess, like maybe TransferInProgress.
+			dataVolumeCopy.Status.Phase = cdiv1.UploadReady
+		case string(corev1.PodFailed):
+			dataVolumeCopy.Status.Phase = cdiv1.Failed
+		case string(corev1.PodSucceeded):
+			dataVolumeCopy.Status.Phase = cdiv1.Succeeded
+		}
+	}
+}
+
 func (c *DataVolumeController) updateDataVolumeStatus(dataVolume *cdiv1.DataVolume, pvc *corev1.PersistentVolumeClaim) error {
 	dataVolumeCopy := dataVolume.DeepCopy()
 	var err error
@@ -370,6 +388,11 @@ func (c *DataVolumeController) updateDataVolumeStatus(dataVolume *cdiv1.DataVolu
 				dataVolumeCopy.Status.Phase = cdiv1.CloneScheduled
 				c.updateCloneStatusPhase(pvc, dataVolumeCopy)
 			}
+			_, ok = pvc.Annotations[AnnUploadRequest]
+			if ok {
+				dataVolumeCopy.Status.Phase = cdiv1.UploadScheduled
+				c.updateUploadStatusPhase(pvc, dataVolumeCopy)
+			}
 
 		case corev1.ClaimLost:
 			dataVolumeCopy.Status.Phase = cdiv1.Failed
@@ -493,7 +516,8 @@ func newPersistentVolumeClaim(dataVolume *cdiv1.DataVolume) (*corev1.PersistentV
 		} else {
 			annotations[AnnCloneRequest] = dataVolume.Namespace + "/" + dataVolume.Spec.Source.PVC.Name
 		}
-
+	} else if dataVolume.Spec.Source.UPLOAD != nil {
+		annotations[AnnUploadRequest] = ""
 	} else {
 		return nil, errors.Errorf("no source set for datavolume")
 	}

pkg/controller/datavolume-controller.go

@@ -106,6 +106,22 @@ func newCloneDataVolume(name string) *cdiv1.DataVolume {
 	}
 }
 
+func newUploadDataVolume(name string) *cdiv1.DataVolume {
+	return &cdiv1.DataVolume{
+		TypeMeta: metav1.TypeMeta{APIVersion: cdiv1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: cdiv1.DataVolumeSpec{
+			Source: cdiv1.DataVolumeSource{
+				UPLOAD: &cdiv1.DataVolumeSourceUpload{},
+			},
+			PVC: &corev1.PersistentVolumeClaimSpec{},
+		},
+	}
+}
+
 func (f *fixture) newController() (*DataVolumeController, informers.SharedInformerFactory, kubeinformers.SharedInformerFactory) {
 	f.client = fake.NewSimpleClientset(f.objects...)
 	f.kubeclient = k8sfake.NewSimpleClientset(f.kubeobjects...)
@@ -543,3 +559,106 @@ func TestCloneClaimLost(t *testing.T) {
 	f.expectUpdateDataVolumeStatusAction(result)
 	f.run(getKey(dataVolume, t))
 }
+
+// Upload tests
+func TestUploadScheduled(t *testing.T) {
+	f := newFixture(t)
+	dataVolume := newUploadDataVolume("upload-datavolume")
+	pvc, _ := newPersistentVolumeClaim(dataVolume)
+
+	dataVolume.Status.Phase = cdiv1.Pending
+	pvc.Status.Phase = corev1.ClaimBound
+	pvc.Annotations[AnnUploadRequest] = ""
+
+	f.dataVolumeLister = append(f.dataVolumeLister, dataVolume)
+	f.objects = append(f.objects, dataVolume)
+	f.pvcLister = append(f.pvcLister, pvc)
+	f.kubeobjects = append(f.kubeobjects, pvc)
+
+	result := dataVolume.DeepCopy()
+	result.Status.Phase = cdiv1.UploadScheduled
+	f.expectUpdateDataVolumeStatusAction(result)
+	f.run(getKey(dataVolume, t))
+}
+
+func TestUploadReady(t *testing.T) {
+	f := newFixture(t)
+	dataVolume := newUploadDataVolume("upload-datavolume")
+	pvc, _ := newPersistentVolumeClaim(dataVolume)
+
+	dataVolume.Status.Phase = cdiv1.Pending
+	pvc.Status.Phase = corev1.ClaimBound
+	pvc.Annotations[AnnUploadRequest] = ""
+	pvc.Annotations[AnnPodPhase] = "Running"
+
+	f.dataVolumeLister = append(f.dataVolumeLister, dataVolume)
+	f.objects = append(f.objects, dataVolume)
+	f.pvcLister = append(f.pvcLister, pvc)
+	f.kubeobjects = append(f.kubeobjects, pvc)
+
+	result := dataVolume.DeepCopy()
+	result.Status.Phase = cdiv1.UploadReady
+	f.expectUpdateDataVolumeStatusAction(result)
+	f.run(getKey(dataVolume, t))
+}
+
+func TestUploadSucceeded(t *testing.T) {
+	f := newFixture(t)
+	dataVolume := newUploadDataVolume("upload-datavolume")
+	pvc, _ := newPersistentVolumeClaim(dataVolume)
+
+	dataVolume.Status.Phase = cdiv1.Pending
+	pvc.Status.Phase = corev1.ClaimBound
+	pvc.Annotations[AnnUploadRequest] = ""
+	pvc.Annotations[AnnPodPhase] = "Succeeded"
+
+	f.dataVolumeLister = append(f.dataVolumeLister, dataVolume)
+	f.objects = append(f.objects, dataVolume)
+	f.pvcLister = append(f.pvcLister, pvc)
+	f.kubeobjects = append(f.kubeobjects, pvc)
+
+	result := dataVolume.DeepCopy()
+	result.Status.Phase = cdiv1.Succeeded
+	f.expectUpdateDataVolumeStatusAction(result)
+	f.run(getKey(dataVolume, t))
+}
+
+func TestUploadPodFailed(t *testing.T) {
+	f := newFixture(t)
+	dataVolume := newCloneDataVolume("upload-datavolume")
+	pvc, _ := newPersistentVolumeClaim(dataVolume)
+
+	dataVolume.Status.Phase = cdiv1.Pending
+	pvc.Status.Phase = corev1.ClaimBound
+	pvc.Annotations[AnnUploadRequest] = ""
+	pvc.Annotations[AnnPodPhase] = "Failed"
+
+	f.dataVolumeLister = append(f.dataVolumeLister, dataVolume)
+	f.objects = append(f.objects, dataVolume)
+	f.pvcLister = append(f.pvcLister, pvc)
+	f.kubeobjects = append(f.kubeobjects, pvc)
+
+	result := dataVolume.DeepCopy()
+	result.Status.Phase = cdiv1.Failed
+	f.expectUpdateDataVolumeStatusAction(result)
+	f.run(getKey(dataVolume, t))
+}
+
+func TestUploadClaimLost(t *testing.T) {
+	f := newFixture(t)
+	dataVolume := newUploadDataVolume("upload-datavolume")
+	pvc, _ := newPersistentVolumeClaim(dataVolume)
+
+	dataVolume.Status.Phase = cdiv1.Pending
+	pvc.Status.Phase = corev1.ClaimLost
+
+	f.dataVolumeLister = append(f.dataVolumeLister, dataVolume)
+	f.objects = append(f.objects, dataVolume)
+	f.pvcLister = append(f.pvcLister, pvc)
+	f.kubeobjects = append(f.kubeobjects, pvc)
+
+	result := dataVolume.DeepCopy()
+	result.Status.Phase = cdiv1.Failed
+	f.expectUpdateDataVolumeStatusAction(result)
+	f.run(getKey(dataVolume, t))
+}