From 8be54050aed1ef6d97c93f0ce28dc0c55ebcfde6 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 20 Aug 2026 12:07:06 +0400 Subject: [PATCH 1/4] =?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/4] 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/4] 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/4] 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.