From a117d6c2caa6297ea8b61f3f4635ef64f20d1a10 Mon Sep 17 00:00:00 2001 From: Srinivas Chippagiri Date: Wed, 28 Oct 2026 11:42:36 -0700 Subject: [PATCH 1/4] feat(fluentd): support a PodDisruptionBudget for the Fluentd StatefulSet Add spec.podDisruptionBudget (minAvailable or maxUnavailable) to Fluentd. In collector mode, the operator creates and updates a PodDisruptionBudget selecting the StatefulSet's pods, and removes it when the field is unset or in agent mode. Signed-off-by: Srinivas Chippagiri --- apis/fluentd/v1alpha1/fluentd_types.go | 16 ++++++ .../fluent-operator-clusterRole.yaml | 11 ++++ controllers/fluent_controller_finalizer.go | 29 ++++++++++ controllers/fluentd_controller.go | 14 +++++ pkg/operator/pdb.go | 31 +++++++++++ pkg/operator/pdb_test.go | 54 +++++++++++++++++++ 6 files changed, 155 insertions(+) create mode 100644 pkg/operator/pdb.go create mode 100644 pkg/operator/pdb_test.go diff --git a/apis/fluentd/v1alpha1/fluentd_types.go b/apis/fluentd/v1alpha1/fluentd_types.go index bf02608f3..5ae61b852 100644 --- a/apis/fluentd/v1alpha1/fluentd_types.go +++ b/apis/fluentd/v1alpha1/fluentd_types.go @@ -23,6 +23,7 @@ import ( corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" ) const ( @@ -48,6 +49,9 @@ type FluentdSpec struct { // Numbers of the Fluentd instance // Applicable when the mode is "collector", and will be ignored when the mode is "agent" Replicas *int32 `json:"replicas,omitempty"` + // PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. + // Applicable when the mode is "collector", and will be ignored when the mode is "agent" + PodDisruptionBudget *PodDisruptionBudget `json:"podDisruptionBudget,omitempty"` // Numbers of the workers in Fluentd instance Workers *int32 `json:"workers,omitempty"` // Global logging verbosity @@ -201,3 +205,15 @@ type FluentdList struct { metav1.ListMeta `json:"metadata,omitempty"` Items []Fluentd `json:"items"` } + +// PodDisruptionBudget limits the number of Fluentd pods that are down at the same time during a voluntary +// disruption, such as a node drain. Set exactly one of minAvailable and maxUnavailable. +// +kubebuilder:validation:XValidation:rule="has(self.minAvailable) != has(self.maxUnavailable)",message="set exactly one of minAvailable and maxUnavailable" +type PodDisruptionBudget struct { + // Number or percentage of Fluentd pods that must stay available. + // +kubebuilder:validation:XIntOrString + MinAvailable *intstr.IntOrString `json:"minAvailable,omitempty"` + // Number or percentage of Fluentd pods that can be unavailable. + // +kubebuilder:validation:XIntOrString + MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"` +} diff --git a/charts/fluent-operator/templates/fluent-operator-clusterRole.yaml b/charts/fluent-operator/templates/fluent-operator-clusterRole.yaml index 232022c83..139e15a04 100644 --- a/charts/fluent-operator/templates/fluent-operator-clusterRole.yaml +++ b/charts/fluent-operator/templates/fluent-operator-clusterRole.yaml @@ -7,6 +7,17 @@ metadata: app.kubernetes.io/name: fluent-operator name: {{ .Values.operator.rbac.clusterRole.name }} rules: + - apiGroups: + - policy + resources: + - poddisruptionbudgets + verbs: + - create + - delete + - get + - list + - patch + - watch - apiGroups: - apps resources: diff --git a/controllers/fluent_controller_finalizer.go b/controllers/fluent_controller_finalizer.go index b1b0745a9..0712a0270 100644 --- a/controllers/fluent_controller_finalizer.go +++ b/controllers/fluent_controller_finalizer.go @@ -7,6 +7,7 @@ import ( appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrl "sigs.k8s.io/controller-runtime" @@ -111,6 +112,20 @@ func (r *FluentdReconciler) delete(ctx context.Context, fd *fluentdv1alpha1.Flue return nil } +// deletePodDisruptionBudget removes the Fluentd PodDisruptionBudget, if there is one. +func (r *FluentdReconciler) deletePodDisruptionBudget(ctx context.Context, fd *fluentdv1alpha1.Fluentd) error { + pdb := policyv1.PodDisruptionBudget{ + ObjectMeta: metav1.ObjectMeta{ + Name: fd.Name, + Namespace: fd.Namespace, + }, + } + if err := r.Delete(ctx, &pdb); err != nil && !errors.IsNotFound(err) { + return err + } + return nil +} + func (r *FluentdReconciler) mutate(obj client.Object, fd *fluentdv1alpha1.Fluentd) controllerutil.MutateFn { switch o := obj.(type) { case *rbacv1.ClusterRole: @@ -197,6 +212,20 @@ func (r *FluentdReconciler) mutate(obj client.Object, fd *fluentdv1alpha1.Fluent o.Spec.Template.Annotations["kubectl.kubernetes.io/restartedAt"] = restartedAt } + if err := ctrl.SetControllerReference(fd, o, r.Scheme); err != nil { + return err + } + return nil + } + case *policyv1.PodDisruptionBudget: + expected := operator.MakeFluentdPodDisruptionBudget(*fd) + + return func() error { + if expected == nil { + return nil + } + o.Labels = expected.Labels + o.Spec = expected.Spec if err := ctrl.SetControllerReference(fd, o, r.Scheme); err != nil { return err } diff --git a/controllers/fluentd_controller.go b/controllers/fluentd_controller.go index 7889707a4..47ddd83ad 100644 --- a/controllers/fluentd_controller.go +++ b/controllers/fluentd_controller.go @@ -24,6 +24,7 @@ import ( "github.com/go-logr/logr" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -52,6 +53,7 @@ type FluentdReconciler struct { // +kubebuilder:rbac:groups=fluentd.fluent.io,resources=fluentds/finalizers,verbs=update // +kubebuilder:rbac:groups=apps,resources=daemonsets;statefulsets,verbs=get;list;watch;create;patch;delete // +kubebuilder:rbac:groups=core,resources=secrets,verbs=get +// +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;patch;delete // +kubebuilder:rbac:groups=core,resources=serviceaccounts;services,verbs=get;list;watch;create;patch;delete // +kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles;clusterrolebindings,verbs=create;get;list;watch;patch @@ -137,6 +139,9 @@ func (r *FluentdReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct if err = r.Delete(ctx, &sts); err != nil && !errors.IsNotFound(err) { return ctrl.Result{}, err } + if err = r.deletePodDisruptionBudget(ctx, &fd); err != nil { + return ctrl.Result{}, err + } } else { // Deploy Fluentd StatefulSet sts := operator.MakeStatefulSet(fd) @@ -153,6 +158,14 @@ func (r *FluentdReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct if err = r.Delete(ctx, &ds); err != nil && !errors.IsNotFound(err) { return ctrl.Result{}, err } + // Deploy or remove the Fluentd PodDisruptionBudget + if pdb := operator.MakeFluentdPodDisruptionBudget(fd); pdb != nil { + if _, err = controllerutil.CreateOrPatch(ctx, r.Client, pdb, r.mutate(pdb, &fd)); err != nil { + return ctrl.Result{}, err + } + } else if err = r.deletePodDisruptionBudget(ctx, &fd); err != nil { + return ctrl.Result{}, err + } } // Deploy Fluentd Service if !fd.Spec.DisableService { @@ -248,5 +261,6 @@ func (r *FluentdReconciler) SetupWithManager(mgr ctrl.Manager) error { Owns(&appsv1.DaemonSet{}). Owns(&appsv1.StatefulSet{}). Owns(&corev1.Service{}). + Owns(&policyv1.PodDisruptionBudget{}). Complete(r) } diff --git a/pkg/operator/pdb.go b/pkg/operator/pdb.go new file mode 100644 index 000000000..21ae6c759 --- /dev/null +++ b/pkg/operator/pdb.go @@ -0,0 +1,31 @@ +package operator + +import ( + policyv1 "k8s.io/api/policy/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + fluentdv1alpha1 "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1" +) + +// MakeFluentdPodDisruptionBudget returns the PodDisruptionBudget for the Fluentd StatefulSet, selecting its pods with +// the StatefulSet's own selector. It returns nil when the Fluentd spec does not ask for one. +func MakeFluentdPodDisruptionBudget(fd fluentdv1alpha1.Fluentd) *policyv1.PodDisruptionBudget { + if fd.Spec.PodDisruptionBudget == nil { + return nil + } + + sts := MakeStatefulSet(fd) + + return &policyv1.PodDisruptionBudget{ + ObjectMeta: metav1.ObjectMeta{ + Name: fd.Name, + Namespace: fd.Namespace, + Labels: sts.Labels, + }, + Spec: policyv1.PodDisruptionBudgetSpec{ + MinAvailable: fd.Spec.PodDisruptionBudget.MinAvailable, + MaxUnavailable: fd.Spec.PodDisruptionBudget.MaxUnavailable, + Selector: sts.Spec.Selector, + }, + } +} diff --git a/pkg/operator/pdb_test.go b/pkg/operator/pdb_test.go new file mode 100644 index 000000000..d9fb5c3e2 --- /dev/null +++ b/pkg/operator/pdb_test.go @@ -0,0 +1,54 @@ +package operator + +import ( + "testing" + + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + + fluentdv1alpha1 "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1" +) + +func newCollectorFluentd(pdb *fluentdv1alpha1.PodDisruptionBudget) fluentdv1alpha1.Fluentd { + replicas := int32(3) + return fluentdv1alpha1.Fluentd{ + ObjectMeta: metav1.ObjectMeta{Name: "fluentd", Namespace: "logging", Labels: map[string]string{"team": "obs"}}, + Spec: fluentdv1alpha1.FluentdSpec{ + Replicas: &replicas, + PodDisruptionBudget: pdb, + }, + } +} + +func TestMakeFluentdPodDisruptionBudget(t *testing.T) { + g := NewGomegaWithT(t) + + t.Run("no budget requested", func(t *testing.T) { + g.Expect(MakeFluentdPodDisruptionBudget(newCollectorFluentd(nil))).To(BeNil()) + }) + + t.Run("minAvailable selects the statefulset pods", func(t *testing.T) { + minAvailable := intstr.FromInt32(2) + fd := newCollectorFluentd(&fluentdv1alpha1.PodDisruptionBudget{MinAvailable: &minAvailable}) + + pdb := MakeFluentdPodDisruptionBudget(fd) + + g.Expect(pdb).NotTo(BeNil()) + g.Expect(pdb.Name).To(Equal("fluentd")) + g.Expect(pdb.Namespace).To(Equal("logging")) + g.Expect(pdb.Spec.MinAvailable).To(Equal(&minAvailable)) + g.Expect(pdb.Spec.MaxUnavailable).To(BeNil()) + g.Expect(pdb.Spec.Selector).To(Equal(MakeStatefulSet(fd).Spec.Selector)) + }) + + t.Run("maxUnavailable as a percentage", func(t *testing.T) { + maxUnavailable := intstr.FromString("50%") + fd := newCollectorFluentd(&fluentdv1alpha1.PodDisruptionBudget{MaxUnavailable: &maxUnavailable}) + + pdb := MakeFluentdPodDisruptionBudget(fd) + + g.Expect(pdb.Spec.MinAvailable).To(BeNil()) + g.Expect(pdb.Spec.MaxUnavailable).To(Equal(&maxUnavailable)) + }) +} From ebf736f7fdb9467fbb2a195e421e91ce202a192a Mon Sep 17 00:00:00 2001 From: Srinivas Chippagiri Date: Wed, 28 Oct 2026 11:47:21 -0700 Subject: [PATCH 2/4] chore: regenerate deepcopy, CRDs, RBAC, manifests and docs for the Fluentd PodDisruptionBudget Signed-off-by: Srinivas Chippagiri --- .../fluentd/v1alpha1/zz_generated.deepcopy.go | 31 +++++++++++++++++ .../templates/fluentd.fluent.io_fluentds.yaml | 23 +++++++++++++ .../crds/fluentd.fluent.io_fluentds.yaml | 23 +++++++++++++ .../crd/bases/fluentd.fluent.io_fluentds.yaml | 23 +++++++++++++ config/rbac/role.yaml | 11 ++++++ docs/fluentd.md | 13 +++++++ manifests/setup/setup.yaml | 34 +++++++++++++++++++ 7 files changed, 158 insertions(+) diff --git a/apis/fluentd/v1alpha1/zz_generated.deepcopy.go b/apis/fluentd/v1alpha1/zz_generated.deepcopy.go index ee3604099..d9cdc7d27 100644 --- a/apis/fluentd/v1alpha1/zz_generated.deepcopy.go +++ b/apis/fluentd/v1alpha1/zz_generated.deepcopy.go @@ -28,6 +28,7 @@ import ( rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" ) // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. @@ -829,6 +830,11 @@ func (in *FluentdSpec) DeepCopyInto(out *FluentdSpec) { *out = new(int32) **out = **in } + if in.PodDisruptionBudget != nil { + in, out := &in.PodDisruptionBudget, &out.PodDisruptionBudget + *out = new(PodDisruptionBudget) + (*in).DeepCopyInto(*out) + } if in.Workers != nil { in, out := &in.Workers, &out.Workers *out = new(int32) @@ -1173,3 +1179,28 @@ func (in *OutputStatus) DeepCopy() *OutputStatus { in.DeepCopyInto(out) return out } + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *PodDisruptionBudget) DeepCopyInto(out *PodDisruptionBudget) { + *out = *in + if in.MinAvailable != nil { + in, out := &in.MinAvailable, &out.MinAvailable + *out = new(intstr.IntOrString) + **out = **in + } + if in.MaxUnavailable != nil { + in, out := &in.MaxUnavailable, &out.MaxUnavailable + *out = new(intstr.IntOrString) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodDisruptionBudget. +func (in *PodDisruptionBudget) DeepCopy() *PodDisruptionBudget { + if in == nil { + return nil + } + out := new(PodDisruptionBudget) + in.DeepCopyInto(out) + return out +} diff --git a/charts/fluent-operator-fluentd-crds/templates/fluentd.fluent.io_fluentds.yaml b/charts/fluent-operator-fluentd-crds/templates/fluentd.fluent.io_fluentds.yaml index 20eafc429..6cd55cd50 100644 --- a/charts/fluent-operator-fluentd-crds/templates/fluentd.fluent.io_fluentds.yaml +++ b/charts/fluent-operator-fluentd-crds/templates/fluentd.fluent.io_fluentds.yaml @@ -3171,6 +3171,29 @@ spec: type: string description: NodeSelector type: object + podDisruptionBudget: + description: |- + PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. + Applicable when the mode is "collector", and will be ignored when the mode is "agent" + properties: + maxUnavailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that can be + unavailable. + x-kubernetes-int-or-string: true + minAvailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that must stay + available. + x-kubernetes-int-or-string: true + type: object + x-kubernetes-validations: + - message: set exactly one of minAvailable and maxUnavailable + rule: has(self.minAvailable) != has(self.maxUnavailable) positionDB: description: |- Storage for position db. You will use it if tail input is enabled. diff --git a/charts/fluent-operator/crds/fluentd.fluent.io_fluentds.yaml b/charts/fluent-operator/crds/fluentd.fluent.io_fluentds.yaml index e757692c8..5e1992d8e 100644 --- a/charts/fluent-operator/crds/fluentd.fluent.io_fluentds.yaml +++ b/charts/fluent-operator/crds/fluentd.fluent.io_fluentds.yaml @@ -3169,6 +3169,29 @@ spec: type: string description: NodeSelector type: object + podDisruptionBudget: + description: |- + PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. + Applicable when the mode is "collector", and will be ignored when the mode is "agent" + properties: + maxUnavailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that can be + unavailable. + x-kubernetes-int-or-string: true + minAvailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that must stay + available. + x-kubernetes-int-or-string: true + type: object + x-kubernetes-validations: + - message: set exactly one of minAvailable and maxUnavailable + rule: has(self.minAvailable) != has(self.maxUnavailable) positionDB: description: |- Storage for position db. You will use it if tail input is enabled. diff --git a/config/crd/bases/fluentd.fluent.io_fluentds.yaml b/config/crd/bases/fluentd.fluent.io_fluentds.yaml index 480df7f19..84a6edac7 100644 --- a/config/crd/bases/fluentd.fluent.io_fluentds.yaml +++ b/config/crd/bases/fluentd.fluent.io_fluentds.yaml @@ -3170,6 +3170,29 @@ spec: type: string description: NodeSelector type: object + podDisruptionBudget: + description: |- + PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. + Applicable when the mode is "collector", and will be ignored when the mode is "agent" + properties: + maxUnavailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that can be + unavailable. + x-kubernetes-int-or-string: true + minAvailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that must stay + available. + x-kubernetes-int-or-string: true + type: object + x-kubernetes-validations: + - message: set exactly one of minAvailable and maxUnavailable + rule: has(self.minAvailable) != has(self.maxUnavailable) positionDB: description: |- Storage for position db. You will use it if tail input is enabled. diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index abcbd7d48..a97736b0a 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -118,6 +118,17 @@ rules: - fluentds/finalizers verbs: - update +- apiGroups: + - policy + resources: + - poddisruptionbudgets + verbs: + - create + - delete + - get + - list + - patch + - watch - apiGroups: - rbac.authorization.k8s.io resources: diff --git a/docs/fluentd.md b/docs/fluentd.md index 3490e5d60..cca584b4d 100644 --- a/docs/fluentd.md +++ b/docs/fluentd.md @@ -35,6 +35,7 @@ This Document documents the types introduced by the fluentd Operator. * [Output](#output) * [OutputList](#outputlist) * [OutputSpec](#outputspec) +* [PodDisruptionBudget](#poddisruptionbudget) # BufferVolume @@ -337,6 +338,7 @@ FluentdSpec defines the desired state of Fluentd | defaultOutputSelector | Select cluster output plugins used to send all logs that did not match any route to the matching outputs | *[metav1.LabelSelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.31/#labelselector-v1-meta) | | disableService | By default will build the related service according to the globalinputs definition. | bool | | replicas | Numbers of the Fluentd instance Applicable when the mode is \"collector\", and will be ignored when the mode is \"agent\" | *int32 | +| podDisruptionBudget | PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. Applicable when the mode is \"collector\", and will be ignored when the mode is \"agent\" | *[PodDisruptionBudget](#poddisruptionbudget) | | workers | Numbers of the workers in Fluentd instance | *int32 | | logLevel | Global logging verbosity | string | | image | Fluentd image. | string | @@ -447,3 +449,14 @@ OutputSpec defines the desired state of Output | outputs | | [][output.Output](plugins/fluentd/output/output.md) | [Back to TOC](#table-of-contents) +# PodDisruptionBudget + +PodDisruptionBudget limits the number of Fluentd pods that are down at the same time during a voluntary disruption, such as a node drain. Set exactly one of minAvailable and maxUnavailable. + + +| Field | Description | Scheme | +| ----- | ----------- | ------ | +| minAvailable | Number or percentage of Fluentd pods that must stay available. | *intstr.IntOrString | +| maxUnavailable | Number or percentage of Fluentd pods that can be unavailable. | *intstr.IntOrString | + +[Back to TOC](#table-of-contents) diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml index 61fa9328c..eb9a06840 100644 --- a/manifests/setup/setup.yaml +++ b/manifests/setup/setup.yaml @@ -29525,6 +29525,29 @@ spec: type: string description: NodeSelector type: object + podDisruptionBudget: + description: |- + PodDisruptionBudget for the Fluentd pods, created and kept in sync by the operator. + Applicable when the mode is "collector", and will be ignored when the mode is "agent" + properties: + maxUnavailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that can be + unavailable. + x-kubernetes-int-or-string: true + minAvailable: + anyOf: + - type: integer + - type: string + description: Number or percentage of Fluentd pods that must stay + available. + x-kubernetes-int-or-string: true + type: object + x-kubernetes-validations: + - message: set exactly one of minAvailable and maxUnavailable + rule: has(self.minAvailable) != has(self.maxUnavailable) positionDB: description: |- Storage for position db. You will use it if tail input is enabled. @@ -43703,6 +43726,17 @@ rules: - fluentds/finalizers verbs: - update +- apiGroups: + - policy + resources: + - poddisruptionbudgets + verbs: + - create + - delete + - get + - list + - patch + - watch - apiGroups: - rbac.authorization.k8s.io resources: From fdb52bf45ab5bc37addc43e3b6a8ba4d22f7d136 Mon Sep 17 00:00:00 2001 From: Srinivas Chippagiri Date: Thu, 29 Oct 2026 11:54:34 -0700 Subject: [PATCH 3/4] fix(fluentd): only delete a PodDisruptionBudget the operator controls deletePodDisruptionBudget deleted any PodDisruptionBudget named after the Fluentd resource, on every reconcile without a budget and always in agent mode. A PodDisruptionBudget written by hand under that name, the workaround before the operator could manage one, would be deleted on upgrade. Read it from the cache and only delete it when the Fluentd resource controls it. Also create a Gomega instance per subtest in the PodDisruptionBudget tests, so failures are reported against the right subtest. Signed-off-by: Srinivas Chippagiri --- controllers/fluent_controller_finalizer.go | 14 ++- controllers/fluentd_pdb_test.go | 102 +++++++++++++++++++++ pkg/operator/pdb_test.go | 5 +- 3 files changed, 114 insertions(+), 7 deletions(-) create mode 100644 controllers/fluentd_pdb_test.go diff --git a/controllers/fluent_controller_finalizer.go b/controllers/fluent_controller_finalizer.go index 0712a0270..75485c5d8 100644 --- a/controllers/fluent_controller_finalizer.go +++ b/controllers/fluent_controller_finalizer.go @@ -113,12 +113,16 @@ func (r *FluentdReconciler) delete(ctx context.Context, fd *fluentdv1alpha1.Flue } // deletePodDisruptionBudget removes the Fluentd PodDisruptionBudget, if there is one. +// deletePodDisruptionBudget removes the PodDisruptionBudget the operator created for fd. A PodDisruptionBudget +// with the same name that the operator doesn't control, such as one written by hand before the operator could +// manage it, is left alone. PodDisruptionBudgets are watched, so the Get is served from the cache. func (r *FluentdReconciler) deletePodDisruptionBudget(ctx context.Context, fd *fluentdv1alpha1.Fluentd) error { - pdb := policyv1.PodDisruptionBudget{ - ObjectMeta: metav1.ObjectMeta{ - Name: fd.Name, - Namespace: fd.Namespace, - }, + var pdb policyv1.PodDisruptionBudget + if err := r.Get(ctx, client.ObjectKey{Namespace: fd.Namespace, Name: fd.Name}, &pdb); err != nil { + return client.IgnoreNotFound(err) + } + if !metav1.IsControlledBy(&pdb, fd) { + return nil } if err := r.Delete(ctx, &pdb); err != nil && !errors.IsNotFound(err) { return err diff --git a/controllers/fluentd_pdb_test.go b/controllers/fluentd_pdb_test.go new file mode 100644 index 000000000..bc3beec00 --- /dev/null +++ b/controllers/fluentd_pdb_test.go @@ -0,0 +1,102 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controllers + +import ( + "context" + "testing" + + policyv1 "k8s.io/api/policy/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/scheme" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + fluentdv1alpha1 "github.com/fluent/fluent-operator/v3/apis/fluentd/v1alpha1" +) + +func newFluentdTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := scheme.AddToScheme(s); err != nil { + t.Fatalf("failed to add client-go scheme: %v", err) + } + if err := fluentdv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("failed to add fluentd scheme: %v", err) + } + return s +} + +// The operator only deletes a PodDisruptionBudget it controls: one written by hand +// under the Fluentd resource's name, the workaround before the operator could manage +// one, survives turning the feature off and running in agent mode. +func TestDeletePodDisruptionBudgetOnlyDeletesItsOwn(t *testing.T) { + s := newFluentdTestScheme(t) + fd := &fluentdv1alpha1.Fluentd{ + ObjectMeta: metav1.ObjectMeta{Name: "fluentd", Namespace: "logging", UID: "fluentd-uid"}, + } + key := client.ObjectKey{Namespace: fd.Namespace, Name: fd.Name} + + for _, tc := range []struct { + name string + pdb *policyv1.PodDisruptionBudget + wantDeleted bool + }{ + {name: "no PodDisruptionBudget"}, + { + name: "owned by the Fluentd resource", + pdb: &policyv1.PodDisruptionBudget{ObjectMeta: metav1.ObjectMeta{Name: fd.Name, Namespace: fd.Namespace}}, + wantDeleted: true, + }, + { + name: "written by hand", + pdb: &policyv1.PodDisruptionBudget{ObjectMeta: metav1.ObjectMeta{Name: fd.Name, Namespace: fd.Namespace}}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + builder := fake.NewClientBuilder().WithScheme(s) + if tc.pdb != nil { + if tc.wantDeleted { + if err := ctrl.SetControllerReference(fd, tc.pdb, s); err != nil { + t.Fatal(err) + } + } + builder = builder.WithObjects(tc.pdb) + } + cli := builder.Build() + r := &FluentdReconciler{Client: cli, Log: logf.Log, Scheme: s} + + if err := r.deletePodDisruptionBudget(context.Background(), fd); err != nil { + t.Fatalf("deletePodDisruptionBudget returned error: %v", err) + } + + err := cli.Get(context.Background(), key, &policyv1.PodDisruptionBudget{}) + switch { + case tc.pdb == nil || tc.wantDeleted: + if !errors.IsNotFound(err) { + t.Fatalf("PodDisruptionBudget still exists: %v", err) + } + case err != nil: + t.Fatalf("PodDisruptionBudget the operator does not control was deleted: %v", err) + } + }) + } +} diff --git a/pkg/operator/pdb_test.go b/pkg/operator/pdb_test.go index d9fb5c3e2..5f3330619 100644 --- a/pkg/operator/pdb_test.go +++ b/pkg/operator/pdb_test.go @@ -22,13 +22,13 @@ func newCollectorFluentd(pdb *fluentdv1alpha1.PodDisruptionBudget) fluentdv1alph } func TestMakeFluentdPodDisruptionBudget(t *testing.T) { - g := NewGomegaWithT(t) - t.Run("no budget requested", func(t *testing.T) { + g := NewGomegaWithT(t) g.Expect(MakeFluentdPodDisruptionBudget(newCollectorFluentd(nil))).To(BeNil()) }) t.Run("minAvailable selects the statefulset pods", func(t *testing.T) { + g := NewGomegaWithT(t) minAvailable := intstr.FromInt32(2) fd := newCollectorFluentd(&fluentdv1alpha1.PodDisruptionBudget{MinAvailable: &minAvailable}) @@ -43,6 +43,7 @@ func TestMakeFluentdPodDisruptionBudget(t *testing.T) { }) t.Run("maxUnavailable as a percentage", func(t *testing.T) { + g := NewGomegaWithT(t) maxUnavailable := intstr.FromString("50%") fd := newCollectorFluentd(&fluentdv1alpha1.PodDisruptionBudget{MaxUnavailable: &maxUnavailable}) From ea9e307e532dd1f67712f916e8db34c65e80b08a Mon Sep 17 00:00:00 2001 From: Srinivas Chippagiri Date: Thu, 29 Oct 2026 19:18:49 -0700 Subject: [PATCH 4/4] Remove the stale deletePodDisruptionBudget doc comment Signed-off-by: Srinivas Chippagiri --- controllers/fluent_controller_finalizer.go | 1 - 1 file changed, 1 deletion(-) diff --git a/controllers/fluent_controller_finalizer.go b/controllers/fluent_controller_finalizer.go index 75485c5d8..e7defa08b 100644 --- a/controllers/fluent_controller_finalizer.go +++ b/controllers/fluent_controller_finalizer.go @@ -112,7 +112,6 @@ func (r *FluentdReconciler) delete(ctx context.Context, fd *fluentdv1alpha1.Flue return nil } -// deletePodDisruptionBudget removes the Fluentd PodDisruptionBudget, if there is one. // deletePodDisruptionBudget removes the PodDisruptionBudget the operator created for fd. A PodDisruptionBudget // with the same name that the operator doesn't control, such as one written by hand before the operator could // manage it, is left alone. PodDisruptionBudgets are watched, so the Get is served from the cache.