[WIP] Broker CR reconciler

acceptance/features/broker-crd.feature

@@ -0,0 +1,19 @@
+@serial
+Feature: Broker CRD lifecycle
+  @skip:gke @skip:aks @skip:eks
+  Scenario: Adopt pods from orphan-deleted StatefulSet and decommission a broker
+    Given I create a basic cluster "broker-test" with 3 nodes
+    And cluster "broker-test" is stable with 3 nodes
+    When I pause reconciliation on cluster "broker-test"
+    And I orphan-delete the StatefulSet for cluster "broker-test"
+    And I create Broker CRs for cluster "broker-test"
+    And I grant roll-grants to all Broker CRs for cluster "broker-test"
+    Then all Broker CRs for cluster "broker-test" should be Running
+    And cluster "broker-test" admin API should show 3 brokers
+    When I update Broker "broker-test-0" pod template with env "ROTATION_TEST=applied" for cluster "broker-test"
+    And I grant roll-grants to all Broker CRs for cluster "broker-test"
+    Then Broker "broker-test-0" pod should have env "ROTATION_TEST" = "applied"
+    And all Broker CRs for cluster "broker-test" should be Running
+    When I set decommission on Broker "broker-test-2"
+    Then Broker "broker-test-2" should reach phase "Decommissioned"
+    And cluster "broker-test" admin API should show 2 brokers

acceptance/main_test.go

@@ -384,6 +384,7 @@ func operatorCmdFlags() []string {
 		"--unbind-pvcs-after=5s",
 		"--cluster-connection-timeout=500ms",
 		"--enable-shadowlinks",
+		"--enable-broker",
 	}
 	if testutil.AcceptanceSetupNodePools() > 0 {
 		flags = append(flags, "--enable-v2-nodepools=true")

acceptance/steps/broker.go

@@ -0,0 +1,338 @@
+// 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 steps
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	runtimeclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/yaml"
+
+	framework "github.com/redpanda-data/redpanda-operator/harpoon"
+	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
+)
+
+func pauseReconciliation(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+	var cluster redpandav1alpha2.Redpanda
+	require.NoError(t, t.Get(ctx, key, &cluster))
+
+	patch := runtimeclient.MergeFrom(cluster.DeepCopy())
+	if cluster.Annotations == nil {
+		cluster.Annotations = map[string]string{}
+	}
+	cluster.Annotations["cluster.redpanda.com/managed"] = "false"
+	require.NoError(t, t.Patch(ctx, &cluster, patch))
+	t.Logf("Paused reconciliation on cluster %q", clusterName)
+}
+
+func orphanDeleteStatefulSet(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+	var sts appsv1.StatefulSet
+	require.NoError(t, t.Get(ctx, key, &sts))
+	require.NoError(t, t.Delete(ctx, &sts, runtimeclient.PropagationPolicy(metav1.DeletePropagationOrphan)))
+	t.Logf("Orphan-deleted StatefulSet %q", clusterName)
+}
+
+func createBrokerCRsForCluster(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+
+	var sts appsv1.StatefulSet
+	// STS may already be gone if orphan-delete propagated; list pods by label instead.
+	var pods corev1.PodList
+	err := t.Get(ctx, key, &sts)
+	if err == nil {
+		sel, selErr := metav1.LabelSelectorAsSelector(sts.Spec.Selector)
+		require.NoError(t, selErr)
+		require.NoError(t, t.List(ctx, &pods, runtimeclient.InNamespace(key.Namespace), runtimeclient.MatchingLabelsSelector{Selector: sel}))
+	} else {
+		require.NoError(t, t.List(ctx, &pods, runtimeclient.InNamespace(key.Namespace), runtimeclient.MatchingLabels{
+			"app.kubernetes.io/instance": clusterName,
+			"app.kubernetes.io/name":     "redpanda",
+		}))
+	}
+	require.NotEmpty(t, pods.Items, "no pods found for cluster %q", clusterName)
+
+	sort.Slice(pods.Items, func(i, j int) bool { return pods.Items[i].Name < pods.Items[j].Name })
+
+	for i, pod := range pods.Items {
+		brokerName := fmt.Sprintf("%s-%d", key.Name, i)
+
+		specBytes, err := yaml.Marshal(pod.Spec)
+		require.NoError(t, err)
+		checksum := fmt.Sprintf("%x", sha256.Sum256(specBytes))
+
+		// Collect existing PVC names from the pod's volumes.
+		// For StatefulSet pods, the ClaimName is already the full PVC name
+		// (e.g. "datadir-broker-test-0"), not the VolumeClaimTemplate name.
+		var existingClaims []redpandav1alpha2.ExistingClaim
+		for _, vol := range pod.Spec.Volumes {
+			if vol.PersistentVolumeClaim != nil {
+				existingClaims = append(existingClaims, redpandav1alpha2.ExistingClaim{
+					Name: vol.PersistentVolumeClaim.ClaimName,
+				})
+			}
+		}
+
+		broker := &redpandav1alpha2.Broker{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      brokerName,
+				Namespace: key.Namespace,
+			},
+			Spec: redpandav1alpha2.BrokerSpec{
+				ClusterRef: redpandav1alpha2.ClusterRef{
+					Name: clusterName,
+				},
+				NetworkIndex: ptr.To(int32(i)),
+				PodTemplate: redpandav1alpha2.BrokerPodTemplate{
+					Labels:      pod.Labels,
+					Annotations: map[string]string{"config.redpanda.com/checksum": checksum},
+					Spec:        pod.Spec,
+				},
+				Storage: redpandav1alpha2.BrokerStorage{
+					ExistingClaims: existingClaims,
+				},
+			},
+		}
+		require.NoError(t, t.Create(ctx, broker))
+		t.Logf("Created Broker CR %q (networkIndex=%d)", brokerName, i)
+	}
+
+	t.Cleanup(func(ctx context.Context) {
+		t := framework.T(ctx)
+		var brokers redpandav1alpha2.BrokerList
+		_ = t.List(ctx, &brokers, runtimeclient.InNamespace(key.Namespace))
+		for i := range brokers.Items {
+			_ = t.Delete(ctx, &brokers.Items[i])
+		}
+	})
+}
+
+func grantRollGrants(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+	var brokers redpandav1alpha2.BrokerList
+	require.NoError(t, t.List(ctx, &brokers, runtimeclient.InNamespace(key.Namespace)))
+
+	deadline := strconv.FormatInt(time.Now().Add(10*time.Minute).Unix(), 10)
+	for i := range brokers.Items {
+		b := &brokers.Items[i]
+		checksum := b.Spec.PodTemplate.Annotations["config.redpanda.com/checksum"]
+		require.NotEmpty(t, checksum, "broker %q missing config checksum", b.Name)
+
+		patch := runtimeclient.MergeFrom(b.DeepCopy())
+		if b.Annotations == nil {
+			b.Annotations = map[string]string{}
+		}
+		b.Annotations["operator.redpanda.com/roll-grant"] = checksum + "/" + deadline
+		require.NoError(t, t.Patch(ctx, b, patch))
+		t.Logf("Granted roll-grant to Broker %q", b.Name)
+	}
+}
+
+func allBrokerCRsRunning(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+	require.Eventually(t, func() bool {
+		var brokers redpandav1alpha2.BrokerList
+		if err := t.List(ctx, &brokers, runtimeclient.InNamespace(key.Namespace)); err != nil {
+			return false
+		}
+		if len(brokers.Items) == 0 {
+			return false
+		}
+		for _, b := range brokers.Items {
+			if b.Status.Phase != redpandav1alpha2.BrokerPhaseRunning {
+				t.Logf("Broker %q phase=%s (want Running)", b.Name, b.Status.Phase)
+				return false
+			}
+		}
+		return true
+	}, 5*time.Minute, 5*time.Second, "not all Broker CRs reached Running phase")
+}
+
+func setDecommissionOnBroker(ctx context.Context, t framework.TestingT, brokerName string) {
+	key := t.ResourceKey(brokerName)
+	var broker redpandav1alpha2.Broker
+	require.NoError(t, t.Get(ctx, key, &broker))
+
+	patch := runtimeclient.MergeFrom(broker.DeepCopy())
+	broker.Spec.Decommission = true
+	require.NoError(t, t.Patch(ctx, &broker, patch))
+	t.Logf("Set decommission=true on Broker %q", brokerName)
+}
+
+func brokerShouldReachPhase(ctx context.Context, t framework.TestingT, brokerName, phase string) {
+	key := t.ResourceKey(brokerName)
+	var broker redpandav1alpha2.Broker
+	require.Eventually(t, func() bool {
+		if err := t.Get(ctx, key, &broker); err != nil {
+			return false
+		}
+		t.Logf("Broker %q phase=%s (want %s)", brokerName, broker.Status.Phase, phase)
+		return string(broker.Status.Phase) == phase
+	}, 5*time.Minute, 5*time.Second, "Broker %q never reached phase %q", brokerName, phase)
+}
+
+func updateBrokerPodTemplateEnv(ctx context.Context, t framework.TestingT, brokerName, envKeyValue, _ string) {
+	key := t.ResourceKey(brokerName)
+	var broker redpandav1alpha2.Broker
+	require.NoError(t, t.Get(ctx, key, &broker))
+
+	parts := strings.SplitN(envKeyValue, "=", 2)
+	require.Len(t, parts, 2, "env must be KEY=VALUE, got %q", envKeyValue)
+
+	broker.Spec.PodTemplate.Spec.Containers[0].Env = append(
+		broker.Spec.PodTemplate.Spec.Containers[0].Env,
+		corev1.EnvVar{Name: parts[0], Value: parts[1]},
+	)
+
+	specBytes, err := yaml.Marshal(broker.Spec.PodTemplate.Spec)
+	require.NoError(t, err)
+	checksum := fmt.Sprintf("%x", sha256.Sum256(specBytes))
+	broker.Spec.PodTemplate.Annotations["config.redpanda.com/checksum"] = checksum
+
+	require.NoError(t, t.Update(ctx, &broker))
+	t.Logf("Updated Broker %q pod template: added env %s, new checksum %s", brokerName, envKeyValue, checksum[:12])
+}
+
+func brokerPodShouldHaveEnv(ctx context.Context, t framework.TestingT, brokerName, envName, envValue string) {
+	key := t.ResourceKey(brokerName)
+	require.Eventually(t, func() bool {
+		var broker redpandav1alpha2.Broker
+		if err := t.Get(ctx, key, &broker); err != nil {
+			return false
+		}
+		var pod corev1.Pod
+		if err := t.Get(ctx, runtimeclient.ObjectKey{Name: broker.PodName(), Namespace: key.Namespace}, &pod); err != nil {
+			return false
+		}
+		for _, c := range pod.Spec.Containers {
+			for _, e := range c.Env {
+				if e.Name == envName && e.Value == envValue {
+					return true
+				}
+			}
+		}
+		t.Logf("Broker %q pod %q does not yet have env %s=%s", brokerName, pod.Name, envName, envValue)
+		return false
+	}, 5*time.Minute, 5*time.Second, "Broker %q pod never got env %s=%s", brokerName, envName, envValue)
+}
+
+func brokerShouldNotBeInMaintenanceMode(ctx context.Context, t framework.TestingT, brokerName, clusterName string) {
+	key := t.ResourceKey(brokerName)
+	var broker redpandav1alpha2.Broker
+	require.NoError(t, t.Get(ctx, key, &broker))
+	require.NotNil(t, broker.Status.BrokerID, "broker %q has no broker ID", brokerName)
+	brokerID := int(*broker.Status.BrokerID)
+
+	clients := clientsForCluster(ctx, clusterName)
+	admin := clients.RedpandaAdmin(ctx)
+
+	brokers, err := admin.Brokers(ctx)
+	require.NoError(t, err)
+
+	for _, b := range brokers {
+		if b.NodeID == brokerID {
+			draining := b.Maintenance != nil && b.Maintenance.Draining
+			t.Logf("Broker %q (nodeID=%d) maintenance=%+v", brokerName, brokerID, b.Maintenance)
+			require.False(t, draining, "broker %q (nodeID=%d) is still in maintenance mode", brokerName, brokerID)
+			return
+		}
+	}
+	t.Fatalf("broker %q (nodeID=%d) not found in admin API response", brokerName, brokerID)
+}
+
+func brokerPVCsShouldBeOwnedByBrokerCR(ctx context.Context, t framework.TestingT, brokerName, _ string) {
+	key := t.ResourceKey(brokerName)
+	var broker redpandav1alpha2.Broker
+	require.NoError(t, t.Get(ctx, key, &broker))
+
+	require.NotEmpty(t, broker.Spec.Storage.ExistingClaims, "Broker %q has no ExistingClaims", brokerName)
+	for _, ec := range broker.Spec.Storage.ExistingClaims {
+		var pvc corev1.PersistentVolumeClaim
+		require.NoError(t, t.Get(ctx, runtimeclient.ObjectKey{Name: ec.Name, Namespace: key.Namespace}, &pvc))
+
+		ctrl := metav1.GetControllerOf(&pvc)
+		require.NotNil(t, ctrl, "PVC %q has no controller ownerRef", ec.Name)
+		require.Equal(t, broker.Name, ctrl.Name, "PVC %q controller ownerRef name mismatch", ec.Name)
+		require.Equal(t, "Broker", ctrl.Kind, "PVC %q controller ownerRef kind mismatch", ec.Name)
+		t.Logf("PVC %q is owned by Broker %q", ec.Name, broker.Name)
+	}
+}
+
+func allBrokerCRsShouldHaveConditions(ctx context.Context, t framework.TestingT, clusterName string) {
+	key := t.ResourceKey(clusterName)
+	var brokers redpandav1alpha2.BrokerList
+	require.NoError(t, t.List(ctx, &brokers, runtimeclient.InNamespace(key.Namespace)))
+	require.NotEmpty(t, brokers.Items)
+
+	for _, b := range brokers.Items {
+		for _, condType := range []string{"Ready", "PodScheduled", "BrokerRegistered", "StorageBound"} {
+			found := false
+			for _, c := range b.Status.Conditions {
+				if c.Type == condType {
+					found = true
+					require.Equal(t, string(metav1.ConditionTrue), string(c.Status),
+						"Broker %q condition %s is %s (want True)", b.Name, condType, c.Status)
+					break
+				}
+			}
+			require.True(t, found, "Broker %q missing condition %s", b.Name, condType)
+		}
+		t.Logf("Broker %q has all 4 conditions True", b.Name)
+	}
+}
+
+func brokerShouldHaveCondition(ctx context.Context, t framework.TestingT, brokerName, condType, expectedStatus string) {
+	key := t.ResourceKey(brokerName)
+	want := metav1.ConditionStatus(expectedStatus)
+	require.Eventually(t, func() bool {
+		var broker redpandav1alpha2.Broker
+		if err := t.Get(ctx, key, &broker); err != nil {
+			return false
+		}
+		cond := apimeta.FindStatusCondition(broker.Status.Conditions, condType)
+		if cond == nil {
+			t.Logf("Broker %q missing condition %s", brokerName, condType)
+			return false
+		}
+		if cond.Status != want {
+			t.Logf("Broker %q condition %s=%s (want %s)", brokerName, condType, cond.Status, want)
+			return false
+		}
+		return true
+	}, 2*time.Minute, 5*time.Second, "Broker %q condition %s never reached %s", brokerName, condType, expectedStatus)
+}
+
+func clusterAdminAPIShouldShowBrokers(ctx context.Context, t framework.TestingT, clusterName string, count int) {
+	clients := clientsForCluster(ctx, clusterName)
+	admin := clients.RedpandaAdmin(ctx)
+
+	require.Eventually(t, func() bool {
+		brokers, err := admin.Brokers(ctx)
+		if err != nil {
+			t.Logf("Failed to get brokers: %v", err)
+			return false
+		}
+		t.Logf("Cluster %q has %d brokers (want %d)", clusterName, len(brokers), count)
+		return len(brokers) == count
+	}, 5*time.Minute, 5*time.Second, "cluster %q never had %d brokers", clusterName, count)
+}

acceptance/steps/register.go

@@ -176,6 +176,22 @@ func init() {
 	framework.RegisterStep(`^service "([^"]*)" should not have field managers:$`, checkResourceNoFieldManagers)
 	framework.RegisterStep(`^cluster "([^"]*)" should have sync error:$`, checkClusterHasSyncError)
 
+	// Broker CRD scenario steps
+	framework.RegisterStep(`^I pause reconciliation on cluster "([^"]*)"$`, pauseReconciliation)
+	framework.RegisterStep(`^I orphan-delete the StatefulSet for cluster "([^"]*)"$`, orphanDeleteStatefulSet)
+	framework.RegisterStep(`^I create Broker CRs for cluster "([^"]*)"$`, createBrokerCRsForCluster)
+	framework.RegisterStep(`^I grant roll-grants to all Broker CRs for cluster "([^"]*)"$`, grantRollGrants)
+	framework.RegisterStep(`^all Broker CRs for cluster "([^"]*)" should be Running$`, allBrokerCRsRunning)
+	framework.RegisterStep(`^I set decommission on Broker "([^"]*)"$`, setDecommissionOnBroker)
+	framework.RegisterStep(`^Broker "([^"]*)" should reach phase "([^"]*)"$`, brokerShouldReachPhase)
+	framework.RegisterStep(`^cluster "([^"]*)" admin API should show (\d+) brokers$`, clusterAdminAPIShouldShowBrokers)
+	framework.RegisterStep(`^I update Broker "([^"]*)" pod template with env "([^"]*)" for cluster "([^"]*)"$`, updateBrokerPodTemplateEnv)
+	framework.RegisterStep(`^Broker "([^"]*)" pod should have env "([^"]*)" = "([^"]*)"$`, brokerPodShouldHaveEnv)
+	framework.RegisterStep(`^Broker "([^"]*)" should not be in maintenance mode in cluster "([^"]*)"$`, brokerShouldNotBeInMaintenanceMode)
+	framework.RegisterStep(`^Broker "([^"]*)" PVCs should be owned by its Broker CR in cluster "([^"]*)"$`, brokerPVCsShouldBeOwnedByBrokerCR)
+	framework.RegisterStep(`^all Broker CRs for cluster "([^"]*)" should have conditions Ready, PodScheduled, BrokerRegistered, StorageBound as True$`, allBrokerCRsShouldHaveConditions)
+	framework.RegisterStep(`^Broker "([^"]*)" should have condition "([^"]*)" as "([^"]*)"$`, brokerShouldHaveCondition)
+
 	// Debug steps
 	framework.RegisterStep(`^I become debuggable$`, sleepALongTime)
 }