STRDTORC-772: concord-task: using withItems with fork (walmartlabs#756)

Author: brig

CHANGELOG.md

@@ -13,6 +13,11 @@ Concord) repositories.
 
 - docker: updated Ansible to 2.6.7.
 
+### Breaking
+
+- concord-tasks: IN parameter `jobs` renamed to `forks` to avoid
+confusion with OUT parameter `jobs`.
+
 
 
 ## [0.95.0] - 2018-11-03

examples/fork_join/concord.yml

@@ -8,7 +8,7 @@ flows:
       # disable the `onCancel` handler, because it's going to handle
       # the parent's cancellation only
       disableOnCancel: true
-      jobs:
+      forks:
       # spawn multiple jobs with different parameters
       - entryPoint: aJob
         arguments:

it/server/src/test/java/com/walmartlabs/concord/it/server/ConcordTaskIT.java

@@ -251,4 +251,72 @@ public void testStartChildFinishedWithError() throws Exception {
         byte[] ab = getLog(pir.getLogFileName());
         assertLog(".*Child process.*FAILED.*BOOOM.*", ab);
     }
+
+    @Test(timeout = 60000)
+    public void testForkWithItemsWithOutVariable() throws Exception {
+        String orgName = "org_" + randomString();
+
+        OrganizationsApi orgApi = new OrganizationsApi(getApiClient());
+        orgApi.createOrUpdate(new OrganizationEntry().setName(orgName));
+
+        String projectName = "project_" + randomString();
+
+        ProjectsApi projectsApi = new ProjectsApi(getApiClient());
+        projectsApi.createOrUpdate(orgName, new ProjectEntry()
+                .setName(projectName)
+                .setVisibility(ProjectEntry.VisibilityEnum.PUBLIC)
+                .setAcceptsRawPayload(true));
+
+        byte[] payload = archive(ProcessRbacIT.class.getResource("concordTaskForkWithItemsWithOut").toURI());
+        Map<String, Object> input = new HashMap<>();
+        input.put("archive", payload);
+        input.put("org", orgName);
+        input.put("project", projectName);
+
+        StartProcessResponse spr = start(input);
+
+        ProcessApi processApi = new ProcessApi(getApiClient());
+        ProcessEntry pir = waitForStatus(processApi, spr.getInstanceId(), ProcessEntry.StatusEnum.FAILED);
+
+        // ---
+
+        byte[] ab = getLog(pir.getLogFileName());
+        assertLog(".*color=RED.*", ab);
+        assertLog(".*color=WHITE.*", ab);
+        assertLog(".*Done.*\\[\\[.*\\], \\[.*\\]\\] is completed.*", ab);
+    }
+
+    @Test(timeout = 60000)
+    public void testForkWithItems() throws Exception {
+        String orgName = "org_" + randomString();
+
+        OrganizationsApi orgApi = new OrganizationsApi(getApiClient());
+        orgApi.createOrUpdate(new OrganizationEntry().setName(orgName));
+
+        String projectName = "project_" + randomString();
+
+        ProjectsApi projectsApi = new ProjectsApi(getApiClient());
+        projectsApi.createOrUpdate(orgName, new ProjectEntry()
+                .setName(projectName)
+                .setVisibility(ProjectEntry.VisibilityEnum.PUBLIC)
+                .setAcceptsRawPayload(true));
+
+        byte[] payload = archive(ProcessRbacIT.class.getResource("concordTaskForkWithItems").toURI());
+        Map<String, Object> input = new HashMap<>();
+        input.put("archive", payload);
+        input.put("org", orgName);
+        input.put("project", projectName);
+
+        StartProcessResponse spr = start(input);
+
+        ProcessApi processApi = new ProcessApi(getApiClient());
+        ProcessEntry pir = waitForStatus(processApi, spr.getInstanceId(), ProcessEntry.StatusEnum.FAILED);
+
+        // ---
+
+        byte[] ab = getLog(pir.getLogFileName());
+        assertLog(".*color=RED.*", ab);
+        assertLog(".*color=WHITE.*", ab);
+        assertLog(".*Done.*\\[\\[.*\\], \\[.*\\]\\] is completed.*", ab);
+    }
 }

it/server/src/test/resources/com/walmartlabs/concord/it/server/concordTaskForkWithItems/concord.yml

@@ -0,0 +1,23 @@
+configuration:
+  arguments:
+    forkItems:
+    - entryPoint: sayHello
+      arguments:
+        color: "RED"
+    - entryPoint: sayHello
+      arguments:
+        color: "WHITE"
+
+flows:
+  default:
+  - task: concord
+    in:
+      action: fork
+      entryPoint: ${item.entryPoint}
+      arguments: ${item.arguments}
+    withItems: ${forkItems}
+
+  - log: "Done! ${jobs} is completed"
+
+  sayHello:
+  - log: "FORK: Hello, ${color}"

it/server/src/test/resources/com/walmartlabs/concord/it/server/concordTaskForkWithItemsWithOut/concord.yml

@@ -0,0 +1,25 @@
+configuration:
+  arguments:
+    forkItems:
+    - entryPoint: sayHello
+      arguments:
+        color: "RED"
+    - entryPoint: sayHello
+      arguments:
+        color: "WHITE"
+
+flows:
+  default:
+  - task: concord
+    in:
+      action: fork
+      entryPoint: ${item.entryPoint}
+      arguments: ${item.arguments}
+    out:
+      myJobs: ${jobs}
+    withItems: ${forkItems}
+
+  - log: "Done! ${myJobs} is completed"
+
+  sayHello:
+  - log: "FORK: Hello, ${color}"

it/server/src/test/resources/com/walmartlabs/concord/it/server/killCascade/concord.yml

@@ -5,7 +5,7 @@ flows:
     in:
       action: fork
       disableOnCancel: true
-      jobs:
+      forks:
       - entryPoint: aJob
         arguments:
           color: "red"

it/server/src/test/resources/com/walmartlabs/concord/it/server/processWithChildren/concord.yml

@@ -9,7 +9,7 @@ flows:
       # disable the `onCancel` handler, because it's going to handle
       # the parent's cancellation only
       disableOnCancel: true
-      jobs:
+      forks:
       # spawn multiple jobs with different parameters
       - entryPoint: aJob
         arguments:

plugins/tasks/concord/src/main/java/com/walmartlabs/concord/client/ConcordTask.java

@@ -25,6 +25,7 @@
 import com.walmartlabs.concord.common.IOUtils;
 import com.walmartlabs.concord.sdk.Constants;
 import com.walmartlabs.concord.sdk.Context;
+import com.walmartlabs.concord.sdk.ContextUtils;
 import com.walmartlabs.concord.sdk.InjectVariable;
 import org.apache.commons.compress.archivers.zip.ZipArchiveOutputStream;
 import org.slf4j.Logger;
@@ -68,6 +69,7 @@ public class ConcordTask extends AbstractConcordTask {
     private static final String DISABLE_ON_CANCEL_KEY = "disableOnCancel";
     private static final String DISABLE_ON_FAILURE_KEY = "disableOnFailure";
     private static final String ENTRY_POINT_KEY = "entryPoint";
+    private static final String FORKS_KEY = "forks";
     private static final String INSTANCE_ID_KEY = "instanceId";
     private static final String INSTANCES_KEY = "instances";
     private static final String JOB_OUT_KEY = "jobOut";
@@ -360,30 +362,21 @@ private Object getOutVars(Context ctx, String childId) throws Exception {
         });
     }
 
-    @SuppressWarnings("unchecked")
     private void fork(Context ctx) throws Exception {
-        List<Map<String, Object>> jobs;
-
-        Object v = ctx.getVariable(JOBS_KEY);
-        if (v != null) {
-            if (v instanceof List) {
-                jobs = (List<Map<String, Object>>) v;
-            } else {
-                throw new IllegalArgumentException("'" + JOBS_KEY + "' must be a list");
-            }
-        } else {
+        List<Map<String, Object>> jobs = ContextUtils.getList(ctx, FORKS_KEY, null);
+        if (jobs == null) {
             jobs = Collections.singletonList(createJobCfg(ctx, null));
         }
 
+        if (jobs.isEmpty()) {
+            throw new IllegalArgumentException("'" + FORKS_KEY + "' can't be an empty list");
+        }
+
         List<String> jobIds = forkMany(ctx, jobs);
         ctx.setVariable(JOBS_KEY, jobIds);
     }
 
     private List<String> forkMany(Context ctx, List<Map<String, Object>> jobs) throws Exception {
-        if (jobs.isEmpty()) {
-            throw new IllegalArgumentException("'" + JOBS_KEY + "' can't be an empty list");
-        }
-
         List<String> ids = new ArrayList<>();
 
         for (Map<String, Object> job : jobs) {