feat(cron): Added timezone support to cron workflows. Closes argoproj…

…#1931 (argoproj#1986)

Dockerfile

@@ -85,6 +85,8 @@ ENTRYPOINT [ "argoexec" ]
 # workflow-controller
 ####################################################################################################
 FROM scratch as workflow-controller
+# Add timezone data
+COPY --from=argo-build /usr/share/zoneinfo /usr/share/zoneinfo
 COPY --from=argo-build /go/src/github.com/argoproj/argo/dist/workflow-controller-linux-amd64 /bin/workflow-controller
 ENTRYPOINT [ "workflow-controller" ]
 

Dockerfile.dev

@@ -54,6 +54,8 @@ ENTRYPOINT [ "argoexec" ]
 # workflow-controller
 ####################################################################################################
 FROM scratch as workflow-controller
+# Add timezone data
+COPY --from=builder /usr/share/zoneinfo /usr/share/zoneinfo
 COPY workflow-controller /bin/
 ENTRYPOINT [ "workflow-controller" ]
 

api/openapi-spec/swagger.json

@@ -303,6 +303,10 @@
  "description": "Suspend is a flag that will stop new CronWorkflows from running if set to true",
  "type": "boolean"
  },
+ "timezone": {
+ "description": "Timezone is the timezone against which the cron schedule will be calculated, e.g. \"Asia/Tokyo\". Default is machine's local time.",
+ "type": "string"
+ },
  "workflowSpec": {
  "description": "WorkflowSpec is the spec of the workflow to be run",
  "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.WorkflowSpec"

cmd/argo/commands/cron/get.go

@@ -67,6 +67,9 @@ func printCronWorkflowTemplate(wf *wfv1.CronWorkflow, outFmt string) {
  fmt.Printf(fmtStr, "Created:", humanize.Timestamp(wf.ObjectMeta.CreationTimestamp.Time))
  fmt.Printf(fmtStr, "Schedule:", wf.Spec.Schedule)
  fmt.Printf(fmtStr, "Suspended:", wf.Spec.Suspend)
+ if wf.Spec.Timezone != "" {
+ fmt.Printf(fmtStr, "Timezone:", wf.Spec.Timezone)
+ }
  if wf.Spec.StartingDeadlineSeconds != nil {
  fmt.Printf(fmtStr, "StartingDeadlineSeconds:", *wf.Spec.StartingDeadlineSeconds)
  }

cmd/server/cronworkflow/cron-workflow.swagger.json

@@ -3032,6 +3032,10 @@
  "type": "integer",
  "format": "int32",
  "title": "FailedJobsHistoryLimit is the number of successful jobs to be kept at a time"
+ },
+ "timezone": {
+ "type": "string",
+ "description": "Timezone is the timezone against which the cron schedule will be calculated, e.g. \"Asia/Tokyo\". Default is machine's local time."
  }
  }
  },

examples/cron-workflow.yaml

@@ -4,6 +4,12 @@ metadata:
  name: hello-world
 spec:
  schedule: "* * * * *"
+ timezone: "America/Los_Angeles" # Default to local machine timezone
+ startingDeadlineSeconds: 0
+ concurrencyPolicy: "Replace" # Default to "Allow"
+ successfulJobsHistoryLimit: 4 # Default 3
+ failedJobsHistoryLimit: 4 # Default 1
+ suspend: false # Set to "true" to suspend scheduling
  workflowSpec:
  entrypoint: whalesay
  templates:

pkg/apis/workflow/v1alpha1/cron_workflow_types.go

@@ -40,6 +40,8 @@ type CronWorkflowSpec struct {
  SuccessfulJobsHistoryLimit *int32 `json:"successfulJobsHistoryLimit,omitempty" protobuf:"varint,6,opt,name=successfulJobsHistoryLimit"`
  // FailedJobsHistoryLimit is the number of successful jobs to be kept at a time
  FailedJobsHistoryLimit *int32 `json:"failedJobsHistoryLimit,omitempty" protobuf:"varint,7,opt,name=failedJobsHistoryLimit"`
+ // Timezone is the timezone against which the cron schedule will be calculated, e.g. "Asia/Tokyo". Default is machine's local time.
+ Timezone string `json:"timezone,omitempty" protobuf:"bytes,8,opt,name=timezone"`
 }
 
 type CronWorkflowStatus struct {

test/e2e/cron_test.go

@@ -1,6 +1,7 @@
 package e2e
 
 import (
+ "fmt"
  wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
  "github.com/argoproj/argo/test/e2e/fixtures"
  "github.com/argoproj/argo/workflow/common"
@@ -12,6 +13,7 @@ import (
  "k8s.io/apimachinery/pkg/fields"
  "k8s.io/apimachinery/pkg/labels"
  "k8s.io/apimachinery/pkg/selection"
+ "strconv"
  "testing"
  "time"
 )
@@ -32,6 +34,52 @@ func (s *CronSuite) TestBasic() {
  })
 }
 
+func (s *CronSuite) TestBasicTimezone() {
+ // This test works by scheduling a CronWorkflow for the next minute, but using the local time of another timezone
+ // then seeing if the Workflow was ran within the next minute. Since this test would be trivial if the selected
+ // timezone was the same as the local timezone, a little-used timezone is used.
+ testTimezone := "Pacific/Niue"
+ testLocation, err := time.LoadLocation(testTimezone)
+ if err != nil {
+ s.T().Fatal(err)
+ }
+ hour, min, _ := time.Now().In(testLocation).Clock()
+ min++
+ if min == 60 {
+ min = 0
+ hour = (hour + 1) % 24
+ }
+ scheduleInTestTimezone := strconv.Itoa(min) + " " + strconv.Itoa(hour) + " * * *"
+ s.Given().
+ CronWorkflow(fmt.Sprintf(`
+apiVersion: argoproj.io/v1alpha1
+kind: CronWorkflow
+metadata:
+ name: test-cron-wf-basic
+ labels:
+ argo-e2e: true
+spec:
+ schedule: "%s"
+ timezone: "%s"
+ workflowSpec:
+ entrypoint: whalesay
+ templates:
+ - name: whalesay
+ container:
+ image: python:alpine3.6
+ imagePullPolicy: IfNotPresent
+ command: ["sh", -c]
+ args: ["echo hello"]
+`, scheduleInTestTimezone, testTimezone)).
+ When().
+ CreateCronWorkflow().
+ Wait(1 * time.Minute).
+ Then().
+ ExpectCron(func(t *testing.T, cronWf *wfv1.CronWorkflow) {
+ assert.True(t, cronWf.Status.LastScheduledTime.Time.After(time.Now().Add(-1*time.Minute)))
+ })
+}
+
 func (s *CronSuite) TestSuspend() {
  s.Given().
  CronWorkflow("@testdata/basic.yaml").

ui/src/app/cron-workflows/components/cron-workflow-summary-panel.tsx

@@ -19,6 +19,7 @@ export const CronWorkflowSummaryPanel = (props: Props) => {
  {title: 'Name', value: props.cronWorkflow.metadata.name},
  {title: 'Namespace', value: props.cronWorkflow.metadata.namespace},
  {title: 'Schedule', value: props.cronWorkflow.spec.schedule},
+ {title: 'Timezone', value: props.cronWorkflow.spec.timezone},
  {
  title: 'Concurrency Policy',
  value: props.cronWorkflow.spec.concurrencyPolicy ? props.cronWorkflow.spec.concurrencyPolicy : 'Allow'

ui/src/models/cron-workflows.ts

@@ -17,6 +17,7 @@ export interface CronWorkflowSpec {
  startingDeadlineSeconds?: number;
  successfulJobsHistoryLimit?: number;
  failedJobsHistoryLimit?: number;
+ timezone?: string;
 }
 
 export interface CronWorkflowStatus {

workflow/cron/controller.go

@@ -123,8 +123,7 @@ func (cc *Controller) processNextCronItem() bool {
  return true
  }
 
- cronWfIf := cc.wfClientset.ArgoprojV1alpha1().CronWorkflows(cronWf.Namespace)
- cronWorkflowOperationCtx, err := newCronWfOperationCtx(cronWf, cc.wfClientset, cronWfIf)
+ cronWorkflowOperationCtx, err := newCronWfOperationCtx(cronWf, cc.wfClientset)
  if err != nil {
  log.Error(err)
  return true
@@ -142,7 +141,12 @@ func (cc *Controller) processNextCronItem() bool {
  delete(cc.nameEntryIDMap, key.(string))
  }
 
- entryId, err := cc.cron.AddJob(cronWf.Spec.Schedule, cronWorkflowOperationCtx)
+ cronSchedule := cronWf.Spec.Schedule
+ if cronWf.Spec.Timezone != "" {
+ cronSchedule = "CRON_TZ=" + cronWf.Spec.Timezone + " " + cronSchedule
+ }
+
+ entryId, err := cc.cron.AddJob(cronSchedule, cronWorkflowOperationCtx)
  if err != nil {
  log.Errorf("could not schedule CronWorkflow: %s", err)
  return true

workflow/cron/operator.go

@@ -26,13 +26,13 @@ type cronWfOperationCtx struct {
  cronWfIf typed.CronWorkflowInterface
 }
 
-func newCronWfOperationCtx(cronWorkflow *v1alpha1.CronWorkflow, wfClientset versioned.Interface, cronWfIf typed.CronWorkflowInterface) (*cronWfOperationCtx, error) {
+func newCronWfOperationCtx(cronWorkflow *v1alpha1.CronWorkflow, wfClientset versioned.Interface) (*cronWfOperationCtx, error) {
  return &cronWfOperationCtx{
  name: cronWorkflow.ObjectMeta.Name,
  cronWf: cronWorkflow,
  wfClientset: wfClientset,
  wfClient: wfClientset.ArgoprojV1alpha1().Workflows(cronWorkflow.Namespace),
- cronWfIf: cronWfIf,
+ cronWfIf: wfClientset.ArgoprojV1alpha1().CronWorkflows(cronWorkflow.Namespace),
  }, nil
 }