fix: fetch pods status for pending Deployment,DaemonSet and StatefulSet (#392)

* fetch pod status for pending Deployment,DaemonSet and StatefulSet

* fix linter

* fix selector

* add limit

* add progress timestamp to annotation

* refactor

* linter

Author: zreigz

internal/controller/clusterdrain_controller.go

@@ -34,6 +34,7 @@ const defaultBatchSize = 50
 const (
 	drainAnnotation    = "deployment.plural.sh/drain"
 	healthStatusJitter = 5
+	threshold          = 15
 )
 
 // ClusterDrainReconciler reconciles a ClusterDrain object
@@ -218,28 +219,42 @@ func healthStatusDelay() time.Duration {
 }
 
 func waitForHealthStatus(ctx context.Context, c client.Client, obj *unstructured.Unstructured) error {
-	startTime := time.Now()
+	timeout := threshold * time.Minute            // Timeout duration
+	ticker := time.NewTicker(healthStatusDelay()) // Ticker to periodically check health status
+	defer ticker.Stop()
+
+	// Create a timeout channel that will trigger after the specified timeout
+	timeoutChan := time.After(timeout)
+
 	for {
-		time.Sleep(healthStatusDelay())
-		if err := c.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
-			return err
-		}
-		status := common.ToStatus(obj)
-		if status == nil {
-			return fmt.Errorf("status is nil")
-		}
+		select {
+		case <-ticker.C:
+			// Fetch the latest object
+			if err := c.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
+				return err
+			}
 
-		switch *status {
-		case console.ComponentStateRunning:
-			return nil
-		case console.ComponentStateFailed:
-			return fmt.Errorf("component %s failed", obj.GetName())
-		}
+			// Check the status of the object
+			status := common.ToStatus(obj)
+			if status == nil {
+				return fmt.Errorf("status is nil")
+			}
 
-		if time.Since(startTime).Minutes() > 5 {
-			return fmt.Errorf("timeout after %f minutes", time.Since(startTime).Minutes())
+			// Handle the different states of the component
+			switch *status {
+			case console.ComponentStateRunning:
+				return nil
+			case console.ComponentStateFailed:
+				return fmt.Errorf("component %s failed", obj.GetName())
+			}
+
+		case <-timeoutChan:
+			return fmt.Errorf("timeout after %f minutes", timeout.Seconds()/60)
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
+
 }
 
 func splitIntoWaves[T any](items []T, batchSize int) [][]T {

internal/utils/kubernetes.go

@@ -52,6 +52,24 @@ func TryAddControllerRef(ctx context.Context, client ctrlruntimeclient.Client, o
 	})
 }
 
+func TryToUpdate(ctx context.Context, client ctrlruntimeclient.Client, object ctrlruntimeclient.Object) error {
+	key := ctrlruntimeclient.ObjectKeyFromObject(object)
+
+	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		original := object.DeepCopyObject().(ctrlruntimeclient.Object)
+		if err := client.Get(ctx, key, object); err != nil {
+			return fmt.Errorf("could not fetch current %s/%s state, got error: %w", object.GetName(), object.GetNamespace(), err)
+		}
+
+		if reflect.DeepEqual(object, original) {
+			return nil
+		}
+
+		return client.Patch(ctx, original, ctrlruntimeclient.MergeFrom(object))
+	})
+
+}
+
 func TryAddOwnerRef(ctx context.Context, client ctrlruntimeclient.Client, owner ctrlruntimeclient.Object, object ctrlruntimeclient.Object, scheme *runtime.Scheme) error {
 	key := ctrlruntimeclient.ObjectKeyFromObject(object)
 

pkg/common/common.go

@@ -1,15 +1,22 @@
 package common
 
 import (
+	"context"
+	"time"
+
+	"github.com/pluralsh/deployment-operator/internal/utils"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	ctrclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 )
 
 const (
-	ManagedByLabel  = "plural.sh/managed-by"
-	AgentLabelValue = "agent"
+	ManagedByLabel             = "plural.sh/managed-by"
+	AgentLabelValue            = "agent"
+	LastProgressTimeAnnotation = "plural.sh/last-progress-time"
 )
 
 func ManagedByAgentLabelSelector() labels.Selector {
@@ -34,3 +41,37 @@ func Unmarshal(s string) (map[string]interface{}, error) {
 
 	return result, nil
 }
+
+func GetLastProgressTimestamp(ctx context.Context, k8sClient ctrclient.Client, obj *unstructured.Unstructured) (progressTime metav1.Time, err error) {
+	progressTime = metav1.Now()
+
+	if obj.GetAnnotations() == nil {
+		obj.SetAnnotations(make(map[string]string))
+	}
+	annotations := obj.GetAnnotations()
+	timeStr, ok := annotations[LastProgressTimeAnnotation]
+
+	defer func() {
+		if !ok {
+			err = utils.TryToUpdate(ctx, k8sClient, obj)
+			if err != nil {
+				return
+			}
+			key := ctrclient.ObjectKeyFromObject(obj)
+			err = k8sClient.Get(ctx, key, obj)
+		}
+	}()
+
+	if !ok {
+		annotations[LastProgressTimeAnnotation] = progressTime.Time.Format(time.RFC3339)
+		obj.SetAnnotations(annotations)
+		return
+	}
+	parsedTime, err := time.Parse(time.RFC3339, timeStr)
+	if err != nil {
+		return
+	}
+	progressTime = metav1.Time{Time: parsedTime}
+
+	return
+}

pkg/common/health.go

@@ -429,6 +429,16 @@ func getPodHealth(obj *unstructured.Unstructured) (*HealthStatus, error) {
 	}
 }
 
+func FindPodStatusCondition(conditions []corev1.PodCondition, conditionType corev1.PodConditionType) *corev1.PodCondition {
+	for i := range conditions {
+		if conditions[i].Type == conditionType {
+			return &conditions[i]
+		}
+	}
+
+	return nil
+}
+
 func getCorev1PodHealth(pod *corev1.Pod) (*HealthStatus, error) {
 	// This logic cannot be applied when the pod.Spec.RestartPolicy is: corev1.RestartPolicyOnFailure,
 	// corev1.RestartPolicyNever, otherwise it breaks the resource hook logic.
@@ -449,6 +459,17 @@ func getCorev1PodHealth(pod *corev1.Pod) (*HealthStatus, error) {
 			}
 		}
 
+		if cond := FindPodStatusCondition(pod.Status.Conditions, corev1.PodScheduled); cond != nil {
+			if status == "" && cond.Status == corev1.ConditionFalse {
+				// status older than 5min
+				cutoffTime := metav1.NewTime(time.Now().Add(-5 * time.Minute))
+				if cond.LastTransitionTime.Before(&cutoffTime) {
+					status = HealthStatusDegraded
+					messages = append(messages, cond.Message)
+				}
+			}
+		}
+
 		if status != "" {
 			return &HealthStatus{
 				Status:  status,

pkg/scraper/component_scraper.go

@@ -6,10 +6,9 @@ import (
 	"sync"
 	"time"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"github.com/pluralsh/deployment-operator/internal/utils"
 
 	"github.com/cert-manager/cert-manager/pkg/apis/certmanager"
-
 	"github.com/pluralsh/deployment-operator/internal/helpers"
 	"github.com/pluralsh/deployment-operator/pkg/common"
 	agentcommon "github.com/pluralsh/deployment-operator/pkg/common"
@@ -18,6 +17,7 @@ import (
 	"github.com/pluralsh/polly/containers"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
@@ -132,15 +132,15 @@ func RunAiInsightComponentScraperInBackgroundOrDie(ctx context.Context, k8sClien
 	}
 }
 
-func setUnhealthyComponents(ctx context.Context, k8sClient ctrclient.Client, gvk schema.GroupVersionKind, opts ...ctrclient.ListOption) error {
+func setUnhealthyComponents(ctx context.Context, k8sClient ctrclient.Client, gvk schema.GroupVersionKind) error {
 	pager := listResources(ctx, k8sClient, gvk)
 	for pager.HasNext() {
 		items, err := pager.NextPage()
 		if err != nil {
 			return err
 		}
 		for _, item := range items {
-			health, err := common.GetResourceHealth(&item)
+			health, err := getResourceHealthStatus(ctx, k8sClient, &item)
 			if err != nil {
 				return err
 			}
@@ -156,6 +156,36 @@ func setUnhealthyComponents(ctx context.Context, k8sClient ctrclient.Client, gvk
 	return nil
 }
 
+func getResourceHealthStatus(ctx context.Context, k8sClient ctrclient.Client, obj *unstructured.Unstructured) (*common.HealthStatus, error) {
+	health, err := common.GetResourceHealth(obj)
+	if err != nil {
+		return nil, err
+	}
+
+	progressTime, err := common.GetLastProgressTimestamp(ctx, k8sClient, obj)
+	if err != nil {
+		return nil, err
+	}
+
+	// remove entry if no longer progressing
+	if health.Status != common.HealthStatusProgressing {
+		// cleanup progress timestamp
+		annotations := obj.GetAnnotations()
+		delete(annotations, common.LastProgressTimeAnnotation)
+		obj.SetAnnotations(annotations)
+		return health, utils.TryToUpdate(ctx, k8sClient, obj)
+	}
+
+	// mark as failed if it exceeds a threshold
+	cutoffTime := metav1.NewTime(time.Now().Add(-30 * time.Minute))
+
+	if progressTime.Before(&cutoffTime) {
+		health.Status = common.HealthStatusDegraded
+	}
+
+	return health, nil
+}
+
 func listResources(ctx context.Context, k8sClient ctrclient.Client, gvk schema.GroupVersionKind) *algorithms.Pager[unstructured.Unstructured] {
 	var opts []ctrclient.ListOption
 	manageByOperatorLabels := map[string]string{