You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by vi...@apache.org on 2026/05/11 09:22:30 UTC
(superset-kubernetes-operator) branch main updated: feat(lifecycle): add cron scheduling for clone task (#41)
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/superset-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 9c7e5e6 feat(lifecycle): add cron scheduling for clone task (#41)
9c7e5e6 is described below
commit 9c7e5e687bb97330cd9a9f94d2ef0f238ab61abd
Author: Ville Brofeldt <33...@users.noreply.github.com>
AuthorDate: Mon May 11 02:22:25 2026 -0700
feat(lifecycle): add cron scheduling for clone task (#41)
---
api/v1alpha1/superset_types.go | 27 ++-
api/v1alpha1/zz_generated.deepcopy.go | 31 ++-
.../crds/superset.apache.org_supersets.yaml | 22 ++
.../crd/bases/superset.apache.org_supersets.yaml | 22 ++
docs/reference/api-reference.md | 28 +++
docs/user-guide/lifecycle.md | 51 ++++-
go.mod | 1 +
go.sum | 2 +
internal/controller/clone_test.go | 238 ++++++++++++++++++++-
internal/controller/superset_controller.go | 89 ++++++++
internal/schedule/schedule.go | 87 ++++++++
internal/schedule/schedule_test.go | 150 +++++++++++++
12 files changed, 741 insertions(+), 7 deletions(-)
diff --git a/api/v1alpha1/superset_types.go b/api/v1alpha1/superset_types.go
index 6d9f450..562a5eb 100644
--- a/api/v1alpha1/superset_types.go
+++ b/api/v1alpha1/superset_types.go
@@ -270,6 +270,25 @@ type BaseTaskSpec struct {
Disabled *bool `json:"disabled,omitempty"`
}
+// SchedulableBaseTaskSpec extends BaseTaskSpec with cron-based scheduling.
+// Tasks that embed this type can be periodically re-executed without external
+// triggers. The schedule is additive to the manual trigger field.
+type SchedulableBaseTaskSpec struct {
+ BaseTaskSpec `json:",inline"`
+
+ // CronSchedule is a 5-field cron expression (minute hour day-of-month month
+ // day-of-week) that triggers periodic re-execution of this task and all
+ // downstream tasks. When the clock crosses a cron boundary, the task
+ // checksum changes and the lifecycle pipeline re-runs.
+ //
+ // Uses standard cron syntax. Examples: "0 2 * * *" (daily 2 AM UTC),
+ // "0 */6 * * *" (every 6 hours), "30 1 * * 1" (Mondays 1:30 AM UTC).
+ // +optional
+ // +kubebuilder:validation:MinLength=9
+ // +kubebuilder:validation:MaxLength=256
+ CronSchedule *string `json:"cronSchedule,omitempty"`
+}
+
// LifecycleSpec defines lifecycle management configuration for database migrations
// and application initialization tasks.
// +kubebuilder:validation:XValidation:rule="!has(self.init) || !has(self.init.command) || size(self.init.command) == 0 || (!has(self.init.adminUser) && !has(self.init.loadExamples))",message="init.command is mutually exclusive with init.adminUser and init.loadExamples"
@@ -385,7 +404,7 @@ type PodRetentionSpec struct {
// Only allowed in Development or Staging mode.
// Triggers on source config changes and the trigger field (inherited from BaseTaskSpec).
type CloneTaskSpec struct {
- BaseTaskSpec `json:",inline"`
+ SchedulableBaseTaskSpec `json:",inline"`
// Source database to clone from (typically production, read-only user).
Source CloneSourceSpec `json:"source"`
@@ -625,6 +644,12 @@ type TaskRefStatus struct {
Image string `json:"image,omitempty"`
// +optional
Message string `json:"message,omitempty"`
+ // LastScheduledAt is the cron tick that triggered the most recent scheduled run.
+ // +optional
+ LastScheduledAt *metav1.Time `json:"lastScheduledAt,omitempty"`
+ // NextScheduleAt is the next future cron tick when the schedule will fire.
+ // +optional
+ NextScheduleAt *metav1.Time `json:"nextScheduleAt,omitempty"`
}
// UpgradeContext tracks the current upgrade operation.
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
index 2f3e6dd..a399e32 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -361,7 +361,7 @@ func (in *CloneSourceSpec) DeepCopy() *CloneSourceSpec {
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *CloneTaskSpec) DeepCopyInto(out *CloneTaskSpec) {
*out = *in
- in.BaseTaskSpec.DeepCopyInto(&out.BaseTaskSpec)
+ in.SchedulableBaseTaskSpec.DeepCopyInto(&out.SchedulableBaseTaskSpec)
in.Source.DeepCopyInto(&out.Source)
if in.ExcludeTables != nil {
in, out := &in.ExcludeTables, &out.ExcludeTables
@@ -1492,6 +1492,27 @@ func (in *ScalableComponentSpec) DeepCopy() *ScalableComponentSpec {
return out
}
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SchedulableBaseTaskSpec) DeepCopyInto(out *SchedulableBaseTaskSpec) {
+ *out = *in
+ in.BaseTaskSpec.DeepCopyInto(&out.BaseTaskSpec)
+ if in.CronSchedule != nil {
+ in, out := &in.CronSchedule, &out.CronSchedule
+ *out = new(string)
+ **out = **in
+ }
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SchedulableBaseTaskSpec.
+func (in *SchedulableBaseTaskSpec) DeepCopy() *SchedulableBaseTaskSpec {
+ if in == nil {
+ return nil
+ }
+ out := new(SchedulableBaseTaskSpec)
+ in.DeepCopyInto(out)
+ return out
+}
+
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ServiceAccountSpec) DeepCopyInto(out *ServiceAccountSpec) {
*out = *in
@@ -2480,6 +2501,14 @@ func (in *TaskRefStatus) DeepCopyInto(out *TaskRefStatus) {
in, out := &in.CompletedAt, &out.CompletedAt
*out = (*in).DeepCopy()
}
+ if in.LastScheduledAt != nil {
+ in, out := &in.LastScheduledAt, &out.LastScheduledAt
+ *out = (*in).DeepCopy()
+ }
+ if in.NextScheduleAt != nil {
+ in, out := &in.NextScheduleAt, &out.NextScheduleAt
+ *out = (*in).DeepCopy()
+ }
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TaskRefStatus.
diff --git a/charts/superset-operator/crds/superset.apache.org_supersets.yaml b/charts/superset-operator/crds/superset.apache.org_supersets.yaml
index f2d50fe..827c07c 100644
--- a/charts/superset-operator/crds/superset.apache.org_supersets.yaml
+++ b/charts/superset-operator/crds/superset.apache.org_supersets.yaml
@@ -11973,6 +11973,10 @@ spec:
type: string
type: array
x-kubernetes-list-type: atomic
+ cronSchedule:
+ maxLength: 256
+ minLength: 9
+ type: string
disabled:
type: boolean
excludeTableData:
@@ -35608,8 +35612,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
@@ -35635,8 +35645,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
@@ -35662,8 +35678,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
diff --git a/config/crd/bases/superset.apache.org_supersets.yaml b/config/crd/bases/superset.apache.org_supersets.yaml
index f2d50fe..827c07c 100644
--- a/config/crd/bases/superset.apache.org_supersets.yaml
+++ b/config/crd/bases/superset.apache.org_supersets.yaml
@@ -11973,6 +11973,10 @@ spec:
type: string
type: array
x-kubernetes-list-type: atomic
+ cronSchedule:
+ maxLength: 256
+ minLength: 9
+ type: string
disabled:
type: boolean
excludeTableData:
@@ -35608,8 +35612,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
@@ -35635,8 +35645,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
@@ -35662,8 +35678,14 @@ spec:
type: string
image:
type: string
+ lastScheduledAt:
+ format: date-time
+ type: string
message:
type: string
+ nextScheduleAt:
+ format: date-time
+ type: string
podName:
type: string
startedAt:
diff --git a/docs/reference/api-reference.md b/docs/reference/api-reference.md
index 8bef937..9f3b812 100644
--- a/docs/reference/api-reference.md
+++ b/docs/reference/api-reference.md
@@ -84,6 +84,7 @@ _Appears in:_
- [CloneTaskSpec](#clonetaskspec)
- [InitTaskSpec](#inittaskspec)
- [MigrateTaskSpec](#migratetaskspec)
+- [SchedulableBaseTaskSpec](#schedulablebasetaskspec)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
@@ -257,6 +258,7 @@ _Appears in:_
| `timeout` _[Duration](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1#Duration)_ | Maximum timeout per attempt. | | Optional: \{\} <br /> |
| `maxRetries` _integer_ | Maximum number of retries before permanent failure. | 3 | Minimum: 1 <br />Optional: \{\} <br /> |
| `disabled` _boolean_ | Disabled skips this task entirely when true. | | Optional: \{\} <br /> |
+| `cronSchedule` _string_ | CronSchedule is a 5-field cron expression (minute hour day-of-month month<br />day-of-week) that triggers periodic re-execution of this task and all<br />downstream tasks. When the clock crosses a cron boundary, the task<br />checksum changes and the lifecycle pipeline re-runs.<br />Uses standard cron syntax. Examples: "0 2 * * *" (daily 2 AM UTC),<br />"0 */6 * * *" (every 6 hours), "30 1 * * 1" (Mondays 1:30 AM UTC). | | MaxLength: 256 <br />MinLength: 9 <b [...]
| `source` _[CloneSourceSpec](#clonesourcespec)_ | Source database to clone from (typically production, read-only user). | | |
| `excludeTables` _string array_ | Tables to exclude entirely from the dump (schema and data). | | Optional: \{\} <br /> |
| `excludeTableData` _string array_ | Tables where schema is dumped but data is not. Useful for large tables<br />needed by migrations but not for testing (e.g., "logs", "query"). | | Optional: \{\} <br /> |
@@ -949,6 +951,30 @@ _Appears in:_
| `podDisruptionBudget` _[PDBSpec](#pdbspec)_ | PodDisruptionBudget for protecting availability during voluntary disruptions. Overrides spec.podDisruptionBudget. | | Optional: \{\} <br /> |
+#### SchedulableBaseTaskSpec
+
+
+
+SchedulableBaseTaskSpec extends BaseTaskSpec with cron-based scheduling.
+Tasks that embed this type can be periodically re-executed without external
+triggers. The schedule is additive to the manual trigger field.
+
+
+
+_Appears in:_
+- [CloneTaskSpec](#clonetaskspec)
+
+| Field | Description | Default | Validation |
+| --- | --- | --- | --- |
+| `command` _string array_ | Command override for the task pod. | | Optional: \{\} <br /> |
+| `trigger` _string_ | Trigger is an opaque string. Changing its value forces a re-run of this<br />task and all downstream tasks. Use a timestamp, UUID, or CI build ID. | | Optional: \{\} <br /> |
+| `requiresDrain` _boolean_ | RequiresDrain controls whether components must be scaled to zero before<br />this task runs. When true, the operator deletes all component child CRs<br />before executing the task pod, preventing database connection conflicts.<br />Defaults vary per task type: true for clone and migrate, false for init. | | Optional: \{\} <br /> |
+| `timeout` _[Duration](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1#Duration)_ | Maximum timeout per attempt. | | Optional: \{\} <br /> |
+| `maxRetries` _integer_ | Maximum number of retries before permanent failure. | 3 | Minimum: 1 <br />Optional: \{\} <br /> |
+| `disabled` _boolean_ | Disabled skips this task entirely when true. | | Optional: \{\} <br /> |
+| `cronSchedule` _string_ | CronSchedule is a 5-field cron expression (minute hour day-of-month month<br />day-of-week) that triggers periodic re-execution of this task and all<br />downstream tasks. When the clock crosses a cron boundary, the task<br />checksum changes and the lifecycle pipeline re-runs.<br />Uses standard cron syntax. Examples: "0 2 * * *" (daily 2 AM UTC),<br />"0 */6 * * *" (every 6 hours), "30 1 * * 1" (Mondays 1:30 AM UTC). | | MaxLength: 256 <br />MinLength: 9 <b [...]
+
+
#### ServiceAccountSpec
@@ -1534,6 +1560,8 @@ _Appears in:_
| `podName` _string_ | | | Optional: \{\} <br /> |
| `image` _string_ | | | Optional: \{\} <br /> |
| `message` _string_ | | | Optional: \{\} <br /> |
+| `lastScheduledAt` _[Time](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1#Time)_ | LastScheduledAt is the cron tick that triggered the most recent scheduled run. | | Optional: \{\} <br /> |
+| `nextScheduleAt` _[Time](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1#Time)_ | NextScheduleAt is the next future cron tick when the schedule will fire. | | Optional: \{\} <br /> |
#### UpgradeContext
diff --git a/docs/user-guide/lifecycle.md b/docs/user-guide/lifecycle.md
index d8f3ecd..0ece15f 100644
--- a/docs/user-guide/lifecycle.md
+++ b/docs/user-guide/lifecycle.md
@@ -51,7 +51,7 @@ Each task has hardcoded trigger inputs — what it watches for changes:
| Task | Watches | Re-runs when... |
|------|---------|-----------------|
-| Clone | `trigger` field, source config, excludes | Trigger value changes, or source DB config changes |
+| Clone | `trigger` field, `cronSchedule` tick, source config, excludes | Trigger value changes, schedule tick boundary crossed, or source DB config changes |
| Migrate | Image (resolved lifecycle image) | Image tag or repository changes |
| Init | Config checksum (rendered Python config) | Any config-affecting field changes |
@@ -83,6 +83,55 @@ spec:
disabled: true # user manages migrations externally
```
+### Scheduled Execution
+
+Tasks that support scheduling (currently clone) accept a `cronSchedule` field —
+a standard 5-field cron expression that triggers periodic re-execution:
+
+```yaml
+spec:
+ environment: Staging
+ lifecycle:
+ clone:
+ cronSchedule: "0 2 * * *" # daily at 2 AM UTC
+ source:
+ host: postgres-prod.db.svc
+ database: superset_prod
+ username: prod_reader
+ passwordFrom:
+ name: prod-reader-creds
+ key: password
+```
+
+When a schedule is configured, the operator automatically re-runs the full
+lifecycle pipeline (clone → migrate → init) each time a cron tick boundary is
+crossed. The `trigger` field remains functional for manual overrides on top of
+the schedule — both contribute independently.
+
+**How it works:**
+
+- The operator computes the "current tick" (most recent past time matching the
+ expression) and includes it in the task checksum
+- When the clock crosses a cron boundary, the tick changes, the checksum changes,
+ and the pipeline re-runs
+- The operator requeues itself to wake at the next cron tick
+- If the operator is down during a scheduled tick, it catches up on the next
+ reconcile
+- If the pipeline is still running when a tick fires, it completes normally; the
+ new tick is detected afterward and triggers one re-run (no backlog accumulation)
+
+**Status reporting:**
+
+The clone task status includes `lastScheduledAt` (the tick that triggered the
+most recent run) and `nextScheduleAt` (the next future tick).
+
+**Alternative — external CronJob:**
+
+For teams that prefer external scheduling, a Kubernetes CronJob can patch
+the `trigger` field on a cron schedule. This requires a CronJob resource,
+ServiceAccount, RoleBinding, and a kubectl image, but keeps the scheduling
+logic outside the operator.
+
When disabled, the task's CR is deleted and it does not participate in the
pipeline. Downstream tasks still run but don't receive propagation from the
disabled task.
diff --git a/go.mod b/go.mod
index 48256c8..373e640 100644
--- a/go.mod
+++ b/go.mod
@@ -6,6 +6,7 @@ require (
github.com/Masterminds/semver/v3 v3.4.0
github.com/onsi/ginkgo/v2 v2.28.2
github.com/onsi/gomega v1.39.1
+ github.com/robfig/cron/v3 v3.0.1
github.com/stretchr/testify v1.11.1
k8s.io/api v0.35.1
k8s.io/apimachinery v0.35.1
diff --git a/go.sum b/go.sum
index d2a6e01..fca465d 100644
--- a/go.sum
+++ b/go.sum
@@ -118,6 +118,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+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/internal/controller/clone_test.go b/internal/controller/clone_test.go
index 87a85d2..06dce61 100644
--- a/internal/controller/clone_test.go
+++ b/internal/controller/clone_test.go
@@ -21,6 +21,7 @@ package controller
import (
"strings"
"testing"
+ "time"
corev1 "k8s.io/api/core/v1"
@@ -146,8 +147,10 @@ func TestBuildCloneCommand_CustomCommand(t *testing.T) {
superset := &supersetv1alpha1.Superset{}
superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
Clone: &supersetv1alpha1.CloneTaskSpec{
- BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{
- Command: []string{"/bin/sh", "-c", "custom-clone-script.sh"},
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{
+ BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{
+ Command: []string{"/bin/sh", "-c", "custom-clone-script.sh"},
+ },
},
Source: supersetv1alpha1.CloneSourceSpec{
Host: "pg-prod.svc",
@@ -409,8 +412,8 @@ func TestIsTaskEnabled(t *testing.T) {
name: "clone explicitly disabled",
spec: &supersetv1alpha1.LifecycleSpec{
Clone: &supersetv1alpha1.CloneTaskSpec{
- BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{Disabled: common.Ptr(true)},
- Source: supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"},
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{Disabled: common.Ptr(true)}},
+ Source: supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"},
},
},
taskType: taskTypeClone,
@@ -752,3 +755,230 @@ func TestPipelineChain_CustomTaskSlotsBetweenStages(t *testing.T) {
t.Error("migrate should be stable when nothing upstream changed")
}
}
+
+func TestCloneInputs_ScheduleTickChangesChecksum(t *testing.T) {
+ // When the clock crosses a cron boundary, the clone checksum should change.
+ before := time.Date(2026, 5, 11, 1, 59, 0, 0, time.UTC)
+ after := time.Date(2026, 5, 11, 2, 1, 0, 0, time.UTC)
+
+ cronExpr := "0 2 * * *"
+ source := supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"}
+
+ r1 := &SupersetReconciler{Now: func() time.Time { return before }}
+ r2 := &SupersetReconciler{Now: func() time.Time { return after }}
+
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{CronSchedule: &cronExpr},
+ Source: source,
+ },
+ }
+
+ inputs1 := r1.cloneInputs(superset)
+ inputs2 := r2.cloneInputs(superset)
+
+ checksum1 := r1.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs1)
+ checksum2 := r2.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs2)
+
+ if checksum1 == checksum2 {
+ t.Error("clone checksum should change when crossing a cron boundary")
+ }
+}
+
+func TestCloneInputs_ScheduleAndTrigger_BothContribute(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 0, 0, 0, time.UTC)
+ r := &SupersetReconciler{Now: func() time.Time { return now }}
+
+ cronExpr := "0 * * * *"
+ source := supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"}
+ trigger1 := "v1"
+ trigger2 := "v2"
+
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{
+ BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{Trigger: &trigger1},
+ CronSchedule: &cronExpr,
+ },
+ Source: source,
+ },
+ }
+
+ inputs1 := r.cloneInputs(superset)
+ checksum1 := r.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs1)
+
+ // Change trigger only.
+ superset.Spec.Lifecycle.Clone.Trigger = &trigger2
+ inputs2 := r.cloneInputs(superset)
+ checksum2 := r.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs2)
+
+ if checksum1 == checksum2 {
+ t.Error("changing trigger should change checksum even with same schedule tick")
+ }
+}
+
+func TestCloneInputs_NoSchedule_StableChecksum(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 0, 0, 0, time.UTC)
+ r := &SupersetReconciler{Now: func() time.Time { return now }}
+
+ source := supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"}
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ Source: source,
+ },
+ }
+
+ inputs1 := r.cloneInputs(superset)
+ inputs2 := r.cloneInputs(superset)
+
+ checksum1 := r.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs1)
+ checksum2 := r.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, inputs2)
+
+ if checksum1 != checksum2 {
+ t.Error("checksum should be stable when no schedule is set")
+ }
+}
+
+func TestCloneInputs_ScheduleStableWithinWindow(t *testing.T) {
+ // Two reconciles within the same cron window produce the same checksum.
+ t1 := time.Date(2026, 5, 11, 14, 10, 0, 0, time.UTC)
+ t2 := time.Date(2026, 5, 11, 14, 50, 0, 0, time.UTC)
+
+ cronExpr := "0 * * * *"
+ source := supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"}
+
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{CronSchedule: &cronExpr},
+ Source: source,
+ },
+ }
+
+ r1 := &SupersetReconciler{Now: func() time.Time { return t1 }}
+ r2 := &SupersetReconciler{Now: func() time.Time { return t2 }}
+
+ checksum1 := r1.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, r1.cloneInputs(superset))
+ checksum2 := r2.computeStepChecksum("uid", taskTypeClone, []string{"cmd"}, r2.cloneInputs(superset))
+
+ if checksum1 != checksum2 {
+ t.Error("checksum should be stable within the same cron window")
+ }
+}
+
+func TestPipelineChain_ScheduleTickPropagatesDownstream(t *testing.T) {
+ before := time.Date(2026, 5, 11, 1, 59, 0, 0, time.UTC)
+ after := time.Date(2026, 5, 11, 2, 1, 0, 0, time.UTC)
+
+ cronExpr := "0 2 * * *"
+ source := supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"}
+
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{CronSchedule: &cronExpr},
+ Source: source,
+ },
+ }
+
+ cloneCmd := []string{"/bin/sh", "-c", "pg_dump | psql"}
+ migrateCmd := []string{"/bin/sh", "-c", "superset db upgrade"}
+ initCmd := []string{"/bin/sh", "-c", "superset init"}
+ parentUID := "test-uid"
+
+ // Before boundary.
+ r1 := &SupersetReconciler{Now: func() time.Time { return before }}
+ cloneChecksum1 := r1.computeStepChecksum(parentUID, taskTypeClone, cloneCmd, r1.cloneInputs(superset))
+ migrateChecksum1 := r1.computeStepChecksum(cloneChecksum1, taskTypeMigrate, migrateCmd, struct {
+ Image string
+ Trigger string
+ }{"img:4.0", ""})
+ initChecksum1 := r1.computeStepChecksum(migrateChecksum1, taskTypeInit, initCmd, struct {
+ ConfigChecksum string
+ Trigger string
+ }{"cfg", ""})
+
+ // After boundary.
+ r2 := &SupersetReconciler{Now: func() time.Time { return after }}
+ cloneChecksum2 := r2.computeStepChecksum(parentUID, taskTypeClone, cloneCmd, r2.cloneInputs(superset))
+ migrateChecksum2 := r2.computeStepChecksum(cloneChecksum2, taskTypeMigrate, migrateCmd, struct {
+ Image string
+ Trigger string
+ }{"img:4.0", ""})
+ initChecksum2 := r2.computeStepChecksum(migrateChecksum2, taskTypeInit, initCmd, struct {
+ ConfigChecksum string
+ Trigger string
+ }{"cfg", ""})
+
+ if cloneChecksum1 == cloneChecksum2 {
+ t.Error("clone checksum should change after boundary")
+ }
+ if migrateChecksum1 == migrateChecksum2 {
+ t.Error("migrate should cascade from clone schedule tick change")
+ }
+ if initChecksum1 == initChecksum2 {
+ t.Error("init should cascade from clone schedule tick change")
+ }
+}
+
+func TestScheduleRequeue_ComputesCorrectDuration(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ r := &SupersetReconciler{Now: func() time.Time { return now }}
+
+ cronExpr := "0 * * * *" // hourly at :00
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{CronSchedule: &cronExpr},
+ Source: supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"},
+ },
+ }
+
+ requeue := r.nextScheduleRequeue(superset)
+ // Next tick is 15:00, so 30 minutes + 1s buffer.
+ expected := 30*time.Minute + time.Second
+ if requeue != expected {
+ t.Errorf("expected requeue %v, got %v", expected, requeue)
+ }
+}
+
+func TestScheduleRequeue_NoSchedule(t *testing.T) {
+ r := &SupersetReconciler{Now: func() time.Time { return time.Now() }}
+
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ Source: supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"},
+ },
+ }
+
+ requeue := r.nextScheduleRequeue(superset)
+ if requeue != 0 {
+ t.Errorf("expected 0 requeue with no schedule, got %v", requeue)
+ }
+}
+
+func TestScheduleRequeue_DisabledClone(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ r := &SupersetReconciler{Now: func() time.Time { return now }}
+
+ cronExpr := "0 * * * *"
+ superset := &supersetv1alpha1.Superset{}
+ superset.Spec.Lifecycle = &supersetv1alpha1.LifecycleSpec{
+ Clone: &supersetv1alpha1.CloneTaskSpec{
+ SchedulableBaseTaskSpec: supersetv1alpha1.SchedulableBaseTaskSpec{
+ BaseTaskSpec: supersetv1alpha1.BaseTaskSpec{Disabled: common.Ptr(true)},
+ CronSchedule: &cronExpr,
+ },
+ Source: supersetv1alpha1.CloneSourceSpec{Host: "h", Database: "d", Username: "u"},
+ },
+ }
+
+ requeue := r.nextScheduleRequeue(superset)
+ if requeue != 0 {
+ t.Errorf("expected 0 requeue with disabled clone, got %v", requeue)
+ }
+}
diff --git a/internal/controller/superset_controller.go b/internal/controller/superset_controller.go
index 48dd12c..961f70d 100644
--- a/internal/controller/superset_controller.go
+++ b/internal/controller/superset_controller.go
@@ -46,6 +46,7 @@ import (
naming "github.com/apache/superset-kubernetes-operator/internal/common"
supersetconfig "github.com/apache/superset-kubernetes-operator/internal/config"
"github.com/apache/superset-kubernetes-operator/internal/resolution"
+ "github.com/apache/superset-kubernetes-operator/internal/schedule"
)
// SupersetReconciler reconciles a Superset object.
@@ -53,6 +54,7 @@ type SupersetReconciler struct {
client.Client
Scheme *runtime.Scheme
Recorder events.EventRecorder
+ Now func() time.Time
}
// +kubebuilder:rbac:groups=superset.apache.org,resources=supersets,verbs=get;list;watch
@@ -181,6 +183,11 @@ func (r *SupersetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
return ctrl.Result{}, fmt.Errorf("updating status: %w", err)
}
+ // Phase 6: Schedule-based requeue for periodic lifecycle tasks.
+ if requeue := r.nextScheduleRequeue(superset); requeue > 0 {
+ return ctrl.Result{RequeueAfter: requeue}, nil
+ }
+
return ctrl.Result{}, nil
}
@@ -632,6 +639,7 @@ func (r *SupersetReconciler) reconcileLifecycleTask(
Image: child.Status.Image,
Message: child.Status.Message,
}
+ r.projectScheduleStatus(superset, taskType, taskRef)
switch taskType {
case taskTypeClone:
superset.Status.Lifecycle.Clone = taskRef
@@ -913,11 +921,13 @@ func (r *SupersetReconciler) cloneInputs(superset *supersetv1alpha1.Superset) an
clone := superset.Spec.Lifecycle.Clone
return struct {
Trigger string
+ ScheduleTick string
Source supersetv1alpha1.CloneSourceSpec
ExcludeTables []string
ExcludeTableData []string
}{
Trigger: derefOrDefault(clone.Trigger, ""),
+ ScheduleTick: r.scheduleTick(clone.CronSchedule),
Source: clone.Source,
ExcludeTables: clone.ExcludeTables,
ExcludeTableData: clone.ExcludeTableData,
@@ -955,6 +965,85 @@ func (r *SupersetReconciler) initInputs(superset *supersetv1alpha1.Superset, con
}
}
+func (r *SupersetReconciler) now() time.Time {
+ if r.Now != nil {
+ return r.Now()
+ }
+ return time.Now()
+}
+
+func (r *SupersetReconciler) scheduleTick(cronSchedule *string) string {
+ if cronSchedule == nil || *cronSchedule == "" {
+ return ""
+ }
+ return schedule.CurrentTick(*cronSchedule, r.now())
+}
+
+func (r *SupersetReconciler) nextScheduleRequeue(superset *supersetv1alpha1.Superset) time.Duration {
+ if superset.Spec.Lifecycle == nil {
+ return 0
+ }
+ var earliest time.Time
+ for _, expr := range r.activeSchedules(superset) {
+ next := schedule.NextTick(expr, r.now())
+ if next.IsZero() {
+ continue
+ }
+ if earliest.IsZero() || next.Before(earliest) {
+ earliest = next
+ }
+ }
+ if earliest.IsZero() {
+ return 0
+ }
+ d := earliest.Sub(r.now()) + time.Second
+ if d < time.Second {
+ return time.Second
+ }
+ return d
+}
+
+func (r *SupersetReconciler) activeSchedules(superset *supersetv1alpha1.Superset) []string {
+ lc := superset.Spec.Lifecycle
+ var out []string
+ if lc.Clone != nil && lc.Clone.CronSchedule != nil && !isDisabled(lc.Clone.Disabled) {
+ out = append(out, *lc.Clone.CronSchedule)
+ }
+ return out
+}
+
+func isDisabled(disabled *bool) bool {
+ return disabled != nil && *disabled
+}
+
+func (r *SupersetReconciler) projectScheduleStatus(superset *supersetv1alpha1.Superset, taskType string, taskRef *supersetv1alpha1.TaskRefStatus) {
+ if superset.Spec.Lifecycle == nil {
+ return
+ }
+ var cronSchedule *string
+ switch taskType {
+ case taskTypeClone:
+ if superset.Spec.Lifecycle.Clone != nil {
+ cronSchedule = superset.Spec.Lifecycle.Clone.CronSchedule
+ }
+ }
+ if cronSchedule == nil || *cronSchedule == "" {
+ return
+ }
+ now := r.now()
+ if tick := schedule.CurrentTick(*cronSchedule, now); tick != "" {
+ parsed, err := time.Parse(time.RFC3339, tick)
+ if err == nil {
+ t := metav1.NewTime(parsed)
+ taskRef.LastScheduledAt = &t
+ }
+ }
+ if next := schedule.NextTick(*cronSchedule, now); !next.IsZero() {
+ t := metav1.NewTime(next)
+ taskRef.NextScheduleAt = &t
+ }
+}
+
func (r *SupersetReconciler) taskMaxRetries(superset *supersetv1alpha1.Superset, taskType string) *int32 {
if superset.Spec.Lifecycle == nil {
return nil
diff --git a/internal/schedule/schedule.go b/internal/schedule/schedule.go
new file mode 100644
index 0000000..899d3b4
--- /dev/null
+++ b/internal/schedule/schedule.go
@@ -0,0 +1,87 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 schedule
+
+import (
+ "fmt"
+ "time"
+
+ "github.com/robfig/cron/v3"
+)
+
+var parser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+
+// CurrentTick returns the most recent past time matching the cron expression,
+// formatted as RFC3339 UTC (e.g., "2026-05-12T02:00:00Z").
+// Returns "" if the expression is invalid or no tick exists within the
+// lookback window (~2 years).
+func CurrentTick(expr string, now time.Time) string {
+ sched, err := parser.Parse(expr)
+ if err != nil {
+ return ""
+ }
+ prev := findPrevTick(sched, now)
+ if prev.IsZero() {
+ return ""
+ }
+ return prev.UTC().Format(time.RFC3339)
+}
+
+// NextTick returns the next future time matching the cron expression.
+// Returns zero time if the expression is invalid.
+func NextTick(expr string, now time.Time) time.Time {
+ sched, err := parser.Parse(expr)
+ if err != nil {
+ return time.Time{}
+ }
+ return sched.Next(now)
+}
+
+// Validate checks whether a cron expression is parseable.
+// Returns an error describing the problem, or nil if valid.
+func Validate(expr string) error {
+ _, err := parser.Parse(expr)
+ if err != nil {
+ return fmt.Errorf("invalid cron expression %q: %w", expr, err)
+ }
+ return nil
+}
+
+// findPrevTick finds the most recent time matching the schedule that is <= now.
+// Uses geometric doubling of the lookback window for efficiency.
+func findPrevTick(sched cron.Schedule, now time.Time) time.Time {
+ lookback := time.Minute
+ for range 20 {
+ start := now.Add(-lookback)
+ t := sched.Next(start)
+ if !t.After(now) {
+ // Found a tick in the window. Iterate forward to find the latest one <= now.
+ prev := t
+ for {
+ t = sched.Next(t)
+ if t.After(now) {
+ return prev
+ }
+ prev = t
+ }
+ }
+ lookback *= 2
+ }
+ return time.Time{}
+}
diff --git a/internal/schedule/schedule_test.go b/internal/schedule/schedule_test.go
new file mode 100644
index 0000000..9e7d4e2
--- /dev/null
+++ b/internal/schedule/schedule_test.go
@@ -0,0 +1,150 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 schedule
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestCurrentTick_Hourly(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ tick := CurrentTick("0 * * * *", now)
+ assert.Equal(t, "2026-05-11T14:00:00Z", tick)
+}
+
+func TestCurrentTick_Daily(t *testing.T) {
+ now := time.Date(2026, 5, 11, 10, 0, 0, 0, time.UTC)
+ tick := CurrentTick("0 2 * * *", now)
+ assert.Equal(t, "2026-05-11T02:00:00Z", tick)
+}
+
+func TestCurrentTick_BeforeFirstDailyTick(t *testing.T) {
+ now := time.Date(2026, 5, 11, 1, 0, 0, 0, time.UTC)
+ tick := CurrentTick("0 2 * * *", now)
+ assert.Equal(t, "2026-05-10T02:00:00Z", tick)
+}
+
+func TestCurrentTick_EveryMinute(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 45, 0, time.UTC)
+ tick := CurrentTick("* * * * *", now)
+ assert.Equal(t, "2026-05-11T14:30:00Z", tick)
+}
+
+func TestCurrentTick_ExactlyOnBoundary(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 0, 0, 0, time.UTC)
+ tick := CurrentTick("0 * * * *", now)
+ // At exactly the boundary, the tick is the previous occurrence
+ // because Next(now-lookback) finds 14:00 which is not AFTER now (it equals now).
+ assert.Equal(t, "2026-05-11T14:00:00Z", tick)
+}
+
+func TestCurrentTick_Monthly(t *testing.T) {
+ now := time.Date(2026, 5, 15, 10, 0, 0, 0, time.UTC)
+ tick := CurrentTick("0 0 1 * *", now)
+ assert.Equal(t, "2026-05-01T00:00:00Z", tick)
+}
+
+func TestCurrentTick_MonthlyBeforeFirstDay(t *testing.T) {
+ now := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
+ // At midnight on the 1st — the tick is exactly now.
+ tick := CurrentTick("0 0 1 * *", now)
+ assert.Equal(t, "2026-06-01T00:00:00Z", tick)
+}
+
+func TestCurrentTick_Weekly(t *testing.T) {
+ // 2026-05-11 is a Monday (day 1)
+ now := time.Date(2026, 5, 14, 12, 0, 0, 0, time.UTC) // Thursday
+ tick := CurrentTick("30 1 * * 1", now) // Mondays at 1:30
+ assert.Equal(t, "2026-05-11T01:30:00Z", tick)
+}
+
+func TestCurrentTick_InvalidExpr(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ tick := CurrentTick("not a cron", now)
+ assert.Equal(t, "", tick)
+}
+
+func TestCurrentTick_EmptyExpr(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ tick := CurrentTick("", now)
+ assert.Equal(t, "", tick)
+}
+
+func TestCurrentTick_EveryFiveMinutes(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 37, 0, 0, time.UTC)
+ tick := CurrentTick("*/5 * * * *", now)
+ assert.Equal(t, "2026-05-11T14:35:00Z", tick)
+}
+
+func TestCurrentTick_EverySixHours(t *testing.T) {
+ now := time.Date(2026, 5, 11, 7, 0, 0, 0, time.UTC)
+ tick := CurrentTick("0 */6 * * *", now)
+ assert.Equal(t, "2026-05-11T06:00:00Z", tick)
+}
+
+func TestCurrentTick_StableWithinWindow(t *testing.T) {
+ // Two times within the same cron window should return the same tick.
+ expr := "0 2 * * *"
+ t1 := time.Date(2026, 5, 11, 10, 0, 0, 0, time.UTC)
+ t2 := time.Date(2026, 5, 11, 23, 59, 59, 0, time.UTC)
+ assert.Equal(t, CurrentTick(expr, t1), CurrentTick(expr, t2))
+}
+
+func TestCurrentTick_ChangesAcrossBoundary(t *testing.T) {
+ expr := "0 2 * * *"
+ before := time.Date(2026, 5, 11, 1, 59, 59, 0, time.UTC)
+ after := time.Date(2026, 5, 11, 2, 0, 1, 0, time.UTC)
+ assert.NotEqual(t, CurrentTick(expr, before), CurrentTick(expr, after))
+}
+
+func TestNextTick_Hourly(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ next := NextTick("0 * * * *", now)
+ expected := time.Date(2026, 5, 11, 15, 0, 0, 0, time.UTC)
+ assert.Equal(t, expected, next)
+}
+
+func TestNextTick_Daily(t *testing.T) {
+ now := time.Date(2026, 5, 11, 3, 0, 0, 0, time.UTC)
+ next := NextTick("0 2 * * *", now)
+ expected := time.Date(2026, 5, 12, 2, 0, 0, 0, time.UTC)
+ assert.Equal(t, expected, next)
+}
+
+func TestNextTick_InvalidExpr(t *testing.T) {
+ now := time.Date(2026, 5, 11, 14, 30, 0, 0, time.UTC)
+ next := NextTick("invalid", now)
+ assert.True(t, next.IsZero())
+}
+
+func TestValidate_Valid(t *testing.T) {
+ assert.NoError(t, Validate("0 2 * * *"))
+ assert.NoError(t, Validate("*/5 * * * *"))
+ assert.NoError(t, Validate("0 */6 * * 1-5"))
+}
+
+func TestValidate_Invalid(t *testing.T) {
+ require.Error(t, Validate("not valid"))
+ require.Error(t, Validate(""))
+ require.Error(t, Validate("* * *")) // too few fields
+}