Merge branch 'master' into ib/log-db

Author: ibodrov

it/runtime-v2/src/test/java/com/walmartlabs/concord/it/runtime/v2/ConcordConfiguration.java

@@ -9,9 +9,9 @@
  * 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.
@@ -60,7 +60,7 @@ public static ConcordRule configure() {
                 .useLocalMavenRepository(true)
                 .extraConfigurationSupplier(() -> """
                     concord-agent {
-                        dependencyResolveTimeout = "5 seconds"
+                        dependencyResolveTimeout = "20 seconds"
                         logMaxDelay = "250 milliseconds"
                         pollInterval = "250 milliseconds"
                         prefork {

it/runtime-v2/src/test/java/com/walmartlabs/concord/it/runtime/v2/ProcessIT.java

@@ -35,6 +35,7 @@
 import static com.walmartlabs.concord.it.common.ITUtils.randomString;
 import static com.walmartlabs.concord.it.runtime.v2.Utils.resourceToString;
 import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 public class ProcessIT extends AbstractTest {
 
@@ -196,6 +197,25 @@ public void testOutVariables() throws Exception {
         assertFalse(data.containsKey("z"));
     }
 
+    @Test
+    public void testOutVariablesForFailedProcess() throws Exception {
+        Payload payload = new Payload()
+                .archive(resource("outForFailed"))
+                .out("x", "y.some.boolean", "z");
+
+        ConcordProcess proc = concord.processes().start(payload);
+        expectStatus(proc, ProcessEntry.StatusEnum.FAILED);
+
+        // ---
+
+        Map<String, Object> data = proc.getOutVariables();
+        assertNotNull(data);
+
+        assertEquals(123, data.get("x"));
+        assertEquals(true, data.get("y.some.boolean"));
+        assertFalse(data.containsKey("z"));
+    }
+
     @Test
     public void testThrowWithPayload() throws Exception {
         Payload payload = new Payload()
@@ -729,6 +749,23 @@ public void testDryRunModeNotSupportedByScript() throws Exception {
         proc.assertLog(".*Error @ line: 6, col: 7. Dry-run mode is not supported for this 'script' step.*");
     }
 
+    @Test
+    public void testThrowParallelWithPayload() throws Exception {
+        Payload payload = new Payload()
+                .archive(resource("parallelExceptionPayload"));
+
+        ConcordProcess proc = concord.processes().start(payload);
+        expectStatus(proc, ProcessEntry.StatusEnum.FINISHED);
+
+        // ---
+        Map<String, Object> data = proc.getOutVariables();
+        List<Map<String, Object>> exceptions = (List<Map<String, Object>>) ConfigurationUtils.get(data, "exceptions");
+
+        assertNotNull(exceptions);
+        assertEquals(List.of("BOOM1", "BOOM2"), exceptions.stream().map(e -> e.get("message")).toList());
+        assertEquals(List.of(Map.of("key", 1), Map.of("key", 2)), exceptions.stream().map(e -> e.get("payload")).toList());
+    }
+
     private List<ProcessEventEntry> getProcessElementEvents(ConcordProcess proc) throws Exception {
         ProcessEventsApi processEventsApi = new ProcessEventsApi(concord.apiClient());
         return processEventsApi.listProcessEvents(proc.instanceId(), "ELEMENT", null, null, null, "pre", null, null);

it/runtime-v2/src/test/resources/com/walmartlabs/concord/it/runtime/v2/outForFailed/concord.yml

@@ -0,0 +1,14 @@
+configuration:
+  runtime: "concord-v2"
+
+flows:
+  default:
+  - set:
+      x: 123
+      y:
+        some:
+          nested: ["data", "in", "arrays"]
+          boolean: true
+          number: 234
+
+  - throw: BOOM

it/runtime-v2/src/test/resources/com/walmartlabs/concord/it/runtime/v2/parallelExceptionPayload/concord.yml

@@ -0,0 +1,22 @@
+configuration:
+  runtime: "concord-v2"
+  out:
+    - exceptions
+
+flows:
+  default:
+  - try:
+    - task: "throw"
+      in:
+        exception: "BOOM${item}"
+        payload:
+          key: "${item}"
+    loop:
+      items:
+        - 1
+        - 2
+      mode: parallel
+      parallelism: 2
+    error:
+      - set:
+          exceptions: ${lastError.cause.exceptions.stream().map(e -> e.cause).toList()}

runtime/v2/runner-test/src/test/java/com/walmartlabs/concord/runtime/v2/runner/LogSegmentsTest.java

@@ -625,7 +625,7 @@ public void callWithRetryTest() throws Exception {
                 "(concord.yaml) @ line: 3, col: 7, thread: 0, flow: inner");
         assertSegmentStatusError(log, 1);
 
-        assertSystemSegment(log, "[WARN ] Last error: com.walmartlabs.concord.runtime.v2.sdk.UserDefinedException: FAIL. Waiting for 1000ms before retry (attempt #0)");
+        assertSystemSegment(log, "[WARN ] Last error: FAIL. Waiting for 1000ms before retry (attempt #0)");
 
         assertSegmentLog(log, 2, "[INFO ] in inner flow");
         assertSegmentStatusOk(log, 2);

runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/OutVariablesProcessor.java

@@ -9,9 +9,9 @@
  * 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.
@@ -20,6 +20,7 @@
  * =====
  */
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.walmartlabs.concord.runtime.v2.sdk.EvalContext;
 import com.walmartlabs.concord.runtime.v2.sdk.EvalContextFactory;
@@ -43,6 +44,9 @@
  */
 public class OutVariablesProcessor implements ExecutionListener {
 
+    private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {
+    };
+
     private final ObjectMapper objectMapper;
     private final PersistenceService persistenceService;
     private final List<String> outVariables;
@@ -84,7 +88,13 @@ public void afterProcessEnds(Runtime runtime, State state, Frame lastFrame) {
             return;
         }
 
+        Map<String, Object> currentOut = persistenceService.loadPersistedFile(Constants.Files.OUT_VALUES_FILE_NAME,
+                in -> objectMapper.readValue(in, MAP_TYPE));
+
+        Map<String, Object> result = new HashMap<>(currentOut != null ? currentOut : Collections.emptyMap());
+        result.putAll(outValues);
+
         persistenceService.persistFile(Constants.Files.OUT_VALUES_FILE_NAME,
-                out -> objectMapper.writeValue(out, outValues));
+                out -> objectMapper.writeValue(out, result));
     }
 }

runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/Runner.java

@@ -26,9 +26,7 @@
 import com.walmartlabs.concord.runtime.common.injector.InstanceId;
 import com.walmartlabs.concord.runtime.v2.model.ProcessDefinition;
 import com.walmartlabs.concord.runtime.v2.runner.compiler.CompilerUtils;
-import com.walmartlabs.concord.runtime.v2.runner.vm.SaveLastErrorCommand;
-import com.walmartlabs.concord.runtime.v2.runner.vm.UpdateLocalsCommand;
-import com.walmartlabs.concord.runtime.v2.runner.vm.VMUtils;
+import com.walmartlabs.concord.runtime.v2.runner.vm.*;
 import com.walmartlabs.concord.runtime.v2.sdk.Compiler;
 import com.walmartlabs.concord.runtime.v2.sdk.ProcessConfiguration;
 import com.walmartlabs.concord.svm.*;
@@ -77,7 +75,7 @@ public ProcessSnapshot start(ProcessConfiguration processConfiguration, ProcessD
         // install the exception handler into the root frame
         // takes care of all unhandled errors bubbling up
         VMUtils.assertNearestRoot(state, state.getRootThreadId())
-                .setExceptionHandler(new SaveLastErrorCommand());
+                .setExceptionHandler(new BlockCommand(new SaveOutVariablesOnErrorCommand(), new SaveLastErrorCommand()));
 
         VM vm = createVM(processDefinition);
         // update the global variables using the input map by running a special command

runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/remote/TaskCallEventRecordingListener.java

@@ -28,6 +28,7 @@
 import com.walmartlabs.concord.runtime.v2.model.Location;
 import com.walmartlabs.concord.runtime.v2.model.Step;
 import com.walmartlabs.concord.runtime.v2.runner.EventReportingService;
+import com.walmartlabs.concord.runtime.v2.runner.SensitiveDataHolder;
 import com.walmartlabs.concord.runtime.v2.runner.tasks.TaskCallEvent;
 import com.walmartlabs.concord.runtime.v2.runner.tasks.TaskCallListener;
 import com.walmartlabs.concord.runtime.v2.sdk.*;
@@ -64,7 +65,9 @@ public void onEvent(TaskCallEvent event) {
 
         List<Object> inVars = event.input();
         if (inVars != null && eventConfiguration.recordTaskInVars()) {
-            Map<String, Object> vars = maskVars(convertInput(hideSensitiveData(inVars, event.inputAnnotations())), eventConfiguration.inVarsBlacklist());
+            Map<String, Object> input = convertInput(processSensitiveDataAnnotations(inVars, event.inputAnnotations()));
+            input = processSensitiveData(input);
+            Map<String, Object> vars = maskVars(input, eventConfiguration.inVarsBlacklist());
             if (eventConfiguration.truncateInVars()) {
                 vars = ObjectTruncater.truncateMap(vars, eventConfiguration.truncateMaxStringLength(), eventConfiguration.truncateMaxArrayLength(), eventConfiguration.truncateMaxDepth());
             }
@@ -75,7 +78,9 @@ public void onEvent(TaskCallEvent event) {
 
         Object outVars = event.result();
         if (outVars != null && eventConfiguration.recordTaskOutVars()) {
-            Map<String, Object> vars = maskVars(asMapOrNull(outVars), eventConfiguration.outVarsBlacklist());
+            Map<String, Object> output = asMapOrNull(outVars);
+            output = processSensitiveData(output);
+            Map<String, Object> vars = maskVars(output, eventConfiguration.outVarsBlacklist());
             if (eventConfiguration.truncateOutVars()) {
                 vars = ObjectTruncater.truncateMap(vars, eventConfiguration.truncateMaxStringLength(), eventConfiguration.truncateMaxArrayLength(), eventConfiguration.truncateMaxDepth());
             }
@@ -86,7 +91,9 @@ public void onEvent(TaskCallEvent event) {
 
         Object metaVars = event.meta();
         if (metaVars != null && eventConfiguration.recordTaskMeta()) {
-            Map<String, Object> meta = maskVars(asMapOrNull(metaVars), eventConfiguration.metaBlacklist());
+            Map<String, Object> rawMeta = asMapOrNull(metaVars);
+            Map<String, Object> meta = processSensitiveData(rawMeta);
+            meta = maskVars(meta, eventConfiguration.metaBlacklist());
             if (eventConfiguration.truncateMeta()) {
                 meta = ObjectTruncater.truncateMap(meta, eventConfiguration.truncateMaxStringLength(), eventConfiguration.truncateMaxArrayLength(), eventConfiguration.truncateMaxDepth());
             }
@@ -170,6 +177,34 @@ static Map<String, Object> maskVars(Map<String, Object> vars, Collection<String>
         return result;
     }
 
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    static <T> T processSensitiveData(T v) {
+        Set<String> sensitiveStrings = SensitiveDataHolder.getInstance().get();
+        if (sensitiveStrings.isEmpty()) {
+            return v;
+        }
+
+        if (v instanceof String s) {
+            for (String sensitiveString : sensitiveStrings) {
+                s = s.replace(sensitiveString, MASK);
+            }
+            return (T) s;
+        } else if (v instanceof List<?> l) {
+            List<Object> result = new ArrayList<>(l.size());
+            for (Object vv : l) {
+                vv = processSensitiveData(vv);
+                result.add(vv);
+            }
+            return (T) result;
+        } else if (v instanceof Map m) {
+            Map<String, Object> result = new HashMap<>(m);
+            result.replaceAll((k, vv) -> processSensitiveData(vv));
+            return (T) result;
+        }
+
+        return v;
+    }
+
     @SuppressWarnings("unchecked")
     private static Map<String, Object> ensureModifiable(Map<String, Object> m, int depth, String[] path) {
         if (depth == 0) {
@@ -217,7 +252,7 @@ private static Map<String, Object> convertInput(List<Object> input) {
         return result;
     }
 
-    private static List<Object> hideSensitiveData(List<Object> input, List<List<Annotation>> annotations) {
+    private static List<Object> processSensitiveDataAnnotations(List<Object> input, List<List<Annotation>> annotations) {
         if (annotations.isEmpty()) {
             return input;
         }

runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/LoggedException.java

@@ -40,4 +40,9 @@ public String toString() {
     public String getMessage() {
         return getCause().getMessage();
     }
+
+    @Override
+    public StackTraceElement[] getStackTrace() {
+        return new StackTraceElement[0];
+    }
 }