From 7307c4817cbaa80380228118f0d6ebd5e4c93734 Mon Sep 17 00:00:00 2001 From: Gianluca Mardente Date: Wed, 15 Jul 2026 19:30:17 +0200 Subject: [PATCH] fix: clean up of resources from deleted clusters. Clean up ClusterProfile/ConfigMap/Secret and EventReport left behind by deleted clusters If a Cluster/SveltosCluster is deleted (or event-manager is down when it happens), the EventReport, ClusterProfile, ConfigMap, and Secret instances that were created because of it could be left behind indefinitely. The existing cleanup paths were triggered off Cluster delete watches or off EventTrigger reconciles. The cluster ones can be missed the transition entirely if event-manager wasn't running at the time. This PR fixes that. A new `pollStaleEventReports` background loop (5-minute tick, only on the default/non-sharded instance, independent of EventReportMode): lists every Cluster/SveltosCluster cluster-wide and every EventReport, and for any EventReport whose cluster no longer exists, removes every EventTrigger's instantiated ClusterProfile/ConfigMap/Secret for that cluster plus the stale EventReport itself. --- cmd/main.go | 2 +- controllers/eventreport_collection.go | 137 ++++++++++++++++++++- controllers/eventreport_collection_test.go | 113 +++++++++++++++++ controllers/eventtrigger_controller.go | 12 +- controllers/eventtrigger_deployer.go | 35 ++++-- controllers/export_test.go | 1 + controllers/suite_test.go | 10 ++ 7 files changed, 294 insertions(+), 16 deletions(-) diff --git a/cmd/main.go b/cmd/main.go index 1f1df6a..0bc69e6 100644 --- a/cmd/main.go +++ b/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) diff --git a/controllers/eventreport_collection.go b/controllers/eventreport_collection.go index c9a8798..f86b4a1 100644 --- a/controllers/eventreport_collection.go +++ b/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. diff --git a/controllers/eventreport_collection_test.go b/controllers/eventreport_collection_test.go index ca32255..8c65e61 100644 --- a/controllers/eventreport_collection_test.go +++ b/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) { diff --git a/controllers/eventtrigger_controller.go b/controllers/eventtrigger_controller.go index a021a13..a918063 100644 --- a/controllers/eventtrigger_controller.go +++ b/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 } diff --git a/controllers/eventtrigger_deployer.go b/controllers/eventtrigger_deployer.go index 8fcfca7..b2a1095 100644 --- a/controllers/eventtrigger_deployer.go +++ b/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") diff --git a/controllers/export_test.go b/controllers/export_test.go index 44c903e..baacda9 100644 --- a/controllers/export_test.go +++ b/controllers/export_test.go @@ -26,6 +26,7 @@ var ( RemoveEventReportsFromCluster = removeEventReportsFromCluster CollectAndProcessEventReportsFromCluster = collectAndProcessEventReportsFromCluster CollectAndProcessAllEventReports = collectAndProcessAllEventReports + RemoveStaleEventReportsOnce = removeStaleEventReportsOnce DeleteEventReport = deleteEventReport ProcessEventTriggerForCluster = processEventTriggerForCluster UpdateEventReportStatus = updateEventReportStatus diff --git a/controllers/suite_test.go b/controllers/suite_test.go index 8c23823..114d2f8 100644 --- a/controllers/suite_test.go +++ b/controllers/suite_test.go @@ -35,6 +35,7 @@ import ( "github.com/projectsveltos/event-manager/controllers" "github.com/projectsveltos/event-manager/internal/test/helpers" + "github.com/projectsveltos/libsveltos/lib/clusterproxy" libsveltoscrd "github.com/projectsveltos/libsveltos/lib/crd" "github.com/projectsveltos/libsveltos/lib/k8s_utils" ) @@ -130,6 +131,15 @@ var _ = BeforeSuite(func() { if synced := testEnv.GetCache().WaitForCacheSync(ctx); !synced { time.Sleep(time.Second) } + + // clusterproxy caches CAPI CRD presence once, in a process-wide flag that is never reset. + // Later tests exercise EventTriggerReconciler.Reconcile with a fake client that has no + // CustomResourceDefinition object, which would otherwise be the first caller and permanently + // (and incorrectly) cache "CAPI not present" for the rest of this test binary. Priming it + // here, with the real testEnv client (which does have the CRD), locks in the correct answer + // first. + _, err = clusterproxy.GetListOfClusters(ctx, testEnv.Client, "", "", ctrl.Log) + Expect(err).To(BeNil()) }) var _ = AfterSuite(func() {