Dag descriptor converter

.gitignore

@@ -8,3 +8,4 @@ dependency-reduced-pom.xml
 .DS_Store
 /rill-flow-web/src/main/resources/static/
 .vscode/
+logs

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/dag/DAG.java

@@ -20,9 +20,9 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.weibo.rill.flow.interfaces.model.mapping.Mapping;
 import com.weibo.rill.flow.interfaces.model.resource.BaseResource;
-import com.weibo.rill.flow.olympicene.core.model.strategy.CallbackConfig;
 import com.weibo.rill.flow.interfaces.model.strategy.Timeline;
 import com.weibo.rill.flow.interfaces.model.task.BaseTask;
+import com.weibo.rill.flow.olympicene.core.model.strategy.CallbackConfig;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
@@ -39,6 +39,7 @@
     private String version;
     private DAGType type;
     private Timeline timeline;
+    @Setter
     private List<BaseTask> tasks;
     @Setter
     private List<BaseResource> resources;
@@ -48,6 +49,9 @@
     @Setter
     private Map<String, List<Mapping>> commonMapping;
     private String inputSchema;
+    private Map<String, Object> output;
+    @Setter
+    private String endTaskName;
 
     @JsonCreator
     public DAG(@JsonProperty("workspace") String workspace,
@@ -62,7 +66,8 @@
                @JsonProperty("commonMapping") Map<String, List<Mapping>> commonMapping,
                @JsonProperty("namespace") String namespace,
                @JsonProperty("service") String service,
-               @JsonProperty("inputSchema") String inputSchema) {
+               @JsonProperty("inputSchema") String inputSchema,
+               @JsonProperty("output") Map<String, Object> output) {
         this.workspace = StringUtils.isBlank(workspace) ? namespace : workspace;
         this.dagName = StringUtils.isBlank(dagName) ? service : dagName;
         this.version = version;
@@ -74,5 +79,6 @@
         this.defaultContext = defaultContext;
         this.commonMapping = commonMapping;
         this.inputSchema = inputSchema;
+        this.output = output;
     }
 }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/dag/DescriptorPO.java

@@ -0,0 +1,10 @@
+package com.weibo.rill.flow.olympicene.core.model.dag;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class DescriptorPO {
+    private String descriptor;
+}

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/dag/DescriptorVO.java

@@ -0,0 +1,10 @@
+package com.weibo.rill.flow.olympicene.core.model.dag;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class DescriptorVO {
+    private String descriptor;
+}

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ChoiceTask.java

@@ -45,6 +45,7 @@
             @JsonProperty("description") String description,
             @JsonProperty("category") String category,
             @JsonProperty("next") String next,
+            @JsonProperty("input") Map<String, Object> input,
             @JsonProperty("inputMappings") List<Mapping> inputMappings,
             @JsonProperty("outputMappings") List<Mapping> outputMappings,
             @JsonProperty("choices") List<Choice> choices,
@@ -55,7 +56,8 @@
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
             @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
+                timeline, isKeyCallback, keyExp, parameters, templateId, input);
         this.choices = choices;
     }
 

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ForeachTask.java

@@ -58,8 +58,10 @@ public ForeachTask(
             @JsonProperty("isKeyCallback") boolean isKeyCallback,
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
-            @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+            @JsonProperty("templateId") String templateId,
+            @JsonProperty("input") Map<String, Object> input) {
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
+                timeline, isKeyCallback, keyExp, parameters, templateId, input);
         this.synchronization = synchronization;
         this.iterationMapping = iterationMapping;
         this.tasks = tasks;

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/PassTask.java

@@ -25,6 +25,7 @@
 import com.weibo.rill.flow.interfaces.model.strategy.Timeline;
 import com.weibo.rill.flow.interfaces.model.task.BaseTask;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 import java.util.ArrayList;
@@ -35,6 +36,7 @@
 @Getter
 @Setter
 @JsonTypeName("pass")
+@NoArgsConstructor
 public class PassTask extends BaseTask {
 
     @JsonCreator
@@ -51,8 +53,10 @@
                     @JsonProperty("isKeyCallback") boolean isKeyCallback,
                     @JsonProperty("keyExp") String keyExp,
                     @JsonProperty("parameters") Map<String, Object> parameters,
-                    @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+                    @JsonProperty("templateId") String templateId,
+                    @JsonProperty("input") Map<String, Object> input) {
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
+                timeline, isKeyCallback, keyExp, parameters, templateId, input);
         Optional.ofNullable(timeline).ifPresent(it -> it.setTimeoutInSeconds(null));
     }
 

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ReturnTask.java

@@ -52,8 +52,10 @@
                       @JsonProperty("isKeyCallback") boolean isKeyCallback,
                       @JsonProperty("keyExp") String keyExp,
                       @JsonProperty("parameters") Map<String, Object> parameters,
-                      @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+                      @JsonProperty("templateId") String templateId,
+                      @JsonProperty("input") Map<String, Object> input) {
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
+                timeline, isKeyCallback, keyExp, parameters, templateId, input);
         Optional.ofNullable(timeline).ifPresent(it -> it.setTimeoutInSeconds(null));
         this.conditions = conditions;
     }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/SuspenseTask.java

@@ -55,8 +55,10 @@
                         @JsonProperty("isKeyCallback") boolean isKeyCallback,
                         @JsonProperty("keyExp") String keyExp,
                         @JsonProperty("parameters") Map<String, Object> parameters,
-                        @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, tolerance, inputMappings, outputMappings, progress, degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+                        @JsonProperty("templateId") String templateId,
+                        @JsonProperty("input") Map<String, Object> input) {
+        super(name, title, description, category, next, tolerance, inputMappings, outputMappings, progress, degrade,
+                timeline, isKeyCallback, keyExp, parameters, templateId, input);
         this.conditions = conditions;
         this.interruptions = interruptions;
     }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/SwitchTask.java

@@ -51,9 +51,10 @@ public SwitchTask(
             @JsonProperty("isKeyCallback") boolean isKeyCallback,
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
-            @JsonProperty("templateId") String templateId) {
+            @JsonProperty("templateId") String templateId,
+            @JsonProperty("input") Map<String, Object> input) {
         super(name, title, description, category, null, false, inputMappings, null, progress,
-                degrade, timeline, isKeyCallback, keyExp, parameters, templateId);
+                degrade, timeline, isKeyCallback, keyExp, parameters, templateId, input);
         this.switches = switches;
         this.setNext(switches.stream().map(Switch::getNext).collect(Collectors.joining(",")));
     }

rill-flow-dag/olympicene-core/src/test/groovy/com/weibo/rill/flow/olympicene/core/helper/DAGWalkHelperTest.groovy

@@ -20,7 +20,7 @@ class DAGWalkHelperTest extends Specification {
 
     def "task status of parent task should be correct"() {
         given:
-        ForeachTask baseTask = new ForeachTask('base_1', '', '',TaskCategory.FOREACH.getValue(), null, null, null, null, null, null, null, null, null, false, null, null, null)
+        ForeachTask baseTask = new ForeachTask('base_1', '', '',TaskCategory.FOREACH.getValue(), null, null, null, null, null, null, null, null, null, false, null, null, null, null)
 
         TaskInfo parentTask = new TaskInfo(name: 'parent',
                 task: baseTask,

rill-flow-dag/olympicene-ddl/src/main/java/com/weibo/rill/flow/olympicene/ddl/serialize/ObjectMapperFactory.java

@@ -53,6 +53,7 @@ public class ObjectMapperFactory {
     }
 
     private static final YAMLFactory YAML_FACTORY = new YAMLFactory()
+            .configure(YAMLGenerator.Feature.USE_NATIVE_TYPE_ID, false)
             .configure(YAMLGenerator.Feature.WRITE_DOC_START_MARKER, false)
             .configure(YAMLGenerator.Feature.USE_NATIVE_OBJECT_ID, false);
     private static final ObjectMapper YAML_MAPPER = new ObjectMapper(YAML_FACTORY);
@@ -63,6 +64,7 @@ public class ObjectMapperFactory {
         YAML_MAPPER.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
         YAML_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
         YAML_MAPPER.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL, true);
+        YAML_MAPPER.setSerializationInclusion(JsonInclude.Include.NON_NULL);
         YAML_MAPPER.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
         YAML_MAPPER.registerSubtypes(
                 new NamedType(FunctionTask.class, "function"),

rill-flow-dag/olympicene-storage/src/test/groovy/com/weibo/rill/flow/olympicene/storage/save/impl/DAGInfoDAOTest.groovy

@@ -20,7 +20,7 @@ class DAGInfoDAOTest extends Specification {
     def setup() {
         dagInfo.executionId = executionId
         dagInfo.dagStatus = DAGStatus.NOT_STARTED
-        dagInfo.dag = new DAG("workspace", "dagName", "1.0.0", DAGType.FLOW, null, Lists.newArrayList(), null, null, null, null, "ns", "service", null)
+        dagInfo.dag = new DAG("workspace", "dagName", "1.0.0", DAGType.FLOW, null, Lists.newArrayList(), null, null, null, null, "ns", "service", null, null)
     }
 
     def "updateDagInfo redis eval shardingKey check"() {

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/mappings/JSONPathInputOutputMapping.java

@@ -42,7 +42,7 @@
 @Slf4j
 public class JSONPathInputOutputMapping implements InputOutputMapping, JSONPath {
     Configuration conf = Configuration.builder().options(Option.DEFAULT_PATH_LEAF_TO_NULL).build();
-    private static final Pattern JSONPATH_PATTERN = Pattern.compile("\\[\"(.*?)\"]|\\['(.*?)']");
+    private static final Pattern JSONPATH_PATTERN = Pattern.compile("\\[(.*?)]");
 
     @Value("${rill.flow.function.trigger.uri}")
     private String rillFlowFunctionTriggerUri;
@@ -201,26 +201,69 @@
         while (matcher.find()) {
             if (matcher.group(1) != null) {
                 jsonPathParts.add(matcher.group(1));
-            } else if (matcher.group(2) != null) {
-                jsonPathParts.add(matcher.group(2));
             }
         }
 
-        jsonPathParts.remove(jsonPathParts.size() - 1);
-
         Object current = map;
-        for (String part: jsonPathParts) {
+        for (int i = 0; i < jsonPathParts.size() - 1; i++) {
+            String part = jsonPathParts.get(i);
+            if (part.startsWith("\"") || part.startsWith("'")) {
+                part = part.substring(1, part.length() - 1);
+            }
             if (current instanceof Map) {
-                Map<String, Object> mapCurrent = (Map<String, Object>) current;
-                current = mapCurrent.computeIfAbsent(part, k -> new HashMap<>());
-            } else {
-                break;
+                current = processMapJsonPathPart(current, part, jsonPathParts, i);
+            } else if (current instanceof List) {
+                current = processListJsonPathPart(current, part, jsonPathParts, i);
             }
         }
 
         return JsonPath.using(conf).parse(map).set(path, value).json();
     }
 
+    private Object processListJsonPathPart(Object current, String part, List<String> jsonPathParts, int i) {
+        List<Object> listCurrent = (List<Object>) current;
+        int index = Integer.parseInt(part);
+        Object insertPosition = listCurrent.get(index);
+        if (jsonPathParts.get(i + 1).matches("\\d+")) {
+            // 1. 下一个元素是数字，也就是数组的索引，所以需要创建数组并且填充到索引位置
+            List<Object> nextArray = createAndFillNextArrayPart(insertPosition, jsonPathParts, i);
+            listCurrent.set(index, nextArray);
+        } else if (i + 1 < jsonPathParts.size() && insertPosition == null) {
+            // 2. 下一个元素不是数字，则创建 map
+            listCurrent.set(index, new HashMap<>());
+        }
+        return listCurrent.get(index);
+    }
+
+    private Object processMapJsonPathPart(Object current, String part, List<String> jsonPathParts, int i) {
+        Map<String, Object> mapCurrent = (Map<String, Object>) current;
+        Object currentValue = mapCurrent.get(part);
+        if (jsonPathParts.get(i + 1).matches("\\d+")) {
+            List<Object> nextArray = createAndFillNextArrayPart(currentValue, jsonPathParts, i);
+            mapCurrent.put(part, nextArray);
+        } else if (i + 1 < jsonPathParts.size() && currentValue == null) {
+            mapCurrent.put(part, new HashMap<>());
+        }
+        return mapCurrent.get(part);
+    }
+
+    /**
+     * 为下一个元素创建数组类型对象，并用 null 值填充指定元素个数
+     */
+    private List<Object> createAndFillNextArrayPart(Object nextPart, List<String> jsonPathParts, int i) {
+        List<Object> nextArray;
+        if (nextPart instanceof List) {
+            nextArray = (List<Object>) nextPart;
+        } else {
+            nextArray = new ArrayList<>();
+        }
+        int nextIndex = Integer.parseInt(jsonPathParts.get(i + 1));
+        for (int j = nextArray.size(); j <= nextIndex; j++) {
+            nextArray.add(null);
+        }
+        return nextArray;
+    }
+
     @Override
     public Map<String, Map<String, Object>> delete(Map<String, Map<String, Object>> map, String path) {
         if (map == null) {

rill-flow-dag/olympicene-traversal/src/main/java/com/weibo/rill/flow/olympicene/traversal/runners/PassTaskRunner.java

@@ -20,11 +20,11 @@
 import com.google.common.collect.Sets;
 import com.weibo.rill.flow.interfaces.model.task.TaskInfo;
 import com.weibo.rill.flow.interfaces.model.task.TaskStatus;
+import com.weibo.rill.flow.olympicene.core.model.task.ExecutionResult;
 import com.weibo.rill.flow.olympicene.core.model.task.TaskCategory;
 import com.weibo.rill.flow.olympicene.core.runtime.DAGContextStorage;
 import com.weibo.rill.flow.olympicene.core.runtime.DAGInfoStorage;
 import com.weibo.rill.flow.olympicene.core.runtime.DAGStorageProcedure;
-import com.weibo.rill.flow.olympicene.core.model.task.ExecutionResult;
 import com.weibo.rill.flow.olympicene.core.switcher.SwitcherManager;
 import com.weibo.rill.flow.olympicene.traversal.helper.ContextHelper;
 import com.weibo.rill.flow.olympicene.traversal.mappings.InputOutputMapping;
@@ -60,7 +60,7 @@ protected ExecutionResult doRun(String executionId, TaskInfo taskInfo, Map<Strin
         log.info("pass task begin to run executionId:{}, taskInfoName:{}", executionId, taskInfo.getName());
         if (CollectionUtils.isNotEmpty(taskInfo.getTask().getOutputMappings())) {
             Map<String, Object> context = ContextHelper.getInstance().getContext(dagContextStorage, executionId, taskInfo);
-            outputMappings(context, new HashMap<>(), new HashMap<>(), taskInfo.getTask().getOutputMappings());
+            outputMappings(context, new HashMap<>(), input, taskInfo.getTask().getOutputMappings());
             saveContext(executionId, context, Sets.newHashSet(taskInfo));
         }
 

rill-flow-dag/olympicene-traversal/src/test/groovy/com/weibo/rill/flow/olympicene/traversal/json/JSONPathTest.groovy

@@ -73,6 +73,44 @@ class JSONPathTest extends Specification {
         ['input': ['type': 'gif']] | _
     }
 
+    def "set value when there is an array in the path"() {
+        given:
+        String path = "\$.input.meta.user[1].id"
+        long value = 1
+
+        when:
+        mapping.setValue(map, value, path)
+
+        then:
+        mapping.getValue(map, path) == value
+
+        where:
+        map                                        | _
+        [:]                                        | _
+        ['context': 123]                           | _
+        ['input': ['type': 'gif']]                 | _
+        ['input': ['meta': ['user': [['id': 1]]]]] | _
+    }
+
+    def "set value when the last part of the path is an array"() {
+        given:
+        String path = "\$.input.meta.user[1]"
+        long value = 1
+
+        when:
+        mapping.setValue(map, value, path)
+
+        then:
+        mapping.getValue(map, path) == value
+
+        where:
+        map                                        | _
+        [:]                                        | _
+        ['context': 123]                           | _
+        ['input': ['type': 'gif']]                 | _
+        ['input': ['meta': ['user': [['id': 1]]]]] | _
+    }
+
     def "set value intermediate route test contains dot"() {
         given:
         String path = "\$.input.meta.user.[\"id.key\"]"

rill-flow-impl/src/test/groovy/com/weibo/rill/flow/impl/service/DagInfoDeserializeServiceImplTest.groovy

@@ -21,7 +21,7 @@ class DagInfoDeserializeServiceImplTest extends Specification {
     def setup() {
         dagInfo.executionId = executionId
         dagInfo.dagStatus = DAGStatus.NOT_STARTED
-        dagInfo.dag = new DAG("workspace", "dagName", "1.0.0", DAGType.FLOW, null, Lists.newArrayList(), null, null, null, null, "ns", "service", null)
+        dagInfo.dag = new DAG("workspace", "dagName", "1.0.0", DAGType.FLOW, null, Lists.newArrayList(), null, null, null, null, "ns", "service", null, null)
     }
 
     def "delDagInfo invoke setting if time above zero"() {

rill-flow-interfaces/src/main/java/com/weibo/rill/flow/interfaces/model/mapping/Mapping.java

@@ -18,6 +18,8 @@
 
 import lombok.*;
 
+import java.util.Objects;
+
 @AllArgsConstructor
 @NoArgsConstructor
 @Getter
@@ -50,4 +52,19 @@ public Mapping(String source, String target) {
         this.source = source;
         this.target = target;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        Mapping mapping = (Mapping) o;
+        return Objects.equals(reference, mapping.reference) && Objects.equals(source, mapping.source)
+                && Objects.equals(transform, mapping.transform) && Objects.equals(target, mapping.target)
+                && Objects.equals(variable, mapping.variable);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(reference, source, transform, target, variable);
+    }
 }