operator: let layered CRs delete after their cluster is removed + opt-in GC owner ref (UX-1351)

acceptance/features/role-crds.feature

@@ -297,6 +297,37 @@ Feature: Role CRDs
     And role "swap-role" should not have member "olduser2" in cluster "sasl"
     And RedpandaRole "swap-role" should have status field "managedPrincipals" set to "true"
 
+  @skip:gke @skip:aks @skip:eks
+  Scenario: Opt into garbage collection with the cluster via owner reference
+    Given there is no role "gc-role" in cluster "sasl"
+    And there are the following pre-existing users in cluster "sasl"
+      | name   | password | mechanism     |
+      | gcuser | password | SCRAM-SHA-256 |
+    When I apply Kubernetes manifest:
+    """
+    # The delete-with-cluster annotation opts the role into garbage collection:
+    # the operator stamps an owner reference pointing at the referenced cluster
+    # so Kubernetes deletes the role when the cluster is deleted (which also lets
+    # the namespace terminate cleanly). Without the annotation the role survives
+    # cluster deletion and can be removed on its own schedule.
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: RedpandaRole
+    metadata:
+      name: gc-role
+      annotations:
+        operator.redpanda.com/delete-with-cluster: "true"
+    spec:
+      cluster:
+        clusterRef:
+          name: sasl
+      principals:
+        - User:gcuser
+    """
+    And role "gc-role" is successfully synced
+    Then role "gc-role" should exist in cluster "sasl"
+    And the Kubernetes object of type "RedpandaRole.v1alpha2.cluster.redpanda.com" with name "gc-role" has an OwnerReference pointing to the cluster "sasl"
+
   @skip:gke @skip:aks @skip:eks
   Scenario: Manage roles with internal flag
     Given there is no role "__admin-role-k8s" in cluster "sasl"

acceptance/steps/k8s.go

@@ -22,13 +22,15 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/util/jsonpath"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	framework "github.com/redpanda-data/redpanda-operator/harpoon"
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
+	vectorizedv1alpha1 "github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1"
 )
 
 // this is a nasty hack due to the fact that we can't disable the linter for typecheck
@@ -46,8 +48,6 @@ func podWillEventuallyBeInPhase(ctx context.Context, t framework.TestingT, podNa
 }
 
 func kubernetesObjectHasClusterOwner(ctx context.Context, t framework.TestingT, groupVersionKind, resourceName, clusterName string) {
-	var cluster redpandav1alpha2.Redpanda
-
 	gvk, _ := schema.ParseKindArg(groupVersionKind)
 	obj, err := t.Scheme().New(*gvk)
 	require.NoError(t, err)
@@ -57,9 +57,12 @@ func kubernetesObjectHasClusterOwner(ctx context.Context, t framework.TestingT,
 	require.NoError(t, t.Get(ctx, t.ResourceKey(resourceName), o))
 
 	require.Eventually(t, func() bool {
-		require.NoError(t, t.Get(ctx, t.ResourceKey(clusterName), &cluster))
 		require.NoError(t, t.Get(ctx, t.ResourceKey(resourceName), o))
-		cluster.SetGroupVersionKind(redpandav1alpha2.SchemeGroupVersion.WithKind("Redpanda"))
+
+		// The referenced cluster (and therefore the expected owner reference GVK)
+		// depends on the variant: the vectorized variant rewrites clusterRefs to
+		// the v1 Cluster, every other variant uses the v2 Redpanda.
+		expected := expectedClusterOwnerReference(ctx, t, clusterName)
 
 		references := o.GetOwnerReferences()
 		if len(references) != 1 {
@@ -68,13 +71,9 @@ func kubernetesObjectHasClusterOwner(ctx context.Context, t framework.TestingT,
 		}
 
 		actual := references[0]
-		expected := cluster.OwnerShipRefObj()
-
-		matchesAPIVersion := actual.APIVersion == expected.APIVersion
-		matchesKind := actual.Kind == expected.Kind
-		matchesName := actual.Name == expected.Name
-
-		matches := matchesAPIVersion && matchesKind && matchesName
+		matches := actual.APIVersion == expected.APIVersion &&
+			actual.Kind == expected.Kind &&
+			actual.Name == expected.Name
 
 		t.Logf(`Checking object contains cluster owner reference? (actual: %s/%s -> %s) (expected: %s/%s -> %s)`, actual.Kind, actual.APIVersion, actual.Name, expected.Kind, expected.APIVersion, expected.Name)
 		return matches
@@ -85,6 +84,27 @@ func kubernetesObjectHasClusterOwner(ctx context.Context, t framework.TestingT,
 	t.Logf("Object has cluster owner reference for %q", clusterName)
 }
 
+// expectedClusterOwnerReference returns the owner reference a garbage-collected
+// object should carry for clusterName, accounting for the active variant: the
+// vectorized variant resolves clusterRefs to the v1 Cluster, all others to the
+// v2 Redpanda.
+func expectedClusterOwnerReference(ctx context.Context, t framework.TestingT, clusterName string) metav1.OwnerReference {
+	if getVersion(t, "") == "vectorized" {
+		var cluster vectorizedv1alpha1.Cluster
+		require.NoError(t, t.Get(ctx, t.ResourceKey(clusterName), &cluster))
+		return metav1.OwnerReference{
+			APIVersion: vectorizedv1alpha1.GroupVersion.String(),
+			Kind:       "Cluster",
+			Name:       cluster.Name,
+		}
+	}
+
+	var cluster redpandav1alpha2.Redpanda
+	require.NoError(t, t.Get(ctx, t.ResourceKey(clusterName), &cluster))
+	cluster.SetGroupVersionKind(redpandav1alpha2.SchemeGroupVersion.WithKind("Redpanda"))
+	return cluster.OwnerShipRefObj()
+}
+
 type recordedVariable string
 
 func recordVariable(ctx context.Context, t framework.TestingT, jsonPath, groupVersionKind, resourceName, variableName string) context.Context {

operator/internal/controller/redpanda/owner_reference_test.go

@@ -0,0 +1,111 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package redpanda
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
+	vectorizedv1alpha1 "github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1"
+)
+
+// TestOwnerReferenceForCluster pins the GC owner-reference builder used for the
+// opt-in auto-cleanup: deleting the referenced cluster CR should cascade-delete
+// the layered CR. The reference must (a) carry the correct GVK for v1/v2/stretch
+// refs, (b) be set only when the cluster lives in the same namespace as the
+// object (Kubernetes GC ignores cross-namespace owner refs), and (c) never block
+// the cluster's own deletion (controller=false, blockOwnerDeletion=false).
+func TestOwnerReferenceForCluster(t *testing.T) {
+	const ns = "team-a"
+
+	v2Cluster := &redpandav1alpha2.Redpanda{
+		ObjectMeta: metav1.ObjectMeta{Name: "rp", Namespace: ns, UID: types.UID("uid-v2")},
+	}
+	v1Cluster := &vectorizedv1alpha1.Cluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "legacy", Namespace: ns, UID: types.UID("uid-v1")},
+	}
+
+	v2Ref := &redpandav1alpha2.ClusterRef{Name: "rp"} // group/kind default to v2
+	v1Ref := &redpandav1alpha2.ClusterRef{
+		Name:  "legacy",
+		Group: ptr.To("redpanda.vectorized.io"),
+		Kind:  ptr.To("Cluster"),
+	}
+
+	tests := []struct {
+		name            string
+		ref             *redpandav1alpha2.ClusterRef
+		cluster         client.Object
+		objectNamespace string
+		wantNil         bool
+		wantAPIVersion  string
+		wantKind        string
+		wantName        string
+		wantUID         types.UID
+	}{
+		{
+			name:            "v2 Redpanda, same namespace",
+			ref:             v2Ref,
+			cluster:         v2Cluster,
+			objectNamespace: ns,
+			wantAPIVersion:  "cluster.redpanda.com/v1alpha2",
+			wantKind:        "Redpanda",
+			wantName:        "rp",
+			wantUID:         types.UID("uid-v2"),
+		},
+		{
+			name:            "v1 vectorized Cluster, same namespace",
+			ref:             v1Ref,
+			cluster:         v1Cluster,
+			objectNamespace: ns,
+			wantAPIVersion:  "redpanda.vectorized.io/v1alpha1",
+			wantKind:        "Cluster",
+			wantName:        "legacy",
+			wantUID:         types.UID("uid-v1"),
+		},
+		{
+			name:            "cross-namespace yields no owner ref",
+			ref:             v2Ref,
+			cluster:         v2Cluster,
+			objectNamespace: "other-team",
+			wantNil:         true,
+		},
+		{
+			name:            "nil cluster yields no owner ref",
+			ref:             v2Ref,
+			cluster:         nil,
+			objectNamespace: ns,
+			wantNil:         true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := ownerReferenceForCluster(tt.ref, tt.cluster, tt.objectNamespace)
+			if tt.wantNil {
+				require.Nil(t, got)
+				return
+			}
+			require.NotNil(t, got)
+			require.Equal(t, tt.wantAPIVersion, ptr.Deref(got.APIVersion, ""))
+			require.Equal(t, tt.wantKind, ptr.Deref(got.Kind, ""))
+			require.Equal(t, tt.wantName, ptr.Deref(got.Name, ""))
+			require.Equal(t, tt.wantUID, ptr.Deref(got.UID, ""))
+			require.False(t, ptr.Deref(got.Controller, true), "owner ref must not be a controller ref")
+			require.False(t, ptr.Deref(got.BlockOwnerDeletion, true), "owner ref must not block cluster deletion")
+		})
+	}
+}

operator/internal/controller/redpanda/resource_controller.go

@@ -19,13 +19,16 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/redpanda-data/common-go/otelutil/log"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metav1apply "k8s.io/client-go/applyconfigurations/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
 
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
+	vectorizedv1alpha1 "github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/lifecycle"
 	internalclient "github.com/redpanda-data/redpanda-operator/operator/pkg/client"
 	"github.com/redpanda-data/redpanda-operator/pkg/multicluster"
@@ -157,8 +160,7 @@ func ignoreAllConnectionErrors(logger logr.Logger, err error) error {
 	// able to clean ourselves up anyway.
 	if internalclient.IsTerminalClientError(err) ||
 		internalclient.IsConfigurationError(err) ||
-		internalclient.IsInvalidClusterError(err) ||
-		isNotFoundInChain(err) ||
+		isClusterGone(err) ||
 		isNetworkDialError(err) {
 		// We use Info rather than Error here because we don't want
 		// to ignore the verbosity settings. This is really only for
@@ -169,6 +171,65 @@ func ignoreAllConnectionErrors(logger logr.Logger, err error) error {
 	return err
 }
 
+// isClusterGone reports whether err means the referenced cluster can no longer
+// be resolved, so finalizer cleanup has nothing to act on and should proceed.
+// This covers the cluster CR being deleted (NotFound / ErrInvalidClusterRef) as
+// well as the whole kind being torn down: when the v1 (vectorized) operator is
+// uninstalled the CRD is removed and the lookup returns a *meta.NoKindMatchError
+// rather than NotFound, and a stripped RBAC for the type returns Forbidden.
+// Treating these as "cluster gone" keeps a RedpandaRole (or any layered CR)
+// from hanging in Terminating after its cluster is fully removed.
+func isClusterGone(err error) bool {
+	return internalclient.IsInvalidClusterError(err) ||
+		isNotFoundInChain(err) ||
+		meta.IsNoMatchError(err) ||
+		apierrors.IsForbidden(err)
+}
+
+// DeleteWithClusterAnnotation opts a layered CR into garbage collection when its
+// referenced cluster is deleted. When set to "true", the reconciler stamps an
+// owner reference pointing at the resolved cluster so Kubernetes deletes the CR
+// along with the cluster. Absent (the default), the CR survives cluster deletion
+// and can be deleted on its own schedule.
+const DeleteWithClusterAnnotation = "operator.redpanda.com/delete-with-cluster"
+
+// ownerReferenceForCluster builds a garbage-collection owner reference linking a
+// layered CR to its referenced cluster, or nil when one must not be set.
+//
+// It returns nil when the cluster is unresolved or lives in a different
+// namespace than the object: Kubernetes garbage collection ignores
+// cross-namespace owner references. The reference is deliberately non-controller
+// and non-blocking (controller=false, blockOwnerDeletion=false) so it acts only
+// as a GC edge and never blocks the cluster's own deletion.
+func ownerReferenceForCluster(ref *redpandav1alpha2.ClusterRef, cluster client.Object, objectNamespace string) *metav1apply.OwnerReferenceApplyConfiguration {
+	if ref == nil || cluster == nil {
+		return nil
+	}
+	if cluster.GetNamespace() != objectNamespace {
+		return nil
+	}
+
+	var apiVersion, kind string
+	switch {
+	case ref.IsV1():
+		apiVersion, kind = vectorizedv1alpha1.GroupVersion.String(), "Cluster"
+	case ref.IsStretchCluster():
+		apiVersion, kind = redpandav1alpha2.GroupVersion.String(), "StretchCluster"
+	case ref.IsV2():
+		apiVersion, kind = redpandav1alpha2.GroupVersion.String(), "Redpanda"
+	default:
+		return nil
+	}
+
+	return metav1apply.OwnerReference().
+		WithAPIVersion(apiVersion).
+		WithKind(kind).
+		WithName(cluster.GetName()).
+		WithUID(cluster.GetUID()).
+		WithController(false).
+		WithBlockOwnerDeletion(false)
+}
+
 // isNotFoundInChain walks the error chain to check if a "not found" error
 // is wrapped anywhere. This handles both K8s NotFound (missing secrets/configmaps)
 // and cloud secret not found errors.