fix: Ensure CronWorkflows are persisted once per operation (argoproj#…

…4172)
jlegrone · Oct 1, 2020 · 6265c70 · 6265c70
1 parent 2a992ae
commit 6265c70
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 10 deletions.
diff --git a/workflow/cron/controller.go b/workflow/cron/controller.go
@@ -255,6 +255,8 @@ func (cc *Controller) processNextWorkflowItem() bool {
  return true
  }
 
+ defer woc.persistUpdate()
+
  // If the workflow is completed or was deleted, remove it from Active Workflows
  if wf.Status.Fulfilled() || !wfExists {
  log.Warnf("Workflow '%s' from CronWorkflow '%s' completed", wf.Name, woc.cronWf.Name)

diff --git a/workflow/cron/operator.go b/workflow/cron/operator.go
@@ -35,6 +35,7 @@ type cronWfOperationCtx struct {
  cronWfIf typed.CronWorkflowInterface
  log *log.Entry
  metrics *metrics.Metrics
+ updated bool
 }
 
 func newCronWfOperationCtx(cronWorkflow *v1alpha1.CronWorkflow, wfClientset versioned.Interface, wfLister util.WorkflowLister, metrics *metrics.Metrics) *cronWfOperationCtx {
@@ -51,10 +52,13 @@ func newCronWfOperationCtx(cronWorkflow *v1alpha1.CronWorkflow, wfClientset vers
  "namespace": cronWorkflow.ObjectMeta.Namespace,
  }),
  metrics: metrics,
+ updated: false,
  }
 }
 
 func (woc *cronWfOperationCtx) Run() {
+ defer woc.persistUpdate()
+
  woc.log.Infof("Running %s", woc.name)
 
  err := woc.validateCronWorkflow()
@@ -87,7 +91,7 @@ func (woc *cronWfOperationCtx) Run() {
  woc.cronWf.Status.Active = append(woc.cronWf.Status.Active, getWorkflowObjectReference(wf, runWf))
  woc.cronWf.Status.LastScheduledTime = &v1.Time{Time: time.Now()}
  woc.cronWf.Status.Conditions.RemoveCondition(v1alpha1.ConditionTypeSubmissionError)
- woc.persistUpdate()
+ woc.updated = true
 }
 
 func (woc *cronWfOperationCtx) validateCronWorkflow() error {
@@ -98,6 +102,7 @@ func (woc *cronWfOperationCtx) validateCronWorkflow() error {
  woc.reportCronWorkflowError(v1alpha1.ConditionTypeSpecError, fmt.Sprint(err))
  } else {
  woc.cronWf.Status.Conditions.RemoveCondition(v1alpha1.ConditionTypeSpecError)
+ woc.updated = true
  }
  return err
 }
@@ -116,24 +121,26 @@ func getWorkflowObjectReference(wf *v1alpha1.Workflow, runWf *v1alpha1.Workflow)
 }
 
 func (woc *cronWfOperationCtx) persistUpdate() {
- if woc.origCronWf.ResourceVersion != woc.cronWf.ResourceVersion {
+ if !woc.updated {
+ return
+ } else if woc.origCronWf.ResourceVersion != woc.cronWf.ResourceVersion {
  woc.log.Error("cannot update cron workflow with mismatched resource versions")
  return
  }
- cronWf, err := woc.cronWfIf.Update(woc.cronWf)
+
+ _, err := woc.cronWfIf.Update(woc.cronWf)
  if err != nil {
  if !errors.IsConflict(err) {
  woc.log.WithError(err).Error("failed to update CronWorkflow")
  return
  }
  var reapplyErr error
- cronWf, reapplyErr = woc.reapplyUpdate()
+ _, reapplyErr = woc.reapplyUpdate()
  if err != nil {
  woc.log.WithError(reapplyErr).WithField("original error", err).Error("failed to update CronWorkflow after reapply attempt")
  return
  }
  }
- woc.cronWf = cronWf
 }
 
 func (woc *cronWfOperationCtx) reapplyUpdate() (*v1alpha1.CronWorkflow, error) {
@@ -286,13 +293,13 @@ func (woc *cronWfOperationCtx) reconcileDeletedWfs() error {
  return fmt.Errorf("unable to list workflows: %s", err)
  }
 
- currentWfs := make(map[types.UID]bool)
+ currentWfs := make(map[types.UID]*v1alpha1.Workflow)
  for _, wf := range wfList {
- currentWfs[wf.UID] = true
+ currentWfs[wf.UID] = wf
  }
 
  for _, objectRef := range woc.cronWf.Status.Active {
- if found := currentWfs[objectRef.UID]; !found {
+ if wf, found := currentWfs[objectRef.UID]; !found || wf.Status.Fulfilled() {
  woc.removeFromActiveList(objectRef.UID)
  }
  }
@@ -305,7 +312,7 @@ func (woc *cronWfOperationCtx) removeActiveWf(wf *v1alpha1.Workflow) {
  return
  }
  woc.removeFromActiveList(wf.ObjectMeta.UID)
- woc.persistUpdate()
+ woc.updated = true
 }
 
 func (woc *cronWfOperationCtx) removeFromActiveList(uid types.UID) {
@@ -316,6 +323,7 @@ func (woc *cronWfOperationCtx) removeFromActiveList(uid types.UID) {
  }
  }
  woc.cronWf.Status.Active = newActive
+ woc.updated = true
 }
 
 func (woc *cronWfOperationCtx) enforceHistoryLimit() {
@@ -397,5 +405,5 @@ func (woc *cronWfOperationCtx) reportCronWorkflowError(conditionType v1alpha1.Co
  Status: v1.ConditionTrue,
  })
  woc.metrics.CronWorkflowSubmissionError()
- woc.persistUpdate()
+ woc.updated = true
 }