fix: Only save memoization cache when node succeeded (#5711)

Signed-off-by: terrytangyuan <[email protected]>
argoproj · Apr 19, 2021 · dba2c04 · dba2c04
1 parent 8e9e6d6
commit dba2c04
Showing 1 changed file with 7 additions and 5 deletions.
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
@@ -874,11 +874,13 @@ func (woc *wfOperationCtx) podReconciliation(ctx context.Context) error {
  woc.wf.Status.Nodes[nodeID] = *newState
  woc.addOutputsToGlobalScope(node.Outputs)
  if node.MemoizationStatus != nil {
- c := woc.controller.cacheFactory.GetCache(controllercache.ConfigMapCache, node.MemoizationStatus.CacheName)
- err := c.Save(ctx, node.MemoizationStatus.Key, node.ID, node.Outputs)
- if err != nil {
- woc.log.WithFields(log.Fields{"nodeID": node.ID}).WithError(err).Error("Failed to save node outputs to cache")
- node.Phase = wfv1.NodeError
+ if node.Succeeded() {
+ c := woc.controller.cacheFactory.GetCache(controllercache.ConfigMapCache, node.MemoizationStatus.CacheName)
+ err := c.Save(ctx, node.MemoizationStatus.Key, node.ID, node.Outputs)
+ if err != nil {
+ woc.log.WithFields(log.Fields{"nodeID": node.ID}).WithError(err).Error("Failed to save node outputs to cache")
+ node.Phase = wfv1.NodeError
+ }
  }
  }
  if node.Phase == wfv1.NodeRunning {