fix: clean up of resources from deleted clusters.

cmd/main.go

@@ -145,7 +145,7 @@ func main() {
 	eventTriggerReconciler := getEventTriggerReconciler(mgr)
 	eventTriggerReconciler.Deployer = d
 
-	eventTriggerController, err = eventTriggerReconciler.SetupWithManager(mgr)
+	eventTriggerController, err = eventTriggerReconciler.SetupWithManager(ctx, mgr)
 	if err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "EventTrigger")
 		os.Exit(1)

controllers/eventreport_collection.go

@@ -196,7 +196,7 @@ func buildEventTriggersForClusterMap(eventTriggers *v1beta1.EventTriggerList,
 }
 
 // Periodically collects EventReports from each cluster (excluding pull mode clusters)
-func collectEventReports(config *rest.Config, c client.Client, s *runtime.Scheme,
+func collectEventReports(ctx context.Context, config *rest.Config, c client.Client, s *runtime.Scheme,
 	shardKey, capiOnboardAnnotation, version string, logger logr.Logger) {
 
 	interval := 10 * time.Second
@@ -209,15 +209,20 @@ func collectEventReports(config *rest.Config, c client.Client, s *runtime.Scheme
 	mgmtClusterSchema = s
 	mgmtClusterConfig = config
 
-	ctx := context.TODO()
 	for {
+		if ctx.Err() != nil {
+			return
+		}
+
 		logger.V(logs.LogDebug).Info("collecting EventTriggers")
 		// get all EventTriggers
 		eventTriggers := &v1beta1.EventTriggerList{}
 		err := c.List(ctx, eventTriggers)
 		if err != nil {
 			logger.V(logs.LogInfo).Error(err, "failed to get eventTriggers")
-			time.Sleep(interval)
+			if sleepOrDone(ctx, interval) {
+				return
+			}
 			continue
 		}
 
@@ -229,7 +234,9 @@ func collectEventReports(config *rest.Config, c client.Client, s *runtime.Scheme
 			shardKey, logger)
 		if err != nil {
 			logger.V(logs.LogInfo).Error(err, "failed to get clusters")
-			time.Sleep(interval)
+			if sleepOrDone(ctx, interval) {
+				return
+			}
 			continue
 		}
 
@@ -270,7 +277,20 @@ func collectEventReports(config *rest.Config, c client.Client, s *runtime.Scheme
 			firstCollection = false
 		}
 
-		time.Sleep(interval)
+		if sleepOrDone(ctx, interval) {
+			return
+		}
+	}
+}
+
+// sleepOrDone waits for either d to elapse or ctx to be canceled, whichever comes first.
+// Returns true if ctx was canceled, so callers can stop instead of looping through shutdown.
+func sleepOrDone(ctx context.Context, d time.Duration) bool {
+	select {
+	case <-ctx.Done():
+		return true
+	case <-time.After(d):
+		return false
 	}
 }
 
@@ -372,6 +392,113 @@ func processEventReportClusters(ctx context.Context, c client.Client,
 	}
 }
 
+// pollStaleEventReports periodically scans every EventReport in the management cluster and, for
+// any whose Cluster/SveltosCluster no longer exists, removes it along with any ClusterProfile/
+// ConfigMap/Secret an EventTrigger instantiated because of it. This runs independent of
+// EventReportMode: a cluster can be in pull mode - and so leave stale EventReports behind in the
+// management cluster - no matter how this instance collects reports. Only the default
+// (non-sharded) instance runs this: cluster existence is a global fact, not something a single
+// shard's cluster subset can safely determine on its own.
+func pollStaleEventReports(ctx context.Context, c client.Client, capiOnboardAnnotation string, logger logr.Logger) {
+	const interval = 5 * time.Minute
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		if err := removeStaleEventReportsOnce(ctx, c, capiOnboardAnnotation, logger); err != nil {
+			logger.V(logs.LogInfo).Error(err, "failed to remove stale EventReports")
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+// removeStaleEventReportsOnce lists every currently existing Cluster/SveltosCluster (cluster-wide,
+// no shard filtering) and every EventReport in the management cluster, then removes any EventReport
+// whose cluster is not in that live set, along with the ClusterProfile/ConfigMap/Secret instances
+// any EventTrigger created because of it.
+func removeStaleEventReportsOnce(ctx context.Context, c client.Client, capiOnboardAnnotation string,
+	logger logr.Logger) error {
+
+	liveClusters, err := clusterproxy.GetListOfClusters(ctx, c, "", capiOnboardAnnotation, logger)
+	if err != nil {
+		return err
+	}
+
+	type clusterKey struct{ ns, name, clusterType string }
+	live := make(map[clusterKey]bool, len(liveClusters))
+	for i := range liveClusters {
+		ref := liveClusters[i]
+		ct := strings.ToLower(string(clusterproxy.GetClusterType(&ref)))
+		live[clusterKey{ref.Namespace, ref.Name, ct}] = true
+	}
+
+	eventReportList := &libsveltosv1beta1.EventReportList{}
+	if err := c.List(ctx, eventReportList); err != nil {
+		return err
+	}
+
+	eventTriggers := &v1beta1.EventTriggerList{}
+	if err := c.List(ctx, eventTriggers); err != nil {
+		return err
+	}
+
+	staleClusters := make(map[clusterKey]bool)
+	for i := range eventReportList.Items {
+		er := &eventReportList.Items[i]
+		if er.Labels == nil {
+			continue
+		}
+		clusterName := er.Labels[libsveltosv1beta1.EventReportClusterNameLabel]
+		clusterType := er.Labels[libsveltosv1beta1.EventReportClusterTypeLabel]
+		clusterNs := er.Namespace
+		if clusterName == "" || clusterType == "" {
+			continue
+		}
+		key := clusterKey{clusterNs, clusterName, clusterType}
+		if !live[key] {
+			staleClusters[key] = true
+		}
+	}
+
+	var retErr error
+	for key := range staleClusters {
+		l := logger.WithValues("cluster", fmt.Sprintf("%s/%s:%s", key.ns, key.name, key.clusterType))
+
+		clusterType := libsveltosv1beta1.ClusterTypeCapi
+		if strings.EqualFold(key.clusterType, string(libsveltosv1beta1.ClusterTypeSveltos)) {
+			clusterType = libsveltosv1beta1.ClusterTypeSveltos
+		}
+
+		for i := range eventTriggers.Items {
+			eventTrigger := &eventTriggers.Items[i]
+			// Passing a nil EventReport and no clusterProfiles/policyRefs to keep means: this
+			// cluster is gone, so every ClusterProfile/ConfigMap/Secret this EventTrigger
+			// instantiated for it is stale. No-op if this EventTrigger never created anything
+			// for this cluster.
+			if err := removeInstantiatedResources(ctx, c, key.ns, key.name, clusterType, eventTrigger,
+				nil, nil, nil, l); err != nil {
+				l.V(logs.LogInfo).Error(err, fmt.Sprintf(
+					"failed to remove instantiated resources for EventTrigger %s", eventTrigger.Name))
+				retErr = err
+			}
+		}
+
+		if err := removeEventReportsFromCluster(ctx, c, key.ns, key.name, clusterType,
+			map[string]bool{}, l); err != nil {
+			l.V(logs.LogInfo).Error(err, "failed to remove stale EventReports")
+			retErr = err
+		}
+	}
+
+	return retErr
+}
+
 // collectAndProcessAllEventReports is used in agentless mode. It fetches all EventReports
 // from the management cluster in a single List call, groups the ones needing attention by
 // cluster, and processes only those clusters — avoiding N per-cluster List calls.

controllers/eventreport_collection_test.go

@@ -37,6 +37,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
+	configv1beta1 "github.com/projectsveltos/addon-controller/api/v1beta1"
 	"github.com/projectsveltos/event-manager/api/v1beta1"
 	"github.com/projectsveltos/event-manager/controllers"
 	libsveltosv1beta1 "github.com/projectsveltos/libsveltos/api/v1beta1"
@@ -858,6 +859,118 @@ spec:
 		Expect(err).To(BeNil())
 		Expect(string(*ceAction)).To(Equal("Delete"))
 	})
+
+	It("removeStaleEventReportsOnce removes EventReport and instantiated resources when cluster no longer exists", func() {
+		clusterNamespace := randomString()
+		clusterName := randomString()
+		clusterType := libsveltosv1beta1.ClusterTypeCapi
+
+		ns := &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{Name: clusterNamespace},
+		}
+		Expect(testEnv.Create(context.TODO(), ns)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, ns)).To(Succeed())
+
+		eventSourceName := randomString()
+		eventReport := getEventReport(eventSourceName, clusterNamespace, clusterName)
+		eventReport.Labels[libsveltosv1beta1.EventReportClusterNameLabel] = clusterName
+		eventReport.Labels[libsveltosv1beta1.EventReportClusterTypeLabel] = strings.ToLower(string(clusterType))
+		Expect(testEnv.Create(context.TODO(), eventReport)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, eventReport)).To(Succeed())
+
+		eventTrigger := &v1beta1.EventTrigger{
+			ObjectMeta: metav1.ObjectMeta{Name: randomString()},
+			Spec:       v1beta1.EventTriggerSpec{EventSourceName: eventSourceName},
+		}
+		Expect(testEnv.Create(context.TODO(), eventTrigger)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, eventTrigger)).To(Succeed())
+
+		// clusterProfile simulates a ClusterProfile instantiated by this EventTrigger because of
+		// this EventReport, back when clusterNamespace/clusterName still existed.
+		clusterProfile := &configv1beta1.ClusterProfile{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: randomString(),
+				Labels: controllers.GetInstantiatedObjectLabels(clusterNamespace, clusterName, eventTrigger.Name,
+					nil, clusterType),
+			},
+		}
+		Expect(testEnv.Create(context.TODO(), clusterProfile)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, clusterProfile)).To(Succeed())
+
+		// removeInstantiatedResources defers ConfigMap/Secret removal until any stale
+		// ClusterProfile is fully gone, so this converges over a few calls. The call's own
+		// error return is ignored: it aggregates failures across every stale cluster found
+		// in the shared testEnv (other specs may leave their own stale leftovers behind), so
+		// it does not reflect whether this test's own objects were handled correctly.
+		Eventually(func() bool {
+			_ = controllers.RemoveStaleEventReportsOnce(context.TODO(), testEnv.Client, "", logger)
+
+			err := testEnv.Get(context.TODO(),
+				types.NamespacedName{Namespace: eventReport.Namespace, Name: eventReport.Name},
+				&libsveltosv1beta1.EventReport{})
+			if err == nil || !apierrors.IsNotFound(err) {
+				return false
+			}
+
+			err = testEnv.Get(context.TODO(),
+				types.NamespacedName{Name: clusterProfile.Name}, &configv1beta1.ClusterProfile{})
+			return err != nil && apierrors.IsNotFound(err)
+		}, timeout, pollingInterval).Should(BeTrue())
+	})
+
+	It("removeStaleEventReportsOnce leaves EventReport and instantiated resources alone when cluster still exists", func() {
+		cluster := prepareCluster()
+		clusterType := libsveltosv1beta1.ClusterTypeCapi
+
+		// prepareCluster does not itself wait for the Cluster to be visible in the cache (only
+		// for the Machine and kubeconfig Secret it creates), so wait here: otherwise
+		// clusterproxy.GetListOfClusters can race the informer and momentarily miss this
+		// cluster, making removeStaleEventReportsOnce see it as gone.
+		Expect(waitForObject(context.TODO(), testEnv.Client, cluster)).To(Succeed())
+
+		eventSourceName := randomString()
+		eventReport := getEventReport(eventSourceName, cluster.Namespace, cluster.Name)
+		eventReport.Labels[libsveltosv1beta1.EventReportClusterNameLabel] = cluster.Name
+		eventReport.Labels[libsveltosv1beta1.EventReportClusterTypeLabel] = strings.ToLower(string(clusterType))
+		Expect(testEnv.Create(context.TODO(), eventReport)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, eventReport)).To(Succeed())
+
+		eventTrigger := &v1beta1.EventTrigger{
+			ObjectMeta: metav1.ObjectMeta{Name: randomString()},
+			Spec:       v1beta1.EventTriggerSpec{EventSourceName: eventSourceName},
+		}
+		Expect(testEnv.Create(context.TODO(), eventTrigger)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, eventTrigger)).To(Succeed())
+
+		clusterProfile := &configv1beta1.ClusterProfile{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: randomString(),
+				Labels: controllers.GetInstantiatedObjectLabels(cluster.Namespace, cluster.Name, eventTrigger.Name,
+					eventReport, clusterType),
+			},
+		}
+		Expect(testEnv.Create(context.TODO(), clusterProfile)).To(Succeed())
+		Expect(waitForObject(context.TODO(), testEnv.Client, clusterProfile)).To(Succeed())
+
+		// The call's own error return is ignored here too, for the same reason as above: it
+		// aggregates failures across every stale cluster in the shared testEnv, unrelated to
+		// this test's own (live) cluster. What matters is that this test's objects survive
+		// repeated calls.
+		Consistently(func() bool {
+			_ = controllers.RemoveStaleEventReportsOnce(context.TODO(), testEnv.Client, "", logger)
+
+			err := testEnv.Get(context.TODO(),
+				types.NamespacedName{Namespace: eventReport.Namespace, Name: eventReport.Name},
+				&libsveltosv1beta1.EventReport{})
+			if err != nil {
+				return false
+			}
+
+			err = testEnv.Get(context.TODO(),
+				types.NamespacedName{Name: clusterProfile.Name}, &configv1beta1.ClusterProfile{})
+			return err == nil
+		}, timeout, pollingInterval).Should(BeTrue())
+	})
 })
 
 func validateEventReports(eventSourceName string, cluster *clusterv1.Cluster, clusterType *libsveltosv1beta1.ClusterType) {

controllers/eventtrigger_controller.go

@@ -304,7 +304,7 @@ func (r *EventTriggerReconciler) reconcileNormal(
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *EventTriggerReconciler) SetupWithManager(mgr ctrl.Manager) (controller.Controller, error) {
+func (r *EventTriggerReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) (controller.Controller, error) {
 	c, err := ctrl.NewControllerManagedBy(mgr).
 		For(&v1beta1.EventTrigger{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		WithOptions(controller.Options{
@@ -364,10 +364,18 @@ func (r *EventTriggerReconciler) SetupWithManager(mgr ctrl.Manager) (controller.
 	*/
 
 	if r.EventReportMode == CollectFromManagementCluster {
-		go collectEventReports(mgr.GetConfig(), mgr.GetClient(), mgr.GetScheme(), r.ShardKey,
+		go collectEventReports(ctx, mgr.GetConfig(), mgr.GetClient(), mgr.GetScheme(), r.ShardKey,
 			r.CapiOnboardAnnotation, getVersion(), mgr.GetLogger())
 	}
 
+	if r.ShardKey == "" {
+		// Cluster existence is a global fact, not something a single shard's cluster subset can
+		// safely determine on its own, so only the default (non-sharded) instance runs this.
+		// Runs regardless of EventReportMode: a cluster can be in pull mode - and so leave stale
+		// EventReports in the management cluster - independent of how this instance collects reports.
+		go pollStaleEventReports(ctx, mgr.GetClient(), r.CapiOnboardAnnotation, mgr.GetLogger())
+	}
+
 	return c, nil
 }
 

controllers/eventtrigger_deployer.go

@@ -332,20 +332,39 @@ func undeployEventTriggerResourcesFromCluster(ctx context.Context, c client.Clie
 		return err
 	}
 
-	isPullMode, err := clusterproxy.IsClusterInPullMode(ctx, c, clusterNamespace, clusterName,
-		clusterType, logger)
+	// This was queued while the cluster was still present; time may have passed since then,
+	// so check again. If the cluster is gone by now, there is no agent left to undeploy from
+	// or to ack a pull mode removal request.
+	var clusterStillExists bool
+	cluster, err := clusterproxy.GetCluster(ctx, c, clusterNamespace, clusterName, clusterType)
 	if err != nil {
-		logger.V(logs.LogInfo).Error(err, "failed to verify if Cluster is in pull mode")
-		return err
+		if apierrors.IsNotFound(err) {
+			clusterStillExists = false
+		} else {
+			return err
+		}
+	} else {
+		clusterStillExists = cluster.GetDeletionTimestamp().IsZero()
 	}
 
-	if isPullMode {
-		logger.V(logs.LogDebug).Info("Undeploy eventTrigger in pull mode")
-		err := undeployEventTriggerInPullMode(ctx, c, clusterNamespace, clusterName, eventTrigger, logger)
+	if clusterStillExists {
+		isPullMode, err := clusterproxy.IsClusterInPullMode(ctx, c, clusterNamespace, clusterName,
+			clusterType, logger)
 		if err != nil {
-			logger.V(logs.LogInfo).Error(err, "failed to undeploy eventTrigger in pull mode")
+			logger.V(logs.LogInfo).Error(err, "failed to verify if Cluster is in pull mode")
 			return err
 		}
+
+		if isPullMode {
+			logger.V(logs.LogDebug).Info("Undeploy eventTrigger in pull mode")
+			err := undeployEventTriggerInPullMode(ctx, c, clusterNamespace, clusterName, eventTrigger, logger)
+			if err != nil {
+				logger.V(logs.LogInfo).Error(err, "failed to undeploy eventTrigger in pull mode")
+				return err
+			}
+		}
+	} else {
+		logger.V(logs.LogDebug).Info("cluster is gone, skipping pull mode undeploy")
 	}
 
 	logger.V(logs.LogDebug).Info("Undeploy eventTrigger. Removing EventSource")

controllers/export_test.go

@@ -26,6 +26,7 @@ var (
 	RemoveEventReportsFromCluster            = removeEventReportsFromCluster
 	CollectAndProcessEventReportsFromCluster = collectAndProcessEventReportsFromCluster
 	CollectAndProcessAllEventReports         = collectAndProcessAllEventReports
+	RemoveStaleEventReportsOnce              = removeStaleEventReportsOnce
 	DeleteEventReport                        = deleteEventReport
 	ProcessEventTriggerForCluster            = processEventTriggerForCluster
 	UpdateEventReportStatus                  = updateEventReportStatus