operator: add PodDisruptionBudget support to Pipeline CRD

Add spec.budget field to Pipeline with maxUnavailable/minAvailable
options, following the convention used by Strimzi and Prometheus
Operator. The PDB is rendered by the Syncer alongside the Deployment
and ConfigMap, so it is automatically garbage-collected on CR deletion.

CRD validation enforces exactly one of maxUnavailable or minAvailable
via CEL rule. RBAC updated for policy/poddisruptionbudgets.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: david-yu

operator/api/redpanda/v1alpha2/pipeline_types.go

@@ -12,6 +12,7 @@ package v1alpha2
 import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
 
 	"github.com/redpanda-data/redpanda-operator/operator/pkg/functional"
@@ -178,11 +179,33 @@ type PipelineSpec struct {
 	// +optional
 	Zones []string `json:"zones,omitempty"`
 
+	// Budget configures a PodDisruptionBudget for the pipeline Deployment,
+	// protecting pipeline pods from voluntary disruptions such as node drains
+	// and cluster autoscaler evictions. When not set, no PDB is created.
+	// +optional
+	Budget *PipelineBudget `json:"budget,omitempty"`
+
 	// ClusterSource is a reference to the Redpanda cluster this pipeline connects to.
 	// +optional
 	ClusterSource *ClusterSource `json:"cluster,omitempty"`
 }
 
+// PipelineBudget configures a PodDisruptionBudget for the pipeline.
+// Exactly one of MaxUnavailable or MinAvailable must be specified.
+// +kubebuilder:validation:XValidation:message="exactly one of maxUnavailable or minAvailable must be set",rule="has(self.maxUnavailable) != has(self.minAvailable)"
+type PipelineBudget struct {
+	// MaxUnavailable is the maximum number of pipeline pods that can be
+	// unavailable during a voluntary disruption. Can be an absolute number
+	// (e.g. 1) or a percentage (e.g. "25%").
+	// +optional
+	MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"`
+	// MinAvailable is the minimum number of pipeline pods that must remain
+	// available during a voluntary disruption. Can be an absolute number
+	// (e.g. 2) or a percentage (e.g. "75%").
+	// +optional
+	MinAvailable *intstr.IntOrString `json:"minAvailable,omitempty"`
+}
+
 // PipelineStatus defines the observed state of a Connect resource.
 type PipelineStatus struct {
 	// ObservedGeneration is the last observed generation of the Connect resource.

operator/chart/files/rbac/pipeline.ClusterRole.yaml

@@ -76,6 +76,18 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - policy
+    resources:
+      - poddisruptionbudgets
+    verbs:
+      - get
+      - list
+      - watch
+      - create
+      - update
+      - patch
+      - delete
   - apiGroups:
       - monitoring.coreos.com
     resources:

operator/config/crd/bases/cluster.redpanda.com_pipelines.yaml

@@ -66,6 +66,34 @@ spec:
                   per-pipeline annotations taking precedence. Useful for integrations like
                   Datadog autodiscovery that rely on pod annotations.
                 type: object
+              budget:
+                description: |-
+                  Budget configures a PodDisruptionBudget for the pipeline Deployment,
+                  protecting pipeline pods from voluntary disruptions such as node drains
+                  and cluster autoscaler evictions. When not set, no PDB is created.
+                properties:
+                  maxUnavailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: |-
+                      MaxUnavailable is the maximum number of pipeline pods that can be
+                      unavailable during a voluntary disruption. Can be an absolute number
+                      (e.g. 1) or a percentage (e.g. "25%").
+                    x-kubernetes-int-or-string: true
+                  minAvailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: |-
+                      MinAvailable is the minimum number of pipeline pods that must remain
+                      available during a voluntary disruption. Can be an absolute number
+                      (e.g. 2) or a percentage (e.g. "75%").
+                    x-kubernetes-int-or-string: true
+                type: object
+                x-kubernetes-validations:
+                - message: exactly one of maxUnavailable or minAvailable must be set
+                  rule: has(self.maxUnavailable) != has(self.minAvailable)
               cluster:
                 description: ClusterSource is a reference to the Redpanda cluster
                   this pipeline connects to.

operator/config/rbac/itemized/pipeline.yaml

@@ -76,6 +76,18 @@ rules:
   - get
   - list
   - watch
+- apiGroups:
+  - policy
+  resources:
+  - poddisruptionbudgets
+  verbs:
+  - get
+  - list
+  - watch
+  - create
+  - update
+  - patch
+  - delete
 - apiGroups:
   - monitoring.coreos.com
   resources:

operator/internal/controller/pipeline/controller_test.go

@@ -27,10 +27,12 @@ import (
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
@@ -667,6 +669,87 @@ func TestRender_Deployment_Zones(t *testing.T) {
 	assert.Equal(t, zoneTopologyKey, dp.Spec.Template.Spec.TopologySpreadConstraints[0].TopologyKey)
 }
 
+// PodDisruptionBudget tests.
+
+func TestRender_PDB_NotConfigured(t *testing.T) {
+	pipeline := &redpandav1alpha2.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "no-pdb", Namespace: "default"},
+		Spec:       redpandav1alpha2.PipelineSpec{ConfigYAML: "input:\n  stdin: {}\noutput:\n  stdout: {}\n"},
+	}
+
+	r := &render{pipeline: pipeline, labels: Labels(pipeline)}
+	objs, err := r.Render(t.Context())
+	require.NoError(t, err)
+
+	// Should only have ConfigMap + Deployment, no PDB.
+	for _, obj := range objs {
+		assert.NotEqual(t, "PodDisruptionBudget", obj.GetObjectKind().GroupVersionKind().Kind)
+	}
+}
+
+func TestRender_PDB_MaxUnavailable(t *testing.T) {
+	maxUnavail := intstr.FromInt32(1)
+	pipeline := &redpandav1alpha2.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "pdb-max", Namespace: "default"},
+		Spec: redpandav1alpha2.PipelineSpec{
+			ConfigYAML: "input:\n  stdin: {}\noutput:\n  stdout: {}\n",
+			Budget: &redpandav1alpha2.PipelineBudget{
+				MaxUnavailable: &maxUnavail,
+			},
+		},
+	}
+
+	labels := Labels(pipeline)
+	r := &render{pipeline: pipeline, labels: labels}
+	objs, err := r.Render(t.Context())
+	require.NoError(t, err)
+
+	// Find the PDB.
+	var pdb *policyv1.PodDisruptionBudget
+	for _, obj := range objs {
+		if p, ok := obj.(*policyv1.PodDisruptionBudget); ok {
+			pdb = p
+		}
+	}
+	require.NotNil(t, pdb, "expected a PodDisruptionBudget in rendered objects")
+	assert.Equal(t, "pdb-max", pdb.Name)
+	assert.Equal(t, "default", pdb.Namespace)
+	assert.Equal(t, labels, pdb.Labels)
+	assert.Equal(t, labels, pdb.Spec.Selector.MatchLabels)
+	require.NotNil(t, pdb.Spec.MaxUnavailable)
+	assert.Equal(t, int32(1), pdb.Spec.MaxUnavailable.IntVal)
+	assert.Nil(t, pdb.Spec.MinAvailable)
+}
+
+func TestRender_PDB_MinAvailable(t *testing.T) {
+	minAvail := intstr.FromString("75%")
+	pipeline := &redpandav1alpha2.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "pdb-min", Namespace: "default"},
+		Spec: redpandav1alpha2.PipelineSpec{
+			ConfigYAML: "input:\n  stdin: {}\noutput:\n  stdout: {}\n",
+			Budget: &redpandav1alpha2.PipelineBudget{
+				MinAvailable: &minAvail,
+			},
+		},
+	}
+
+	labels := Labels(pipeline)
+	r := &render{pipeline: pipeline, labels: labels}
+	objs, err := r.Render(t.Context())
+	require.NoError(t, err)
+
+	var pdb *policyv1.PodDisruptionBudget
+	for _, obj := range objs {
+		if p, ok := obj.(*policyv1.PodDisruptionBudget); ok {
+			pdb = p
+		}
+	}
+	require.NotNil(t, pdb, "expected a PodDisruptionBudget in rendered objects")
+	require.NotNil(t, pdb.Spec.MinAvailable)
+	assert.Equal(t, "75%", pdb.Spec.MinAvailable.StrVal)
+	assert.Nil(t, pdb.Spec.MaxUnavailable)
+}
+
 // License validation unit tests.
 
 func TestValidateLicenseNoPath(t *testing.T) {

operator/internal/controller/pipeline/render.go

@@ -18,6 +18,7 @@ import (
 	"github.com/redpanda-data/common-go/kube"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -52,6 +53,7 @@ func Types() []kube.Object {
 	return []kube.Object{
 		&appsv1.Deployment{},
 		&corev1.ConfigMap{},
+		&policyv1.PodDisruptionBudget{},
 		&monitoringv1.PodMonitor{},
 	}
 }
@@ -71,6 +73,10 @@ func (r *render) Render(_ context.Context) ([]kube.Object, error) {
 
 	objs := []kube.Object{cm, dp}
 
+	if pdb := r.podDisruptionBudget(); pdb != nil {
+		objs = append(objs, pdb)
+	}
+
 	if pm := r.podMonitor(); pm != nil {
 		objs = append(objs, pm)
 	}
@@ -264,6 +270,40 @@ func (r *render) deployment() *appsv1.Deployment {
 	}
 }
 
+func (r *render) podDisruptionBudget() *policyv1.PodDisruptionBudget {
+	budget := r.pipeline.Spec.Budget
+	if budget == nil {
+		return nil
+	}
+
+	pdb := &policyv1.PodDisruptionBudget{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "policy/v1",
+			Kind:       "PodDisruptionBudget",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        r.pipeline.Name,
+			Namespace:   r.pipeline.Namespace,
+			Labels:      r.labels,
+			Annotations: r.annotations(),
+		},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			Selector: &metav1.LabelSelector{
+				MatchLabels: r.labels,
+			},
+		},
+	}
+
+	if budget.MaxUnavailable != nil {
+		pdb.Spec.MaxUnavailable = budget.MaxUnavailable
+	}
+	if budget.MinAvailable != nil {
+		pdb.Spec.MinAvailable = budget.MinAvailable
+	}
+
+	return pdb
+}
+
 // buildClusterConnectionResources returns the env vars, volumes, and volume
 // mounts needed to connect to a Redpanda cluster via clusterRef.
 func buildClusterConnectionResources(cc *clusterConnection) ([]corev1.EnvVar, []corev1.Volume, []corev1.VolumeMount) {