diff --git a/api/v1alpha2/etcddefragpolicy_types.go b/api/v1alpha2/etcddefragpolicy_types.go new file mode 100644 index 00000000..a16e8d15 --- /dev/null +++ b/api/v1alpha2/etcddefragpolicy_types.go @@ -0,0 +1,143 @@ +/* +Copyright 2023 Timofey Larkin. + +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 v1alpha2 + +import ( + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// ConcurrencyPolicy decides what a due tick does when a run stamped by the same +// policy is still in flight. +// +kubebuilder:validation:Enum=Allow;Forbid +type ConcurrencyPolicy string + +const ( + // AllowConcurrent stamps the due run even while a previous one is still + // active. EtcdDefrag serializes per cluster on its own, so the new run + // simply queues behind the active one. + AllowConcurrent ConcurrencyPolicy = "Allow" + // ForbidConcurrent skips the due tick while a run stamped by this policy is + // still active, rather than letting runs pile up. The default. + ForbidConcurrent ConcurrencyPolicy = "Forbid" +) + +// EtcdDefragPolicySpec is the desired state of an EtcdDefragPolicy: a recurring +// schedule that stamps out EtcdDefrag runs against one EtcdCluster, so the +// operator absorbs the cadence instead of relying on an external CronJob. +type EtcdDefragPolicySpec struct { + // ClusterRef names the EtcdCluster (same namespace) each stamped EtcdDefrag + // targets. + ClusterRef corev1.LocalObjectReference `json:"clusterRef"` + + // Schedule is a standard five-field cron expression, interpreted in UTC, + // naming when a run is stamped (e.g. "0 3 * * *" for 03:00 daily). + // +kubebuilder:validation:MinLength=1 + Schedule string `json:"schedule"` + + // Suspend pauses stamping. Runs already in flight are left alone; clearing + // it resumes at the next scheduled tick (missed ticks are not backfilled). + // +optional + Suspend *bool `json:"suspend,omitempty"` + + // ConcurrencyPolicy decides what a due tick does when a previous stamped run + // is still active. Defaults to Forbid. + // +optional + ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"` + + // StartingDeadlineSeconds bounds how late a missed tick may still be started. + // If the operator was down (or the tick forbidden) and more than this many + // seconds have passed since the scheduled time, that tick is skipped rather + // than started late. Absent means no deadline. + // +kubebuilder:validation:Minimum=0 + // +optional + StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"` + + // HistoryLimit caps how many finished (Complete/Failed) EtcdDefrags stamped + // by this policy are retained; the oldest beyond the limit are deleted. + // Absent leaves cleanup to each run's ttlSecondsAfterFinished. + // +kubebuilder:validation:Minimum=0 + // +optional + HistoryLimit *int32 `json:"historyLimit,omitempty"` + + // Rule is stamped verbatim into each EtcdDefrag; it decides which members a + // run touches. Absent stamps runs with no rule (the default gate). + // +optional + Rule *DefragRule `json:"rule,omitempty"` + + // TTLSecondsAfterFinished is stamped into each EtcdDefrag so a stamped run + // garbage-collects itself once finished. Complements HistoryLimit. + // +kubebuilder:validation:Minimum=0 + // +optional + TTLSecondsAfterFinished *int32 `json:"ttlSecondsAfterFinished,omitempty"` +} + +// EtcdDefragPolicyStatus is the observed state of an EtcdDefragPolicy. +type EtcdDefragPolicyStatus struct { + // LastScheduleTime is the scheduled time of the most recent tick the policy + // acted on (stamped or deliberately skipped). It anchors the next tick, so a + // tick is never acted on twice. + // +optional + LastScheduleTime *metav1.Time `json:"lastScheduleTime,omitempty"` + + // LastSuccessfulTime is when a stamped run most recently reached Complete. + // +optional + LastSuccessfulTime *metav1.Time `json:"lastSuccessfulTime,omitempty"` + + // Active references the stamped EtcdDefrags that have not yet finished. + // +optional + // +listType=atomic + Active []corev1.LocalObjectReference `json:"active,omitempty"` + + // Conditions represent the latest observations — notably why stamping is + // paused (Suspended) or not happening (InvalidSchedule). + // +optional + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// +kubebuilder:object:root=true +// +kubebuilder:subresource:status +// +kubebuilder:printcolumn:name="Cluster",type=string,JSONPath=`.spec.clusterRef.name` +// +kubebuilder:printcolumn:name="Schedule",type=string,JSONPath=`.spec.schedule` +// +kubebuilder:printcolumn:name="Suspend",type=boolean,JSONPath=`.spec.suspend` +// +kubebuilder:printcolumn:name="Last Schedule",type=date,JSONPath=`.status.lastScheduleTime` +// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp` + +// EtcdDefragPolicy is the Schema for the etcddefragpolicies API. It stamps out +// EtcdDefrag runs on a cron schedule so the operator drives recurring +// defragmentation itself. Each run is a discrete, auditable EtcdDefrag owned by +// the policy. +type EtcdDefragPolicy struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec EtcdDefragPolicySpec `json:"spec,omitempty"` + Status EtcdDefragPolicyStatus `json:"status,omitempty"` +} + +// +kubebuilder:object:root=true + +// EtcdDefragPolicyList contains a list of EtcdDefragPolicy. +type EtcdDefragPolicyList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []EtcdDefragPolicy `json:"items"` +} + +func init() { + SchemeBuilder.Register(&EtcdDefragPolicy{}, &EtcdDefragPolicyList{}) +} diff --git a/api/v1alpha2/zz_generated.deepcopy.go b/api/v1alpha2/zz_generated.deepcopy.go index 0c2cca5b..8a3ebe45 100644 --- a/api/v1alpha2/zz_generated.deepcopy.go +++ b/api/v1alpha2/zz_generated.deepcopy.go @@ -414,6 +414,141 @@ func (in *EtcdDefragList) DeepCopyObject() runtime.Object { return nil } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EtcdDefragPolicy) DeepCopyInto(out *EtcdDefragPolicy) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EtcdDefragPolicy. +func (in *EtcdDefragPolicy) DeepCopy() *EtcdDefragPolicy { + if in == nil { + return nil + } + out := new(EtcdDefragPolicy) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EtcdDefragPolicy) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EtcdDefragPolicyList) DeepCopyInto(out *EtcdDefragPolicyList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]EtcdDefragPolicy, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EtcdDefragPolicyList. +func (in *EtcdDefragPolicyList) DeepCopy() *EtcdDefragPolicyList { + if in == nil { + return nil + } + out := new(EtcdDefragPolicyList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EtcdDefragPolicyList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EtcdDefragPolicySpec) DeepCopyInto(out *EtcdDefragPolicySpec) { + *out = *in + out.ClusterRef = in.ClusterRef + if in.Suspend != nil { + in, out := &in.Suspend, &out.Suspend + *out = new(bool) + **out = **in + } + if in.StartingDeadlineSeconds != nil { + in, out := &in.StartingDeadlineSeconds, &out.StartingDeadlineSeconds + *out = new(int64) + **out = **in + } + if in.HistoryLimit != nil { + in, out := &in.HistoryLimit, &out.HistoryLimit + *out = new(int32) + **out = **in + } + if in.Rule != nil { + in, out := &in.Rule, &out.Rule + *out = new(DefragRule) + (*in).DeepCopyInto(*out) + } + if in.TTLSecondsAfterFinished != nil { + in, out := &in.TTLSecondsAfterFinished, &out.TTLSecondsAfterFinished + *out = new(int32) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EtcdDefragPolicySpec. +func (in *EtcdDefragPolicySpec) DeepCopy() *EtcdDefragPolicySpec { + if in == nil { + return nil + } + out := new(EtcdDefragPolicySpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EtcdDefragPolicyStatus) DeepCopyInto(out *EtcdDefragPolicyStatus) { + *out = *in + if in.LastScheduleTime != nil { + in, out := &in.LastScheduleTime, &out.LastScheduleTime + *out = (*in).DeepCopy() + } + if in.LastSuccessfulTime != nil { + in, out := &in.LastSuccessfulTime, &out.LastSuccessfulTime + *out = (*in).DeepCopy() + } + if in.Active != nil { + in, out := &in.Active, &out.Active + *out = make([]v1.LocalObjectReference, len(*in)) + copy(*out, *in) + } + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EtcdDefragPolicyStatus. +func (in *EtcdDefragPolicyStatus) DeepCopy() *EtcdDefragPolicyStatus { + if in == nil { + return nil + } + out := new(EtcdDefragPolicyStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *EtcdDefragSpec) DeepCopyInto(out *EtcdDefragSpec) { *out = *in diff --git a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml new file mode 100644 index 00000000..9f188cea --- /dev/null +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml @@ -0,0 +1,282 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.18.0 + name: etcddefragpolicies.etcd-operator.cozystack.io +spec: + group: etcd-operator.cozystack.io + names: + kind: EtcdDefragPolicy + listKind: EtcdDefragPolicyList + plural: etcddefragpolicies + singular: etcddefragpolicy + scope: Namespaced + versions: + - additionalPrinterColumns: + - jsonPath: .spec.clusterRef.name + name: Cluster + type: string + - jsonPath: .spec.schedule + name: Schedule + type: string + - jsonPath: .spec.suspend + name: Suspend + type: boolean + - jsonPath: .status.lastScheduleTime + name: Last Schedule + type: date + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha2 + schema: + openAPIV3Schema: + description: |- + EtcdDefragPolicy is the Schema for the etcddefragpolicies API. It stamps out + EtcdDefrag runs on a cron schedule so the operator drives recurring + defragmentation itself. Each run is a discrete, auditable EtcdDefrag owned by + the policy. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: |- + EtcdDefragPolicySpec is the desired state of an EtcdDefragPolicy: a recurring + schedule that stamps out EtcdDefrag runs against one EtcdCluster, so the + operator absorbs the cadence instead of relying on an external CronJob. + properties: + clusterRef: + description: |- + ClusterRef names the EtcdCluster (same namespace) each stamped EtcdDefrag + targets. + properties: + name: + default: "" + description: |- + Name of the referent. + This field is effectively required, but due to backwards compatibility is + allowed to be empty. Instances of this type with an empty value here are + almost certainly wrong. + More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names + type: string + type: object + x-kubernetes-map-type: atomic + concurrencyPolicy: + description: |- + ConcurrencyPolicy decides what a due tick does when a previous stamped run + is still active. Defaults to Forbid. + enum: + - Allow + - Forbid + type: string + historyLimit: + description: |- + HistoryLimit caps how many finished (Complete/Failed) EtcdDefrags stamped + by this policy are retained; the oldest beyond the limit are deleted. + Absent leaves cleanup to each run's ttlSecondsAfterFinished. + format: int32 + minimum: 0 + type: integer + rule: + description: |- + Rule is stamped verbatim into each EtcdDefrag; it decides which members a + run touches. Absent stamps runs with no rule (the default gate). + properties: + all: + description: |- + All defragments every member unconditionally, regardless of size — the + explicit "do it now". Mutually exclusive with the threshold fields below. + type: boolean + freeSpaceAbove: + anyOf: + - type: integer + - type: string + description: |- + FreeSpaceAbove defragments a member whose reclaimable space + (DbSize-DbSizeInUse) exceeds this. The primary, always-applied gate. + Absent means the built-in default (200Mi). + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + minReclaim: + anyOf: + - type: integer + - type: string + description: |- + MinReclaim floors the quota arm: even under quota pressure, skip a member + that would reclaim less than this. Only meaningful with QuotaUsageAbove, + and must not exceed FreeSpaceAbove. Absent means the built-in default + (32Mi). + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + quotaUsageAbove: + description: |- + QuotaUsageAbove: when DbSize exceeds this fraction of the backend quota + (approaching NOSPACE), lower the reclaimable floor to MinReclaim so small + wins are taken under pressure. A member is never defragmented when its + reclaimable space is below MinReclaim. Integer percent 1..100 with a "%" + suffix, e.g. "80%". + pattern: ^([1-9][0-9]?|100)%$ + type: string + type: object + x-kubernetes-validations: + - message: rule.all cannot be combined with freeSpaceAbove/quotaUsageAbove/minReclaim + rule: '!(has(self.all) && self.all) || (!has(self.freeSpaceAbove) + && !has(self.quotaUsageAbove) && !has(self.minReclaim))' + - message: freeSpaceAbove must be greater than 0 + rule: '!has(self.freeSpaceAbove) || quantity(string(self.freeSpaceAbove)).isGreaterThan(quantity(''0''))' + - message: minReclaim must be greater than 0 + rule: '!has(self.minReclaim) || quantity(string(self.minReclaim)).isGreaterThan(quantity(''0''))' + - message: minReclaim is only meaningful with quotaUsageAbove + rule: '!has(self.minReclaim) || has(self.quotaUsageAbove)' + - message: minReclaim must not exceed freeSpaceAbove + rule: '!(has(self.minReclaim) && has(self.freeSpaceAbove)) || quantity(string(self.minReclaim)).compareTo(quantity(string(self.freeSpaceAbove))) + <= 0' + schedule: + description: |- + Schedule is a standard five-field cron expression, interpreted in UTC, + naming when a run is stamped (e.g. "0 3 * * *" for 03:00 daily). + minLength: 1 + type: string + startingDeadlineSeconds: + description: |- + StartingDeadlineSeconds bounds how late a missed tick may still be started. + If the operator was down (or the tick forbidden) and more than this many + seconds have passed since the scheduled time, that tick is skipped rather + than started late. Absent means no deadline. + format: int64 + minimum: 0 + type: integer + suspend: + description: |- + Suspend pauses stamping. Runs already in flight are left alone; clearing + it resumes at the next scheduled tick (missed ticks are not backfilled). + type: boolean + ttlSecondsAfterFinished: + description: |- + TTLSecondsAfterFinished is stamped into each EtcdDefrag so a stamped run + garbage-collects itself once finished. Complements HistoryLimit. + format: int32 + minimum: 0 + type: integer + required: + - clusterRef + - schedule + type: object + status: + description: EtcdDefragPolicyStatus is the observed state of an EtcdDefragPolicy. + properties: + active: + description: Active references the stamped EtcdDefrags that have not + yet finished. + items: + description: |- + LocalObjectReference contains enough information to let you locate the + referenced object inside the same namespace. + properties: + name: + default: "" + description: |- + Name of the referent. + This field is effectively required, but due to backwards compatibility is + allowed to be empty. Instances of this type with an empty value here are + almost certainly wrong. + More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names + type: string + type: object + x-kubernetes-map-type: atomic + type: array + x-kubernetes-list-type: atomic + conditions: + description: |- + Conditions represent the latest observations — notably why stamping is + paused (Suspended) or not happening (InvalidSchedule). + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + lastScheduleTime: + description: |- + LastScheduleTime is the scheduled time of the most recent tick the policy + acted on (stamped or deliberately skipped). It anchors the next tick, so a + tick is never acted on twice. + format: date-time + type: string + lastSuccessfulTime: + description: LastSuccessfulTime is when a stamped run most recently + reached Complete. + format: date-time + type: string + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/charts/etcd-operator/files/manager-role-rules.yaml b/charts/etcd-operator/files/manager-role-rules.yaml index fe21bed2..bd1b889d 100644 --- a/charts/etcd-operator/files/manager-role-rules.yaml +++ b/charts/etcd-operator/files/manager-role-rules.yaml @@ -94,6 +94,7 @@ - etcd-operator.cozystack.io resources: - etcdclusters/status + - etcddefragpolicies/status - etcddefrags/status - etcdmembers/status - etcdsnapshots/status @@ -104,9 +105,8 @@ - apiGroups: - etcd-operator.cozystack.io resources: - - etcddefrags + - etcddefragpolicies verbs: - - delete - get - list - patch @@ -115,6 +115,7 @@ - apiGroups: - etcd-operator.cozystack.io resources: + - etcddefrags - etcdmembers - etcdsnapshots verbs: diff --git a/controllers/etcddefragpolicy_controller.go b/controllers/etcddefragpolicy_controller.go new file mode 100644 index 00000000..b07d8f10 --- /dev/null +++ b/controllers/etcddefragpolicy_controller.go @@ -0,0 +1,344 @@ +/* +Copyright 2023 Timofey Larkin. + +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 +*/ + +package controllers + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/robfig/cron/v3" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/record" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/log" + + lll "github.com/cozystack/etcd-operator/api/v1alpha2" +) + +const ( + // defragPolicyCondition is the single condition type on an EtcdDefragPolicy: + // True while the policy is actively scheduling, False (with the reason) when + // suspended or holding an unparseable schedule. + defragPolicyCondition = "Active" + + // defragPolicyMaxCatchup bounds how many missed ticks the controller walks + // after being down; beyond it a long backlog is collapsed into a single run + // rather than replaying every slot. + defragPolicyMaxCatchup = 100 +) + +// EtcdDefragPolicyReconciler stamps out EtcdDefrag runs on a cron schedule so +// the operator drives recurring defragmentation itself. Each run is a discrete +// EtcdDefrag owned by the policy (so it cascades on delete) and labelled with +// the policy name (so the controller can find its own runs). +type EtcdDefragPolicyReconciler struct { + client.Client + Scheme *runtime.Scheme + + // Recorder emits scheduling events. Tests may leave it nil. + Recorder record.EventRecorder + + // now is the clock, overridable in tests. nil means time.Now. + now func() time.Time +} + +//+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcddefragpolicies,verbs=get;list;watch;update;patch +//+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcddefragpolicies/status,verbs=get;update;patch +//+kubebuilder:rbac:groups=etcd-operator.cozystack.io,resources=etcddefrags,verbs=get;list;watch;create;delete + +func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + logger := log.FromContext(ctx) + + pol := &lll.EtcdDefragPolicy{} + if err := r.Get(ctx, req.NamespacedName, pol); err != nil { + return ctrl.Result{}, client.IgnoreNotFound(err) + } + + // Observe the runs this policy owns. + var runs lll.EtcdDefragList + if err := r.List(ctx, &runs, client.InNamespace(pol.Namespace), + client.MatchingLabels{LabelDefragPolicy: pol.Name}); err != nil { + return ctrl.Result{}, err + } + active, finished := partitionDefragRuns(runs.Items) + pol.Status.Active = defragRunRefs(active) + if t := latestSuccessfulTime(finished); t != nil { + pol.Status.LastSuccessfulTime = t + } + + // Trim finished history to HistoryLimit (each run's own + // ttlSecondsAfterFinished is the other cleanup path). + if pol.Spec.HistoryLimit != nil { + if err := r.gcHistory(ctx, finished, int(*pol.Spec.HistoryLimit)); err != nil { + return ctrl.Result{}, err + } + } + + if pol.Spec.Suspend != nil && *pol.Spec.Suspend { + setDefragPolicyCondition(pol, metav1.ConditionFalse, "Suspended", "scheduling is suspended") + return ctrl.Result{}, r.Status().Update(ctx, pol) + } + + sched, err := parseUTCSchedule(pol.Spec.Schedule) + if err != nil { + setDefragPolicyCondition(pol, metav1.ConditionFalse, "InvalidSchedule", + fmt.Sprintf("cannot parse schedule %q: %v", pol.Spec.Schedule, err)) + // Only a spec change can fix this; the watch re-triggers, so don't requeue. + return ctrl.Result{}, r.Status().Update(ctx, pol) + } + setDefragPolicyCondition(pol, metav1.ConditionTrue, "Scheduled", "policy is scheduling runs") + + now := r.clock() + earliest := pol.CreationTimestamp.Time + if pol.Status.LastScheduleTime != nil { + earliest = pol.Status.LastScheduleTime.Time + } + due, next := nextSchedule(sched, earliest, now, defragPolicyMaxCatchup) + + if due == nil { + if err := r.Status().Update(ctx, pol); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: requeueFor(next, now)}, nil + } + tick := *due + + // A tick too far in the past (operator was down, or held by Forbid) is + // skipped rather than started late. + if d := pol.Spec.StartingDeadlineSeconds; d != nil && now.Sub(tick) > time.Duration(*d)*time.Second { + r.event(pol, corev1.EventTypeWarning, "MissedSchedule", + fmt.Sprintf("skipped scheduled time %s: past the %ds starting deadline", tickString(tick), *d)) + pol.Status.LastScheduleTime = &metav1.Time{Time: tick} + if err := r.Status().Update(ctx, pol); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: requeueFor(next, now)}, nil + } + + if concurrencyPolicy(pol) == lll.ForbidConcurrent && len(active) > 0 { + r.event(pol, corev1.EventTypeNormal, "ConcurrencyForbidden", + fmt.Sprintf("skipped scheduled time %s: %d run(s) still active", tickString(tick), len(active))) + pol.Status.LastScheduleTime = &metav1.Time{Time: tick} + if err := r.Status().Update(ctx, pol); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: requeueFor(next, now)}, nil + } + + run := r.buildDefrag(pol, tick) + if err := controllerutil.SetControllerReference(pol, run, r.Scheme); err != nil { + return ctrl.Result{}, err + } + if err := r.Create(ctx, run); err != nil { + if !apierrors.IsAlreadyExists(err) { + return ctrl.Result{}, err + } + // The deterministic name means a re-reconcile of the same tick is a + // no-op rather than a duplicate run. + logger.Info("defrag already stamped for this tick", "tick", tickString(tick), "name", run.Name) + } else { + r.event(pol, corev1.EventTypeNormal, "StampedRun", + fmt.Sprintf("stamped EtcdDefrag %q for scheduled time %s", run.Name, tickString(tick))) + pol.Status.Active = append(pol.Status.Active, corev1.LocalObjectReference{Name: run.Name}) + } + pol.Status.LastScheduleTime = &metav1.Time{Time: tick} + if err := r.Status().Update(ctx, pol); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: requeueFor(next, now)}, nil +} + +// buildDefrag renders the EtcdDefrag stamped for a tick. The name is +// deterministic in the scheduled time so a re-reconcile of the same tick +// collides (IsAlreadyExists) instead of double-stamping. +func (r *EtcdDefragPolicyReconciler) buildDefrag(pol *lll.EtcdDefragPolicy, tick time.Time) *lll.EtcdDefrag { + return &lll.EtcdDefrag{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("%s-%d", pol.Name, tick.Unix()), + Namespace: pol.Namespace, + Labels: map[string]string{ + LabelDefragPolicy: pol.Name, + LabelCluster: pol.Spec.ClusterRef.Name, + }, + }, + Spec: lll.EtcdDefragSpec{ + ClusterRef: pol.Spec.ClusterRef, + Rule: pol.Spec.Rule.DeepCopy(), + TTLSecondsAfterFinished: copyInt32(pol.Spec.TTLSecondsAfterFinished), + }, + } +} + +func (r *EtcdDefragPolicyReconciler) gcHistory(ctx context.Context, finished []lll.EtcdDefrag, limit int) error { + if len(finished) <= limit { + return nil + } + sort.Slice(finished, func(i, j int) bool { + return defragFinishTime(&finished[i]).Before(defragFinishTime(&finished[j])) + }) + for i := 0; i < len(finished)-limit; i++ { + if err := r.Delete(ctx, &finished[i]); err != nil && !apierrors.IsNotFound(err) { + return err + } + } + return nil +} + +func (r *EtcdDefragPolicyReconciler) clock() time.Time { + if r.now != nil { + return r.now() + } + return time.Now() +} + +func (r *EtcdDefragPolicyReconciler) event(obj client.Object, eventType, reason, msg string) { + if r.Recorder != nil { + r.Recorder.Event(obj, eventType, reason, msg) + } +} + +func (r *EtcdDefragPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error { + if r.now == nil { + r.now = time.Now + } + return ctrl.NewControllerManagedBy(mgr). + For(&lll.EtcdDefragPolicy{}). + Owns(&lll.EtcdDefrag{}). + Complete(r) +} + +// ── pure helpers ──────────────────────────────────────────────────────────── + +// parseUTCSchedule parses a standard five-field cron expression in UTC. A +// user-supplied CRON_TZ/TZ prefix is honoured as-is; otherwise UTC is forced so +// the schedule does not silently follow the operator process's local zone. +func parseUTCSchedule(schedule string) (cron.Schedule, error) { + spec := strings.TrimSpace(schedule) + if !strings.Contains(spec, "TZ=") { + spec = "CRON_TZ=UTC " + spec + } + return cron.ParseStandard(spec) +} + +// nextSchedule returns the most recent scheduled time at or before now that is +// strictly after earliest (nil if the next tick is still in the future), and +// the next tick after now. A backlog longer than maxCatchup is collapsed into a +// single run stamped at now. +func nextSchedule(sched cron.Schedule, earliest, now time.Time, maxCatchup int) (due *time.Time, next time.Time) { + t := sched.Next(earliest) + if t.After(now) { + return nil, t + } + last := t + for n := 0; ; n++ { + t = sched.Next(t) + if t.After(now) { + break + } + if n >= maxCatchup { + last = now + return &last, sched.Next(now) + } + last = t + } + return &last, t +} + +// requeueFor is the delay until next, floored so a just-passed boundary still +// yields a positive requeue. +func requeueFor(next, now time.Time) time.Duration { + if d := next.Sub(now); d > 0 { + return d + } + return time.Second +} + +func concurrencyPolicy(pol *lll.EtcdDefragPolicy) lll.ConcurrencyPolicy { + if pol.Spec.ConcurrencyPolicy == "" { + return lll.ForbidConcurrent + } + return pol.Spec.ConcurrencyPolicy +} + +func partitionDefragRuns(items []lll.EtcdDefrag) (active, finished []lll.EtcdDefrag) { + for i := range items { + switch items[i].Status.Phase { + case lll.EtcdDefragPhaseComplete, lll.EtcdDefragPhaseFailed: + finished = append(finished, items[i]) + default: + active = append(active, items[i]) + } + } + return active, finished +} + +func defragRunRefs(items []lll.EtcdDefrag) []corev1.LocalObjectReference { + if len(items) == 0 { + return nil + } + out := make([]corev1.LocalObjectReference, 0, len(items)) + for i := range items { + out = append(out, corev1.LocalObjectReference{Name: items[i].Name}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} + +func latestSuccessfulTime(items []lll.EtcdDefrag) *metav1.Time { + var best *metav1.Time + for i := range items { + d := &items[i] + if d.Status.Phase != lll.EtcdDefragPhaseComplete || d.Status.CompletedAt == nil { + continue + } + if best == nil || d.Status.CompletedAt.After(best.Time) { + best = d.Status.CompletedAt + } + } + return best +} + +// defragFinishTime orders finished runs for history GC: completion time, or the +// creation time when a run finished without stamping CompletedAt. +func defragFinishTime(d *lll.EtcdDefrag) time.Time { + if d.Status.CompletedAt != nil { + return d.Status.CompletedAt.Time + } + return d.CreationTimestamp.Time +} + +func setDefragPolicyCondition(pol *lll.EtcdDefragPolicy, status metav1.ConditionStatus, reason, msg string) { + setCondition(&pol.Status.Conditions, metav1.Condition{ + Type: defragPolicyCondition, + Status: status, + Reason: reason, + Message: msg, + ObservedGeneration: pol.Generation, + }) +} + +func tickString(t time.Time) string { return t.UTC().Format(time.RFC3339) } + +func copyInt32(p *int32) *int32 { + if p == nil { + return nil + } + v := *p + return &v +} diff --git a/controllers/etcddefragpolicy_controller_test.go b/controllers/etcddefragpolicy_controller_test.go new file mode 100644 index 00000000..49dccda7 --- /dev/null +++ b/controllers/etcddefragpolicy_controller_test.go @@ -0,0 +1,299 @@ +/* +Copyright 2023 Timofey Larkin. + +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 +*/ + +package controllers + +import ( + "context" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/record" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + lll "github.com/cozystack/etcd-operator/api/v1alpha2" +) + +var epoch = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + +func TestParseUTCSchedule(t *testing.T) { + if _, err := parseUTCSchedule("0 3 * * *"); err != nil { + t.Fatalf("valid schedule rejected: %v", err) + } + // UTC is forced: a schedule with no TZ is evaluated in UTC regardless of the + // process zone. "0 0 * * *" from 12:00 UTC lands on the next UTC midnight. + sched, err := parseUTCSchedule("0 0 * * *") + if err != nil { + t.Fatal(err) + } + got := sched.Next(time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)) + if want := time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC); !got.Equal(want) { + t.Errorf("Next = %s, want %s", got, want) + } + if _, err := parseUTCSchedule("not a schedule"); err == nil { + t.Error("expected an error for an unparseable schedule") + } +} + +func TestNextSchedule(t *testing.T) { + sched, err := parseUTCSchedule("0 * * * *") // top of every hour + if err != nil { + t.Fatal(err) + } + + // Next tick still in the future: nothing due. + if due, next := nextSchedule(sched, epoch, epoch.Add(30*time.Minute), 100); due != nil { + t.Errorf("due = %s, want nil (next tick is in the future)", due) + } else if want := epoch.Add(time.Hour); !next.Equal(want) { + t.Errorf("next = %s, want %s", next, want) + } + + // One tick due: the most recent boundary at or before now. + if due, next := nextSchedule(sched, epoch, epoch.Add(90*time.Minute), 100); due == nil { + t.Fatal("due = nil, want the 01:00 tick") + } else if !due.Equal(epoch.Add(time.Hour)) { + t.Errorf("due = %s, want %s", due, epoch.Add(time.Hour)) + } else if !next.Equal(epoch.Add(2 * time.Hour)) { + t.Errorf("next = %s, want %s", next, epoch.Add(2*time.Hour)) + } + + // A long backlog collapses to a single run stamped at now. + now := epoch.Add(1000 * time.Hour) + if due, _ := nextSchedule(sched, epoch, now, 100); due == nil || !due.Equal(now) { + t.Errorf("due = %v, want collapse to now (%s)", due, now) + } +} + +// ── controller ────────────────────────────────────────────────────────────── + +func defragPolicy(name, schedule string, opts ...func(*lll.EtcdDefragPolicy)) *lll.EtcdDefragPolicy { + p := &lll.EtcdDefragPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", CreationTimestamp: metav1.NewTime(epoch)}, + Spec: lll.EtcdDefragPolicySpec{ClusterRef: corev1.LocalObjectReference{Name: "c1"}, Schedule: schedule}, + } + for _, o := range opts { + o(p) + } + return p +} + +func policyReconciler(t *testing.T, now time.Time, objs ...client.Object) (*EtcdDefragPolicyReconciler, client.Client) { + t.Helper() + c, s := newTestClient(t, objs...) + return &EtcdDefragPolicyReconciler{Client: c, Scheme: s, Recorder: record.NewFakeRecorder(20), now: func() time.Time { return now }}, c +} + +func listPolicyRuns(t *testing.T, c client.Client, policy string) []lll.EtcdDefrag { + t.Helper() + var runs lll.EtcdDefragList + if err := c.List(context.Background(), &runs, client.InNamespace("ns"), client.MatchingLabels{LabelDefragPolicy: policy}); err != nil { + t.Fatalf("list runs: %v", err) + } + return runs.Items +} + +func reconcilePolicy(t *testing.T, r *EtcdDefragPolicyReconciler, name string) ctrl.Result { + t.Helper() + res, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: nn(name, "ns")}) + if err != nil { + t.Fatalf("reconcile: %v", err) + } + return res +} + +// A due tick stamps one EtcdDefrag, owned by and labelled with the policy, +// carrying the policy's rule/ttl, and records lastScheduleTime. +func TestDefragPolicy_StampsWhenDue(t *testing.T) { + ttl := int32(3600) + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { + p.Spec.Rule = &lll.DefragRule{All: true} + p.Spec.TTLSecondsAfterFinished = &ttl + }) + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol) + + reconcilePolicy(t, r, "p") + + runs := listPolicyRuns(t, c, "p") + if len(runs) != 1 { + t.Fatalf("stamped %d runs, want 1", len(runs)) + } + run := runs[0] + if run.Spec.ClusterRef.Name != "c1" || run.Spec.Rule == nil || !run.Spec.Rule.All { + t.Errorf("stamped run spec = %+v, want clusterRef c1 + rule.all", run.Spec) + } + if run.Spec.TTLSecondsAfterFinished == nil || *run.Spec.TTLSecondsAfterFinished != ttl { + t.Errorf("stamped ttl = %v, want %d", run.Spec.TTLSecondsAfterFinished, ttl) + } + if run.Labels[LabelCluster] != "c1" { + t.Errorf("missing cluster label: %v", run.Labels) + } + if len(run.OwnerReferences) != 1 || run.OwnerReferences[0].Name != "p" { + t.Errorf("owner refs = %+v, want the policy", run.OwnerReferences) + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if got.Status.LastScheduleTime == nil || !got.Status.LastScheduleTime.Time.Equal(epoch.Add(time.Hour)) { + t.Errorf("lastScheduleTime = %v, want 01:00", got.Status.LastScheduleTime) + } + if len(got.Status.Active) != 1 { + t.Errorf("status.active = %v, want the stamped run", got.Status.Active) + } +} + +// Before the first tick, nothing is stamped and the policy requeues. +func TestDefragPolicy_NotDueYet(t *testing.T) { + pol := defragPolicy("p", "0 0 * * *") // daily midnight + r, c := policyReconciler(t, epoch.Add(time.Hour), pol) + + res := reconcilePolicy(t, r, "p") + if len(listPolicyRuns(t, c, "p")) != 0 { + t.Fatalf("stamped a run before the first tick") + } + if res.RequeueAfter <= 0 { + t.Errorf("expected a requeue toward the next tick, got %+v", res) + } +} + +// A suspended policy stamps nothing and reports Suspended. +func TestDefragPolicy_Suspended(t *testing.T) { + suspend := true + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.Suspend = &suspend }) + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol) + + reconcilePolicy(t, r, "p") + if len(listPolicyRuns(t, c, "p")) != 0 { + t.Fatalf("suspended policy stamped a run") + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if cond := findPolicyCond(got); cond == nil || cond.Reason != "Suspended" { + t.Errorf("condition = %+v, want Suspended", cond) + } +} + +// An unparseable schedule reports InvalidSchedule and stamps nothing. +func TestDefragPolicy_InvalidSchedule(t *testing.T) { + pol := defragPolicy("p", "every blue moon") + r, c := policyReconciler(t, epoch.Add(time.Hour), pol) + + reconcilePolicy(t, r, "p") + if len(listPolicyRuns(t, c, "p")) != 0 { + t.Fatalf("stamped a run on an invalid schedule") + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if cond := findPolicyCond(got); cond == nil || cond.Reason != "InvalidSchedule" { + t.Errorf("condition = %+v, want InvalidSchedule", cond) + } +} + +// With the default Forbid policy, a due tick is skipped while a previous run is +// still active — no second run is stamped, but the tick is consumed. +func TestDefragPolicy_ForbidConcurrent(t *testing.T) { + pol := defragPolicy("p", "0 * * * *") + active := activeRun("p-existing", "p") + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol, active) + + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 1 { + t.Fatalf("Forbid stamped a concurrent run: %d runs", len(runs)) + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if got.Status.LastScheduleTime == nil { + t.Errorf("a forbidden tick should still advance lastScheduleTime") + } +} + +// With Allow, a due tick is stamped even while a previous run is active. +func TestDefragPolicy_AllowConcurrent(t *testing.T) { + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.ConcurrencyPolicy = lll.AllowConcurrent }) + active := activeRun("p-existing", "p") + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol, active) + + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 2 { + t.Fatalf("Allow did not stamp a concurrent run: %d runs", len(runs)) + } +} + +// A tick older than StartingDeadlineSeconds is skipped rather than started +// late: nothing is stamped, but the tick is consumed (lastScheduleTime advances +// to it) so the controller does not retry the stale slot. +func TestDefragPolicy_MissedDeadline(t *testing.T) { + deadline := int64(60) + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.StartingDeadlineSeconds = &deadline }) + // now is 01:30, so the 01:00 tick is 30m old — well past the 60s deadline. + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol) + + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 0 { + t.Fatalf("stamped a run past the starting deadline: %d runs", len(runs)) + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if got.Status.LastScheduleTime == nil || !got.Status.LastScheduleTime.Time.Equal(epoch.Add(time.Hour)) { + t.Errorf("lastScheduleTime = %v, want the missed 01:00 tick consumed", got.Status.LastScheduleTime) + } +} + +// A tick within StartingDeadlineSeconds is stamped normally: the deadline only +// suppresses runs older than its window. +func TestDefragPolicy_WithinDeadline(t *testing.T) { + deadline := int64(7200) // 2h, comfortably wider than the 30m-old tick + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.StartingDeadlineSeconds = &deadline }) + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol) + + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 1 { + t.Fatalf("a tick within the deadline should stamp one run, got %d", len(runs)) + } +} + +// HistoryLimit trims the oldest finished runs, keeping the newest. +func TestDefragPolicy_HistoryLimit(t *testing.T) { + limit := int32(1) + pol := defragPolicy("p", "0 0 * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.HistoryLimit = &limit }) // not due + old1 := finishedRun("p-1", "p", epoch.Add(1*time.Hour)) + old2 := finishedRun("p-2", "p", epoch.Add(2*time.Hour)) + newest := finishedRun("p-3", "p", epoch.Add(3*time.Hour)) + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol, old1, old2, newest) + + reconcilePolicy(t, r, "p") + runs := listPolicyRuns(t, c, "p") + if len(runs) != 1 { + t.Fatalf("history GC kept %d runs, want 1", len(runs)) + } + if runs[0].Name != "p-3" { + t.Errorf("GC kept %q, want the newest p-3", runs[0].Name) + } +} + +func activeRun(name, policy string) *lll.EtcdDefrag { + return &lll.EtcdDefrag{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: policy}}, + Status: lll.EtcdDefragStatus{Phase: lll.EtcdDefragPhaseRunning}, + } +} + +func finishedRun(name, policy string, completed time.Time) *lll.EtcdDefrag { + ct := metav1.NewTime(completed) + return &lll.EtcdDefrag{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: policy}}, + Status: lll.EtcdDefragStatus{Phase: lll.EtcdDefragPhaseComplete, CompletedAt: &ct}, + } +} + +func findPolicyCond(pol *lll.EtcdDefragPolicy) *metav1.Condition { + for i := range pol.Status.Conditions { + if pol.Status.Conditions[i].Type == defragPolicyCondition { + return &pol.Status.Conditions[i] + } + } + return nil +} diff --git a/controllers/helpers.go b/controllers/helpers.go index 087038a7..1202f78f 100644 --- a/controllers/helpers.go +++ b/controllers/helpers.go @@ -16,6 +16,10 @@ const ( // LabelCluster is the label key used to associate resources with an EtcdCluster. LabelCluster = "etcd-operator.cozystack.io/cluster" + // LabelDefragPolicy tags an EtcdDefrag with the EtcdDefragPolicy that + // stamped it, so the policy controller can find its own runs. + LabelDefragPolicy = "etcd-operator.cozystack.io/defrag-policy" + // LabelRole identifies the etcd-side raft role of a member's Pod. The // only value the operator emits today is RoleVoter; learners carry no // LabelRole at all so the per-cluster PodDisruptionBudget can select diff --git a/controllers/testing_helpers_test.go b/controllers/testing_helpers_test.go index fd4353c8..11a1410b 100644 --- a/controllers/testing_helpers_test.go +++ b/controllers/testing_helpers_test.go @@ -301,7 +301,7 @@ func newTestClient(t *testing.T, objs ...client.Object) (client.Client, *runtime c := fake.NewClientBuilder(). WithScheme(s). WithObjects(objs...). - WithStatusSubresource(&lll.EtcdCluster{}, &lll.EtcdMember{}, &lll.EtcdSnapshot{}, &lll.EtcdDefrag{}). + WithStatusSubresource(&lll.EtcdCluster{}, &lll.EtcdMember{}, &lll.EtcdSnapshot{}, &lll.EtcdDefrag{}, &lll.EtcdDefragPolicy{}). Build() return c, s } diff --git a/docs/etcd-defrag.md b/docs/etcd-defrag.md index 03a552bb..9c74a636 100644 --- a/docs/etcd-defrag.md +++ b/docs/etcd-defrag.md @@ -9,11 +9,10 @@ It is a one-shot, run-to-completion record, modeled on [`EtcdSnapshot`](concepts the operator drives it through `status.phase` and it never re-runs. **Scheduling.** `EtcdDefrag` is the *run*; what *triggers* a run is separate. -Today, recurring defragmentation is driven by creating `EtcdDefrag` objects from -outside (a `CronJob`, a GitOps cron). A companion `EtcdDefragPolicy` kind — a -cadence (`schedule`) and/or a condition (`when`) that stamps out `EtcdDefrag` -runs — is planned so the operator absorbs that scheduling itself; it is not part -of this API PR. +For recurring defragmentation, [`EtcdDefragPolicy`](#recurring-runs-etcddefragpolicy) +stamps out `EtcdDefrag` objects on a cron schedule so the operator drives the +cadence itself; you can also create `EtcdDefrag` objects from outside (a +`CronJob`, a GitOps cron) if you prefer to own the schedule elsewhere. ## Why in the operator (not a bare CronJob) @@ -131,10 +130,54 @@ no `spec` knobs: backoff up to the deadline; a failed per-member RPC is retried a bounded number of times then marked `Failed` (a failing leader fails the run). - **Retry across runs:** terminal phases (`Complete`/`Failed`) are sticky — an - `EtcdDefrag` never re-runs itself. A retry is a *new* `EtcdDefrag`: the external - scheduler's next tick for periodic use, or a re-create for a one-shot. Each - attempt is a discrete, auditable object (GC'd via `ttlSecondsAfterFinished`) - rather than hidden retry state. + `EtcdDefrag` never re-runs itself. A retry is a *new* `EtcdDefrag`: an + [`EtcdDefragPolicy`](#recurring-runs-etcddefragpolicy) tick for periodic use, or + a re-create for a one-shot. Each attempt is a discrete, auditable object (GC'd + via `ttlSecondsAfterFinished`) rather than hidden retry state. + +## Recurring runs (`EtcdDefragPolicy`) + +`EtcdDefragPolicy` schedules `EtcdDefrag` runs on a cron cadence. Each tick +stamps a new `EtcdDefrag` — owned by the policy (so it cascades on delete) — and +the run then follows all the safety rules above. The policy only *triggers* +runs; it never defragments directly. + +```yaml +apiVersion: etcd-operator.cozystack.io/v1alpha2 +kind: EtcdDefragPolicy +metadata: + name: nightly + namespace: team-a +spec: + clusterRef: + name: etcd + schedule: "0 3 * * *" # standard five-field cron, evaluated in UTC + concurrencyPolicy: Forbid # skip a tick while a previous run is still active (default) + ttlSecondsAfterFinished: 3600 + historyLimit: 3 # keep the last 3 finished runs + rule: + freeSpaceAbove: 200Mi + quotaUsageAbove: 80% + minReclaim: 32Mi +``` + +- **`schedule`** is a standard five-field cron expression in UTC. Prefix it with + `CRON_TZ=` to use another zone. +- **`concurrencyPolicy`** is `Forbid` (default — a tick is skipped while a + stamped run is still active) or `Allow` (stamp anyway; `EtcdDefrag`'s own + per-cluster serialization queues it behind the active run). +- **`suspend: true`** pauses stamping without deleting the policy; missed ticks + are not backfilled on resume. +- **`startingDeadlineSeconds`** skips a tick that is already older than the + deadline (e.g. after the operator was down) instead of starting it late. +- **`historyLimit`** caps retained finished runs; `ttlSecondsAfterFinished` (per + run) is the other cleanup path. +- **`rule`** / **`ttlSecondsAfterFinished`** are copied verbatim into each + stamped `EtcdDefrag`. + +`status.lastScheduleTime` anchors the next tick (so a tick is never acted on +twice), `status.lastSuccessfulTime` records the last `Complete`, and +`status.active` lists runs still in flight. ## Relationship to capacity metrics diff --git a/go.mod b/go.mod index 22046840..eac9e79b 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2 github.com/aws/smithy-go v1.26.0 github.com/dustin/go-humanize v1.0.1 + github.com/robfig/cron/v3 v3.0.1 github.com/spf13/cobra v1.10.2 go.etcd.io/etcd/api/v3 v3.6.11 go.etcd.io/etcd/client/v3 v3.6.11 diff --git a/go.sum b/go.sum index 87ba20da..3d820c27 100644 --- a/go.sum +++ b/go.sum @@ -152,6 +152,8 @@ github.com/prometheus/common v0.62.0 h1:xasJaQlnWAeyHdUBeGjXmutelfJHWMRr+Fg4QszZ github.com/prometheus/common v0.62.0/go.mod h1:vyBcEuLSvWos9B1+CyL7JZ2up+uFzXhkqml0W5zIY1I= github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= diff --git a/main.go b/main.go index 8a1d18a4..91efa088 100644 --- a/main.go +++ b/main.go @@ -260,6 +260,14 @@ func main() { setupLog.Error(err, "unable to create controller", "controller", "EtcdDefrag") os.Exit(1) } + if err = (&controllers.EtcdDefragPolicyReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + Recorder: mgr.GetEventRecorderFor("etcd-operator"), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "EtcdDefragPolicy") + os.Exit(1) + } //+kubebuilder:scaffold:builder if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {