Add two SEL functions to support getting a specific param from a subw…

…orkflow and getting a step's end time. (#71)

maestro-common/src/main/java/com/netflix/maestro/models/Constants.java

@@ -262,6 +262,9 @@ public static WorkflowVersion of(String version) {
   /** Step instance status param name key used in SEL to retrieve the step status. */
   public static final String STEP_STATUS_PARAM = "MAESTRO_STEP_STATUS";
 
+  /** Step instance end time param name key used in SEL to retrieve the step end time. */
+  public static final String STEP_END_TIME_PARAM = "MAESTRO_STEP_END_TIME";
+
   /**
    * Match all instances. Special value to denote a breakpoint which is set to match all wf
    * instances.

maestro-common/src/testFixtures/resources/fixtures/execution/sample-step-runtime-summary-1.json

@@ -4,6 +4,7 @@
   "step_instance_id": 123,
   "step_attempt_id": 2,
   "step_instance_uuid": "bar",
+  "type": "subworkflow",
   "step_run_params": {
     "foo": {
       "value": "bar",
@@ -58,20 +59,6 @@
         "total_step_count": 1
       },
       "type": "SUBWORKFLOW"
-    },
-    "maestro_foreach": {
-      "foreach_workflow_id": "inline-wf",
-      "foreach_identity": "foo",
-      "total_loop_count": 10,
-      "next_loop_index": 0,
-      "foreach_overview": {
-        "checkpoint": 6,
-        "stats": {
-          "CREATED": 5,
-          "SUCCEEDED": 1
-        }
-      },
-      "type": "FOREACH"
     }
   },
   "dependencies": {

maestro-common/src/testFixtures/resources/fixtures/execution/sample-step-runtime-summary-3.json

@@ -0,0 +1,104 @@
+{
+  "step_id": "foo",
+  "step_name": "step1",
+  "step_instance_id": 123,
+  "step_attempt_id": 2,
+  "step_instance_uuid": "bar",
+  "type": "foreach",
+  "step_run_params": {
+    "foo": {
+      "value": "bar",
+      "type": "STRING",
+      "mode": "mutable"
+    }
+  },
+  "params": {
+    "param1": {
+      "value": "foo",
+      "type": "STRING",
+      "evaluated_result": "foo",
+      "evaluated_time": 1608171805392
+    }
+  },
+  "synced": true,
+  "runtime_state": {
+    "status": "SUCCEEDED",
+    "end_time": 1608171805401,
+    "modify_time": 1608171805401
+  },
+  "timeline": [
+    {
+      "timestamp": 1609272999666,
+      "type": "LOG",
+      "level": "INFO",
+      "message": "hello"
+    }
+  ],
+  "pending_records": [
+    {
+      "event_time": 1608171805401,
+      "new_status": "SUCCEEDED",
+      "old_status": "NOT_CREATED"
+    }
+  ],
+  "artifacts": {
+    "artifact1": {
+      "value": 1,
+      "foo": "bar"
+    },
+    "maestro_foreach": {
+      "foreach_workflow_id": "inline-wf",
+      "foreach_identity": "foo",
+      "total_loop_count": 10,
+      "next_loop_index": 0,
+      "foreach_overview": {
+        "checkpoint": 6,
+        "stats": {
+          "CREATED": 5,
+          "SUCCEEDED": 1
+        }
+      },
+      "type": "FOREACH"
+    }
+  },
+  "dependencies": {
+    "SIGNAL": {
+      "type": "SIGNAL",
+      "statuses": [
+        {
+          "params": {
+            "value": {
+              "name": {
+                "value": "signal/a/1",
+                "type": "STRING"
+              },
+              "_step_dependency_sub_type": {
+                "value": "input_signal",
+                "type": "STRING"
+              },
+              "foo": {
+                "parameter": {
+                  "value": "bar",
+                  "type": "STRING"
+                },
+                "operator": "=",
+                "type": "SIGNAL"
+              }
+            },
+            "type": "MAP",
+            "evaluated_result": {
+              "name": "signal/a/1",
+              "foo": "bar"
+            },
+            "evaluated_time": 1617728847801
+          },
+          "signal_reference": {
+            "signal_instance_id": "fake_signal_id",
+            "timestamp": 1617728849086
+          },
+          "status": "MATCHED"
+        }
+      ]
+    }
+  }
+}

maestro-engine/src/main/java/com/netflix/maestro/engine/eval/MaestroParamExtension.java

@@ -18,14 +18,17 @@
 import com.netflix.maestro.engine.utils.StepHelper;
 import com.netflix.maestro.engine.validations.DryRunValidator;
 import com.netflix.maestro.exceptions.MaestroInternalError;
+import com.netflix.maestro.exceptions.MaestroInvalidExpressionException;
 import com.netflix.maestro.exceptions.MaestroValidationException;
 import com.netflix.maestro.models.Constants;
 import com.netflix.maestro.models.artifact.Artifact;
 import com.netflix.maestro.models.artifact.ForeachArtifact;
+import com.netflix.maestro.models.artifact.SubworkflowArtifact;
 import com.netflix.maestro.models.definition.StepType;
 import com.netflix.maestro.models.initiator.Initiator;
 import com.netflix.maestro.models.initiator.ManualInitiator;
 import com.netflix.maestro.models.initiator.SignalInitiator;
+import com.netflix.maestro.models.instance.StepInstance;
 import com.netflix.maestro.models.parameter.ParamType;
 import com.netflix.maestro.models.parameter.Parameter;
 import com.netflix.maestro.utils.Checks;
@@ -72,6 +75,9 @@ public class MaestroParamExtension extends AbstractParamExtension {
   /** Function to get the signal metadata for return default value if not found. */
   static final String GET_FROM_SIGNAL_OR_DEFAULT = "getFromSignalOrDefault";
 
+  /** Function to get the data from the subworkflow instance. */
+  static final String GET_FROM_SUBWORKFLOW = "getFromSubworkflow";
+
   /** Function to get the foreach metadata. */
   static final String GET_FROM_FOREACH = "getFromForeach";
 
@@ -132,6 +138,8 @@ protected Object callWithThreeArgs(String methodName, String arg1, String arg2,
         throw new IllegalStateException(
             methodName + " must return an array instead of " + res.getClass());
       }
+    } else if (GET_FROM_SUBWORKFLOW.equals(methodName)) {
+      return getFromSubworkflow(arg1, arg2, arg3);
     } else if (GET_FROM_SIGNAL_OR_DEFAULT.equals(methodName)) {
       return getFromSignalOrDefault(arg1, arg2, arg3);
     }
@@ -167,6 +175,12 @@ private Object fromStep(String stepId, String paramName) {
     StepRuntimeSummary runtimeSummary = validateAndGet(stepId);
     if (Constants.STEP_STATUS_PARAM.equals(paramName)) {
       return runtimeSummary.getRuntimeState().getStatus().name();
+    } else if (Constants.STEP_END_TIME_PARAM.equals(paramName)) {
+      return Checks.notNull(
+          runtimeSummary.getRuntimeState().getEndTime(),
+          "ERROR: step [%s]'s [%s] is not set yet.",
+          stepId,
+          paramName);
     }
     Parameter stepParam =
         Checks.notNull(
@@ -365,6 +379,52 @@ private Object fromForeach(String foreachStepId, String stepId, String paramName
     }
   }
 
+  Object getFromSubworkflow(String subworkflowStepId, String stepId, String paramName) {
+    try {
+      return executor
+          .submit(() -> fromSubworkflow(subworkflowStepId, stepId, paramName))
+          .get(TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS);
+    } catch (Exception e) {
+      throw new MaestroInternalError(
+          e,
+          "getFromSubworkflow throws an exception for subworkflowStepId=[%s], stepId=[%s], paramName=[%s]",
+          subworkflowStepId,
+          stepId,
+          paramName);
+    }
+  }
+
+  private Object fromSubworkflow(String subworkflowStepId, String stepId, String paramName) {
+    StepRuntimeSummary runtimeSummary = validateAndGet(subworkflowStepId);
+    Checks.checkTrue(
+        runtimeSummary.getType() == StepType.SUBWORKFLOW,
+        "step [%s] is a type of [%s] instead of subworkflow step, cannot call getFromSubworkflow",
+        subworkflowStepId,
+        runtimeSummary.getType());
+
+    SubworkflowArtifact artifact =
+        Checks.notNull(
+                runtimeSummary.getArtifacts().get(Artifact.Type.SUBWORKFLOW.key()),
+                "Cannot load param [%s] of step [%s] from subworkflow [%s] as it is not initialized",
+                paramName,
+                stepId,
+                subworkflowStepId)
+            .asSubworkflow();
+
+    StepInstance stepInstance =
+        stepInstanceDao.getStepInstance(
+            artifact.getSubworkflowId(),
+            artifact.getSubworkflowInstanceId(),
+            artifact.getSubworkflowRunId(),
+            stepId,
+            Constants.LATEST_INSTANCE_RUN);
+
+    if (stepInstance.getParams() == null || !stepInstance.getParams().containsKey(paramName)) {
+      throw new MaestroInvalidExpressionException("Cannot find the param name: [%s]", paramName);
+    }
+    return stepInstance.getParams().get(paramName).getEvaluatedResult();
+  }
+
   Long nextUniqueId() {
     try {
       Thread.sleep(ThreadLocalRandom.current().nextInt(RANDOM_JITTER_DELAY));

maestro-engine/src/test/java/com/netflix/maestro/engine/eval/MaestroParamExtensionTest.java

@@ -15,6 +15,7 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.when;
 
 import com.netflix.maestro.AssertHelper;
@@ -32,6 +33,7 @@
 import com.netflix.maestro.models.initiator.ManualInitiator;
 import com.netflix.maestro.models.initiator.SignalInitiator;
 import com.netflix.maestro.models.initiator.SubworkflowInitiator;
+import com.netflix.maestro.models.instance.StepInstance;
 import com.netflix.maestro.models.parameter.MapParameter;
 import com.netflix.maestro.models.parameter.ParamType;
 import com.netflix.maestro.models.parameter.Parameter;
@@ -51,6 +53,10 @@
 public class MaestroParamExtensionTest extends MaestroEngineBaseTest {
   private static final String TEST_STEP_RUNTIME_SUMMARY =
       "fixtures/execution/sample-step-runtime-summary-2.json";
+  private static final String TEST_SUBWORKFLOW_STEP_RUNTIME_SUMMARY =
+      "fixtures/execution/sample-step-runtime-summary-1.json";
+  private static final String TEST_STEP_INSTANCE =
+      "fixtures/instances/sample-step-instance-succeeded.json";
 
   @Mock MaestroStepInstanceDao stepInstanceDao;
   @Mock InstanceWrapper instanceWrapper;
@@ -78,6 +84,8 @@ public void testGetFromStep() throws Exception {
         .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
     assertEquals("foo", paramExtension.getFromStep("step1", "param1"));
     assertEquals("SUCCEEDED", paramExtension.getFromStep("step1", Constants.STEP_STATUS_PARAM));
+    assertEquals(
+        1608171805401L, paramExtension.getFromStep("step1", Constants.STEP_END_TIME_PARAM));
   }
 
   @Test
@@ -134,6 +142,19 @@ public void testGetFromForeach() throws Exception {
     assertArrayEquals(new long[] {12, 0, 0, 0, 0, 0}, res);
   }
 
+  @Test
+  public void testGetFromSubworkflow() throws Exception {
+    StepRuntimeSummary summary =
+        loadObject(TEST_SUBWORKFLOW_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
+    when(allStepOutputData.get("foo"))
+        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
+    StepInstance stepInSubworkflow = loadObject(TEST_STEP_INSTANCE, StepInstance.class);
+    when(stepInstanceDao.getStepInstance(any(), anyLong(), anyLong(), any(), any()))
+        .thenReturn(stepInSubworkflow);
+    long res = (Long) paramExtension.getFromSubworkflow("foo", "job1", "sleep_seconds");
+    assertEquals(15, res);
+  }
+
   @Test
   public void testInvalidGetFromStep() throws Exception {
     StepRuntimeSummary summary = loadObject(TEST_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
@@ -264,6 +285,36 @@ public void testInvalidGetFromForeach() throws Exception {
         () -> paramExtension.getFromForeach("foreach-job", "job1", "sleep_seconds"));
   }
 
+  @Test
+  public void testInvalidGetFromSubworkflow() throws Exception {
+    AssertHelper.assertThrows(
+        "Cannot find the referenced step id",
+        MaestroInternalError.class,
+        "getFromSubworkflow throws an exception",
+        () -> paramExtension.getFromSubworkflow("non-existing-job", "job1", "sleep_seconds"));
+
+    StepRuntimeSummary summary = loadObject(TEST_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
+    when(allStepOutputData.get("foo"))
+        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
+    AssertHelper.assertThrows(
+        "step type is not subworkflow",
+        MaestroInternalError.class,
+        "getFromSubworkflow throws an exception",
+        () -> paramExtension.getFromSubworkflow("foo", "job1", "sleep_seconds"));
+
+    summary = loadObject(TEST_SUBWORKFLOW_STEP_RUNTIME_SUMMARY, StepRuntimeSummary.class);
+    when(allStepOutputData.get("foo"))
+        .thenReturn(Collections.singletonMap("maestro_step_runtime_summary", summary));
+    StepInstance stepInSubworkflow = loadObject(TEST_STEP_INSTANCE, StepInstance.class);
+    when(stepInstanceDao.getStepInstance(any(), anyLong(), anyLong(), any(), any()))
+        .thenReturn(stepInSubworkflow);
+    AssertHelper.assertThrows(
+        "param name does not exist",
+        MaestroInternalError.class,
+        "getFromSubworkflow throws an exception",
+        () -> paramExtension.getFromSubworkflow("foo", "job1", "not-existing"));
+  }
+
   @Test
   public void testNextUniqueId() {
     Long expected = 750762533885116445L;

maestro-engine/src/test/java/com/netflix/maestro/engine/execution/StepRuntimeSummaryTest.java

@@ -211,7 +211,7 @@ public void testMergeSubworkflowArtifact() throws Exception {
   public void testMergeForeachArtifact() throws Exception {
     StepRuntimeSummary summary =
         loadObject(
-            "fixtures/execution/sample-step-runtime-summary-1.json", StepRuntimeSummary.class);
+            "fixtures/execution/sample-step-runtime-summary-3.json", StepRuntimeSummary.class);
     ForeachArtifact artifact = summary.getArtifacts().get(Artifact.Type.FOREACH.key()).asForeach();
     assertEquals("inline-wf", artifact.getForeachWorkflowId());
     assertEquals("foo", artifact.getForeachIdentity());
@@ -427,6 +427,10 @@ public void testNoChangeMerge() throws Exception {
     summary.mergeRuntimeUpdate(null, artifacts);
     assertTrue(summary.isSynced());
 
+    summary =
+        loadObject(
+            "fixtures/execution/sample-step-runtime-summary-3.json", StepRuntimeSummary.class);
+    artifacts.clear();
     ForeachArtifact artifact3 = new ForeachArtifact();
     artifact3.setForeachWorkflowId("inline-wf");
     artifact3.setForeachIdentity("foo");