Support multi-stage imports in import populator

pkg/apiserver/webhooks/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/util/validation/field:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
+        "//vendor/k8s.io/utils/pointer:go_default_library",
         "//vendor/kubevirt.io/controller-lifecycle-operator-sdk/api:go_default_library",
     ],
 )

pkg/apiserver/webhooks/populators-validate.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation"
 	k8sfield "k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
 
 	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
 )
@@ -132,21 +133,13 @@ func (wh *populatorValidatingWebhook) validateVolumeImportSource(ar admissionv1.
 		return nil, err
 	}
 
-	// Reject spec updates
 	if ar.Request.Operation == admissionv1.Update {
-		oldSource := cdiv1.VolumeImportSource{}
-		err = json.Unmarshal(ar.Request.OldObject.Raw, &oldSource)
+		cause, err := wh.validateVolumeImportSourceUpdate(ar, &volumeImportSource)
 		if err != nil {
 			return nil, err
 		}
-
-		if !apiequality.Semantic.DeepEqual(volumeImportSource.Spec, oldSource.Spec) {
-			klog.Errorf("Cannot update spec for VolumeImportSource %s/%s", volumeImportSource.GetNamespace(), volumeImportSource.GetName())
-			return []metav1.StatusCause{{
-				Type:    metav1.CauseTypeFieldValueDuplicate,
-				Message: "Cannot update VolumeImportSource Spec",
-				Field:   k8sfield.NewPath("VolumeImportSource").Child("Spec").String(),
-			}}, nil
+		if cause != nil {
+			return cause, nil
 		}
 	}
 
@@ -174,6 +167,15 @@ func (wh *populatorValidatingWebhook) validateVolumeImportSourceSpec(field *k8sf
 		return causes
 	}
 
+	// validate multi-stage import
+	if isMultiStageImport(spec) && (spec.TargetClaim == nil || *spec.TargetClaim == "") {
+		return []metav1.StatusCause{{
+			Type:    metav1.CauseTypeFieldValueInvalid,
+			Message: "Unable to do multi-stage import without specifying a target claim",
+			Field:   field.Child("targetClaim").String(),
+		}}
+	}
+
 	// Validate import sources
 	if http := spec.Source.HTTP; http != nil {
 		return validateHTTPSource(http, field)
@@ -199,3 +201,38 @@ func (wh *populatorValidatingWebhook) validateVolumeImportSourceSpec(field *k8sf
 	// Should never reach this return
 	return nil
 }
+
+func (wh *populatorValidatingWebhook) validateVolumeImportSourceUpdate(ar admissionv1.AdmissionReview, volumeImportSource *cdiv1.VolumeImportSource) ([]metav1.StatusCause, error) {
+	oldSource := cdiv1.VolumeImportSource{}
+	err := json.Unmarshal(ar.Request.OldObject.Raw, &oldSource)
+	if err != nil {
+		return nil, err
+	}
+	newSpec := volumeImportSource.Spec.DeepCopy()
+	oldSpec := oldSource.Spec.DeepCopy()
+
+	// Always admit checkpoint updates for multi-stage migrations.
+	if isMultiStageImport(newSpec) {
+		oldSpec.FinalCheckpoint = pointer.Bool(false)
+		oldSpec.Checkpoints = nil
+		newSpec.FinalCheckpoint = pointer.Bool(false)
+		newSpec.Checkpoints = nil
+	}
+
+	// Reject all other updates
+	if !apiequality.Semantic.DeepEqual(newSpec, oldSpec) {
+		klog.Errorf("Cannot update spec for VolumeImportSource %s/%s", volumeImportSource.GetNamespace(), volumeImportSource.GetName())
+		return []metav1.StatusCause{{
+			Type:    metav1.CauseTypeFieldValueDuplicate,
+			Message: "Cannot update VolumeImportSource Spec",
+			Field:   k8sfield.NewPath("VolumeImportSource").Child("Spec").String(),
+		}}, nil
+	}
+
+	return nil, nil
+}
+
+func isMultiStageImport(spec *cdiv1.VolumeImportSourceSpec) bool {
+	return spec.Source != nil && len(spec.Checkpoints) > 0 &&
+		(spec.Source.VDDK != nil || spec.Source.Imageio != nil)
+}

pkg/apiserver/webhooks/populators-validate_test.go

@@ -112,6 +112,43 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(BeFalse())
 		})
 
+		It("should accept VolumeImportSource spec checkpoints update", func() {
+			source := &cdiv1.ImportSourceType{
+				HTTP: &cdiv1.DataVolumeSourceHTTP{
+					URL: "http://www.example.com",
+				},
+			}
+			importCR := newVolumeImportSource(cdiv1.DataVolumeKubeVirt, source)
+			importCR.Spec.Checkpoints = []cdiv1.DataVolumeCheckpoint{
+				{Current: "test", Previous: ""},
+			}
+			newBytes, _ := json.Marshal(&importCR)
+
+			oldSource := importCR.DeepCopy()
+			oldSource.Spec.Source.HTTP.URL = "http://www.example.es"
+			oldSource.Spec.Checkpoints = nil
+			oldBytes, _ := json.Marshal(oldSource)
+
+			ar := &admissionv1.AdmissionReview{
+				Request: &admissionv1.AdmissionRequest{
+					Operation: admissionv1.Update,
+					Resource: metav1.GroupVersionResource{
+						Group:    cdiv1.SchemeGroupVersion.Group,
+						Version:  cdiv1.SchemeGroupVersion.Version,
+						Resource: "volumeimportsources",
+					},
+					Object: runtime.RawExtension{
+						Raw: newBytes,
+					},
+					OldObject: runtime.RawExtension{
+						Raw: oldBytes,
+					},
+				},
+			}
+			resp := validatePopulatorsAdmissionReview(ar)
+			Expect(resp.Allowed).To(BeFalse())
+		})
+
 		It("should accept VolumeImportSource with HTTP source on create", func() {
 			source := &cdiv1.ImportSourceType{
 				HTTP: &cdiv1.DataVolumeSourceHTTP{
@@ -145,6 +182,59 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(BeFalse())
 		})
 
+		It("should reject VolumeImportSource with incomplete VDDK source", func() {
+			source := &cdiv1.ImportSourceType{
+				VDDK: &cdiv1.DataVolumeSourceVDDK{
+					BackingFile: "",
+					URL:         "",
+					UUID:        "",
+					Thumbprint:  "",
+					SecretRef:   "",
+				},
+			}
+			importCR := newVolumeImportSource(cdiv1.DataVolumeKubeVirt, source)
+			resp := validateVolumeImportSourceCreate(importCR)
+			Expect(resp.Allowed).To(BeFalse())
+		})
+
+		It("should reject multi-stage VolumeImportSource without TargetClaim", func() {
+			source := &cdiv1.ImportSourceType{
+				VDDK: &cdiv1.DataVolumeSourceVDDK{
+					BackingFile: "[iSCSI_Datastore] vm/vm_1.vmdk",
+					URL:         "https://vcenter.corp.com",
+					UUID:        "52260566-b032-36cb-55b1-79bf29e30490",
+					Thumbprint:  "20:6C:8A:5D:44:40:B3:79:4B:28:EA:76:13:60:90:6E:49:D9:D9:A3",
+					SecretRef:   "vddk-credentials",
+				},
+			}
+			importCR := newVolumeImportSource(cdiv1.DataVolumeKubeVirt, source)
+			importCR.Spec.Checkpoints = []cdiv1.DataVolumeCheckpoint{
+				{Current: "test", Previous: ""},
+			}
+			resp := validateVolumeImportSourceCreate(importCR)
+			Expect(resp.Allowed).To(BeFalse())
+		})
+
+		It("should accept multi-stage VolumeImportSource with TargetClaim", func() {
+			source := &cdiv1.ImportSourceType{
+				VDDK: &cdiv1.DataVolumeSourceVDDK{
+					BackingFile: "[iSCSI_Datastore] vm/vm_1.vmdk",
+					URL:         "https://vcenter.corp.com",
+					UUID:        "52260566-b032-36cb-55b1-79bf29e30490",
+					Thumbprint:  "20:6C:8A:5D:44:40:B3:79:4B:28:EA:76:13:60:90:6E:49:D9:D9:A3",
+					SecretRef:   "vddk-credentials",
+				},
+			}
+			importCR := newVolumeImportSource(cdiv1.DataVolumeKubeVirt, source)
+			importCR.Spec.Checkpoints = []cdiv1.DataVolumeCheckpoint{
+				{Current: "test", Previous: ""},
+			}
+			targetClaim := "test-pvc"
+			importCR.Spec.TargetClaim = &targetClaim
+			resp := validateVolumeImportSourceCreate(importCR)
+			Expect(resp.Allowed).To(BeTrue())
+		})
+
 		It("should accept VolumeImportSource with Registry source URL on create", func() {
 			url := "docker://registry:5000/test"
 			source := &cdiv1.ImportSourceType{

pkg/controller/common/BUILD.bazel

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "checkpoint-util.go",
         "runtime-util.go",
         "util.go",
     ],
@@ -27,6 +28,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",