From 8be54050aed1ef6d97c93f0ce28dc0c55ebcfde6 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 20 Aug 2026 12:07:06 +0400 Subject: [PATCH 1/6] =?UTF-8?q?feat(defrag):=20EtcdDefragPolicy=20?= =?UTF-8?q?=E2=80=94=20scheduled=20EtcdDefrag=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a CronJob-style `EtcdDefragPolicy` so the operator drives recurring defragmentation itself, instead of relying on an external CronJob to create `EtcdDefrag` objects (the follow-up flagged as "planned" in the EtcdDefrag docs). The policy reconciler stamps out `EtcdDefrag` runs on a five-field cron schedule (evaluated in UTC; a CRON_TZ prefix is honoured). Each stamped run is owned by the policy (cascades on delete) and labelled with the policy name, and carries the policy's `rule` and `ttlSecondsAfterFinished` — the run then follows all of EtcdDefrag's existing safety rules (per-cluster serialization, health gate, followers-before-leader), so the policy only triggers runs and never defragments directly. Spec: `schedule`, `clusterRef`, `rule`, `ttlSecondsAfterFinished`, `suspend`, `concurrencyPolicy` (Forbid default / Allow), `startingDeadlineSeconds`, and `historyLimit`. Status: `lastScheduleTime` (anchors the next tick so one is never acted on twice), `lastSuccessfulTime`, `active`, and an `Active` condition carrying Suspended / InvalidSchedule reasons. A long backlog after downtime is collapsed into a single run rather than replayed. Adds the API type + generated CRD/deepcopy/RBAC, the controller (wired in main.go, watches its owned EtcdDefrags), robfig/cron/v3, unit + controller tests (schedule parsing/next-tick math, stamp-when-due, not-due, suspend, invalid schedule, Forbid/Allow concurrency, history GC), and docs. Stacked on the EtcdDefrag controller PR. Refs #221. Assisted-By: Claude Opus 4.8 (1M context) Signed-off-by: Andrey Kolkov --- api/v1alpha2/etcddefragpolicy_types.go | 143 ++++++++ api/v1alpha2/zz_generated.deepcopy.go | 135 +++++++ ...rator.cozystack.io_etcddefragpolicies.yaml | 283 ++++++++++++++ .../files/manager-role-rules.yaml | 5 +- controllers/etcddefragpolicy_controller.go | 344 ++++++++++++++++++ .../etcddefragpolicy_controller_test.go | 267 ++++++++++++++ controllers/helpers.go | 4 + controllers/testing_helpers_test.go | 2 +- docs/etcd-defrag.md | 61 +++- go.mod | 1 + go.sum | 2 + main.go | 8 + 12 files changed, 1243 insertions(+), 12 deletions(-) create mode 100644 api/v1alpha2/etcddefragpolicy_types.go create mode 100644 charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml create mode 100644 controllers/etcddefragpolicy_controller.go create mode 100644 controllers/etcddefragpolicy_controller_test.go 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..b80108c0 --- /dev/null +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml @@ -0,0 +1,283 @@ +--- +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..99 with a "%" + suffix, e.g. "80%"; 100% is rejected because a backend never exceeds its + quota (etcd raises NOSPACE first), so the arm could never fire. + pattern: ^[1-9][0-9]?%$ + 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..3302fe76 --- /dev/null +++ b/controllers/etcddefragpolicy_controller_test.go @@ -0,0 +1,267 @@ +/* +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)) + } +} + +// 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 98b029ce..d881cd60 100644 --- a/controllers/testing_helpers_test.go +++ b/controllers/testing_helpers_test.go @@ -333,7 +333,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 f6c380f0..bf246f7d 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 -implemented yet. +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) @@ -146,10 +145,54 @@ no `spec` knobs: reclaimed space still disarms `NOSPACE` on the way out. (Per-member RPC retry is a possible follow-up, not shipped here.) - **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 { From 08a52a2c5d43a331cf9584f500ffa13329ad5c53 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 20 Aug 2026 12:33:13 +0400 Subject: [PATCH 2/6] test(defrag): cover the starting-deadline branch of EtcdDefragPolicy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The controller skips a tick older than StartingDeadlineSeconds — emitting MissedSchedule and consuming the tick without stamping a run — but the case was untested. Add a missed-deadline case (tick past the window: no run, tick still consumed) and a within-deadline mirror (run stamped normally). Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Andrey Kolkov --- .../etcddefragpolicy_controller_test.go | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/controllers/etcddefragpolicy_controller_test.go b/controllers/etcddefragpolicy_controller_test.go index 3302fe76..49dccda7 100644 --- a/controllers/etcddefragpolicy_controller_test.go +++ b/controllers/etcddefragpolicy_controller_test.go @@ -223,6 +223,38 @@ func TestDefragPolicy_AllowConcurrent(t *testing.T) { } } +// 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) From 6f43fc68f50d42ca5d2a457b80579a185cafad8e Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Mon, 24 Aug 2026 17:00:40 +0400 Subject: [PATCH 3/6] fix(defrag): correct EtcdDefragPolicy scheduling and ownership MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on the EtcdDefragPolicy kind: - Schedule becomes a {cron, timezone} struct instead of a string with a CRON_TZ= prefix, so the zone is a first-class, validated field visible to kubectl. Embed time/tzdata in main.go — the distroless image carries no zoneinfo, so any named zone would otherwise fail LoadLocation at runtime. - Bound the catch-up walk by startingDeadlineSeconds: a tick older than the deadline is never returned, so a long outage no longer collapses to a fabricated wall-clock tick that defeats the deadline and destabilizes the run name. A backlog past the walk bound with no deadline parks the policy on a TooManyMissedTicks condition rather than guessing. - Filter listed runs by ownerRef, not the label alone, so a hand-copied run cannot count as active or be deleted by history GC. - Add the clusterRef CEL rule for parity with EtcdDefrag, and surface a rejected Create on the Active condition instead of only in logs. - Constrain the parser to five fields (no descriptors), default concurrencyPolicy in the schema, and key the run name on the tick's minute. Signed-off-by: Andrey Kolkov --- api/v1alpha2/etcddefragpolicy_types.go | 34 +++- api/v1alpha2/zz_generated.deepcopy.go | 16 ++ ...rator.cozystack.io_etcddefragpolicies.yaml | 39 ++++- controllers/etcddefragpolicy_controller.go | 119 ++++++++----- .../etcddefragpolicy_controller_test.go | 161 +++++++++++++++--- main.go | 5 + 6 files changed, 297 insertions(+), 77 deletions(-) diff --git a/api/v1alpha2/etcddefragpolicy_types.go b/api/v1alpha2/etcddefragpolicy_types.go index a16e8d15..bb62b649 100644 --- a/api/v1alpha2/etcddefragpolicy_types.go +++ b/api/v1alpha2/etcddefragpolicy_types.go @@ -36,26 +36,45 @@ const ( ForbidConcurrent ConcurrencyPolicy = "Forbid" ) +// DefragSchedule names when runs are stamped: a five-field cron expression and +// the zone it is read in. The zone is a dedicated field rather than a CRON_TZ +// prefix so it is visible to `kubectl get -o custom-columns` and validated on +// its own. +type DefragSchedule struct { + // Cron is a standard five-field cron expression (e.g. "0 3 * * *" for 03:00). + // Descriptors (@daily) and a TZ=/CRON_TZ= prefix are rejected; use Timezone + // for the zone. + // +kubebuilder:validation:MinLength=1 + // +kubebuilder:validation:XValidation:rule="!self.contains('TZ=')",message="set the zone in schedule.timezone, not a TZ= prefix" + Cron string `json:"cron"` + + // Timezone is an IANA zone name (e.g. "Europe/Moscow") the cron expression is + // read in. Absent reads it in UTC. + // +optional + Timezone string `json:"timezone,omitempty"` +} + // 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. +// +kubebuilder:validation:XValidation:rule="size(self.clusterRef.name) != 0",message="spec.clusterRef.name is required" 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"` + // Schedule names when a run is stamped. + Schedule DefragSchedule `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). + // Suspend pauses stamping. Runs already in flight are left alone. On resume + // the single most recent missed tick may be stamped (subject to + // StartingDeadlineSeconds); earlier missed ticks are never replayed. // +optional Suspend *bool `json:"suspend,omitempty"` // ConcurrencyPolicy decides what a due tick does when a previous stamped run // is still active. Defaults to Forbid. + // +kubebuilder:default=Forbid // +optional ConcurrencyPolicy ConcurrencyPolicy `json:"concurrencyPolicy,omitempty"` @@ -112,7 +131,8 @@ type EtcdDefragPolicyStatus struct { // +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="Schedule",type=string,JSONPath=`.spec.schedule.cron` +// +kubebuilder:printcolumn:name="Timezone",type=string,JSONPath=`.spec.schedule.timezone` // +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` diff --git a/api/v1alpha2/zz_generated.deepcopy.go b/api/v1alpha2/zz_generated.deepcopy.go index 8a3ebe45..7f59359e 100644 --- a/api/v1alpha2/zz_generated.deepcopy.go +++ b/api/v1alpha2/zz_generated.deepcopy.go @@ -171,6 +171,21 @@ func (in *DefragRule) DeepCopy() *DefragRule { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *DefragSchedule) DeepCopyInto(out *DefragSchedule) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DefragSchedule. +func (in *DefragSchedule) DeepCopy() *DefragSchedule { + if in == nil { + return nil + } + out := new(DefragSchedule) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *EtcdCluster) DeepCopyInto(out *EtcdCluster) { *out = *in @@ -477,6 +492,7 @@ func (in *EtcdDefragPolicyList) DeepCopyObject() runtime.Object { func (in *EtcdDefragPolicySpec) DeepCopyInto(out *EtcdDefragPolicySpec) { *out = *in out.ClusterRef = in.ClusterRef + out.Schedule = in.Schedule if in.Suspend != nil { in, out := &in.Suspend, &out.Suspend *out = new(bool) 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 index b80108c0..7efc2e40 100644 --- a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml @@ -18,9 +18,12 @@ spec: - jsonPath: .spec.clusterRef.name name: Cluster type: string - - jsonPath: .spec.schedule + - jsonPath: .spec.schedule.cron name: Schedule type: string + - jsonPath: .spec.schedule.timezone + name: Timezone + type: string - jsonPath: .spec.suspend name: Suspend type: boolean @@ -79,6 +82,7 @@ spec: type: object x-kubernetes-map-type: atomic concurrencyPolicy: + default: Forbid description: |- ConcurrencyPolicy decides what a due tick does when a previous stamped run is still active. Defaults to Forbid. @@ -150,11 +154,26 @@ spec: 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 + description: Schedule names when a run is stamped. + properties: + cron: + description: |- + Cron is a standard five-field cron expression (e.g. "0 3 * * *" for 03:00). + Descriptors (@daily) and a TZ=/CRON_TZ= prefix are rejected; use Timezone + for the zone. + minLength: 1 + type: string + x-kubernetes-validations: + - message: set the zone in schedule.timezone, not a TZ= prefix + rule: '!self.contains(''TZ='')' + timezone: + description: |- + Timezone is an IANA zone name (e.g. "Europe/Moscow") the cron expression is + read in. Absent reads it in UTC. + type: string + required: + - cron + type: object startingDeadlineSeconds: description: |- StartingDeadlineSeconds bounds how late a missed tick may still be started. @@ -166,8 +185,9 @@ spec: 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). + Suspend pauses stamping. Runs already in flight are left alone. On resume + the single most recent missed tick may be stamped (subject to + StartingDeadlineSeconds); earlier missed ticks are never replayed. type: boolean ttlSecondsAfterFinished: description: |- @@ -180,6 +200,9 @@ spec: - clusterRef - schedule type: object + x-kubernetes-validations: + - message: spec.clusterRef.name is required + rule: size(self.clusterRef.name) != 0 status: description: EtcdDefragPolicyStatus is the observed state of an EtcdDefragPolicy. properties: diff --git a/controllers/etcddefragpolicy_controller.go b/controllers/etcddefragpolicy_controller.go index b07d8f10..83aa2a9d 100644 --- a/controllers/etcddefragpolicy_controller.go +++ b/controllers/etcddefragpolicy_controller.go @@ -12,9 +12,9 @@ package controllers import ( "context" + "errors" "fmt" "sort" - "strings" "time" "github.com/robfig/cron/v3" @@ -37,12 +37,17 @@ const ( // 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 bounds how many missed ticks nextSchedule walks to + // find the most recent one. It never replays slots (only the latest tick is + // ever returned); the bound just caps the walk so a large clock jump or a + // long outage surfaces as a condition instead of an unbounded loop. defragPolicyMaxCatchup = 100 ) +// errTooManyMissed reports that more than defragPolicyMaxCatchup ticks are +// unaccounted for — a clock jump or an outage longer than the catch-up window. +var errTooManyMissed = errors.New("too many missed ticks; check the clock or set spec.startingDeadlineSeconds") + // 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 @@ -70,13 +75,16 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, client.IgnoreNotFound(err) } - // Observe the runs this policy owns. + // Observe the runs this policy owns. The label narrows the list server-side; + // the ownerRef is the authority, so a hand-copied run that kept the label but + // isn't controlled by this policy is ignored rather than counted as active or + // deleted by history GC. 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) + active, finished := partitionDefragRuns(ownedRuns(runs.Items, pol)) pol.Status.Active = defragRunRefs(active) if t := latestSuccessfulTime(finished); t != nil { pol.Status.LastSuccessfulTime = t @@ -95,10 +103,10 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, r.Status().Update(ctx, pol) } - sched, err := parseUTCSchedule(pol.Spec.Schedule) + sched, err := parseSchedule(pol.Spec.Schedule) if err != nil { setDefragPolicyCondition(pol, metav1.ConditionFalse, "InvalidSchedule", - fmt.Sprintf("cannot parse schedule %q: %v", pol.Spec.Schedule, err)) + fmt.Sprintf("cannot parse schedule %q: %v", pol.Spec.Schedule.Cron, err)) // Only a spec change can fix this; the watch re-triggers, so don't requeue. return ctrl.Result{}, r.Status().Update(ctx, pol) } @@ -109,27 +117,29 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req 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 + // StartingDeadlineSeconds bounds how late a tick may still start, so a tick + // older than the deadline is skipped anyway: never walk further back than the + // deadline window. This both enforces the deadline and keeps the walk short. + if d := pol.Spec.StartingDeadlineSeconds; d != nil { + if cutoff := now.Add(-time.Duration(*d) * time.Second); cutoff.After(earliest) { + earliest = cutoff } - return ctrl.Result{RequeueAfter: requeueFor(next, now)}, nil } - tick := *due + due, next, err := nextSchedule(sched, earliest, now, defragPolicyMaxCatchup) + if err != nil { + setDefragPolicyCondition(pol, metav1.ConditionFalse, "TooManyMissedTicks", err.Error()) + // A long backlog with no deadline needs operator action (fix the clock, or + // set startingDeadlineSeconds); the watch re-triggers on a spec edit. + return ctrl.Result{}, r.Status().Update(ctx, pol) + } - // 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 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 if concurrencyPolicy(pol) == lll.ForbidConcurrent && len(active) > 0 { r.event(pol, corev1.EventTypeNormal, "ConcurrencyForbidden", @@ -147,6 +157,14 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req } if err := r.Create(ctx, run); err != nil { if !apierrors.IsAlreadyExists(err) { + // A rejected Create (e.g. an invalid clusterRef caught by the stamped + // run's CEL rule) otherwise leaves the policy looking healthy while it + // silently never runs — surface it on the condition, not just in logs. + setDefragPolicyCondition(pol, metav1.ConditionFalse, "StampFailed", + fmt.Sprintf("cannot stamp EtcdDefrag for %s: %v", tickString(tick), err)) + if uerr := r.Status().Update(ctx, pol); uerr != nil { + return ctrl.Result{}, uerr + } return ctrl.Result{}, err } // The deterministic name means a re-reconcile of the same tick is a @@ -166,11 +184,12 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req // 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. +// collides (IsAlreadyExists) instead of double-stamping. Cron's finest +// granularity is one minute, so the name keys on the tick's minute. 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()), + Name: fmt.Sprintf("%s-%d", pol.Name, tick.Unix()/60), Namespace: pol.Namespace, Labels: map[string]string{ LabelDefragPolicy: pol.Name, @@ -225,25 +244,36 @@ func (r *EtcdDefragPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error { // ── 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 +// defragScheduleParser accepts only the standard five fields — no descriptors +// (@daily, @every) — so the parser matches the documented grammar. +var defragScheduleParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) + +// parseSchedule builds a cron schedule from the policy's schedule, read in the +// named zone (UTC when absent). The zone is validated first so a bad one gets a +// clear message instead of a parse error naming an expression the user did not +// write; it reaches the parser as a CRON_TZ prefix, an implementation detail +// rather than user-facing syntax. +func parseSchedule(s lll.DefragSchedule) (cron.Schedule, error) { + tz := s.Timezone + if tz == "" { + tz = "UTC" } - return cron.ParseStandard(spec) + if _, err := time.LoadLocation(tz); err != nil { + return nil, fmt.Errorf("unknown timezone %q: %w", tz, err) + } + return defragScheduleParser.Parse("CRON_TZ=" + tz + " " + s.Cron) } -// nextSchedule returns the most recent scheduled time at or before now that is +// nextSchedule returns the most recent scheduled tick 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) { +// the next tick after now. It walks at most maxCatchup missed ticks; a longer +// backlog returns errTooManyMissed rather than fabricating a non-boundary tick, +// so the caller surfaces a condition instead of silently starting a run the +// deadline was meant to suppress. +func nextSchedule(sched cron.Schedule, earliest, now time.Time, maxCatchup int) (due *time.Time, next time.Time, err error) { t := sched.Next(earliest) if t.After(now) { - return nil, t + return nil, t, nil } last := t for n := 0; ; n++ { @@ -252,12 +282,11 @@ func nextSchedule(sched cron.Schedule, earliest, now time.Time, maxCatchup int) break } if n >= maxCatchup { - last = now - return &last, sched.Next(now) + return nil, time.Time{}, errTooManyMissed } last = t } - return &last, t + return &last, t, nil } // requeueFor is the delay until next, floored so a just-passed boundary still @@ -276,6 +305,18 @@ func concurrencyPolicy(pol *lll.EtcdDefragPolicy) lll.ConcurrencyPolicy { return pol.Spec.ConcurrencyPolicy } +// ownedRuns keeps only the EtcdDefrags this policy actually controls (by +// ownerRef), dropping any that merely carry the policy label. +func ownedRuns(items []lll.EtcdDefrag, pol *lll.EtcdDefragPolicy) []lll.EtcdDefrag { + owned := make([]lll.EtcdDefrag, 0, len(items)) + for i := range items { + if metav1.IsControlledBy(&items[i], pol) { + owned = append(owned, items[i]) + } + } + return owned +} + func partitionDefragRuns(items []lll.EtcdDefrag) (active, finished []lll.EtcdDefrag) { for i := range items { switch items[i].Status.Phase { diff --git a/controllers/etcddefragpolicy_controller_test.go b/controllers/etcddefragpolicy_controller_test.go index 49dccda7..da894d9f 100644 --- a/controllers/etcddefragpolicy_controller_test.go +++ b/controllers/etcddefragpolicy_controller_test.go @@ -12,11 +12,13 @@ package controllers import ( "context" + "errors" "testing" "time" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -26,13 +28,13 @@ import ( 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 { +func TestParseSchedule(t *testing.T) { + if _, err := parseSchedule(lll.DefragSchedule{Cron: "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 * * *") + // No timezone means UTC: "0 0 * * *" from 12:00 UTC lands on the next UTC + // midnight regardless of the process zone. + sched, err := parseSchedule(lll.DefragSchedule{Cron: "0 0 * * *"}) if err != nil { t.Fatal(err) } @@ -40,26 +42,48 @@ func TestParseUTCSchedule(t *testing.T) { 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 { + + // A named zone shifts the boundary: 03:00 Europe/Moscow (UTC+3) is 00:00 UTC. + msk, err := parseSchedule(lll.DefragSchedule{Cron: "0 3 * * *", Timezone: "Europe/Moscow"}) + if err != nil { + t.Fatalf("named timezone rejected (is time/tzdata imported?): %v", err) + } + got = msk.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("Europe/Moscow Next = %s, want %s (00:00 UTC)", got, want) + } + + if _, err := parseSchedule(lll.DefragSchedule{Cron: "not a schedule"}); err == nil { t.Error("expected an error for an unparseable schedule") } + if _, err := parseSchedule(lll.DefragSchedule{Cron: "0 3 * * *", Timezone: "Mars/Olympus"}); err == nil { + t.Error("expected an error for an unknown timezone") + } + // Descriptors are outside the documented five-field grammar. + if _, err := parseSchedule(lll.DefragSchedule{Cron: "@hourly"}); err == nil { + t.Error("expected @hourly to be rejected by the five-field parser") + } } func TestNextSchedule(t *testing.T) { - sched, err := parseUTCSchedule("0 * * * *") // top of every hour + sched, err := parseSchedule(lll.DefragSchedule{Cron: "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 { + if due, next, err := nextSchedule(sched, epoch, epoch.Add(30*time.Minute), 100); err != nil { + t.Fatalf("unexpected error: %v", err) + } else if 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 { + if due, next, err := nextSchedule(sched, epoch, epoch.Add(90*time.Minute), 100); err != nil { + t.Fatalf("unexpected error: %v", err) + } else if 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)) @@ -67,10 +91,10 @@ func TestNextSchedule(t *testing.T) { 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) + // A backlog longer than maxCatchup returns an error rather than fabricating a + // non-boundary tick — the guard that finding 1 was about. + if _, _, err := nextSchedule(sched, epoch, epoch.Add(1000*time.Hour), 100); !errors.Is(err, errTooManyMissed) { + t.Errorf("err = %v, want errTooManyMissed for a backlog past maxCatchup", err) } } @@ -78,8 +102,11 @@ func TestNextSchedule(t *testing.T) { 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}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", UID: policyUID(name), CreationTimestamp: metav1.NewTime(epoch)}, + Spec: lll.EtcdDefragPolicySpec{ + ClusterRef: corev1.LocalObjectReference{Name: "c1"}, + Schedule: lll.DefragSchedule{Cron: schedule}, + }, } for _, o := range opts { o(p) @@ -87,6 +114,21 @@ func defragPolicy(name, schedule string, opts ...func(*lll.EtcdDefragPolicy)) *l return p } +func policyUID(name string) types.UID { return types.UID(name + "-uid") } + +// ownedBy is the controller ownerRef a stamped run carries, matching what +// SetControllerReference writes, so ownedRuns' IsControlledBy check keeps it. +func ownedBy(policy string) []metav1.OwnerReference { + controller := true + return []metav1.OwnerReference{{ + APIVersion: lll.GroupVersion.String(), + Kind: "EtcdDefragPolicy", + Name: policy, + UID: policyUID(policy), + Controller: &controller, + }} +} + func policyReconciler(t *testing.T, now time.Time, objs ...client.Object) (*EtcdDefragPolicyReconciler, client.Client) { t.Helper() c, s := newTestClient(t, objs...) @@ -223,22 +265,37 @@ func TestDefragPolicy_AllowConcurrent(t *testing.T) { } } -// 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. +// A tick older than StartingDeadlineSeconds is never started late: the deadline +// bounds the catch-up window, so the stale 01:00 tick falls outside it and +// nothing is stamped. The next in-window tick will run on its own reconcile. 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") + res := 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)) } + if res.RequeueAfter <= 0 { + t.Errorf("expected a requeue toward the next tick, got %+v", res) + } +} + +// A backlog longer than the catch-up window, with no deadline to bound it, +// parks the policy on TooManyMissedTicks instead of fabricating a run. +func TestDefragPolicy_TooManyMissedTicks(t *testing.T) { + pol := defragPolicy("p", "* * * * *") // every minute: 1000h backlog >> maxCatchup + r, c := policyReconciler(t, epoch.Add(1000*time.Hour), pol) + + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 0 { + t.Fatalf("stamped a run on a too-large backlog: %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) + if cond := findPolicyCond(got); cond == nil || cond.Reason != "TooManyMissedTicks" { + t.Errorf("condition = %+v, want TooManyMissedTicks", cond) } } @@ -274,9 +331,67 @@ func TestDefragPolicy_HistoryLimit(t *testing.T) { } } +// An EtcdDefrag that carries the policy label but is not controlled by the +// policy (e.g. hand-copied YAML) is ignored: it must not count as active under +// Forbid, so the due tick still stamps a genuine run. +func TestDefragPolicy_IgnoresUnownedLabelledRun(t *testing.T) { + pol := defragPolicy("p", "0 * * * *") // default Forbid + imposter := &lll.EtcdDefrag{ + ObjectMeta: metav1.ObjectMeta{Name: "hand-copied", Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: "p"}}, + Status: lll.EtcdDefragStatus{Phase: lll.EtcdDefragPhaseRunning}, + } + r, c := policyReconciler(t, epoch.Add(90*time.Minute), pol, imposter) + + reconcilePolicy(t, r, "p") + stamped := 0 + for _, run := range listPolicyRuns(t, c, "p") { + if run.Name != "hand-copied" { + stamped++ + } + } + if stamped != 1 { + t.Fatalf("owned runs stamped = %d, want 1 (the imposter must not block Forbid)", stamped) + } + got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) + if len(got.Status.Active) != 1 { + t.Errorf("status.active = %v, want only the owned run", got.Status.Active) + } +} + +// A rejected Create surfaces on the Active condition rather than only in logs. +func TestDefragPolicy_StampFailedSurfacesCondition(t *testing.T) { + pol := defragPolicy("p", "0 * * * *") + base, s := newTestClient(t, pol) + r := &EtcdDefragPolicyReconciler{ + Client: &createFailClient{Client: base, err: errors.New("admission rejected")}, + Scheme: s, + Recorder: record.NewFakeRecorder(20), + now: func() time.Time { return epoch.Add(90 * time.Minute) }, + } + + if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: nn("p", "ns")}); err == nil { + t.Fatal("expected the create error to propagate") + } + got := mustGet(t, base, "p", "ns", &lll.EtcdDefragPolicy{}) + if cond := findPolicyCond(got); cond == nil || cond.Reason != "StampFailed" { + t.Errorf("condition = %+v, want StampFailed", cond) + } +} + +// createFailClient fails every Create with a preset error, driving the +// rejected-stamp path without an apiserver. +type createFailClient struct { + client.Client + err error +} + +func (c *createFailClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + return c.err +} + func activeRun(name, policy string) *lll.EtcdDefrag { return &lll.EtcdDefrag{ - ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: policy}}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: policy}, OwnerReferences: ownedBy(policy)}, Status: lll.EtcdDefragStatus{Phase: lll.EtcdDefragPhaseRunning}, } } @@ -284,7 +399,7 @@ func activeRun(name, policy string) *lll.EtcdDefrag { 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}}, + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "ns", Labels: map[string]string{LabelDefragPolicy: policy}, OwnerReferences: ownedBy(policy)}, Status: lll.EtcdDefragStatus{Phase: lll.EtcdDefragPhaseComplete, CompletedAt: &ct}, } } diff --git a/main.go b/main.go index 91efa088..6f82c523 100644 --- a/main.go +++ b/main.go @@ -23,6 +23,11 @@ import ( "os" "strings" + // The distroless base image carries no /usr/share/zoneinfo, so an + // EtcdDefragPolicy timezone would fail time.LoadLocation without the embedded + // database. + _ "time/tzdata" + // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) // to ensure that exec-entrypoint and run can make use of them. _ "k8s.io/client-go/plugin/pkg/client/auth" From dc8f4c6139d41826c9b77b3a2d246daa3a6558c2 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Mon, 24 Aug 2026 17:00:40 +0400 Subject: [PATCH 4/6] docs(defrag): justify the policy and keep the conditional trigger on the roadmap The CronJob comparison argued for EtcdDefrag, not the policy that displaces a kubectl-create CronJob; state the argument that carries (phase-aware concurrency, owner-ref history, per-namespace cost). Restore the note that a condition-triggered mode is contemplated once capacity metrics land, correct the suspend/resume wording to match the code, and document the cascade-delete NOSPACE hazard. Signed-off-by: Andrey Kolkov --- docs/etcd-defrag.md | 45 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/docs/etcd-defrag.md b/docs/etcd-defrag.md index bf246f7d..834f3361 100644 --- a/docs/etcd-defrag.md +++ b/docs/etcd-defrag.md @@ -166,7 +166,9 @@ metadata: spec: clusterRef: name: etcd - schedule: "0 3 * * *" # standard five-field cron, evaluated in UTC + schedule: + cron: "0 3 * * *" # standard five-field cron + timezone: Europe/Moscow # optional IANA zone; UTC when absent concurrencyPolicy: Forbid # skip a tick while a previous run is still active (default) ttlSecondsAfterFinished: 3600 historyLimit: 3 # keep the last 3 finished runs @@ -176,15 +178,20 @@ spec: minReclaim: 32Mi ``` -- **`schedule`** is a standard five-field cron expression in UTC. Prefix it with - `CRON_TZ=` to use another zone. +- **`schedule.cron`** is a standard five-field cron expression (descriptors like + `@daily` are rejected). **`schedule.timezone`** is an IANA zone name it is read + in; absent means UTC. - **`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. +- **`suspend: true`** pauses stamping without deleting the policy. On resume the + single most recent missed tick may be stamped (subject to + `startingDeadlineSeconds`); earlier missed ticks are never replayed. - **`startingDeadlineSeconds`** skips a tick that is already older than the - deadline (e.g. after the operator was down) instead of starting it late. + deadline (e.g. after the operator was down) instead of starting it late. With + no deadline, a backlog longer than the catch-up window parks the policy on a + `TooManyMissedTicks` condition rather than guessing — set a deadline or check + the clock. - **`historyLimit`** caps retained finished runs; `ttlSecondsAfterFinished` (per run) is the other cleanup path. - **`rule`** / **`ttlSecondsAfterFinished`** are copied verbatim into each @@ -194,8 +201,34 @@ spec: twice), `status.lastSuccessfulTime` records the last `Complete`, and `status.active` lists runs still in flight. +Deleting a policy cascades to its runs. A run still `Running` when the policy is +deleted is aborted mid-sweep; if it had already reclaimed space it never disarms +a `NOSPACE` alarm, leaving the backend read-only. Suspend the policy (or delete +with `--cascade=orphan`) to let an in-flight run finish first. + +**Why a policy and not a CronJob.** The safety argument above is about the +`EtcdDefrag` run; it holds whether that run is created by the operator or by a +`kubectl create` in a CronJob. What a CronJob *cannot* express is the scheduling +itself: its `concurrencyPolicy` governs overlapping Jobs, but `kubectl create` +exits in milliseconds while the defrag it asked for runs for minutes, so it can +never skip a tick because last night's sweep is still going — `EtcdDefragPolicy` +gates on `EtcdDefrag.status.phase`, the thing actually still running. A Job also +can't own the CR it created, so the CronJob route leaks `EtcdDefrag` objects; +`historyLimit` plus the owner-ref cascade close that. And it saves a +per-namespace ServiceAccount + Role granting `create` on `etcddefrags` (a +privilege better not handed to a tenant) plus a pinned kubectl image to patch. +The API is a deliberate subset of CronJob — one `historyLimit`, no `Replace` +concurrency — not a clone. + ## Relationship to capacity metrics The capacity metrics and alert rules that tell you *when* a defrag is worth running are tracked separately (see #357); `EtcdDefrag` records sizes in its own `status` during a run rather than as continuously-scraped gauges. + +A condition-triggered mode — a policy that stamps a run when observed +fragmentation crosses a threshold, rather than on a clock — is contemplated once +those metrics land: nothing observes fragmentation between runs today, so +`schedule` is required for now. That mode would be a different feature, not a +replacement for cron, and whether the two are exclusive or combinable ("nightly, +or sooner if fragmentation trips") is left open here. From 1f5c114d999183b59bc8ec5d8381519b622fdcb8 Mon Sep 17 00:00:00 2001 From: Timofei Larkin Date: Tue, 25 Aug 2026 15:48:41 +0300 Subject: [PATCH 5/6] fix(defrag): resume a far-behind policy instead of parking it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bounding the catch-up walk by StartingDeadlineSeconds left it unbounded when no deadline is set: a policy more than defragPolicyMaxCatchup ticks behind returned errTooManyMissed on every pass, and because nothing advanced status.lastScheduleTime the backlog only grew. An hourly policy suspended four days, or any policy created while the operator was down long enough, parked on TooManyMissedTicks until someone edited the spec — a scheduler whose job is to eventually run, permanently not running. Give the no-deadline case a floor derived from the schedule itself, so the walk is bounded either way and a far-behind policy resumes at its most recent tick. Sampling consecutive intervals covers irregular schedules like "0 9,17 * * *", where measuring a single interval would take the short gap and drop a tick that is merely late. Folding the deadline into the cutoff also swallowed the skip: a dropped tick left the policy reading Scheduled with no event and no status trace of the run that never happened. Capture the anchor before the cutoff and emit MissedSchedule for the tick it drops, naming how late it was and whether the deadline or the period floor dropped it. Assisted-By: Claude Opus 5 Signed-off-by: Timofei Larkin --- api/v1alpha2/etcddefragpolicy_types.go | 6 +- ...rator.cozystack.io_etcddefragpolicies.yaml | 6 +- controllers/etcddefragpolicy_controller.go | 38 ++++++++++- .../etcddefragpolicy_controller_test.go | 67 ++++++++++++++++--- docs/etcd-defrag.md | 6 +- 5 files changed, 106 insertions(+), 17 deletions(-) diff --git a/api/v1alpha2/etcddefragpolicy_types.go b/api/v1alpha2/etcddefragpolicy_types.go index bb62b649..3286dc57 100644 --- a/api/v1alpha2/etcddefragpolicy_types.go +++ b/api/v1alpha2/etcddefragpolicy_types.go @@ -108,8 +108,10 @@ type EtcdDefragPolicySpec struct { // 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. + // stamped a run for, or consumed by skipping under ConcurrencyPolicy. It + // anchors the next tick, so a tick is never acted on twice. A tick dropped + // for being too far in the past does not advance it; those are reported as + // MissedSchedule events. // +optional LastScheduleTime *metav1.Time `json:"lastScheduleTime,omitempty"` 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 index 7efc2e40..4ee7ce33 100644 --- a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml @@ -289,8 +289,10 @@ spec: 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. + stamped a run for, or consumed by skipping under ConcurrencyPolicy. It + anchors the next tick, so a tick is never acted on twice. A tick dropped + for being too far in the past does not advance it; those are reported as + MissedSchedule events. format: date-time type: string lastSuccessfulTime: diff --git a/controllers/etcddefragpolicy_controller.go b/controllers/etcddefragpolicy_controller.go index 83aa2a9d..ba7b7d12 100644 --- a/controllers/etcddefragpolicy_controller.go +++ b/controllers/etcddefragpolicy_controller.go @@ -120,10 +120,27 @@ func (r *EtcdDefragPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Req // StartingDeadlineSeconds bounds how late a tick may still start, so a tick // older than the deadline is skipped anyway: never walk further back than the // deadline window. This both enforces the deadline and keeps the walk short. + // With no deadline the floor is one schedule period, so a policy that fell + // arbitrarily far behind resumes at its most recent tick instead of walking + // every slot since it was created. + anchor := earliest + cutoff := now.Add(-lookbackFloor(sched, now)) if d := pol.Spec.StartingDeadlineSeconds; d != nil { - if cutoff := now.Add(-time.Duration(*d) * time.Second); cutoff.After(earliest) { - earliest = cutoff + cutoff = now.Add(-time.Duration(*d) * time.Second) + } + if cutoff.After(earliest) { + earliest = cutoff + } + // A tick the cutoff dropped is a tick deliberately not started. Say so: the + // alternative is a policy reporting Scheduled with no trace of the skip. + if dropped := sched.Next(anchor); earliest.After(anchor) && dropped.Before(earliest) { + why := "further behind than one schedule period" + if pol.Spec.StartingDeadlineSeconds != nil { + why = fmt.Sprintf("past the %ds starting deadline", *pol.Spec.StartingDeadlineSeconds) } + r.event(pol, corev1.EventTypeWarning, "MissedSchedule", + fmt.Sprintf("skipped scheduled time %s: %s late, %s", + tickString(dropped), now.Sub(dropped).Truncate(time.Second), why)) } due, next, err := nextSchedule(sched, earliest, now, defragPolicyMaxCatchup) if err != nil { @@ -289,6 +306,23 @@ func nextSchedule(sched cron.Schedule, earliest, now time.Time, maxCatchup int) return &last, t, nil } +// lookbackFloor bounds how far back nextSchedule walks when no starting +// deadline is set. Sampling consecutive intervals covers irregular schedules +// ("0 9,17 * * *"), and doubling the widest leaves margin for a tick that is +// merely late rather than abandoned. +func lookbackFloor(sched cron.Schedule, now time.Time) time.Duration { + t := sched.Next(now) + var widest time.Duration + for i := 0; i < 4; i++ { + n := sched.Next(t) + if d := n.Sub(t); d > widest { + widest = d + } + t = n + } + return 2 * widest +} + // 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 { diff --git a/controllers/etcddefragpolicy_controller_test.go b/controllers/etcddefragpolicy_controller_test.go index da894d9f..0456be1c 100644 --- a/controllers/etcddefragpolicy_controller_test.go +++ b/controllers/etcddefragpolicy_controller_test.go @@ -13,6 +13,7 @@ package controllers import ( "context" "errors" + "strings" "testing" "time" @@ -283,19 +284,69 @@ func TestDefragPolicy_MissedDeadline(t *testing.T) { } } -// A backlog longer than the catch-up window, with no deadline to bound it, -// parks the policy on TooManyMissedTicks instead of fabricating a run. -func TestDefragPolicy_TooManyMissedTicks(t *testing.T) { +// A policy arbitrarily far behind resumes at its most recent tick rather than +// parking: with no deadline the lookback floor bounds the walk, so one run is +// stamped for the latest tick and the skipped backlog is reported as an event. +// Parking instead would never recover, since nothing advances the anchor. +func TestDefragPolicy_FarBehindResumes(t *testing.T) { pol := defragPolicy("p", "* * * * *") // every minute: 1000h backlog >> maxCatchup - r, c := policyReconciler(t, epoch.Add(1000*time.Hour), pol) + rec := record.NewFakeRecorder(20) + c, s := newTestClient(t, pol) + now := epoch.Add(1000 * time.Hour) + r := &EtcdDefragPolicyReconciler{Client: c, Scheme: s, Recorder: rec, now: func() time.Time { return now }} reconcilePolicy(t, r, "p") - if runs := listPolicyRuns(t, c, "p"); len(runs) != 0 { - t.Fatalf("stamped a run on a too-large backlog: %d runs", len(runs)) + runs := listPolicyRuns(t, c, "p") + if len(runs) != 1 { + t.Fatalf("stamped %d runs on a large backlog, want 1 (the most recent tick)", len(runs)) } got := mustGet(t, c, "p", "ns", &lll.EtcdDefragPolicy{}) - if cond := findPolicyCond(got); cond == nil || cond.Reason != "TooManyMissedTicks" { - t.Errorf("condition = %+v, want TooManyMissedTicks", cond) + if cond := findPolicyCond(got); cond == nil || cond.Reason != "Scheduled" { + t.Errorf("condition = %+v, want Scheduled (the policy must not park)", cond) + } + if got.Status.LastScheduleTime == nil || !got.Status.LastScheduleTime.Time.Equal(now) { + t.Errorf("lastScheduleTime = %v, want the latest tick %s", got.Status.LastScheduleTime, now) + } + if !drainFor(rec, "MissedSchedule") { + t.Error("skipping a backlog should emit a MissedSchedule warning, not pass silently") + } + + // The anchor advanced, so a second pass is a no-op rather than a repeat. + reconcilePolicy(t, r, "p") + if runs := listPolicyRuns(t, c, "p"); len(runs) != 1 { + t.Errorf("second pass stamped again: %d runs, want 1", len(runs)) + } +} + +// A tick dropped by StartingDeadlineSeconds is reported, not silently swallowed: +// otherwise the policy reads Scheduled with no trace of the run that never was. +func TestDefragPolicy_MissedDeadlineIsReported(t *testing.T) { + deadline := int64(60) + pol := defragPolicy("p", "0 * * * *", func(p *lll.EtcdDefragPolicy) { p.Spec.StartingDeadlineSeconds = &deadline }) + rec := record.NewFakeRecorder(20) + c, s := newTestClient(t, pol) + r := &EtcdDefragPolicyReconciler{Client: c, Scheme: s, Recorder: rec, now: func() time.Time { return epoch.Add(90 * time.Minute) }} + + 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)) + } + if !drainFor(rec, "MissedSchedule") { + t.Error("a deadline-skipped tick should emit a MissedSchedule warning") + } +} + +// drainFor reports whether any buffered event mentions reason. +func drainFor(rec *record.FakeRecorder, reason string) bool { + for { + select { + case e := <-rec.Events: + if strings.Contains(e, reason) { + return true + } + default: + return false + } } } diff --git a/docs/etcd-defrag.md b/docs/etcd-defrag.md index 834f3361..91e5a93f 100644 --- a/docs/etcd-defrag.md +++ b/docs/etcd-defrag.md @@ -189,9 +189,9 @@ spec: `startingDeadlineSeconds`); earlier missed ticks are never replayed. - **`startingDeadlineSeconds`** skips a tick that is already older than the deadline (e.g. after the operator was down) instead of starting it late. With - no deadline, a backlog longer than the catch-up window parks the policy on a - `TooManyMissedTicks` condition rather than guessing — set a deadline or check - the clock. + no deadline the window is one schedule period, so a policy that fell far + behind resumes at its most recent tick instead of replaying the backlog. + Either way a skipped tick is reported as a `MissedSchedule` event. - **`historyLimit`** caps retained finished runs; `ttlSecondsAfterFinished` (per run) is the other cleanup path. - **`rule`** / **`ttlSecondsAfterFinished`** are copied verbatim into each From 72cd27a33d27c9604082dd8f6f94453a0016bd39 Mon Sep 17 00:00:00 2001 From: Timofei Larkin Date: Tue, 25 Aug 2026 17:44:07 +0300 Subject: [PATCH 6/6] fix(defrag): bound the policy name and the starting deadline The policy name becomes a label value on every stamped EtcdDefrag, and the controller selects its own runs by that label. Label values cap at 63 characters, so a longer policy name made the controller's own selector unparseable: the List at the top of Reconcile failed before any condition was set, leaving the policy backing off with nothing in its status to say why. Cap the name at 52, which also leaves room for the "-" suffix on the stamped run and matches what CronJob caps its own names to for the same reason. StartingDeadlineSeconds accepted any non-negative int64 but is multiplied out to a time.Duration, which overflows past ~292 years. A value one second past that wrapped to a negative window and pushed the catch-up cutoff into the future, so no tick was ever due and the policy reported Scheduled while silently never running. Cap it at ten years; anything near that already means "no deadline". Both are enforced at admission, covered by CEL tests against a real apiserver. Assisted-By: Claude Opus 5 Signed-off-by: Timofei Larkin --- api/v1alpha2/defragpolicy_cel_test.go | 100 ++++++++++++++++++ api/v1alpha2/etcddefragpolicy_types.go | 11 ++ ...rator.cozystack.io_etcddefragpolicies.yaml | 9 ++ 3 files changed, 120 insertions(+) create mode 100644 api/v1alpha2/defragpolicy_cel_test.go diff --git a/api/v1alpha2/defragpolicy_cel_test.go b/api/v1alpha2/defragpolicy_cel_test.go new file mode 100644 index 00000000..e68dca58 --- /dev/null +++ b/api/v1alpha2/defragpolicy_cel_test.go @@ -0,0 +1,100 @@ +/* +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 v1alpha2_test + +import ( + "context" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + lll "github.com/cozystack/etcd-operator/api/v1alpha2" +) + +func defragPolicy(name string) *lll.EtcdDefragPolicy { + return &lll.EtcdDefragPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: lll.EtcdDefragPolicySpec{ + ClusterRef: corev1.LocalObjectReference{Name: "c1"}, + Schedule: lll.DefragSchedule{Cron: "0 3 * * *"}, + }, + } +} + +// The policy name becomes a label value on every stamped EtcdDefrag, and the +// controller selects its own runs by that label. A name past the 63-character +// label cap makes that selector unparseable, so the reconcile fails on its +// first List — before it can set any condition. Reject it at admission instead. +func TestCEL_DefragPolicyNameLength(t *testing.T) { + skipIfNoEnvtest(t) + ctx := context.Background() + + t.Run("at the cap accepted", func(t *testing.T) { + p := defragPolicy(strings.Repeat("a", 52)) + if err := k8s.Create(ctx, p); err != nil { + t.Fatalf("apiserver rejected a 52-character name: %v", err) + } + _ = k8s.Delete(ctx, p) + }) + + t.Run("past the cap rejected", func(t *testing.T) { + p := defragPolicy(strings.Repeat("b", 53)) + err := k8s.Create(ctx, p) + if err == nil { + _ = k8s.Delete(ctx, p) + t.Fatal("apiserver accepted a 53-character name; expected rejection") + } + if !strings.Contains(err.Error(), "52 characters or fewer") { + t.Fatalf("error did not mention the name cap: %v", err) + } + }) + + // The case that motivated the cap: past 63, the label selector itself is + // invalid, so nothing the controller does can report the problem. + t.Run("past the label cap rejected", func(t *testing.T) { + p := defragPolicy(strings.Repeat("c", 64)) + err := k8s.Create(ctx, p) + if err == nil { + _ = k8s.Delete(ctx, p) + t.Fatal("apiserver accepted a 64-character name; expected rejection") + } + }) +} + +// StartingDeadlineSeconds is multiplied out to a time.Duration, which overflows +// past ~292 years and wraps to a negative window that suppresses every tick. +func TestCEL_DefragPolicyStartingDeadlineBounded(t *testing.T) { + skipIfNoEnvtest(t) + ctx := context.Background() + + t.Run("ordinary deadline accepted", func(t *testing.T) { + p := defragPolicy("deadline-ok") + d := int64(3600) + p.Spec.StartingDeadlineSeconds = &d + if err := k8s.Create(ctx, p); err != nil { + t.Fatalf("apiserver rejected a one-hour deadline: %v", err) + } + _ = k8s.Delete(ctx, p) + }) + + t.Run("overflowing deadline rejected", func(t *testing.T) { + p := defragPolicy("deadline-overflow") + d := int64(9223372037) // one second past what time.Duration can hold + p.Spec.StartingDeadlineSeconds = &d + err := k8s.Create(ctx, p) + if err == nil { + _ = k8s.Delete(ctx, p) + t.Fatal("apiserver accepted an overflowing startingDeadlineSeconds") + } + }) +} diff --git a/api/v1alpha2/etcddefragpolicy_types.go b/api/v1alpha2/etcddefragpolicy_types.go index 3286dc57..40928c2f 100644 --- a/api/v1alpha2/etcddefragpolicy_types.go +++ b/api/v1alpha2/etcddefragpolicy_types.go @@ -82,7 +82,12 @@ type EtcdDefragPolicySpec struct { // 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. + // + // Capped at ten years: the value is multiplied out to a time.Duration, which + // overflows past ~292 years and wraps to a negative window that silently + // suppresses every tick. Anything near the cap already means "no deadline". // +kubebuilder:validation:Minimum=0 + // +kubebuilder:validation:Maximum=315360000 // +optional StartingDeadlineSeconds *int64 `json:"startingDeadlineSeconds,omitempty"` @@ -132,6 +137,12 @@ type EtcdDefragPolicyStatus struct { // +kubebuilder:object:root=true // +kubebuilder:subresource:status +// The name goes into a label value on every stamped EtcdDefrag, and label +// values cap at 63 characters — a longer name makes the controller's own +// label selector unparseable, so it fails before it can report anything. 52 +// leaves room for the "-" suffix on the stamped run, and matches the +// cap CronJob applies to its own names for the same reason. +// +kubebuilder:validation:XValidation:rule="size(self.metadata.name) <= 52",message="metadata.name must be 52 characters or fewer: it becomes a label value on each stamped EtcdDefrag" // +kubebuilder:printcolumn:name="Cluster",type=string,JSONPath=`.spec.clusterRef.name` // +kubebuilder:printcolumn:name="Schedule",type=string,JSONPath=`.spec.schedule.cron` // +kubebuilder:printcolumn:name="Timezone",type=string,JSONPath=`.spec.schedule.timezone` 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 index 4ee7ce33..70c736c0 100644 --- a/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml +++ b/charts/etcd-operator/crd-bases/etcd-operator.cozystack.io_etcddefragpolicies.yaml @@ -180,7 +180,12 @@ spec: 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. + + Capped at ten years: the value is multiplied out to a time.Duration, which + overflows past ~292 years and wraps to a negative window that silently + suppresses every tick. Anything near the cap already means "no deadline". format: int64 + maximum: 315360000 minimum: 0 type: integer suspend: @@ -302,6 +307,10 @@ spec: type: string type: object type: object + x-kubernetes-validations: + - message: 'metadata.name must be 52 characters or fewer: it becomes a label + value on each stamped EtcdDefrag' + rule: size(self.metadata.name) <= 52 served: true storage: true subresources: