LOG-9424: Fix API flooding when MultiCLF shares the same serviceAccount (#3295)

* fix(auth): Change SCC Role naming to prevent conflicts when CLFs share service accounts

When multiple ClusterLogForwarders in the same namespace share the same
spec.serviceAccount, they were creating a shared Role named {sa}-scc.
This caused both CLFs to continuously update the Role's ownerReferences,
alternating between pointing to CLF #1 and CLF #2, flooding the API
server with thousands of update requests.

Changed the Role naming from {sa}-scc to {clfName}-{sa}-scc, making
each CLF's Role unique. The RoleBinding now correctly references this
new Role name format. Both resources keep their owner references for
proper cleanup when a CLF is deleted.

Also removed owner reference from the metadata-reader ClusterRoleBinding
because cluster-scoped resources should not be owned by namespaced
resources, following Kubernetes best practices.

Resolves: LOG-9424

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* feat(reconcile): Add cleanup logic for old SCC Role resources

Add DeleteRole function to enable cleanup of Resources with the old
naming scheme ({sa}-scc) during reconciliation. This ensures smooth
migration from the previous implementation where multiple CLFs sharing
a service account would conflict.

The ReconcileRBAC function now deletes any old Role using the legacy
naming scheme after creating the new one, preventing orphaned resources
during upgrades.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* test(auth): Update tests for new SCC RBAC resource naming scheme

Update existing tests to reflect the new naming scheme where SCC Roles
are named {clfName}-{sa}-scc instead of {sa}-scc. This ensures each
CLF gets a unique Role when sharing a service account.

Add test to verify that metadata-reader ClusterRoleBinding does not
have owner references, ensuring compliance with Kubernetes best
practices (cluster-scoped resources should not be owned by namespaced
resources).

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* check role references before deleting

---------

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: 2 people

internal/auth/rbac.go

@@ -1,6 +1,7 @@
 package auth
 
 import (
+	"context"
 	"fmt"
 	"github.com/openshift/cluster-logging-operator/internal/reconcile"
 	"github.com/openshift/cluster-logging-operator/internal/runtime"
@@ -12,21 +13,50 @@ import (
 
 // ReconcileRBAC reconciles the RBAC specifically for the service account and SCC
 func ReconcileRBAC(k8sClient client.Client, rbacName, saNamespace, saName string, owner metav1.OwnerReference) error {
-	desiredCRB := NewMetaDataReaderClusterRoleBinding(saNamespace, saName, owner)
+	desiredCRB := NewMetaDataReaderClusterRoleBinding(saNamespace, saName)
 	if err := reconcile.ClusterRoleBinding(k8sClient, desiredCRB.Name, func() *rbacv1.ClusterRoleBinding { return desiredCRB }); err != nil {
 		return err
 	}
-	desiredSCCRole := NewServiceAccountSCCRole(saNamespace, saName, owner)
+	desiredSCCRole := NewServiceAccountSCCRole(saNamespace, rbacName, saName, owner)
 	if err := reconcile.Role(k8sClient, desiredSCCRole); err != nil {
 		return err
 	}
 
-	desiredSCCRoleBinding := NewServiceAccountSCCRoleBinding(saNamespace, rbacName, desiredSCCRole.Name, saName, owner)
-	return reconcile.RoleBinding(k8sClient, desiredSCCRoleBinding)
+	desiredSCCRoleBinding := NewServiceAccountSCCRoleBinding(saNamespace, rbacName, saName, owner)
+	if err := reconcile.RoleBinding(k8sClient, desiredSCCRoleBinding); err != nil {
+		return err
+	}
+
+	// Cleanup old resources with previous naming scheme
+	oldRoleName := fmt.Sprintf("%s-scc", saName)
+
+	// List all RoleBindings in the namespace to check if any reference the old role
+	roleBindings := &rbacv1.RoleBindingList{}
+	if err := k8sClient.List(context.TODO(), roleBindings, client.InNamespace(saNamespace)); err != nil {
+		return err
+	}
+
+	// Check if any RoleBinding references the old role
+	hasReferences := false
+	for _, rb := range roleBindings.Items {
+		if rb.RoleRef.Name == oldRoleName {
+			hasReferences = true
+			break
+		}
+	}
+
+	// Only delete the old role if no RoleBindings reference it
+	if !hasReferences {
+		if err := reconcile.DeleteRole(k8sClient, saNamespace, oldRoleName); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 // NewMetaDataReaderClusterRoleBinding stubs a clusterrolebinding to allow reading of pod metadata (i.e. labels)
-func NewMetaDataReaderClusterRoleBinding(saNamespace, saName string, owner metav1.OwnerReference) *rbacv1.ClusterRoleBinding {
+func NewMetaDataReaderClusterRoleBinding(saNamespace, saName string) *rbacv1.ClusterRoleBinding {
 	name := fmt.Sprintf("metadata-reader-%s-%s", saNamespace, saName)
 	desired := runtime.NewClusterRoleBinding(name,
 		rbacv1.RoleRef{
@@ -41,26 +71,26 @@ func NewMetaDataReaderClusterRoleBinding(saNamespace, saName string, owner metav
 		},
 	)
 
-	utils.AddOwnerRefToObject(desired, owner)
 	return desired
 }
 
-func NewServiceAccountSCCRole(namespace, name string, owner metav1.OwnerReference) *rbacv1.Role {
-	name = fmt.Sprintf("%s-scc", name)
+func NewServiceAccountSCCRole(namespace, name, saName string, owner metav1.OwnerReference) *rbacv1.Role {
+	roleName := fmt.Sprintf("%s-%s-scc", name, saName)
 	sccRule := rbacv1.PolicyRule{
 		APIGroups:     []string{"security.openshift.io"},
 		ResourceNames: []string{sccName},
 		Resources:     []string{"securitycontextconstraints"},
 		Verbs:         []string{"use"},
 	}
 
-	desired := runtime.NewRole(namespace, name, sccRule)
+	desired := runtime.NewRole(namespace, roleName, sccRule)
 
 	utils.AddOwnerRefToObject(desired, owner)
 	return desired
 }
 
-func NewServiceAccountSCCRoleBinding(namespace, name, roleName, saName string, owner metav1.OwnerReference) *rbacv1.RoleBinding {
+func NewServiceAccountSCCRoleBinding(namespace, name, saName string, owner metav1.OwnerReference) *rbacv1.RoleBinding {
+	roleName := fmt.Sprintf("%s-%s-scc", name, saName)
 	roleRef := rbacv1.RoleRef{
 		APIGroup: rbacv1.GroupName,
 		Kind:     "Role",
@@ -73,8 +103,8 @@ func NewServiceAccountSCCRoleBinding(namespace, name, roleName, saName string, o
 		Namespace: namespace,
 	}
 
-	name = fmt.Sprintf("%s-scc", name)
-	desired := runtime.NewRoleBinding(namespace, name, roleRef, subject)
+	roleBindingName := fmt.Sprintf("%s-scc", name)
+	desired := runtime.NewRoleBinding(namespace, roleBindingName, roleRef, subject)
 
 	utils.AddOwnerRefToObject(desired, owner)
 	return desired

internal/auth/rbac_test.go

@@ -11,7 +11,7 @@ import (
 
 var _ = Describe("NewMetaDataReaderClusterRoleBinding", func() {
 	It("should stub a well-formed clusterrolebinding", func() {
-		Expect(test.YAMLString(auth.NewMetaDataReaderClusterRoleBinding(constants.OpenshiftNS, "logcollector", metav1.OwnerReference{}))).To(MatchYAML(
+		Expect(test.YAMLString(auth.NewMetaDataReaderClusterRoleBinding(constants.OpenshiftNS, "logcollector"))).To(MatchYAML(
 			`apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
@@ -27,16 +27,21 @@ subjects:
   namespace: openshift-logging
 `))
 	})
+
+	It("should not have owner references", func() {
+		crb := auth.NewMetaDataReaderClusterRoleBinding(constants.OpenshiftNS, "logcollector")
+		Expect(crb.OwnerReferences).To(BeEmpty())
+	})
 })
 
 var _ = Describe("ServiceAccount SCC Role & RoleBinding", func() {
 	It("should stub a well-formed role", func() {
-		Expect(test.YAMLString(auth.NewServiceAccountSCCRole(constants.OpenshiftNS, "my-sa", metav1.OwnerReference{}))).To(MatchYAML(
+		Expect(test.YAMLString(auth.NewServiceAccountSCCRole(constants.OpenshiftNS, "my-clf", "my-sa", metav1.OwnerReference{}))).To(MatchYAML(
 			`apiVersion: rbac.authorization.k8s.io/v1
 kind: Role
 metadata:
   creationTimestamp: null
-  name: my-sa-scc
+  name: my-clf-my-sa-scc
   namespace: openshift-logging
 rules:
 - apiGroups:
@@ -51,20 +56,20 @@ rules:
 	})
 
 	It("should stub a well-formed roleBinding", func() {
-		Expect(test.YAMLString(auth.NewServiceAccountSCCRoleBinding(constants.OpenshiftNS, "my-sa", "customSA-scc", "customSA", metav1.OwnerReference{}))).To(MatchYAML(
+		Expect(test.YAMLString(auth.NewServiceAccountSCCRoleBinding(constants.OpenshiftNS, "my-clf", "my-sa", metav1.OwnerReference{}))).To(MatchYAML(
 			`apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
 metadata:
   creationTimestamp: null
-  name: my-sa-scc
+  name: my-clf-scc
   namespace: openshift-logging
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: Role
-  name: customSA-scc
+  name: my-clf-my-sa-scc
 subjects:
   - kind: ServiceAccount
-    name: customSA
+    name: my-sa
     namespace: openshift-logging
 `))
 	})

internal/reconcile/rbac.go

@@ -7,6 +7,7 @@ import (
 	log "github.com/ViaQ/logerr/v2/log/static"
 	"github.com/openshift/cluster-logging-operator/internal/runtime"
 	rbacv1 "k8s.io/api/rbac/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
@@ -69,3 +70,14 @@ func DeleteClusterRoleBinding(k8sClient client.Client, name string) error {
 	log.V(3).Info("Deleting", "object", object)
 	return k8sClient.Delete(context.TODO(), object)
 }
+
+func DeleteRole(k8sClient client.Client, namespace, name string) error {
+	object := runtime.NewRole(namespace, name)
+	log.V(3).Info("Deleting Role", "namespace", namespace, "name", name)
+	err := k8sClient.Delete(context.TODO(), object)
+	// Ignore NotFound errors - resource is already deleted
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	return err
+}