feat: hub-side level-triggered companion GC (Component 6)

Add CompanionGCReconciler, a level-triggered backstop for referenced-data
companions stranded by interrupted finalization. When the
ReferencedDataController's WD finalizer is interrupted mid-flight (pod restart,
SIGKILL), a hub companion can be left with a referenced-by annotation pointing
at WDs that no longer exist on the hub. Without this controller, the companion
persists forever (leaking data including Secret bytes), its ResourceBinding keeps
a Work alive on the hub, and Karmada continuously re-creates the cell copy —
exactly the cm-pristine/secret-pristine lab scenario.

Design:
- Watches referenced-data-labeled ConfigMaps and Secrets on the hub. The
  federationMgr cache for ConfigMaps and Secrets is restricted to objects
  carrying the ReferencedDataLabel via cache.Options.ByObject (cmd/main.go).
  This is the OOM guard: predicates filter events, not cache contents; without
  the cache-level scope the informer would list-and-watch every ConfigMap/Secret
  on the Karmada hub — the same unscoped-informer pattern that OOMKilled the
  cell CompanionGCReconciler. The label predicate on the controller is kept as
  belt-and-suspenders but is NOT the primary memory guard.
- On each reconcile, parses the referenced-by annotation and checks each
  referenced WD by name in the companion's own hub namespace (ns-{project-uid})
  via HubClient.Get — no MCManager needed because WDs are federated to the hub.
- If ALL referrers are absent: deletes the companion and its ResourceBinding via
  the downstreamCompanionWriter path, driving the full Karmada cascade
  (RB → Work → cell copy deleted permanently).
- Conservative safety: terminating WDs count as present; corrupt annotations
  and malformed WD keys are handled by skip-not-delete.
- A companionGCPeriodicSweep backstop fires every 5 minutes to catch companions
  stranded before the controller started.
- Wired in setupManagementControllers on the federationMgr alongside
  OrphanRBReconciler and InstanceProjector.

Unit tests cover: orphaned CM/Secret deleted + RB torn down; live referrer
preserved; terminating referrer counts as present; all-absent multi-referrer
deleted; partial multi-referrer preserved; corrupt annotation preserved;
empty annotation preserved; unlabeled object unaffected; RB-already-gone
tolerated; periodic sweep drives reconciliation.

Federated e2e (test/e2e/referenced-data-delete-cascade/):
- HAPPY-PATH CASCADE: create WD + ConfigMap/Secret → assert companions on hub +
  cell + RBs present → delete WD → assert hub companion, RBs, cell copies all
  deleted and stay deleted (30 s anti-thrash poll).
- STRANDED COMPANION BACKSTOP: inject a labeled companion with a dead WD key →
  assert CompanionGCReconciler reclaims it within the sweep interval.

The e2e test is correct and self-contained but requires the Kind+Karmada
harness (task e2e:up) which is not runnable headless.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
(cherry picked from commit e419b54)

Author: scotwells

cmd/main.go

@@ -12,15 +12,18 @@ import (
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	"golang.org/x/sync/errgroup"
+	corev1 "k8s.io/api/core/v1"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
@@ -491,14 +494,36 @@ func ignoreCanceled(err error) error {
 // InstanceProjector). Called only when management controllers are enabled and
 // a federation REST config is available.
 func setupManagementControllers(mgr mcmanager.Manager, federationClient client.Client) ([]manager.Runnable, error) {
+	// companionLabelSelector scopes the federation manager's ConfigMap and
+	// Secret informer cache to referenced-data companions only. Without this,
+	// For(&corev1.ConfigMap{}) in CompanionGCReconciler would establish a
+	// cluster-wide ConfigMap+Secret informer that caches every object on the
+	// Karmada hub — the same OOM pattern that killed the cell CompanionGCReconciler.
+	// The label CACHE scope (not the predicate) is the correct OOM guard:
+	// predicates filter events, not cache contents.
+	companionLabelSelector := labels.SelectorFromSet(labels.Set{
+		computev1alpha.ReferencedDataLabel: computev1alpha.ReferencedDataLabelValue,
+	})
+
 	// The federation manager provides a cached, watchable handle to the Karmada
-	// federation control plane. It backs the InstanceProjector's Instance watch
-	// and the WorkloadDeploymentFederator's downstream WorkloadDeployment status
-	// watch. A manager.Manager embeds a cluster.Cluster, so it can be passed
-	// directly anywhere a watchable federation cluster source is required.
+	// federation control plane. It backs the InstanceProjector's Instance watch,
+	// the WorkloadDeploymentFederator's downstream WorkloadDeployment status watch,
+	// and the CompanionGCReconciler. A manager.Manager embeds a cluster.Cluster, so
+	// it can be passed directly anywhere a watchable federation cluster source is
+	// required.
 	federationMgr, err := manager.New(federationRestConfig, manager.Options{
 		Scheme:  scheme,
 		Metrics: metricsserver.Options{BindAddress: "0"},
+		Cache: cache.Options{
+			// Scope ConfigMap and Secret informers to referenced-data companions.
+			// CompanionGCReconciler is the only consumer on federationMgr that
+			// reads these types; nothing else (InstanceProjector, OrphanRBReconciler)
+			// needs non-companion CMs or Secrets from the cache.
+			ByObject: map[client.Object]cache.ByObject{
+				&corev1.ConfigMap{}: {Label: companionLabelSelector},
+				&corev1.Secret{}:    {Label: companionLabelSelector},
+			},
+		},
 	})
 	if err != nil {
 		return nil, fmt.Errorf("federation manager: %w", err)
@@ -535,5 +560,15 @@ func setupManagementControllers(mgr mcmanager.Manager, federationClient client.C
 		return nil, fmt.Errorf("OrphanRBReconciler: %w", err)
 	}
 
+	// CompanionGCReconciler is a level-triggered backstop for stranded hub
+	// companions: labeled ConfigMaps/Secrets whose referenced-by annotation
+	// points at WDs that no longer exist on the hub. On each reconcile it
+	// checks all referrer WDs in the hub namespace; if all are absent the
+	// companion and its ResourceBinding are deleted, driving the Karmada
+	// cascade to clean up Works and cell copies permanently.
+	if err = controller.SetupCompanionGCWithManager(federationMgr, federationClient); err != nil {
+		return nil, fmt.Errorf("CompanionGCReconciler: %w", err)
+	}
+
 	return []manager.Runnable{federationMgr}, nil
 }