Workflow bulk deletion API and action module options (#110)

* Initial commit to add bulk workflow deletion * Added an API endpoint for bulk deletion * Add archiveworkflow param * Added bulk delete action to BulkActionModule * Added archive option to bulk delete * Addition of terminateRemove() Functions (#9) * Added an API endpoint for bulk deletion * Fixed WorkflowBulkServiceTest, added test cases for deleteWorkflow an… (#10) * Fixed WorkflowBulkServiceTest, added test cases for deleteWorkflow and terminateRemove, and added WorkflowResourceTest test for single terminateRemove * pass spotlessjava --------- Co-authored-by: JeffP <[email protected]> --------- Co-authored-by: jeffp1 <[email protected]> Co-authored-by: Ayush Thengne <[email protected]> Co-authored-by: Jeff P <[email protected]> Co-authored-by: Doe1111 <[email protected]>
conductor-oss · Apr 3, 2024 · ff040fe · ff040fe
1 parent 7e7b88b
commit ff040fe
Show file tree

Hide file tree

Showing 10 changed files with 236 additions and 4 deletions.
diff --git a/core/src/main/java/com/netflix/conductor/service/WorkflowBulkService.java b/core/src/main/java/com/netflix/conductor/service/WorkflowBulkService.java
@@ -67,4 +67,23 @@ BulkResponse terminate(
                                     "Cannot process more than {max} workflows. Please use multiple requests.")
                     List<String> workflowIds,
             String reason);
+
+    BulkResponse deleteWorkflow(
+            @NotEmpty(message = "WorkflowIds list cannot be null.")
+                    @Size(
+                            max = MAX_REQUEST_ITEMS,
+                            message =
+                                    "Cannot process more than {max} workflows. Please use multiple requests.")
+                    List<String> workflowIds,
+            boolean archiveWorkflow);
+
+    BulkResponse terminateRemove(
+            @NotEmpty(message = "WorkflowIds list cannot be null.")
+                    @Size(
+                            max = MAX_REQUEST_ITEMS,
+                            message =
+                                    "Cannot process more than {max} workflows. Please use multiple requests.")
+                    List<String> workflowIds,
+            String reason,
+            boolean archiveWorkflow);
 }
diff --git a/core/src/main/java/com/netflix/conductor/service/WorkflowBulkServiceImpl.java b/core/src/main/java/com/netflix/conductor/service/WorkflowBulkServiceImpl.java
@@ -30,9 +30,12 @@ public class WorkflowBulkServiceImpl implements WorkflowBulkService {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(WorkflowBulkService.class);
     private final WorkflowExecutor workflowExecutor;
+    private final WorkflowService workflowService;
 
-    public WorkflowBulkServiceImpl(WorkflowExecutor workflowExecutor) {
+    public WorkflowBulkServiceImpl(
+            WorkflowExecutor workflowExecutor, WorkflowService workflowService) {
         this.workflowExecutor = workflowExecutor;
+        this.workflowService = workflowService;
     }
 
     /**
@@ -164,4 +167,70 @@ public BulkResponse terminate(List<String> workflowIds, String reason) {
         }
         return bulkResponse;
     }
+
+    /**
+     * Removes a list of workflows from the system.
+     *
+     * @param workflowIds List of WorkflowIDs of the workflows you want to remove from system.
+     * @param archiveWorkflow Archives the workflow and associated tasks instead of removing them.
+     */
+    public BulkResponse deleteWorkflow(List<String> workflowIds, boolean archiveWorkflow) {
+        BulkResponse bulkResponse = new BulkResponse();
+        for (String workflowId : workflowIds) {
+            try {
+                workflowService.deleteWorkflow(
+                        workflowId,
+                        archiveWorkflow); // TODO: change this to method that cancels then deletes
+                bulkResponse.appendSuccessResponse(workflowId);
+            } catch (Exception e) {
+                LOGGER.error(
+                        "bulk delete exception, workflowId {}, message: {} ",
+                        workflowId,
+                        e.getMessage(),
+                        e);
+                bulkResponse.appendFailedResponse(workflowId, e.getMessage());
+            }
+        }
+        return bulkResponse;
+    }
+
+    /**
+     * Terminates execution for workflows in a list, then removes each workflow.
+     *
+     * @param workflowIds List of workflow IDs to terminate and delete.
+     * @param reason Reason for terminating the workflow.
+     * @param archiveWorkflow Archives the workflow and associated tasks instead of removing them.
+     * @return bulk response object containing a list of succeeded workflows and a list of failed
+     *     ones with errors
+     */
+    public BulkResponse terminateRemove(
+            List<String> workflowIds, String reason, boolean archiveWorkflow) {
+        BulkResponse bulkResponse = new BulkResponse();
+        for (String workflowId : workflowIds) {
+            try {
+                workflowExecutor.terminateWorkflow(workflowId, reason);
+                bulkResponse.appendSuccessResponse(workflowId);
+            } catch (Exception e) {
+                LOGGER.error(
+                        "bulk terminate exception, workflowId {}, message: {} ",
+                        workflowId,
+                        e.getMessage(),
+                        e);
+                bulkResponse.appendFailedResponse(workflowId, e.getMessage());
+            }
+
+            try {
+                workflowService.deleteWorkflow(workflowId, archiveWorkflow);
+                bulkResponse.appendSuccessResponse(workflowId);
+            } catch (Exception e) {
+                LOGGER.error(
+                        "bulk delete exception, workflowId {}, message: {} ",
+                        workflowId,
+                        e.getMessage(),
+                        e);
+                bulkResponse.appendFailedResponse(workflowId, e.getMessage());
+            }
+        }
+        return bulkResponse;
+    }
 }
diff --git a/core/src/main/java/com/netflix/conductor/service/WorkflowService.java b/core/src/main/java/com/netflix/conductor/service/WorkflowService.java
@@ -242,6 +242,19 @@ void terminateWorkflow(
             @NotEmpty(message = "WorkflowId cannot be null or empty.") String workflowId,
             String reason);
 
+    /**
+     * Terminate workflow execution, and then remove it from the system. Acts as terminate and
+     * remove combined.
+     *
+     * @param workflowId WorkflowId of the workflow
+     * @param reason Reason for terminating the workflow.
+     * @param archiveWorkflow Archives the workflow and associated tasks instead of removing them.
+     */
+    void terminateRemove(
+            @NotEmpty(message = "WorkflowId cannot be null or empty.") String workflowId,
+            String reason,
+            boolean archiveWorkflow);
+
     /**
      * Search for workflows based on payload and given parameters. Use sort options as sort ASCor
      * DESC e.g. sort=name or sort=workflowId:DESC. If order is not specified, defaults to ASC.

diff --git a/core/src/main/java/com/netflix/conductor/service/WorkflowServiceImpl.java b/core/src/main/java/com/netflix/conductor/service/WorkflowServiceImpl.java
@@ -196,6 +196,19 @@ public void deleteWorkflow(String workflowId, boolean archiveWorkflow) {
         executionService.removeWorkflow(workflowId, archiveWorkflow);
     }
 
+    /**
+     * Terminate workflow execution, and then remove it from the system. Acts as terminate and
+     * remove combined.
+     *
+     * @param workflowId WorkflowId of the workflow
+     * @param reason Reason for terminating the workflow.
+     * @param archiveWorkflow Archives the workflow and associated tasks instead of removing them.
+     */
+    public void terminateRemove(String workflowId, String reason, boolean archiveWorkflow) {
+        workflowExecutor.terminateWorkflow(workflowId, reason);
+        executionService.removeWorkflow(workflowId, archiveWorkflow);
+    }
+
     /**
      * Retrieves all the running workflows.
      *

diff --git a/core/src/test/java/com/netflix/conductor/service/WorkflowBulkServiceTest.java b/core/src/test/java/com/netflix/conductor/service/WorkflowBulkServiceTest.java
@@ -50,8 +50,14 @@ WorkflowExecutor workflowExecutor() {
         }
 
         @Bean
-        public WorkflowBulkService workflowBulkService(WorkflowExecutor workflowExecutor) {
-            return new WorkflowBulkServiceImpl(workflowExecutor);
+        WorkflowService workflowService() {
+            return mock(WorkflowService.class);
+        }
+
+        @Bean
+        public WorkflowBulkService workflowBulkService(
+                WorkflowExecutor workflowExecutor, WorkflowService workflowService) {
+            return new WorkflowBulkServiceImpl(workflowExecutor, workflowService);
         }
     }
 
@@ -144,4 +150,28 @@ public void testTerminateNull() {
             throw ex;
         }
     }
+
+    @Test(expected = ConstraintViolationException.class)
+    public void testDeleteWorkflowNull() {
+        try {
+            workflowBulkService.deleteWorkflow(null, false);
+        } catch (ConstraintViolationException ex) {
+            assertEquals(1, ex.getConstraintViolations().size());
+            Set<String> messages = getConstraintViolationMessages(ex.getConstraintViolations());
+            assertTrue(messages.contains("WorkflowIds list cannot be null."));
+            throw ex;
+        }
+    }
+
+    @Test(expected = ConstraintViolationException.class)
+    public void testTerminateRemoveNull() {
+        try {
+            workflowBulkService.terminateRemove(null, null, false);
+        } catch (ConstraintViolationException ex) {
+            assertEquals(1, ex.getConstraintViolations().size());
+            Set<String> messages = getConstraintViolationMessages(ex.getConstraintViolations());
+            assertTrue(messages.contains("WorkflowIds list cannot be null."));
+            throw ex;
+        }
+    }
 }
diff --git a/rest/src/main/java/com/netflix/conductor/rest/controllers/WorkflowBulkResource.java b/rest/src/main/java/com/netflix/conductor/rest/controllers/WorkflowBulkResource.java
@@ -14,6 +14,7 @@
 
 import java.util.List;
 
+import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.PutMapping;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -111,4 +112,33 @@ public BulkResponse terminate(
             @RequestParam(value = "reason", required = false) String reason) {
         return workflowBulkService.terminate(workflowIds, reason);
     }
+
+    /**
+     * Delete the list of workflows.
+     *
+     * @param workflowIds - list of workflow Ids to be deleted
+     * @return bulk reponse object containing a list of successfully deleted workflows
+     */
+    @DeleteMapping("/remove")
+    public BulkResponse deleteWorkflow(
+            @RequestBody List<String> workflowIds,
+            @RequestParam(value = "archiveWorkflow", defaultValue = "true", required = false)
+                    boolean archiveWorkflow) {
+        return workflowBulkService.deleteWorkflow(workflowIds, archiveWorkflow);
+    }
+
+    /**
+     * Terminate then delete the list of workflows.
+     *
+     * @param workflowIds - list of workflow Ids to be deleted
+     * @return bulk response object containing a list of successfully deleted workflows
+     */
+    @DeleteMapping("/terminate-remove")
+    public BulkResponse terminateRemove(
+            @RequestBody List<String> workflowIds,
+            @RequestParam(value = "archiveWorkflow", defaultValue = "true", required = false)
+                    boolean archiveWorkflow,
+            @RequestParam(value = "reason", required = false) String reason) {
+        return workflowBulkService.terminateRemove(workflowIds, reason, archiveWorkflow);
+    }
 }
diff --git a/rest/src/main/java/com/netflix/conductor/rest/controllers/WorkflowResource.java b/rest/src/main/java/com/netflix/conductor/rest/controllers/WorkflowResource.java
@@ -207,6 +207,16 @@ public void terminate(
         workflowService.terminateWorkflow(workflowId, reason);
     }
 
+    @DeleteMapping("/{workflowId}/terminate-remove")
+    @Operation(summary = "Terminate workflow execution and remove the workflow from the system")
+    public void terminateRemove(
+            @PathVariable("workflowId") String workflowId,
+            @RequestParam(value = "reason", required = false) String reason,
+            @RequestParam(value = "archiveWorkflow", defaultValue = "true", required = false)
+                    boolean archiveWorkflow) {
+        workflowService.terminateRemove(workflowId, reason, archiveWorkflow);
+    }
+
     @Operation(
             summary = "Search for workflows based on payload and other parameters",
             description =

diff --git a/rest/src/test/java/com/netflix/conductor/rest/controllers/WorkflowResourceTest.java b/rest/src/test/java/com/netflix/conductor/rest/controllers/WorkflowResourceTest.java
@@ -210,6 +210,13 @@ public void testTerminate() {
         verify(mockWorkflowService, times(1)).terminateWorkflow(anyString(), anyString());
     }
 
+    @Test
+    public void testTerminateRemove() {
+        workflowResource.terminateRemove("w123", "test", false);
+        verify(mockWorkflowService, times(1))
+                .terminateRemove(anyString(), anyString(), anyBoolean());
+    }
+
     @Test
     public void testSearch() {
         workflowResource.search(0, 100, "asc", "*", "*");

diff --git a/ui/src/data/bulkactions.js b/ui/src/data/bulkactions.js
@@ -48,3 +48,20 @@ export const useBulkTerminateWithReasonAction = (callbacks) => {
     });
   }, callbacks);
 };
+
+export const useBulkDeleteAction = (callbacks) => {
+  const fetchContext = useFetchContext();
+
+  return useMutation((mutateParams) => {
+    const path = new Path("/workflow/bulk/remove");
+    path.search.append("archiveWorkflow", mutateParams.archiveWorkflow);
+
+    return fetchWithContext(path, fetchContext, {
+      method: "delete",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: _.get(mutateParams, "body"),
+    });
+  }, callbacks);
+};
diff --git a/ui/src/pages/executions/BulkActionModule.jsx b/ui/src/pages/executions/BulkActionModule.jsx
@@ -21,6 +21,7 @@ import {
   useBulkPauseAction,
   useBulkRetryAction,
   useBulkTerminateWithReasonAction,
+  useBulkDeleteAction,
 } from "../../data/bulkactions";
 
 const useStyles = makeStyles({
@@ -59,6 +60,8 @@ export default function BulkActionModule({ selectedRows }) {
     mutate: terminateWithReasonAction,
     isLoading: terminateWithReasonLoading,
   } = useBulkTerminateWithReasonAction({ onSuccess });
+  const { mutate: deleteAction, isLoading: deleteLoading } = 
+    useBulkDeleteAction({ onSuccess });
 
   const isLoading =
     pauseLoading ||
@@ -67,7 +70,8 @@ export default function BulkActionModule({ selectedRows }) {
     restartLatestLoading ||
     retryLoading ||
     terminateLoading ||
-    terminateWithReasonLoading;
+    terminateWithReasonLoading ||
+    deleteLoading;
 
   function onSuccess(data, variables, context) {
     const retval = {
@@ -133,6 +137,26 @@ export default function BulkActionModule({ selectedRows }) {
               }
             },
           },
+          {
+            label: "Archive",
+            handler: () => {
+              const archiveWorkflow = "true";
+              deleteAction({
+                body: JSON.stringify(selectedIds),
+                archiveWorkflow,
+              });
+            },
+          },
+          {
+            label: "Delete",
+            handler: () => {
+              const archiveWorkflow = "false";
+              deleteAction({
+                body: JSON.stringify(selectedIds),
+                archiveWorkflow,
+              });
+            },
+          },
         ]}
       >
         Bulk Action