runtime-v2: implement loop syntax (#578)

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ConcordJsonSchemaGenerator.java

@@ -21,11 +21,7 @@
  */
 
 import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.*;
 import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonFormatVisitorWrapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -35,16 +31,8 @@
 import com.kjetland.jackson.jsonSchema.SubclassesResolver;
 import com.kjetland.jackson.jsonSchema.SubclassesResolverImpl;
 import com.walmartlabs.concord.imports.Imports;
-import com.walmartlabs.concord.runtime.v2.model.Form;
-import com.walmartlabs.concord.runtime.v2.model.ProcessDefinition;
-import com.walmartlabs.concord.runtime.v2.model.ProcessDefinitionConfiguration;
-import com.walmartlabs.concord.runtime.v2.model.Step;
-import com.walmartlabs.concord.runtime.v2.model.Trigger;
-import com.walmartlabs.concord.runtime.v2.schema.ImportsMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.ProcessDefinitionConfigurationMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.ProcessDefinitionMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.StepMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.TriggerMixIn;
+import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.schema.*;
 
 import java.io.IOException;
 import java.io.OutputStream;

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ProjectSerializerV2.java

@@ -56,6 +56,7 @@ private static ObjectMapper createObjectMapper() {
         SimpleModule module = new SimpleModule()
                 .addSerializer(SimpleOptions.class, new SimpleOptionsSerializer())
                 .addSerializer(Retry.class, new RetryOptionsSerializer())
+                .addSerializer(Loop.class, new LoopOptionsSerializer())
                 .addSerializer(WithItems.class, new WithItemsSerializer())
                 .addSerializer(Checkpoint.class, new CheckpointStepSerializer())
                 .addSerializer(ExitStep.class, new ExitStepSerializer())

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/FlowCallOptions.java

@@ -59,6 +59,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/GroupOfStepsOptions.java

@@ -46,6 +46,9 @@ default List<Step> errorSteps() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     static ImmutableGroupOfStepsOptions.Builder builder() {
         return ImmutableGroupOfStepsOptions.builder();
     }

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/Loop.java

@@ -0,0 +1,55 @@
+package com.walmartlabs.concord.runtime.v2.model;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2019 Walmart Inc.
+ * -----
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * =====
+ */
+
+import org.immutables.value.Value;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.Map;
+
+@Value.Immutable
+@Value.Style(jdkOnly = true)
+public interface Loop extends Serializable {
+
+    long serialVersionUID = 1L;
+
+    static ImmutableLoop.Builder builder() {
+        return  ImmutableLoop.builder();
+    }
+
+    Serializable items();
+
+    @Value.Default
+    default Mode mode() {
+        return Mode.SERIAL;
+    }
+
+    @Value.Default
+    default Map<String, Object> options() {
+        return Collections.emptyMap();
+    }
+
+    enum Mode {
+        SERIAL,
+        PARALLEL
+    }
+}

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinition.java

@@ -24,7 +24,10 @@
 import org.immutables.value.Value;
 
 import java.io.Serializable;
-import java.util.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 @Value.Immutable
 @Value.Style(jdkOnly = true)

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinitionConfiguration.java

@@ -106,6 +106,11 @@ default List<String> out() {
     @Nullable
     String template();
 
+    @Value.Default
+    default int parallelLoopParallelism() {
+        return Runtime.getRuntime().availableProcessors();
+    }
+
     static ImmutableProcessDefinitionConfiguration.Builder builder() {
         return ImmutableProcessDefinitionConfiguration.builder();
     }

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ScriptCallOptions.java

@@ -60,6 +60,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/TaskCallOptions.java

@@ -57,6 +57,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/WithItems.java

@@ -24,6 +24,10 @@
 
 import java.io.Serializable;
 
+/**
+ * @deprecated use {@link com.walmartlabs.concord.runtime.v2.model.Loop}
+ */
+@Deprecated
 @Value.Immutable
 @Value.Style(jdkOnly = true)
 public interface WithItems extends Serializable {

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ConfigurationGrammar.java

@@ -84,7 +84,8 @@ public final class ConfigurationGrammar {
                                     optional("out", stringArrayVal.map(o::addAllOut)),
                                     optional("arguments", mapVal.map(o::arguments)),
                                     optional("debug", booleanVal.map(o::debug)),
-                                    optional("template", stringVal.map(o::template))))
+                                    optional("template", stringVal.map(o::template)),
+                                    optional("parallelLoopParallelism", intVal.map(o::parallelLoopParallelism))))
                             .map(ImmutableProcessDefinitionConfiguration.Builder::build));
 
     public static final Parser<Atom, ProcessDefinitionConfiguration> processCfgVal =

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ExpressionGrammar.java

@@ -22,7 +22,10 @@
 
 import com.fasterxml.jackson.core.JsonToken;
 import com.walmartlabs.concord.runtime.v2.Constants;
-import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.model.Expression;
+import com.walmartlabs.concord.runtime.v2.model.ExpressionOptions;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableExpressionOptions;
+import com.walmartlabs.concord.runtime.v2.model.Step;
 import io.takari.parc.Parser;
 
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/FlowCallGrammar.java

@@ -32,6 +32,7 @@
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -54,6 +55,7 @@ private static Parser<Atom, FlowCallOptions> callOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps))
                 ))

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GrammarV2.java

@@ -31,6 +31,7 @@
 import java.time.Duration;
 import java.time.format.DateTimeParseException;
 import java.util.*;
+import java.util.function.BiPredicate;
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
@@ -96,12 +97,17 @@ public final class GrammarV2 {
     });
 
     public static <E extends Enum<E>> Parser<Atom, E> enumVal(Class<E> enumData) {
+        return enumVal(enumData, String::equals);
+    }
+
+    public static <E extends Enum<E>> Parser<Atom, E> enumVal(Class<E> enumData,
+                                                              BiPredicate<String, String> cmp) {
         return value.map(vv -> {
             String v = vv.getValue(YamlValueType.STRING);
 
-            for (Enum<E> enumVal : enumData.getEnumConstants()) {
-                if (enumVal.name().equals(v)) {
-                    return Enum.valueOf(enumData, v);
+            for (E enumVal : enumData.getEnumConstants()) {
+                if (cmp.test(enumVal.name(), v)) {
+                    return enumVal;
                 }
             }
 

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GroupOfStepsGrammar.java

@@ -24,10 +24,12 @@
 import com.walmartlabs.concord.runtime.v2.model.*;
 import io.takari.parc.Parser;
 
-import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.namedStep;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.with;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static io.takari.parc.Combinators.choice;
 
 public final class GroupOfStepsGrammar {
@@ -47,6 +49,7 @@ private static Parser<Atom, GroupOfStepsOptions> groupOptions(String stepName) {
                         optional("error", stepsVal.map(o::errorSteps)),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("meta", mapVal.map(o::putAllMeta)),
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v)))
                 ))

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/LoopGrammar.java

@@ -0,0 +1,49 @@
+package com.walmartlabs.concord.runtime.v2.parser;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2019 Walmart Inc.
+ * -----
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * =====
+ */
+
+import com.fasterxml.jackson.core.JsonToken;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableLoop;
+import com.walmartlabs.concord.runtime.v2.model.Loop;
+import io.takari.parc.Parser;
+
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+
+public final class LoopGrammar {
+
+    private static final Parser<Atom, Loop> loop =
+            betweenTokens(JsonToken.START_OBJECT, JsonToken.END_OBJECT,
+                    with(Loop::builder,
+                            o -> options(
+                                    mandatory("items", nonNullVal.map(o::items)),
+                                    optional("mode", enumVal(Loop.Mode.class, String::equalsIgnoreCase).map(o::mode)),
+                                    optional("parallelism", intVal.map(v -> o.putOptions("parallelism", v)))
+                            ))
+                            .map(ImmutableLoop.Builder::build));
+
+    public static final Parser<Atom, Loop> loopVal =
+            orError(loop, YamlValueType.LOOP);
+
+    private LoopGrammar() {
+    }
+}

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ParallelGrammar.java

@@ -21,7 +21,6 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.ImmutableParallelBlockOptions;
-import com.walmartlabs.concord.runtime.v2.model.ImmutableTaskCallOptions;
 import com.walmartlabs.concord.runtime.v2.model.ParallelBlock;
 import com.walmartlabs.concord.runtime.v2.model.ParallelBlockOptions;
 import io.takari.parc.Parser;

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ScriptGrammar.java

@@ -21,14 +21,18 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.Constants;
-import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableScriptCallOptions;
+import com.walmartlabs.concord.runtime.v2.model.ScriptCall;
+import com.walmartlabs.concord.runtime.v2.model.ScriptCallOptions;
+import com.walmartlabs.concord.runtime.v2.model.WithItems;
 import io.takari.parc.Parser;
 
 import static com.walmartlabs.concord.runtime.v2.parser.ExpressionGrammar.maybeExpression;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -52,6 +56,7 @@ private static Parser<Atom, ScriptCallOptions> scriptOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps))
                 ))

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/TaskGrammar.java

@@ -32,6 +32,7 @@
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -62,6 +63,7 @@ private static Parser<Atom, TaskCallOptions> taskOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps)),
                         optional("ignoreErrors", booleanVal.map(o::ignoreErrors))

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/YamlValueType.java

@@ -108,6 +108,7 @@ public final class YamlValueType<T> {
     public static final YamlValueType<GithubTriggerExclusiveMode> GITHUB_EXCLUSIVE_MODE = type("GITHUB_EXCLUSIVE_MODE");
     public static final YamlValueType<Map<String, Object>> GITHUB_REPOSITORY_INFO = type("GITHUB_REPOSITORY_INFO");
     public static final YamlValueType<List<Map<String, Object>>> ARRAY_OF_GITHUB_REPOSITORY_INFO = array("REPOSITORY_INFO", GITHUB_REPOSITORY_INFO);
+    public static final YamlValueType<Loop> LOOP = type("LOOP");
 
     private final String name;
 

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/FlowCallStepMixIn.java

@@ -50,11 +50,8 @@ public interface FlowCallStepMixIn extends NamedStep {
     @JsonProperty("error")
     List<Step> error();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("meta")
     Map<String, Object> meta();

runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/GroupOfStepsMixIn.java

@@ -34,11 +34,8 @@ public interface GroupOfStepsMixIn extends NamedStep {
     @JsonSchemaInject(json = "{\"oneOf\": [ {\"type\": \"array\", \"items\" : {\"type\" : \"string\"}}, {\"type\": \"string\"} ]}", merge = false)
     Object out();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("error")
     List<StepMixIn> errorSteps();