fix(controller): skip empty optional step artifacts (#16839)

Signed-off-by: rwong2888 <rwong@forbes.com>

Author: rwong2888

workflow/controller/steps.go

@@ -535,6 +535,9 @@ func (woc *wfOperationCtx) resolveReferences(ctx context.Context, stepGroup []wf
 				}
 				return fmt.Errorf("unable to resolve references: %w", err)
 			}
+			if art.Optional && !resolvedArt.HasLocationOrKey() {
+				continue
+			}
 			resolvedArt.Name = art.Name
 			artifacts = append(artifacts, *resolvedArt)
 		}

workflow/controller/steps_test.go

@@ -12,6 +12,7 @@ import (
 	wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v4/util/logging"
 	"github.com/argoproj/argo-workflows/v4/util/variables"
+	varkeys "github.com/argoproj/argo-workflows/v4/util/variables/keys"
 )
 
 // TestStepsFailedRetries ensures a steps template will recognize exhausted retries
@@ -82,6 +83,28 @@ func TestArtifactResolutionWhenSkipped(t *testing.T) {
 	assert.Equal(t, wfv1.WorkflowSucceeded, woc.wf.Status.Phase)
 }
 
+func TestResolveReferencesSkipsOptionalArtifactFromSkippedStep(t *testing.T) {
+	ctx := logging.TestContext(t.Context())
+	woc := newWoc(ctx, wfv1.Workflow{})
+	scope := createScope(nil)
+	varkeys.StepsNodeRef.OutputsArtifactByName.Set(scope.scope, wfv1.Artifact{}, "generate", "message")
+	stepGroup := []wfv1.WorkflowStep{{
+		Name:     "consume",
+		Template: "consumer",
+		Arguments: wfv1.Arguments{Artifacts: wfv1.Artifacts{{
+			Name:     "message",
+			From:     "{{steps.generate.outputs.artifacts.message}}",
+			Optional: true,
+		}}},
+	}}
+
+	resolvedSteps, err := woc.resolveReferences(ctx, stepGroup, scope)
+
+	require.NoError(t, err)
+	require.Len(t, resolvedSteps, 1)
+	assert.Empty(t, resolvedSteps[0].Arguments.Artifacts)
+}
+
 var stepsWithParamAndGlobalParam = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow