feat(fluentd): support a PodDisruptionBudget for the Fluentd StatefulSet

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"`
+}

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.

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.

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:

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.

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:

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
 			}

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)
 }

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)

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: