fix(api): block device WaitForUserUpload phase (#2178)

- Added a GET method to the upload URL to signal that the upload server is ready to serve.
- Added a readiness probe to the uploader pod.
- Fixed missing output from the data export in the E2E test case.
- Fixed the WaitForUserUpload phase for block devices.

Signed-off-by: Roman Sysoev <roman.sysoev@flant.com>

Author: hardcoretime

images/dvcr-artifact/pkg/uploader/uploader.go

@@ -280,6 +280,12 @@ func (app *uploadServerApp) healthzHandler(w http.ResponseWriter, _ *http.Reques
 }
 
 func (app *uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r *http.Request) bool {
+	// This method is used to signal that ingress is configured and the server can upload user data.
+	if r.Method == http.MethodGet {
+		w.WriteHeader(http.StatusOK)
+		return false
+	}
+
 	if r.Method != http.MethodPost && r.Method != http.MethodPut {
 		w.WriteHeader(http.StatusNotFound)
 		return false

images/virtualization-artifact/pkg/common/pod/pod.go

@@ -103,6 +103,17 @@ func IsPodStarted(pod *corev1.Pod) bool {
 	return true
 }
 
+// IsPodReady returns true if the pod's `Ready` condition status is true; otherwise, it returns false.
+func IsPodReady(pod *corev1.Pod) bool {
+	for _, c := range pod.Status.Conditions {
+		if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+
+	return false
+}
+
 // IsPodComplete returns true if a Pod is in 'Succeeded' phase, false if not.
 func IsPodComplete(pod *corev1.Pod) bool {
 	return pod != nil && pod.Status.Phase == corev1.PodSucceeded

images/virtualization-artifact/pkg/controller/cvi/internal/source/interfaces.go

@@ -66,7 +66,7 @@ type Stat interface {
 	GetDVCRImageName(pod *corev1.Pod) string
 	GetDownloadSpeed(ownerUID types.UID, pod *corev1.Pod) *v1alpha2.StatusSpeed
 	GetProgress(ownerUID types.UID, pod *corev1.Pod, prevProgress string, opts ...service.GetProgressOption) string
-	IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) bool
+	IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) (bool, error)
 	IsUploadStarted(ownerUID types.UID, pod *corev1.Pod) bool
 	CheckPod(pod *corev1.Pod) error
 }

images/virtualization-artifact/pkg/controller/cvi/internal/source/mock.go

@@ -92,6 +92,11 @@ func (ds UploadDataSource) Sync(ctx context.Context, cvi *v1alpha2.ClusterVirtua
 		return reconcile.Result{}, err
 	}
 
+	isUploaderReady, err := ds.statService.IsUploaderReady(pod, svc, ing)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
 	switch {
 	case isDiskProvisioningFinished(condition):
 		log.Info("Cluster virtual image provisioning finished: clean up")
@@ -221,7 +226,7 @@ func (ds UploadDataSource) Sync(ctx context.Context, cvi *v1alpha2.ClusterVirtua
 		}
 
 		log.Info("Provisioning...", "progress", cvi.Status.Progress, "pod.phase", pod.Status.Phase)
-	case ds.statService.IsUploaderReady(pod, svc, ing):
+	case isUploaderReady:
 		cb.
 			Status(metav1.ConditionFalse).
 			Reason(cvicondition.WaitForUserUpload).

images/virtualization-artifact/pkg/controller/cvi/internal/source/upload.go

@@ -272,14 +272,32 @@ func (s StatService) IsImportStarted(ownerUID types.UID, pod *corev1.Pod) bool {
 	return progress.ProgressRaw() > 0
 }
 
-func (s StatService) IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) bool {
+func (s StatService) IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) (bool, error) {
 	if pod == nil || svc == nil || ing == nil {
-		return false
+		return false, nil
+	}
+
+	if !podutil.IsPodReady(pod) {
+		return false, nil
 	}
 
-	ingressIsOK := ing.Annotations[annotations.AnnUploadPath] != "" || ing.Annotations[annotations.AnnUploadURLDeprecated] != ""
+	uploadURL, ok := ing.Annotations[annotations.AnnUploadURL]
+	if ok && uploadURL != "" {
+		client := &http.Client{Timeout: 5 * time.Second}
+		response, err := client.Get(uploadURL)
+		if err != nil {
+			return false, fmt.Errorf("failed to get upload server status: %w", err)
+		}
+		defer response.Body.Close()
+
+		if response.StatusCode == http.StatusOK {
+			return true, nil
+		}
+
+		return false, nil
+	}
 
-	return podutil.IsPodRunning(pod) && podutil.IsPodStarted(pod) && ingressIsOK
+	return ing.Annotations[annotations.AnnUploadPath] != "" || ing.Annotations[annotations.AnnUploadURLDeprecated] != "", nil
 }
 
 func (s StatService) IsUploadStarted(ownerUID types.UID, pod *corev1.Pod) bool {

images/virtualization-artifact/pkg/controller/service/stat_service.go

@@ -22,6 +22,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -36,6 +37,12 @@ const (
 	destinationAuthVol = "dvcr-secret-vol"
 )
 
+// These constants can't be imported from "images/dvcr-artifact/pkg/uploader/uploader.go" due to conflicts with the CDI version.
+const (
+	healthzPort = 8080
+	healthzPath = "/healthz"
+)
+
 type Pod struct {
 	PodSettings *PodSettings
 	Settings    *Settings
@@ -151,6 +158,15 @@ func (p *Pod) makeUploaderContainerSpec() *corev1.Container {
 		SecurityContext: &corev1.SecurityContext{
 			ReadOnlyRootFilesystem: ptr.To(true),
 		},
+		ReadinessProbe: &corev1.Probe{
+			ProbeHandler: corev1.ProbeHandler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Path: healthzPath,
+					Port: intstr.FromInt(healthzPort),
+				},
+			},
+			InitialDelaySeconds: 5,
+		},
 	}
 
 	if p.PodSettings.ResourceRequirements != nil {

images/virtualization-artifact/pkg/controller/uploader/uploader_pod.go

@@ -122,6 +122,11 @@ func (ds UploadDataSource) Sync(ctx context.Context, vd *v1alpha2.VirtualDisk) (
 		vdsupplements.SetPVCName(vd, dv.Status.ClaimName)
 	}
 
+	isUploaderReady, err := ds.statService.IsUploaderReady(pod, svc, ing)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
 	switch {
 	case IsDiskProvisioningFinished(condition):
 		log.Debug("Disk provisioning finished: clean up")
@@ -190,7 +195,7 @@ func (ds UploadDataSource) Sync(ctx context.Context, vd *v1alpha2.VirtualDisk) (
 		}
 
 		if !ds.statService.IsUploadStarted(vd.GetUID(), pod) {
-			if ds.statService.IsUploaderReady(pod, svc, ing) {
+			if isUploaderReady {
 				log.Info("Waiting for the user upload", "pod.phase", pod.Status.Phase)
 
 				vd.Status.Phase = v1alpha2.DiskWaitForUserUpload

images/virtualization-artifact/pkg/controller/vd/internal/source/upload.go

@@ -64,7 +64,7 @@ type Stat interface {
 	step.WaitForPodStepStat
 	step.ReadyContainerRegistryStepStat
 	IsUploadStarted(ownerUID types.UID, pod *corev1.Pod) bool
-	IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) bool
+	IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *netv1.Ingress) (bool, error)
 	GetDownloadSpeed(ownerUID types.UID, pod *corev1.Pod) *v1alpha2.StatusSpeed
 }