Repository: argoproj/argo
Pull request #3615 (2020-07-28 00:22:58)

fix(controller): Panic when outputs in a cache entry are nil
Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [x] I've signed the CLA.
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [x] My builds are green. Try syncing with master if they are not. 
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).



Patch:
diff --git a/workflow/controller/cache/cache.go b/workflow/controller/cache/cache.go
index 2a47830e2..d0edf6ae6 100644
--- a/workflow/controller/cache/cache.go
+++ b/workflow/controller/cache/cache.go
@@ -103,26 +103,26 @@ func (c *configMapCache) Load(key string) (*CacheEntry, error) {
 	defer c.locked.Unlock()
 	cm, err := c.kubeClient.CoreV1().ConfigMaps(c.namespace).Get(c.name, metav1.GetOptions{})
 	if apierr.IsNotFound(err) {
-		c.logError(err, log.Fields{}, "MemoizationCache miss: ConfigMap does not exist")
+		c.logError(err, log.Fields{}, "config map cache miss: config map does not exist")
 		return nil, nil
 	}
 	if err != nil {
-		c.logError(err, log.Fields{}, "Error loading ConfigMap cache")
-		return nil, err
+		c.logError(err, log.Fields{}, "Error loading config map cache")
+		return nil, fmt.Errorf("could not load config map cache: %w", err)
 	}
-	c.logInfo(log.Fields{}, "ConfigMap cache loaded")
+	c.logInfo(log.Fields{}, "config map cache loaded")
 	rawEntry, ok := cm.Data[key]
 	if !ok || rawEntry == "" {
-		c.logInfo(log.Fields{}, "MemoizationCache miss: entry does not exist")
+		c.logInfo(log.Fields{}, "config map cache miss: entry does not exist")
 		return nil, nil
 	}
 	var entry CacheEntry
 	err = json.Unmarshal([]byte(rawEntry), &entry)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("malformed cache entry: could not unmarshal JSON; unable to parse: %w", err)
 	}
 	outputs := entry.Outputs
-	c.logInfo(log.Fields{"key": key}, "ConfigMap cache hit")
+	c.logInfo(log.Fields{"key": key}, "config map cache hit")
 	return &CacheEntry{
 		Outputs: outputs,
 	}, nil
@@ -146,7 +146,7 @@ func (c *configMapCache) Save(key string, nodeId string, value *wfv1.Outputs) er
 		})
 		if err != nil {
 			c.logError(err, log.Fields{"key": key, "nodeId": nodeId}, "Error saving to ConfigMap cache")
-			return err
+			return fmt.Errorf("could not save to config map cache: %w", err)
 		}
 	}
 
@@ -158,7 +158,7 @@ func (c *configMapCache) Save(key string, nodeId string, value *wfv1.Outputs) er
 	entryJSON, err := json.Marshal(newEntry)
 	if err != nil {
 		c.logError(err, log.Fields{"key": key, "nodeId": nodeId}, "Unable to marshal cache entry")
-		return fmt.Errorf("Unable to marshal cache entry: %w", err)
+		return fmt.Errorf("unable to marshal cache entry: %w", err)
 	}
 
 	if cache.Data == nil {
@@ -169,7 +169,7 @@ func (c *configMapCache) Save(key string, nodeId string, value *wfv1.Outputs) er
 	_, err = c.kubeClient.CoreV1().ConfigMaps(c.namespace).Update(cache)
 	if err != nil {
 		c.logError(err, log.Fields{"key": key, "nodeId": nodeId}, "Kubernetes error creating new cache entry")
-		return err
+		return fmt.Errorf("error creating cache entry: %w", err)
 	}
 	return nil
 }
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index fd3b00894..6b49e8f57 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -1443,17 +1443,17 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 			return woc.initializeNodeOrMarkError(node, nodeName, templateScope, orgTmpl, opts.boundaryID, err), err
 		}
 		hit := false
-		outputs := wfv1.Outputs{}
+		outputs := &wfv1.Outputs{}
 		if entry != nil {
 			hit = true
-			outputs = *entry.Outputs
+			outputs = entry.Outputs
 		}
 		memStat := &wfv1.MemoizationStatus{
 			Hit:       hit,
 			Key:       processedTmpl.Memoize.Key,
 			CacheName: processedTmpl.Memoize.Cache.ConfigMap.Name,
 		}
-		node = woc.initializeCacheNode(nodeName, processedTmpl, templateScope, orgTmpl, opts.boundaryID, &outputs, memStat)
+		node = woc.initializeCacheNode(nodeName, processedTmpl, templateScope, orgTmpl, opts.boundaryID, outputs, memStat)
 		woc.wf.Status.Nodes[node.ID] = *node
 		woc.updated = true
 	}


Test patch:
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index ea708b22b..1328fca09 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -4015,12 +4015,52 @@ func TestConfigMapCacheLoadOperate(t *testing.T) {
 	woc.operate()
 
 	status := woc.wf.Status
-	outputs := status.Nodes[""].Outputs
+	assert.Len(t, status.Nodes, 1)
+	outputs := &wfv1.Outputs{}
+	for _, node := range status.Nodes {
+		outputs = node.Outputs
+	}
 	assert.NotNil(t, outputs)
 	assert.Equal(t, "hello", outputs.Parameters[0].Name)
 	assert.Equal(t, sampleOutput, outputs.Parameters[0].Value.StrVal)
 }
 
+func TestConfigMapCacheLoadNilOutputs(t *testing.T) {
+
+	var sampleConfigMapCacheEntry = apiv1.ConfigMap{
+		Data: map[string]string{
+			"hi-there-world": `{"ExpiresAt":"2020-06-18T17:11:05Z","NodeID":"memoize-abx4124-123129321123","Outputs":{}`,
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ConfigMap",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "whalesay-cache",
+			ResourceVersion: "1630732",
+		},
+	}
+	wf := unmarshalWF(workflowCached)
+	cancel, controller := newController()
+	defer cancel()
+
+	_, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.ObjectMeta.Namespace).Create(wf)
+	assert.NoError(t, err)
+	_, err = controller.kubeclientset.CoreV1().ConfigMaps("default").Create(&sampleConfigMapCacheEntry)
+	assert.NoError(t, err)
+
+	woc := newWorkflowOperationCtx(wf, controller)
+	assert.NotPanics(t, woc.operate)
+
+	status := woc.wf.Status
+	assert.Len(t, status.Nodes, 1)
+	outputs := &wfv1.Outputs{}
+	for _, node := range status.Nodes {
+		outputs = node.Outputs
+	}
+	assert.Nil(t, outputs)
+}
+
 func TestConfigMapCacheSaveOperate(t *testing.T) {
 	wf := unmarshalWF(workflowCached)
 	cancel, controller := newController()
@@ -4556,4 +4596,4 @@ func TestFailSuspendedAndPendingNodesAfterShutdown(t *testing.T) {
 			assert.Equal(t, wfv1.NodeFailed, node.Phase)
 		}
 	})
-}
\ No newline at end of file
+}