operator: various sidecar fixes

Prior to this commit the operator sidecar's decommissioner and pvcunbinder
controllers did not work. This was due to:
- RBAC issues, the sidecar did not correctly scope itself to a single namespace.
- Incorrect label selectors hidden within the controllers in question.

Additionally, the statefulset decommissioner's sole test case has been disabled
for quite sometime. There's been zero test coverage of this functionality.

This commit:
- Restores the decommissioner's tests to a working state
- Strips out the "fetcher" to reduce duplication and remove reliance on
  fetching live helm values.
- Replaces baked in filtering with a label selector argument that will be
  constructed by the helm chart.

A follow up commit with chart changes and acceptance tests will be submitted.
It's been made separate to ease the process of backporting to the v2.x.x
branches.

(cherry picked from commit 03dd394)

# Conflicts:
#	operator/cmd/run/run.go
#	operator/go.mod
#	pkg/go.mod

Author: chrisseto

operator/cmd/run/run.go

@@ -17,26 +17,18 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"slices"
 	"strings"
 	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
 	helmkube "helm.sh/helm/v3/pkg/kube"
-	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
-	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/certwatcher"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	kubeClient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
@@ -57,10 +49,10 @@ import (
 	adminutils "github.com/redpanda-data/redpanda-operator/operator/pkg/admin"
 	internalclient "github.com/redpanda-data/redpanda-operator/operator/pkg/client"
 	consolepkg "github.com/redpanda-data/redpanda-operator/operator/pkg/console"
-	pkglabels "github.com/redpanda-data/redpanda-operator/operator/pkg/labels"
 	"github.com/redpanda-data/redpanda-operator/operator/pkg/resources"
 	pkgsecrets "github.com/redpanda-data/redpanda-operator/operator/pkg/secrets"
 	redpandawebhooks "github.com/redpanda-data/redpanda-operator/operator/webhooks/redpanda"
+	"github.com/redpanda-data/redpanda-operator/pkg/pflagutil"
 )
 
 type RedpandaController string
@@ -90,32 +82,6 @@ var availableControllers = []string{
 	DecommissionController.toString(),
 }
 
-type LabelSelectorValue struct {
-	Selector labels.Selector
-}
-
-var _ pflag.Value = ((*LabelSelectorValue)(nil))
-
-func (s *LabelSelectorValue) Set(value string) error {
-	if value == "" {
-		return nil
-	}
-	var err error
-	s.Selector, err = labels.Parse(value)
-	return err
-}
-
-func (s *LabelSelectorValue) String() string {
-	if s.Selector == nil {
-		return ""
-	}
-	return s.Selector.String()
-}
-
-func (s *LabelSelectorValue) Type() string {
-	return "label selector"
-}
-
 // Metrics RBAC permissions
 // +kubebuilder:rbac:groups=authentication.k8s.io,resources=tokenreviews,verbs=create;
 // +kubebuilder:rbac:groups=authorization.k8s.io,resources=subjectaccessreviews,verbs=create;
@@ -148,7 +114,7 @@ func Command() *cobra.Command {
 		operatorMode                        bool
 		ghostbuster                         bool
 		unbindPVCsAfter                     time.Duration
-		unbinderSelector                    LabelSelectorValue
+		unbinderSelector                    pflagutil.LabelSelectorValue
 		allowPVRebinding                    bool
 		autoDeletePVCs                      bool
 		webhookCertPath                     string
@@ -295,21 +261,6 @@ func Command() *cobra.Command {
 	return cmd
 }
 
-type v1Fetcher struct {
-	client kubeClient.Client
-}
-
-func (f *v1Fetcher) FetchLatest(ctx context.Context, name, namespace string) (any, error) {
-	var vectorizedCluster vectorizedv1alpha1.Cluster
-	if err := f.client.Get(ctx, types.NamespacedName{
-		Name:      name,
-		Namespace: namespace,
-	}, &vectorizedCluster); err != nil {
-		return nil, err
-	}
-	return &vectorizedCluster, nil
-}
-
 //nolint:funlen,gocyclo // length looks good
 func Run(
 	ctx context.Context,
@@ -704,7 +655,11 @@ func Run(
 	}
 
 	if enableGhostBrokerDecommissioner {
-		d := decommissioning.NewStatefulSetDecommissioner(mgr, &v1Fetcher{client: mgr.GetClient()},
+		factory := internalclient.NewFactory(mgr.GetConfig(), mgr.GetClient()).WithAdminClientTimeout(rpClientTimeout)
+		adapter := vectorizedDecommissionerAdapter{factory: factory, client: mgr.GetClient()}
+		d := decommissioning.NewStatefulSetDecommissioner(
+			mgr,
+			adapter.getAdminClient,
 			decommissioning.WithSyncPeriod(ghostBrokerDecommissionerSyncPeriod),
 			decommissioning.WithCleanupPVCs(false),
 			// In Operator v1, decommissioning based on pod ordinal is not correct because
@@ -714,105 +669,8 @@ func Run(
 			decommissioning.WithDecommisionOnTooHighOrdinal(false),
 			// Operator v1 supports multiple NodePools, and therefore multiple STS.
 			// This function provides a custom replica count: the desired replicas of all STS, instead of a single STS.
-			decommissioning.WithDesiredReplicasFetcher(func(ctx context.Context, sts *appsv1.StatefulSet) (int32, error) {
-				// Get Cluster CR, so we can then find its StatefulSets for a full count of desired replicas.
-				idx := slices.IndexFunc(
-					sts.OwnerReferences,
-					func(ownerRef metav1.OwnerReference) bool {
-						return ownerRef.APIVersion == vectorizedv1alpha1.GroupVersion.String() && ownerRef.Kind == "Cluster"
-					})
-				if idx == -1 {
-					return 0, nil
-				}
-
-				var vectorizedCluster vectorizedv1alpha1.Cluster
-				if err := mgr.GetClient().Get(ctx, types.NamespacedName{
-					Name:      sts.OwnerReferences[idx].Name,
-					Namespace: sts.Namespace,
-				}, &vectorizedCluster); err != nil {
-					return 0, fmt.Errorf("could not get Cluster: %w", err)
-				}
-
-				// We assume the cluster is fine and synced, checks have been performed in the filter already.
-
-				// Get all nodepool-sts for this Cluster
-				var stsList appsv1.StatefulSetList
-				err := mgr.GetClient().List(ctx, &stsList, &client.ListOptions{
-					LabelSelector: pkglabels.ForCluster(&vectorizedCluster).AsClientSelector(),
-				})
-				if err != nil {
-					return 0, fmt.Errorf("failed to list statefulsets of Cluster: %w", err)
-				}
-
-				if len(stsList.Items) == 0 {
-					return 0, errors.New("found 0 StatefulSets for this Cluster")
-				}
-
-				var allReplicas int32
-				for _, sts := range stsList.Items {
-					allReplicas += ptr.Deref(sts.Spec.Replicas, 0)
-				}
-
-				// Should not happen, but if it actually happens, we don't want to run ghost broker decommissioner.
-				if allReplicas < 3 {
-					return 0, fmt.Errorf("found %d desiredReplicas, but want >= 3", allReplicas)
-				}
-
-				if allReplicas != vectorizedCluster.Status.CurrentReplicas || allReplicas != vectorizedCluster.Status.Replicas {
-					return 0, fmt.Errorf("replicas not synced. status.currentReplicas=%d,status.replicas=%d,allReplicas=%d", vectorizedCluster.Status.CurrentReplicas, vectorizedCluster.Status.Replicas, allReplicas)
-				}
-
-				return allReplicas, nil
-			}),
-			decommissioning.WithFactory(internalclient.NewFactory(mgr.GetConfig(), mgr.GetClient())),
-			decommissioning.WithFilter(func(ctx context.Context, sts *appsv1.StatefulSet) (bool, error) {
-				log := ctrl.LoggerFrom(ctx, "namespace", sts.Namespace).WithName("StatefulSetDecomissioner.Filter")
-				idx := slices.IndexFunc(
-					sts.OwnerReferences,
-					func(ownerRef metav1.OwnerReference) bool {
-						return ownerRef.APIVersion == vectorizedv1alpha1.GroupVersion.String() && ownerRef.Kind == "Cluster"
-					})
-				if idx == -1 {
-					return false, nil
-				}
-
-				var vectorizedCluster vectorizedv1alpha1.Cluster
-				if err := mgr.GetClient().Get(ctx, types.NamespacedName{
-					Name:      sts.OwnerReferences[idx].Name,
-					Namespace: sts.Namespace,
-				}, &vectorizedCluster); err != nil {
-					return false, fmt.Errorf("could not get Cluster: %w", err)
-				}
-
-				managedAnnotationKey := vectorizedv1alpha1.GroupVersion.Group + "/managed"
-				if managed, exists := vectorizedCluster.Annotations[managedAnnotationKey]; exists && managed == "false" {
-					log.V(1).Info("ignoring StatefulSet of unmanaged V1 Cluster", "sts", sts.Name, "namespace", sts.Namespace)
-					return false, nil
-				}
-
-				// Do some "manual" checks, as ClusterlQuiescent condition is always false if a ghost broker causes unhealthy cluster
-				// (and we can therefore not use it to check if the cluster is synced otherwise)
-				if vectorizedCluster.Status.CurrentReplicas != vectorizedCluster.Status.Replicas {
-					log.V(1).Info("replicas are not synced", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace)
-					return false, nil
-				}
-				if vectorizedCluster.Status.Restarting {
-					log.V(1).Info("cluster is restarting", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace)
-					return false, nil
-				}
-
-				if vectorizedCluster.Status.ObservedGeneration != vectorizedCluster.Generation {
-					log.V(1).Info("generation not synced", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace, "generation", vectorizedCluster.Generation, "observedGeneration", vectorizedCluster.Status.ObservedGeneration)
-					return false, nil
-				}
-
-				if vectorizedCluster.Status.DecommissioningNode != nil {
-					log.V(1).Info("decommission in progress", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace, "node", *vectorizedCluster.Status.DecommissioningNode)
-					return false, nil
-				}
-
-				return true, nil
-			}),
+			decommissioning.WithDesiredReplicasFetcher(adapter.desiredReplicas),
+			decommissioning.WithFilter(adapter.filter),
 		)
 		if err := d.SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "StatefulSetDecommissioner")

operator/cmd/run/vectorized.go

@@ -0,0 +1,156 @@
+// Copyright 2025 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 run
+
+import (
+	"context"
+	"fmt"
+	"slices"
+
+	"github.com/cockroachdb/errors"
+	"github.com/redpanda-data/common-go/rpadmin"
+	appsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	vectorizedv1alpha1 "github.com/redpanda-data/redpanda-operator/operator/api/vectorized/v1alpha1"
+	internalclient "github.com/redpanda-data/redpanda-operator/operator/pkg/client"
+	pkglabels "github.com/redpanda-data/redpanda-operator/operator/pkg/labels"
+)
+
+// vectorizedDecommissionerAdapter is a helper struct that implements various methods
+// of mapping StatefulSets through Vectorized Clusters to arguments for the
+// StatefulSetDecommissioner.
+type vectorizedDecommissionerAdapter struct {
+	client  client.Client
+	factory internalclient.ClientFactory
+}
+
+func (b *vectorizedDecommissionerAdapter) desiredReplicas(ctx context.Context, sts *appsv1.StatefulSet) (int32, error) {
+	// Get Cluster CR, so we can then find its StatefulSets for a full count of desired replicas.
+	vectorizedCluster, err := b.getCluster(ctx, sts)
+	if err != nil {
+		return 0, err
+	}
+
+	if vectorizedCluster == nil {
+		return 0, nil
+	}
+
+	// We assume the cluster is fine and synced, checks have been performed in the filter already.
+
+	// Get all nodepool-sts for this Cluster
+	var stsList appsv1.StatefulSetList
+	if err := b.client.List(ctx, &stsList, &client.ListOptions{
+		LabelSelector: pkglabels.ForCluster(vectorizedCluster).AsClientSelector(),
+		Namespace:     vectorizedCluster.Namespace,
+	}); err != nil {
+		return 0, fmt.Errorf("failed to list statefulsets of Cluster: %w", err)
+	}
+
+	if len(stsList.Items) == 0 {
+		return 0, errors.New("found 0 StatefulSets for this Cluster")
+	}
+
+	var allReplicas int32
+	for _, sts := range stsList.Items {
+		allReplicas += ptr.Deref(sts.Spec.Replicas, 0)
+	}
+
+	// Should not happen, but if it actually happens, we don't want to run ghost broker decommissioner.
+	if allReplicas < 3 {
+		return 0, errors.Newf("found %d desiredReplicas, but want >= 3", allReplicas)
+	}
+
+	if allReplicas != vectorizedCluster.Status.CurrentReplicas || allReplicas != vectorizedCluster.Status.Replicas {
+		return 0, errors.Newf("replicas not synced. status.currentReplicas=%d,status.replicas=%d,allReplicas=%d", vectorizedCluster.Status.CurrentReplicas, vectorizedCluster.Status.Replicas, allReplicas)
+	}
+
+	return allReplicas, nil
+}
+
+func (b *vectorizedDecommissionerAdapter) filter(ctx context.Context, sts *appsv1.StatefulSet) (bool, error) {
+	log := ctrl.LoggerFrom(ctx, "namespace", sts.Namespace).WithName("StatefulSetDecomissioner.Filter")
+
+	vectorizedCluster, err := b.getCluster(ctx, sts)
+	if err != nil {
+		return false, err
+	}
+
+	if vectorizedCluster == nil {
+		return false, nil
+	}
+
+	managedAnnotationKey := vectorizedv1alpha1.GroupVersion.Group + "/managed"
+	if managed, exists := vectorizedCluster.Annotations[managedAnnotationKey]; exists && managed == "false" {
+		log.V(1).Info("ignoring StatefulSet of unmanaged V1 Cluster", "sts", sts.Name, "namespace", sts.Namespace)
+		return false, nil
+	}
+
+	// Do some "manual" checks, as ClusterlQuiescent condition is always false if a ghost broker causes unhealthy cluster
+	// (and we can therefore not use it to check if the cluster is synced otherwise)
+	if vectorizedCluster.Status.CurrentReplicas != vectorizedCluster.Status.Replicas {
+		log.V(1).Info("replicas are not synced", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace)
+		return false, nil
+	}
+	if vectorizedCluster.Status.Restarting {
+		log.V(1).Info("cluster is restarting", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace)
+		return false, nil
+	}
+
+	if vectorizedCluster.Status.ObservedGeneration != vectorizedCluster.Generation {
+		log.V(1).Info("generation not synced", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace, "generation", vectorizedCluster.Generation, "observedGeneration", vectorizedCluster.Status.ObservedGeneration)
+		return false, nil
+	}
+
+	if vectorizedCluster.Status.DecommissioningNode != nil {
+		log.V(1).Info("decommission in progress", "cluster", vectorizedCluster.Name, "namespace", vectorizedCluster.Namespace, "node", *vectorizedCluster.Status.DecommissioningNode)
+		return false, nil
+	}
+
+	return true, nil
+}
+
+func (b *vectorizedDecommissionerAdapter) getAdminClient(ctx context.Context, sts *appsv1.StatefulSet) (*rpadmin.AdminAPI, error) {
+	cluster, err := b.getCluster(ctx, sts)
+	if err != nil {
+		return nil, err
+	}
+
+	if cluster == nil {
+		return nil, errors.Newf("failed to resolve %s/%s to vectorized cluster", sts.Namespace, sts.Name)
+	}
+
+	return b.factory.RedpandaAdminClient(ctx, cluster)
+}
+
+func (b *vectorizedDecommissionerAdapter) getCluster(ctx context.Context, sts *appsv1.StatefulSet) (*vectorizedv1alpha1.Cluster, error) {
+	idx := slices.IndexFunc(
+		sts.OwnerReferences,
+		func(ownerRef metav1.OwnerReference) bool {
+			return ownerRef.APIVersion == vectorizedv1alpha1.GroupVersion.String() && ownerRef.Kind == "Cluster"
+		})
+	if idx == -1 {
+		return nil, nil
+	}
+
+	var vectorizedCluster vectorizedv1alpha1.Cluster
+	if err := b.client.Get(ctx, types.NamespacedName{
+		Name:      sts.OwnerReferences[idx].Name,
+		Namespace: sts.Namespace,
+	}, &vectorizedCluster); err != nil {
+		return nil, errors.Wrap(err, "could not get Cluster")
+	}
+
+	return &vectorizedCluster, nil
+}