fix(kubernetes): Fix artifact binding for run job (#3540)

* fix(kubernetes): Fix artifact binding for run job

The Kubernetes V2 Run Job stage doesn't bind artifacts if the manifest
source is text. This is because we only populate artifacts if the
source is an artifact; luckily the logic to populate artifacts is
independent of the aritfact source, so we can run it in either case
by just removing the if condition.

* fix(kubernetes): Fix tests and V1 run job

The prior commit broke the V1 run job stage, as it uses the same
KubernetesJobRunner but doesn't have any of the relevant fields
that ManifestEvaluator tries to evaluate. The root cause of this
is that the context doesn't have a Source, so inherits the default
of Source.Text; this causes us to try to look up the manifest text,
which is null.

Ideally we'd have the V1 run job stage set a different value (maybe
Source.None because we don't even really have a manifest here) or
change the default.  But given the impending deprecation of the V1
provider (and that it's a bit risky to change the default) let's
just work around the issue by short-circuiting manifest evaluation
if the source is Text but there is no text.

Also, the prior commit broke some tests as we're now running
manifestEvaluator.evaluate more than before; fix these tests.

orca-clouddriver/src/main/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/providers/kubernetes/KubernetesJobRunner.java

@@ -17,9 +17,10 @@
 package com.netflix.spinnaker.orca.clouddriver.tasks.providers.kubernetes;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
 import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution;
 import com.netflix.spinnaker.orca.clouddriver.tasks.job.JobRunner;
-import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ManifestContext;
+import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ManifestContext.Source;
 import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ManifestEvaluator;
 import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.RunJobManifestContext;
 import com.netflix.spinnaker.orca.pipeline.util.ArtifactUtils;
@@ -54,20 +55,7 @@ public List<Map> getOperations(StageExecution stage) {
       operation.putAll(stage.getContext());
     }
 
-    RunJobManifestContext runJobManifestContext = stage.mapTo(RunJobManifestContext.class);
-    if (runJobManifestContext.getSource().equals(ManifestContext.Source.Artifact)) {
-      ManifestEvaluator.Result result = manifestEvaluator.evaluate(stage, runJobManifestContext);
-
-      List<Map<Object, Object>> manifests = result.getManifests();
-      if (manifests.size() != 1) {
-        throw new IllegalArgumentException("Run Job only supports manifests with a single Job.");
-      }
-
-      operation.put("source", "text");
-      operation.put("manifest", manifests.get(0));
-      operation.put("requiredArtifacts", result.getRequiredArtifacts());
-      operation.put("optionalArtifacts", result.getOptionalArtifacts());
-    }
+    operation.putAll(getManifestFields(stage));
 
     KubernetesContainerFinder.populateFromStage(operation, stage, artifactUtils);
 
@@ -76,6 +64,34 @@ public List<Map> getOperations(StageExecution stage) {
     return Collections.singletonList(task);
   }
 
+  // Gets the fields relevant to manifests that should be added to the operation
+  private ImmutableMap<String, Object> getManifestFields(StageExecution stage) {
+    RunJobManifestContext runJobManifestContext = stage.mapTo(RunJobManifestContext.class);
+
+    // This short-circuit exists to handle jobs from the Kubernetes V1 provider; these have the
+    // source set to Text (because it's the default and they don't set a source), but also have no
+    // manifests. This will fail if we try to call manifestEvaluator.evaluate() so short-circuit
+    // as the additional fields are not relevant. This workaround can be removed once the V1
+    // provider is removed (currently scheduled for the 1.21 release).
+    if (runJobManifestContext.getSource() == Source.Text
+        && runJobManifestContext.getManifests() == null) {
+      return ImmutableMap.of();
+    }
+
+    ManifestEvaluator.Result result = manifestEvaluator.evaluate(stage, runJobManifestContext);
+
+    List<Map<Object, Object>> manifests = result.getManifests();
+    if (manifests.size() != 1) {
+      throw new IllegalArgumentException("Run Job only supports manifests with a single Job.");
+    }
+
+    return ImmutableMap.of(
+        "source", "text",
+        "manifest", manifests.get(0),
+        "requiredArtifacts", result.getRequiredArtifacts(),
+        "optionalArtifacts", result.getOptionalArtifacts());
+  }
+
   public Map<String, Object> getAdditionalOutputs(StageExecution stage, List<Map> operations) {
     Map<String, Object> outputs = new HashMap<>();
     Map<String, Object> execution = new HashMap<>();

orca-clouddriver/src/test/groovy/com/netflix/spinnaker/orca/clouddriver/tasks/providers/kubernetes/KubernetesJobRunnerSpec.groovy

@@ -17,8 +17,10 @@
 package com.netflix.spinnaker.orca.clouddriver.tasks.providers.kubernetes
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.google.common.collect.ImmutableList
 import com.netflix.spinnaker.kork.artifacts.model.Artifact
 import com.netflix.spinnaker.kork.core.RetrySupport
+import com.netflix.spinnaker.orca.api.pipeline.models.StageExecution
 import com.netflix.spinnaker.orca.clouddriver.OortService
 import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ManifestEvaluator
 import com.netflix.spinnaker.orca.pipeline.model.PipelineExecutionImpl
@@ -31,12 +33,16 @@ import retrofit.mime.TypedString
 import spock.lang.Specification
 
 class KubernetesJobRunnerSpec extends Specification {
-
   def "should return a run job operation if cluster set in context"() {
     given:
     ArtifactUtils artifactUtils = Mock(ArtifactUtils)
     ObjectMapper objectMapper = new ObjectMapper()
-    ManifestEvaluator manifestEvaluator = Mock(ManifestEvaluator)
+    ManifestEvaluator manifestEvaluator = new ManifestEvaluator(
+        Mock(ArtifactUtils),
+        Mock(ContextParameterProcessor),
+        Mock(OortService),
+        new RetrySupport()
+    )
     def stage = new StageExecutionImpl(PipelineExecutionImpl.newPipeline("test"), "runJob", [
       credentials: "abc", cloudProvider: "kubernetes",
       cluster: [
@@ -60,7 +66,12 @@ class KubernetesJobRunnerSpec extends Specification {
     given:
     ArtifactUtils artifactUtils = Mock(ArtifactUtils)
     ObjectMapper objectMapper = new ObjectMapper()
-    ManifestEvaluator manifestEvaluator = Mock(ManifestEvaluator)
+    ManifestEvaluator manifestEvaluator = new ManifestEvaluator(
+        Mock(ArtifactUtils),
+        Mock(ContextParameterProcessor),
+        Mock(OortService),
+        new RetrySupport()
+    )
     def stage = new StageExecutionImpl(PipelineExecutionImpl.newPipeline("test"), "runJob", [
       credentials: "abc", cloudProvider: "kubernetes",
       foo: "bar"
@@ -82,7 +93,14 @@ class KubernetesJobRunnerSpec extends Specification {
     given:
     ArtifactUtils artifactUtils = Mock(ArtifactUtils)
     ObjectMapper objectMapper = new ObjectMapper()
-    ManifestEvaluator manifestEvaluator = Mock(ManifestEvaluator)
+    ManifestEvaluator manifestEvaluator = new ManifestEvaluator(
+        Mock(ArtifactUtils) {
+          getArtifacts(_ as StageExecution) >> ImmutableList.of()
+        },
+        Mock(ContextParameterProcessor),
+        Mock(OortService),
+        new RetrySupport()
+    )
     def stage = new StageExecutionImpl(PipelineExecutionImpl.newPipeline("test"), "runJob", [
       credentials: "abc", cloudProvider: "kubernetes",
       manifest: [