avoid interacting with nil mutatedWfs (flyteorg#477)

* avoid interacting with nil mutatedWfs Signed-off-by: Babis Kiosidis <[email protected]> * only strip crd fields from mutated wfs Signed-off-by: Babis Kiosidis <[email protected]> * test failure in executor Signed-off-by: Babis Kiosidis <[email protected]> Signed-off-by: Babis Kiosidis <[email protected]>
eapolinario · Aug 25, 2022 · 031a005 · 031a005
1 parent 5b28a98
commit 031a005
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 1 deletion.
diff --git a/flytepropeller/pkg/controller/handler.go b/flytepropeller/pkg/controller/handler.go
@@ -248,7 +248,9 @@ func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 		if wfClosureCrdFields != nil {
 			// strip data populated from WorkflowClosureReference
 			w.SubWorkflows, w.Tasks, w.WorkflowSpec = nil, nil, nil
-			mutatedWf.SubWorkflows, mutatedWf.Tasks, mutatedWf.WorkflowSpec = nil, nil, nil
+			if mutatedWf != nil {
+				mutatedWf.SubWorkflows, mutatedWf.Tasks, mutatedWf.WorkflowSpec = nil, nil, nil
+			}
 		}
 
 		if err != nil {

diff --git a/flytepropeller/pkg/controller/handler_test.go b/flytepropeller/pkg/controller/handler_test.go
@@ -894,4 +894,25 @@ func TestPropellerHandler_OffloadedWorkflowClosure(t *testing.T) {
 		err := p.Handle(ctx, namespace, name)
 		assert.Error(t, err)
 	})
+
+	t.Run("TryMutate failure is handled", func(t *testing.T) {
+		scope := promutils.NewTestScope()
+
+		protoStore := &storagemocks.ComposedProtobufStore{}
+		protoStore.OnReadProtobufMatch(mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("foo"))
+		exec.HandleCb = func(ctx context.Context, w *v1alpha1.FlyteWorkflow) error {
+			return fmt.Errorf("foo")
+		}
+		dataStore := storage.NewCompositeDataStore(storage.URLPathConstructor{}, protoStore)
+		p := NewPropellerHandler(ctx, cfg, dataStore, s, exec, scope)
+
+		err := p.Handle(ctx, namespace, name)
+		assert.Error(t, err, "foo")
+
+		r, err := s.Get(ctx, namespace, name)
+		assert.NoError(t, err)
+		assert.Nil(t, r.WorkflowSpec)
+		assert.Nil(t, r.SubWorkflows)
+		assert.Nil(t, r.Tasks)
+	})
 }