Endpoint to get latest step instance views across runs for a given workflow instance

maestro-engine/src/main/java/com/netflix/maestro/engine/dao/MaestroStepInstanceDao.java

@@ -184,6 +184,12 @@ public class MaestroStepInstanceDao extends AbstractDatabaseDao {
           + GET_STEP_FIELD_QUERY_FROM
           + "AND workflow_run_id=?) SELECT * FROM inner_ranked WHERE rank=1";
 
+  private static final String GET_ALL_STEP_INSTANCE_VIEWS_QUERY =
+      INNER_RANK_QUERY_ALL_FIELD_WITH
+          + ", ROW_NUMBER() OVER (PARTITION BY step_id ORDER BY workflow_run_id DESC, step_attempt_id DESC) AS rank"
+          + GET_STEP_FIELD_QUERY_FROM
+          + ") SELECT * FROM inner_ranked WHERE rank=1";
+
   private final MaestroQueueSystem queueSystem;
 
   /**
@@ -1229,6 +1235,37 @@ public Long getNextUniqueId() {
         "Failed to get the next unique id");
   }
 
+  /**
+   * Get the latest step attempts of all the steps across all runs for a given workflow instance
+   * (workflow id, instance id).
+   *
+   * @param workflowId workflow id
+   * @param workflowInstanceId workflow instance id
+   * @return latest step attempts of all the steps across all runs
+   */
+  public List<StepInstance> getAllStepInstanceViews(String workflowId, long workflowInstanceId) {
+    return withMetricLogError(
+        () ->
+            withRetryableQuery(
+                GET_ALL_STEP_INSTANCE_VIEWS_QUERY,
+                stmt -> {
+                  int idx = 0;
+                  stmt.setString(++idx, workflowId);
+                  stmt.setLong(++idx, workflowInstanceId);
+                },
+                result -> {
+                  List<StepInstance> instances = new ArrayList<>();
+                  while (result.next()) {
+                    instances.add(maestroStepFromResult(result));
+                  }
+                  return instances;
+                }),
+        "getAllStepInstanceViews",
+        "Failed to get latest step attempts across all runs for workflow instance [{}][{}]",
+        workflowId,
+        workflowInstanceId);
+  }
+
   /**
    * Get the latest step attempts of all the steps for a given workflow instance run (workflow id,
    * instance id, run id).

maestro-engine/src/test/java/com/netflix/maestro/engine/dao/MaestroStepInstanceDaoTest.java

@@ -54,6 +54,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -321,6 +322,37 @@ public void testGetAllStepInstances() {
     Assertions.assertThat(instance).usingRecursiveComparison().isEqualTo(si);
   }
 
+  @Test
+  public void testGetAllStepInstanceViews() throws Exception {
+    // run 1: job1 (RUNNING) already seeded by setUp; add job2 (RUNNING) to the same run
+    StepInstance job2Run1 = loadObject(TEST_STEP_INSTANCE, StepInstance.class);
+    job2Run1.setStepId("job2");
+    job2Run1.setStepInstanceId(2);
+    stepDao.insertOrUpsertStepInstance(job2Run1, false, null);
+
+    // run 2 (restart from failure): only job1 ran again — job2 succeeded in run 1 and was skipped
+    StepInstance job1Run2 =
+        loadObject("fixtures/instances/sample-step-instance-finishing.json", StepInstance.class);
+    stepDao.insertOrUpsertStepInstance(job1Run2, false, null);
+
+    // should return one entry per step: job1 from run 2, job2 from run 1
+    List<StepInstance> instances = stepDao.getAllStepInstanceViews(TEST_WORKFLOW_ID, 1);
+    instances.sort(Comparator.comparingLong(StepInstance::getStepInstanceId));
+    assertEquals(2, instances.size());
+
+    StepInstance job1Result = instances.get(0);
+    assertEquals("job1", job1Result.getStepId());
+    assertEquals(2, job1Result.getWorkflowRunId());
+    assertEquals(StepInstance.Status.FINISHING, job1Result.getRuntimeState().getStatus());
+    assertEquals("ff4ccce2-0fda-4882-9cd8-12ff90cb5f02", job1Result.getStepUuid());
+
+    StepInstance job2Result = instances.get(1);
+    assertEquals("job2", job2Result.getStepId());
+    assertEquals(1, job2Result.getWorkflowRunId());
+    assertEquals(StepInstance.Status.RUNNING, job2Result.getRuntimeState().getStatus());
+    assertEquals("ff4ccce2-0fda-4882-9cd8-12ff90cb5f06", job2Result.getStepUuid());
+  }
+
   @Test
   public void testGetStepInstances() {
     List<StepInstance> instances = stepDao.getStepInstances(TEST_WORKFLOW_ID, 1, 1, "job1");

maestro-server/src/main/java/com/netflix/maestro/server/controllers/StepInstanceController.java

@@ -32,6 +32,7 @@
     value = "/api/v3/workflows",
     produces = MediaType.APPLICATION_JSON_VALUE,
     consumes = MediaType.APPLICATION_JSON_VALUE)
+@SuppressWarnings("PMD.AvoidDuplicateLiterals")
 public class StepInstanceController {
 
   private final MaestroStepInstanceDao stepInstanceDao;
@@ -88,6 +89,20 @@ public StepInstance getStepInstance(
     return instance;
   }
 
+  @GetMapping(
+      value = "/{workflowId}/instances/{workflowInstanceId}/steps",
+      consumes = MediaType.ALL_VALUE)
+  @Operation(
+      summary = "Get the most recent step instance across all runs for a given workflow instance")
+  public List<StepInstance> getAllStepInstanceViews(
+      @Valid @NotNull @PathVariable("workflowId") String workflowId,
+      @PathVariable("workflowInstanceId") long workflowInstanceId) {
+    List<StepInstance> instances =
+        stepInstanceDao.getAllStepInstanceViews(workflowId, workflowInstanceId);
+    instances.sort(Comparator.comparingLong(StepInstance::getStepInstanceId));
+    return instances;
+  }
+
   @GetMapping(
       value = "/{workflowId}/instances/{workflowInstanceId}/runs/{workflowRunId}/steps",
       consumes = MediaType.ALL_VALUE)

maestro-server/src/test/java/com/netflix/maestro/server/controllers/StepInstanceControllerTest.java

@@ -127,4 +127,19 @@ public void testGetStepInstanceViews() {
     assertEquals(1, ret.size());
     assertEquals(instance2, ret.get(0));
   }
+
+  @Test
+  public void testGetAllStepInstanceViews() {
+    StepInstance instance1 = mock(StepInstance.class);
+    StepInstance instance2 = mock(StepInstance.class);
+    when(instance1.getStepInstanceId()).thenReturn(2L);
+    when(instance2.getStepInstanceId()).thenReturn(1L);
+    when(mockInstanceDao.getAllStepInstanceViews("test-workflow", 1))
+        .thenReturn(Arrays.asList(instance1, instance2));
+    List<StepInstance> ret = stepInstanceController.getAllStepInstanceViews("test-workflow", 1);
+    verify(mockInstanceDao, times(1)).getAllStepInstanceViews("test-workflow", 1);
+    assertEquals(2, ret.size());
+    assertEquals(instance2, ret.get(0));
+    assertEquals(instance1, ret.get(1));
+  }
 }