fix: skip pull-mode clusters with a stale agent heartbeat

controllers/classifier_deployer.go

@@ -948,6 +948,49 @@ func (r *ClassifierReconciler) canProceed(ctx context.Context, classifierScope *
 	return true, nil
 }
 
+// isAgentHealthy returns true if the cluster is NOT in pull mode,
+// OR if it is in pull mode and the heartbeat is current.
+func (r *ClassifierReconciler) isAgentHealthy(ctx context.Context,
+	clusterRef *corev1.ObjectReference, logger logr.Logger) (bool, error) {
+
+	clusterType := clusterproxy.GetClusterType(clusterRef)
+	if clusterType != libsveltosv1beta1.ClusterTypeSveltos {
+		return true, nil
+	}
+
+	isPullMode, err := clusterproxy.IsClusterInPullMode(ctx, r.Client, clusterRef.Namespace,
+		clusterRef.Name, clusterType, logger)
+	if err != nil {
+		msg := fmt.Sprintf("failed to verify if Cluster is in pull mode: %v", err)
+		logger.V(logs.LogDebug).Info(msg)
+		return false, err
+	}
+
+	if !isPullMode {
+		return true, nil
+	}
+
+	sveltosCluster := &libsveltosv1beta1.SveltosCluster{}
+	err = r.Get(ctx,
+		types.NamespacedName{
+			Namespace: clusterRef.Namespace,
+			Name:      clusterRef.Name,
+		}, sveltosCluster)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	// Check if the failure message indicates a heartbeat timeout
+	if pullmode.IsAgentTimeoutError(sveltosCluster) {
+		return false, nil
+	}
+
+	return true, nil
+}
+
 // getCurrentHash gets current hash.
 // It considers Classifier and if mode is ClassifierReportMode == AgentSendReportsNoGateway also
 // the kubeconfig to access management cluster
@@ -1069,6 +1112,19 @@ func (r *ClassifierReconciler) processClassifier(ctx context.Context, classifier
 		clusterInfo.FailureMessage = &failureMessage
 		return clusterInfo, nil
 	}
+	isHealthy, err := r.isAgentHealthy(ctx, cluster, logger)
+	if err != nil {
+		failureMessage := err.Error()
+		clusterInfo.FailureMessage = &failureMessage
+		return clusterInfo, err
+	}
+	if !isHealthy {
+		failureMessage := "agent in managed cluster is not healthy."
+		logger.V(logs.LogInfo).Info(failureMessage)
+		clusterInfo.FailureMessage = &failureMessage
+		clusterInfo.Status = libsveltosv1beta1.SveltosStatusFailedNonRetriable
+		return clusterInfo, nil
+	}
 
 	// Remove any queued entry to cleanup
 	r.Deployer.CleanupEntries(cluster.Namespace, cluster.Name, classifier.Name, f.id,

controllers/classifier_deployer_test.go

@@ -44,6 +44,7 @@ import (
 	libsveltosv1beta1 "github.com/projectsveltos/libsveltos/api/v1beta1"
 	"github.com/projectsveltos/libsveltos/lib/deployer"
 	fakedeployer "github.com/projectsveltos/libsveltos/lib/deployer/fake"
+	"github.com/projectsveltos/libsveltos/lib/pullmode"
 	"github.com/projectsveltos/libsveltos/lib/sveltos_upgrade"
 )
 
@@ -1224,6 +1225,101 @@ metadata:
 		Expect(patches[0].Patch).ToNot(BeEmpty())
 		controllers.SetSveltosAgentConfigMap("")
 	})
+
+	It("isAgentHealthy returns true for a cluster not in pull mode", func() {
+		sveltosCluster := &libsveltosv1beta1.SveltosCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: randomString(),
+				Name:      randomString(),
+			},
+		}
+
+		c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(sveltosCluster).
+			WithStatusSubresource(sveltosCluster).Build()
+
+		reconciler := &controllers.ClassifierReconciler{
+			Client: c,
+			Scheme: scheme,
+		}
+
+		clusterRef := &corev1.ObjectReference{
+			Namespace:  sveltosCluster.Namespace,
+			Name:       sveltosCluster.Name,
+			Kind:       libsveltosv1beta1.SveltosClusterKind,
+			APIVersion: libsveltosv1beta1.GroupVersion.String(),
+		}
+
+		healthy, err := controllers.IsAgentHealthy(reconciler, context.TODO(), clusterRef, logger)
+		Expect(err).To(BeNil())
+		Expect(healthy).To(BeTrue())
+	})
+
+	It("isAgentHealthy returns false for a pull mode cluster whose agent heartbeat timed out", func() {
+		sveltosCluster := &libsveltosv1beta1.SveltosCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: randomString(),
+				Name:      randomString(),
+			},
+			Spec: libsveltosv1beta1.SveltosClusterSpec{
+				PullMode: true,
+			},
+		}
+
+		c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(sveltosCluster).
+			WithStatusSubresource(sveltosCluster).Build()
+
+		heartbeatTimeout := &pullmode.AgentHeartbeatTimeoutError{}
+		failureMessage := heartbeatTimeout.Error()
+		sveltosCluster.Status.FailureMessage = &failureMessage
+		Expect(c.Status().Update(context.TODO(), sveltosCluster)).To(Succeed())
+
+		reconciler := &controllers.ClassifierReconciler{
+			Client: c,
+			Scheme: scheme,
+		}
+
+		clusterRef := &corev1.ObjectReference{
+			Namespace:  sveltosCluster.Namespace,
+			Name:       sveltosCluster.Name,
+			Kind:       libsveltosv1beta1.SveltosClusterKind,
+			APIVersion: libsveltosv1beta1.GroupVersion.String(),
+		}
+
+		healthy, err := controllers.IsAgentHealthy(reconciler, context.TODO(), clusterRef, logger)
+		Expect(err).To(BeNil())
+		Expect(healthy).To(BeFalse())
+	})
+
+	It("isAgentHealthy returns true for a pull mode cluster with a current heartbeat", func() {
+		sveltosCluster := &libsveltosv1beta1.SveltosCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: randomString(),
+				Name:      randomString(),
+			},
+			Spec: libsveltosv1beta1.SveltosClusterSpec{
+				PullMode: true,
+			},
+		}
+
+		c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(sveltosCluster).
+			WithStatusSubresource(sveltosCluster).Build()
+
+		reconciler := &controllers.ClassifierReconciler{
+			Client: c,
+			Scheme: scheme,
+		}
+
+		clusterRef := &corev1.ObjectReference{
+			Namespace:  sveltosCluster.Namespace,
+			Name:       sveltosCluster.Name,
+			Kind:       libsveltosv1beta1.SveltosClusterKind,
+			APIVersion: libsveltosv1beta1.GroupVersion.String(),
+		}
+
+		healthy, err := controllers.IsAgentHealthy(reconciler, context.TODO(), clusterRef, logger)
+		Expect(err).To(BeNil())
+		Expect(healthy).To(BeTrue())
+	})
 })
 
 func prepareCluster() *clusterv1.Cluster {

controllers/export_test.go

@@ -74,6 +74,7 @@ var (
 	GetHandlersForFeature = getHandlersForFeature
 
 	ProcessClassifier                      = (*ClassifierReconciler).processClassifier
+	IsAgentHealthy                         = (*ClassifierReconciler).isAgentHealthy
 	RemoveClassifier                       = (*ClassifierReconciler).removeClassifier
 	RequeueClassifierForCluster            = (*ClassifierReconciler).requeueClassifierForCluster
 	RequeueClassifierForSecret             = (*ClassifierReconciler).requeueClassifierForSecret