fix(vd): uploader

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

Author: Roman Sysoev

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

@@ -280,6 +280,11 @@ func (app *uploadServerApp) healthzHandler(w http.ResponseWriter, _ *http.Reques
 }
 
 func (app *uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r *http.Request) bool {
+	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/ingress/ingress.go

@@ -34,3 +34,8 @@ func MakeOwnerReference(ing *netv1.Ingress) metav1.OwnerReference {
 		Controller:         &isController,
 	}
 }
+
+// IsIngressReady returns true if the load-balancer has at least one ingress point in the status; otherwise, it returns false.
+func IsIngressReady(ing *netv1.Ingress) bool {
+	return len(ing.Status.LoadBalancer.Ingress) != 0
+}

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/service/stat_service.go

@@ -36,6 +36,7 @@ import (
 	"github.com/deckhouse/virtualization-controller/pkg/common/annotations"
 	"github.com/deckhouse/virtualization-controller/pkg/common/humanize_bytes"
 	"github.com/deckhouse/virtualization-controller/pkg/common/imageformat"
+	ingutil "github.com/deckhouse/virtualization-controller/pkg/common/ingress"
 	"github.com/deckhouse/virtualization-controller/pkg/common/percent"
 	podutil "github.com/deckhouse/virtualization-controller/pkg/common/pod"
 	"github.com/deckhouse/virtualization-controller/pkg/controller/conditions"
@@ -277,9 +278,9 @@ func (s StatService) IsUploaderReady(pod *corev1.Pod, svc *corev1.Service, ing *
 		return false
 	}
 
-	ingressIsOK := ing.Annotations[annotations.AnnUploadPath] != "" || ing.Annotations[annotations.AnnUploadURLDeprecated] != ""
+	ingressHasUploadPath := ing.Annotations[annotations.AnnUploadPath] != "" || ing.Annotations[annotations.AnnUploadURLDeprecated] != ""
 
-	return podutil.IsPodRunning(pod) && podutil.IsPodStarted(pod) && ingressIsOK
+	return podutil.IsPodReady(pod) && ingutil.IsIngressReady(ing) && ingressHasUploadPath
 }
 
 func (s StatService) IsUploadStarted(ownerUID types.UID, pod *corev1.Pod) bool {

images/virtualization-artifact/pkg/controller/uploader/uploader_pod.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 {

test/e2e/blockdevice/data_exports.go

@@ -285,20 +285,6 @@ func createUploadDisk(f *framework.Framework, name string) *v1alpha2.VirtualDisk
 	return vd
 }
 
-func retry(maxRetries int, fn func() error) error {
-	var lastErr error
-	for attempt := 1; attempt <= maxRetries; attempt++ {
-		if err := fn(); err != nil {
-			lastErr = err
-			GinkgoWriter.Printf("Attempt %d/%d failed: %v\n", attempt, maxRetries, err)
-			time.Sleep(time.Duration(attempt) * time.Second)
-			continue
-		}
-		return nil
-	}
-	return fmt.Errorf("failed after %d attempts: %w", maxRetries, lastErr)
-}
-
 func uploadFile(f *framework.Framework, vd *v1alpha2.VirtualDisk, filePath string) {
 	err := f.Clients.GenericClient().Get(context.Background(), crclient.ObjectKeyFromObject(vd), vd)
 	Expect(err).NotTo(HaveOccurred())
@@ -314,14 +300,17 @@ func uploadFile(f *framework.Framework, vd *v1alpha2.VirtualDisk, filePath strin
 	}
 	uploadURL := vd.Status.ImageUploadURLs.External
 
-	// During the upload of a VirtualDisk of type 'Upload', there is a bug:
-	//  when the VirtualDisk is in the 'DiskWaitForUserUpload' phase,
-	//  nginx may not be ready yet and can return 413 or 503 errors.
-	//  Once this bug is fixed, the retry mechanism must be removed.
-	const maxRetries = 5
-	err = retry(maxRetries, func() error {
-		return doUploadAttempt(httpClient, uploadURL, filePath)
-	})
+	// uploaderPod, err := getUploaderPod(vd.Name, vd.Namespace, f)
+	// Expect(err).NotTo(HaveOccurred())
+	// untilUploaderPodReady(uploaderPod.Name, uploaderPod.Namespace, f)
+
+	// uploaderIngress, err := getUploaderIngress(vd.Name, vd.Namespace, f)
+	// Expect(err).NotTo(HaveOccurred())
+	// untilUploaderIngressReady(uploaderIngress.Name, uploaderIngress.Namespace, f)
+
+	// time.Sleep(1 * time.Second)
+
+	err = doUploadAttempt(httpClient, uploadURL, filePath)
 	Expect(err).NotTo(HaveOccurred(), "Upload failed")
 }
 
@@ -385,3 +374,77 @@ func checkStorageVolumeDataManagerEnabled() (bool, error) {
 
 	return enabled != nil && *enabled, nil
 }
+
+// func getUploaderPod(vd *v1alpha2.VirtualDisk, f *framework.Framework) (*corev1.Pod, error) {
+// 	pods, err := f.Clients.KubeClient().CoreV1().Pods(vd.Namespace).List(context.Background(), metav1.ListOptions{})
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	for _, pod := range pods.Items {
+// 		for _, ref := range pod.OwnerReferences {
+// 			if ref.Kind == v1alpha2.VirtualDiskKind && ref.Name == vd.Name && strings.HasSuffix(pod.Name, string(vd.UID)) {
+// 				for _, c := range pod.Spec.Containers {
+// 					if c.Name == common.UploaderContainerName {
+// 						return &pod, nil
+// 					}
+// 				}
+// 			}
+// 		}
+// 	}
+
+// 	return nil, fmt.Errorf("could not find uploader pod for virtual disk %s", vd.Name)
+// }
+
+// func untilUploaderPodReady(podName, podNamespace string, f *framework.Framework) {
+// 	GinkgoHelper()
+
+// 	Eventually(func() error {
+// 		pod, err := f.Clients.KubeClient().CoreV1().Pods(podNamespace).Get(context.Background(), podName, metav1.GetOptions{})
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		for _, c := range pod.Status.Conditions {
+// 			if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue {
+// 				return nil
+// 			}
+// 		}
+
+// 		return fmt.Errorf("pod %s is not ready", podName)
+// 	}).WithTimeout(framework.ShortTimeout).WithPolling(framework.PollingInterval).Should(Succeed())
+// }
+
+// func getUploaderIngress(vd *v1alpha2.VirtualDisk, f *framework.Framework) (*netv1.Ingress, error) {
+// 	ings, err := f.Clients.KubeClient().NetworkingV1().Ingresses(vd.Namespace).List(context.Background(), metav1.ListOptions{})
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	for _, ing := range ings.Items {
+// 		for _, ref := range ing.OwnerReferences {
+// 			if ref.Kind == v1alpha2.VirtualDiskKind && ref.Name == vd.Name && strings.HasSuffix(ing.Name, string(vd.UID)) {
+// 				return &ing, nil
+// 			}
+// 		}
+// 	}
+
+// 	return nil, fmt.Errorf("could not find uploader ingress for virtual disk %s", vd.Name)
+// }
+
+// func untilUploaderIngressReady(ingName, ingNamespace string, f *framework.Framework) {
+// 	GinkgoHelper()
+
+// 	Eventually(func() error {
+// 		ing, err := f.Clients.KubeClient().NetworkingV1().Ingresses(ingNamespace).Get(context.Background(), ingName, metav1.GetOptions{})
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		if len(ing.Status.LoadBalancer.Ingress) != 0 {
+// 			return nil
+// 		}
+
+// 		return fmt.Errorf("ingress %s is not ready", ingName)
+// 	}).WithTimeout(framework.MiddleTimeout).WithPolling(framework.PollingInterval).Should(Succeed())
+// }