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/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/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/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/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/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: 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)) + }) +}