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() {