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

@@ -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
+}

operator/cmd/run/vectorized.go

@@ -15,18 +15,26 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/redpanda-data/common-go/rpadmin"
+	rpkadminapi "github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
+	rpkconfig "github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	"github.com/redpanda-data/redpanda-operator/operator/internal/configwatcher"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/controller/decommissioning"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/controller/pvcunbinder"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/probes"
 	internalclient "github.com/redpanda-data/redpanda-operator/operator/pkg/client"
+	"github.com/redpanda-data/redpanda-operator/pkg/pflagutil"
 )
 
 // +kubebuilder:rbac:groups=coordination.k8s.io,namespace=default,resources=leases,verbs=get;list;watch;create;update;patch;delete
@@ -56,6 +64,7 @@ func Command() *cobra.Command {
 		brokerProbeBrokerURL       string
 		runUnbinder                bool
 		unbinderTimeout            time.Duration
+		selector                   pflagutil.LabelSelectorValue
 		panicAfter                 time.Duration
 	)
 
@@ -86,6 +95,7 @@ func Command() *cobra.Command {
 				brokerProbeBrokerURL,
 				runUnbinder,
 				unbinderTimeout,
+				selector.Selector,
 				panicAfter,
 			)
 		},
@@ -102,6 +112,7 @@ func Command() *cobra.Command {
 	// cluster flags
 	cmd.Flags().StringVar(&clusterNamespace, "redpanda-cluster-namespace", "", "The namespace of the cluster that this sidecar manages.")
 	cmd.Flags().StringVar(&clusterName, "redpanda-cluster-name", "", "The name of the cluster that this sidecar manages.")
+	cmd.Flags().Var(&selector, "selector", "Kubernetes label selector that will filter objects to be considered by the all controllers run by the sidecar.")
 
 	// decommission flags
 	cmd.Flags().BoolVar(&runDecommissioner, "run-decommissioner", false, "Specifies if the sidecar should run the broker decommissioner.")
@@ -153,10 +164,14 @@ func Run(
 	brokerProbeBrokerURL string,
 	runUnbinder bool,
 	unbinderTimeout time.Duration,
+	selector labels.Selector,
 	panicAfter time.Duration,
 ) error {
 	setupLog := ctrl.LoggerFrom(ctx).WithName("setup")
 
+	// Required arguments check, in sidecar mode these MUST be specified to
+	// ensure the sidecar only affects the helm deployment that's deployed it.
+
 	if clusterNamespace == "" {
 		err := errors.New("must specify a cluster-namespace parameter")
 		setupLog.Error(err, "no cluster namespace provided")
@@ -169,6 +184,20 @@ func Run(
 		return err
 	}
 
+	if selector == nil || selector.Empty() {
+		// Use a sensible default that's about as correct than the previous
+		// hard coded values. Hardcoding of name=redpanda is incorrect when
+		// nameoverride is used.
+		var err error
+		selector, err = labels.Parse(fmt.Sprintf(
+			"apps.kubernetes.io/component,app.kubernetes.io/name=redpanda,app.kubernetes.io/instance=%s",
+			clusterName,
+		))
+		if err != nil {
+			panic(err)
+		}
+	}
+
 	scheme := runtime.NewScheme()
 
 	for _, fn := range schemes {
@@ -183,35 +212,54 @@ func Run(
 		LeaderElectionID:        clusterName + "." + clusterNamespace + ".redpanda",
 		Scheme:                  scheme,
 		LeaderElectionNamespace: clusterNamespace,
+		Cache: cache.Options{
+			// Only watch the specified namespace, we don't have permissions for watch at the ClusterScope.
+			DefaultNamespaces: map[string]cache.Config{
+				clusterNamespace: {},
+			},
+		},
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to initialize manager")
 		return err
 	}
 
 	if runDecommissioner {
-		fetcher := decommissioning.NewChainedFetcher(
-			// prefer RPK profile first and then move on to fetch from helm values
-			decommissioning.NewRPKProfileFetcher(redpandaYAMLPath),
-			decommissioning.NewHelmFetcher(mgr),
-		)
+		setupLog.Info("broker decommissioner enabled", "namespace", clusterNamespace, "cluster", clusterName, "selector", selector)
 
-		if err := decommissioning.NewStatefulSetDecommissioner(mgr, fetcher, []decommissioning.Option{
-			decommissioning.WithFilter(decommissioning.FilterStatefulSetOwner(clusterNamespace, clusterName)),
+		fs := afero.NewOsFs()
+
+		params := rpkconfig.Params{ConfigFlag: redpandaYAMLPath}
+
+		config, err := params.Load(afero.NewOsFs())
+		if err != nil {
+			return err
+		}
+
+		if err := decommissioning.NewStatefulSetDecommissioner(
+			mgr,
+			func(ctx context.Context, _ *appsv1.StatefulSet) (*rpadmin.AdminAPI, error) {
+				// Always use the config that's loaded from redpanda.yaml, in
+				// sidecar mode no other STS's should be watched.
+				return rpkadminapi.NewClient(ctx, fs, config.VirtualProfile())
+			},
+			decommissioning.WithSelector(selector),
 			decommissioning.WithRequeueTimeout(decommissionRequeueTimeout),
 			decommissioning.WithDelayedCacheInterval(decommissionVoteInterval),
 			decommissioning.WithDelayedCacheMaxCount(decommissionMaxVoteCount),
-		}...).SetupWithManager(mgr); err != nil {
+		).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "StatefulSetDecommissioner")
 			return err
 		}
 	}
 
 	if runUnbinder {
+		setupLog.Info("PVC unbinder enabled", "namespace", clusterNamespace, "selector", selector)
+
 		if err := (&pvcunbinder.Controller{
-			Client:  mgr.GetClient(),
-			Timeout: unbinderTimeout,
-			Filter:  pvcunbinder.FilterPodOwner(clusterNamespace, clusterName),
+			Client:   mgr.GetClient(),
+			Timeout:  unbinderTimeout,
+			Selector: selector,
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "PVCUnbinder")
 			return err

operator/cmd/sidecar/sidecar.go

@@ -42,9 +42,14 @@ require (
 	github.com/redpanda-data/redpanda/src/go/rpk v0.0.0-20240827155712-244863ea0ae8
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/scalalang2/golang-fifo v1.0.2
+<<<<<<< HEAD
 	github.com/spf13/afero v1.11.0
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/pflag v1.0.5
+=======
+	github.com/spf13/afero v1.12.0
+	github.com/spf13/cobra v1.9.1
+>>>>>>> 03dd3942 (operator: various sidecar fixes)
 	github.com/stretchr/testify v1.10.0
 	github.com/testcontainers/testcontainers-go v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/redpanda v0.32.0
@@ -384,10 +389,15 @@ require (
 	github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/cast v1.7.0 // indirect
+<<<<<<< HEAD
 	github.com/spf13/viper v1.18.1 // indirect
 	github.com/stoewer/go-strcase v1.3.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect
+=======
+	github.com/spf13/pflag v1.0.7 // indirect
+	github.com/stoewer/go-strcase v1.3.1 // indirect
+>>>>>>> 03dd3942 (operator: various sidecar fixes)
 	github.com/texttheater/golang-levenshtein v1.0.1 // indirect
 	github.com/thales-e-security/pool v0.0.2 // indirect
 	github.com/theupdateframework/go-tuf v0.7.0 // indirect