Introduce restart workflow feature - backend

rhdhorchestrator · Jun 22, 2023 · 86592a3 · 86592a3
1 parent aa95135
commit 86592a3
Show file tree

Hide file tree

Showing 24 changed files with 1,472 additions and 9 deletions.
diff --git a/integration-tests/src/test/java/com/redhat/parodos/flows/SimpleRestartWorkFlowTest.java b/integration-tests/src/test/java/com/redhat/parodos/flows/SimpleRestartWorkFlowTest.java
diff --git a/parodos-model-api/src/main/java/com/redhat/parodos/workflow/task/BaseWorkFlowTask.java b/parodos-model-api/src/main/java/com/redhat/parodos/workflow/task/BaseWorkFlowTask.java
@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.redhat.parodos.workflow.exception.MissingParameterException;
 import com.redhat.parodos.workflow.task.log.WorkFlowTaskLogger;
@@ -110,6 +111,21 @@ public String getRequiredParameterValue(String parameterName) throws MissingPara
 				});
 	}
 
+	public boolean validateWorkflowParameters(WorkContext workContext) {
+		AtomicBoolean valid = new AtomicBoolean(true);
+		getWorkFlowTaskParameters().stream().filter(param -> !param.isOptional()).forEach(param -> {
+			try {
+				getRequiredParameterValue(param.getKey());
+			}
+			catch (MissingParameterException e) {
+				valid.set(false);
+				log.info("{} is missing from workContext for workflow task {}", param.getKey(), getName());
+			}
+		});
+
+		return valid.get();
+	}
+
 	/**
 	 * Gets an optional parameter. Returns the defaultValue if not found
 	 * @param parameterName

diff --git a/sdk-utils/src/main/java/com/redhat/parodos/sdkutils/RetryExecutorService.java b/sdk-utils/src/main/java/com/redhat/parodos/sdkutils/RetryExecutorService.java
@@ -32,6 +32,12 @@ public T submitWithRetry(Callable<T> task) {
 		// @formatter:on
 	}
 
+	public T submitWithRetry(Callable<T> task, long maxRetryTime) {
+		// @formatter:off
+		return submitWithRetry(task, () -> {}, () -> {}, maxRetryTime, RETRY_DELAY);
+		// @formatter:on
+	}
+
 	/**
 	 * Submit a task to the executor service, retrying on failure until the task
 	 * @param task The task to submit

diff --git a/sdk-utils/src/main/java/com/redhat/parodos/sdkutils/WorkFlowServiceUtils.java b/sdk-utils/src/main/java/com/redhat/parodos/sdkutils/WorkFlowServiceUtils.java
@@ -128,6 +128,11 @@ public static WorkFlowStatusResponseDTO waitWorkflowStatusAsync(WorkflowApi work
 		return waitWorkflowStatusAsync(workflowApi, workFlowExecutionId, StatusEnum.COMPLETED);
 	}
 
+	public static WorkFlowStatusResponseDTO waitWorkflowStatusAsync(WorkflowApi workflowApi, UUID workFlowExecutionId,
+			long maxRetryTimeInMs) {
+		return waitWorkflowStatusAsync(workflowApi, workFlowExecutionId, StatusEnum.COMPLETED, maxRetryTimeInMs);
+	}
+
 	/**
 	 * Invokes @see com.redhat.parodos.sdk.api.WorkflowApi#getStatusAsync(String,
 	 * ApiCallback<WorkFlowStatusResponseDTO>) and retries for 60 seconds.
@@ -158,6 +163,27 @@ public static WorkFlowStatusResponseDTO waitWorkflowStatusAsync(WorkflowApi work
 		return result;
 	}
 
+	public static WorkFlowStatusResponseDTO waitWorkflowStatusAsync(WorkflowApi workflowApi, UUID workFlowExecutionId,
+			StatusEnum expectedStatus, long maxRetryTimeInMs) {
+		WorkFlowStatusResponseDTO result;
+
+		try (var executorService = new RetryExecutorService<WorkFlowStatusResponseDTO>()) {
+			Callable<WorkFlowStatusResponseDTO> task = () -> {
+				WorkFlowStatusResponseDTO status = workflowApi.getStatus(workFlowExecutionId);
+				if (status.getStatus() != expectedStatus) {
+					throw new ApiException("Workflow status is not " + expectedStatus);
+				}
+				return status;
+			};
+
+			result = executorService.submitWithRetry(task, maxRetryTimeInMs);
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Workflow status is not " + expectedStatus, e);
+		}
+		return result;
+	}
+
 	/**
 	 * Finds a project with @see #projectName and @see #projectDescription
 	 * @param projects List to of project to analyze

diff --git a/...examples/src/main/java/com/redhat/parodos/examples/complex/task/AdGroupsWorkFlowTask.java b/...examples/src/main/java/com/redhat/parodos/examples/complex/task/AdGroupsWorkFlowTask.java
@@ -18,6 +18,11 @@ public class AdGroupsWorkFlowTask extends BaseInfrastructureWorkFlowTask {
 	@Override
 	public WorkReport execute(WorkContext workContext) {
 		log.info("AdGroupsWorkFlowTask");
+		if (!validateWorkflowParameters(workContext)) {
+			log.warn("Missing keys in context for workflowExecution {}, context: {}", getMainExecutionId(),
+					workContext);
+			return new DefaultWorkReport(WorkStatus.FAILED, workContext);
+		}
 		return new DefaultWorkReport(WorkStatus.COMPLETED, workContext);
 	}
 

diff --git a/...xamples/src/main/java/com/redhat/parodos/examples/complex/task/NamespaceWorkFlowTask.java b/...xamples/src/main/java/com/redhat/parodos/examples/complex/task/NamespaceWorkFlowTask.java
@@ -18,6 +18,11 @@ public class NamespaceWorkFlowTask extends BaseInfrastructureWorkFlowTask {
 	@Override
 	public WorkReport execute(WorkContext workContext) {
 		log.info("NamespaceWorkFlowTask");
+		if (!validateWorkflowParameters(workContext)) {
+			log.warn("Missing keys in context for workflowExecution {}, context: {}", getMainExecutionId(),
+					workContext);
+			return new DefaultWorkReport(WorkStatus.FAILED, workContext);
+		}
 		return new DefaultWorkReport(WorkStatus.COMPLETED, workContext);
 	}
 

diff --git a/.../src/main/java/com/redhat/parodos/examples/complex/task/SslCertificationWorkFlowTask.java b/.../src/main/java/com/redhat/parodos/examples/complex/task/SslCertificationWorkFlowTask.java
@@ -18,6 +18,11 @@ public class SslCertificationWorkFlowTask extends BaseInfrastructureWorkFlowTask
 	@Override
 	public WorkReport execute(WorkContext workContext) {
 		log.info("SslCertificationWorkFlowTask");
+		if (!validateWorkflowParameters(workContext)) {
+			log.warn("Missing keys in context for workflowExecution {}, context: {}", getMainExecutionId(),
+					workContext);
+			return new DefaultWorkReport(WorkStatus.FAILED, workContext);
+		}
 		return new DefaultWorkReport(WorkStatus.COMPLETED, workContext);
 	}
 

diff --git a/workflow-service-sdk/README.md b/workflow-service-sdk/README.md
@@ -125,6 +125,7 @@ Class | Method | HTTP request | Description
 *WorkflowApi* | [**getStatus**](docs/WorkflowApi.md#getStatus) | **GET** /api/v1/workflows/{workFlowExecutionId}/status | Returns a workflow status
 *WorkflowApi* | [**getStatusByProjectId**](docs/WorkflowApi.md#getStatusByProjectId) | **GET** /api/v1/workflows | Returns workflows by project id
 *WorkflowApi* | [**getWorkflowParameters**](docs/WorkflowApi.md#getWorkflowParameters) | **GET** /api/v1/workflows/{workFlowExecutionId}/context | Returns workflow context parameters
+*WorkflowApi* | [**restartWorkflow**](docs/WorkflowApi.md#restartWorkflow) | **POST** /api/v1/workflows/{workFlowExecutionId} | Restart a workflow execution with same parameters
 *WorkflowApi* | [**updateWorkFlowCheckerTaskStatus**](docs/WorkflowApi.md#updateWorkFlowCheckerTaskStatus) | **POST** /api/v1/workflows/{workFlowExecutionId}/checkers/{workFlowCheckerTaskName} | Updates a workflow checker task status
 *WorkflowDefinitionApi* | [**getWorkFlowDefinitionById**](docs/WorkflowDefinitionApi.md#getWorkFlowDefinitionById) | **GET** /api/v1/workflowdefinitions/{id} | Returns information about a workflow definition by id
 *WorkflowDefinitionApi* | [**getWorkFlowDefinitions**](docs/WorkflowDefinitionApi.md#getWorkFlowDefinitions) | **GET** /api/v1/workflowdefinitions | Returns a list of workflow definition

diff --git a/workflow-service-sdk/api/openapi.yaml b/workflow-service-sdk/api/openapi.yaml
@@ -641,6 +641,48 @@ paths:
       - Workflow
       x-content-type: application/json
       x-accepts: application/json
+  /api/v1/workflows/{workFlowExecutionId}:
+    post:
+      operationId: restartWorkflow
+      parameters:
+      - explode: false
+        in: path
+        name: workFlowExecutionId
+        required: true
+        schema:
+          format: uuid
+          type: string
+        style: simple
+      responses:
+        "202":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/WorkFlowExecutionResponseDTO'
+          description: Accepted
+        "400":
+          content:
+            '*/*':
+              schema:
+                $ref: '#/components/schemas/ErrorMessageDTO'
+          description: Bad Request
+        "401":
+          description: Unauthorized
+        "403":
+          description: Forbidden
+        "404":
+          content: {}
+          description: Not Found
+        "409":
+          content:
+            '*/*':
+              schema:
+                $ref: '#/components/schemas/ErrorMessageDTO'
+          description: Conflict
+      summary: Restart a workflow execution with same parameters
+      tags:
+      - Workflow
+      x-accepts: application/json
   /api/v1/workflows/{workFlowExecutionId}/checkers/{workFlowCheckerTaskName}:
     post:
       operationId: updateWorkFlowCheckerTaskStatus
@@ -1482,14 +1524,20 @@ components:
         workFlowName: workFlowName
         invokingExecutionId: 046b6c7f-0b8a-43b9-b35d-6489e6daee91
         works:
-        - arguments:
+        - works:
+          - null
+          - null
+          arguments:
           - value: value
             key: key
           - value: value
             key: key
           type: TASK
           workName: workName
-        - arguments:
+        - works:
+          - null
+          - null
+          arguments:
           - value: value
             key: key
           - value: value
@@ -1555,6 +1603,7 @@ components:
       type: object
     WorkFlowStatusResponseDTO:
       example:
+        originalExecutionId: 046b6c7f-0b8a-43b9-b35d-6489e6daee91
         workFlowName: workFlowName
         works:
         - alertMessage: alertMessage
@@ -1574,11 +1623,18 @@ components:
           type: TASK
           status: FAILED
         workFlowExecutionId: 046b6c7f-0b8a-43b9-b35d-6489e6daee91
+        restartedCount: 0
         message: message
         status: FAILED
       properties:
         message:
           type: string
+        originalExecutionId:
+          format: uuid
+          type: string
+        restartedCount:
+          format: int32
+          type: integer
         status:
           enum:
           - FAILED
@@ -1634,6 +1690,9 @@ components:
       type: object
     WorkRequestDTO:
       example:
+        works:
+        - null
+        - null
         arguments:
         - value: value
           key: key
@@ -1653,6 +1712,10 @@ components:
           type: string
         workName:
           type: string
+        works:
+          items:
+            $ref: '#/components/schemas/WorkRequestDTO'
+          type: array
       type: object
     WorkStatusResponseDTO:
       example:

diff --git a/workflow-service-sdk/docs/WorkFlowStatusResponseDTO.md b/workflow-service-sdk/docs/WorkFlowStatusResponseDTO.md
@@ -8,6 +8,8 @@
 | Name | Type | Description | Notes |
 |------------ | ------------- | ------------- | -------------|
 |**message** | **String** |  |  [optional] |
+|**originalExecutionId** | **UUID** |  |  [optional] |
+|**restartedCount** | **Integer** |  |  [optional] |
 |**status** | [**StatusEnum**](#StatusEnum) |  |  [optional] |
 |**workFlowExecutionId** | **UUID** |  |  [optional] |
 |**workFlowName** | **String** |  |  [optional] |

diff --git a/workflow-service-sdk/docs/WorkRequestDTO.md b/workflow-service-sdk/docs/WorkRequestDTO.md
@@ -10,6 +10,7 @@
 |**arguments** | [**List&lt;ArgumentRequestDTO&gt;**](ArgumentRequestDTO.md) |  |  [optional] |
 |**type** | [**TypeEnum**](#TypeEnum) |  |  [optional] |
 |**workName** | **String** |  |  [optional] |
+|**works** | [**List&lt;WorkRequestDTO&gt;**](WorkRequestDTO.md) |  |  [optional] |
 
 
 

diff --git a/workflow-service-sdk/docs/WorkflowApi.md b/workflow-service-sdk/docs/WorkflowApi.md
@@ -9,6 +9,7 @@ All URIs are relative to *http://localhost:8080*
 | [**getStatus**](WorkflowApi.md#getStatus) | **GET** /api/v1/workflows/{workFlowExecutionId}/status | Returns a workflow status |
 | [**getStatusByProjectId**](WorkflowApi.md#getStatusByProjectId) | **GET** /api/v1/workflows | Returns workflows by project id |
 | [**getWorkflowParameters**](WorkflowApi.md#getWorkflowParameters) | **GET** /api/v1/workflows/{workFlowExecutionId}/context | Returns workflow context parameters |
+| [**restartWorkflow**](WorkflowApi.md#restartWorkflow) | **POST** /api/v1/workflows/{workFlowExecutionId} | Restart a workflow execution with same parameters |
 | [**updateWorkFlowCheckerTaskStatus**](WorkflowApi.md#updateWorkFlowCheckerTaskStatus) | **POST** /api/v1/workflows/{workFlowExecutionId}/checkers/{workFlowCheckerTaskName} | Updates a workflow checker task status |
 
 
@@ -343,6 +344,71 @@ No authorization required
 | **404** | Not Found |  -  |
 | **409** | Conflict |  -  |
 
+<a name="restartWorkflow"></a>
+# **restartWorkflow**
+> WorkFlowExecutionResponseDTO restartWorkflow(workFlowExecutionId)
+
+Restart a workflow execution with same parameters
+
+### Example
+```java
+// Import classes:
+import com.redhat.parodos.sdk.invoker.ApiClient;
+import com.redhat.parodos.sdk.invoker.ApiException;
+import com.redhat.parodos.sdk.invoker.Configuration;
+import com.redhat.parodos.sdk.invoker.models.*;
+import com.redhat.parodos.sdk.api.WorkflowApi;
+
+public class Example {
+  public static void main(String[] args) {
+    ApiClient defaultClient = Configuration.getDefaultApiClient();
+    defaultClient.setBasePath("http://localhost:8080");
+
+    WorkflowApi apiInstance = new WorkflowApi(defaultClient);
+    UUID workFlowExecutionId = UUID.randomUUID(); // UUID | 
+    try {
+      WorkFlowExecutionResponseDTO result = apiInstance.restartWorkflow(workFlowExecutionId);
+      System.out.println(result);
+    } catch (ApiException e) {
+      System.err.println("Exception when calling WorkflowApi#restartWorkflow");
+      System.err.println("Status code: " + e.getCode());
+      System.err.println("Reason: " + e.getResponseBody());
+      System.err.println("Response headers: " + e.getResponseHeaders());
+      e.printStackTrace();
+    }
+  }
+}
+```
+
+### Parameters
+
+| Name | Type | Description  | Notes |
+|------------- | ------------- | ------------- | -------------|
+| **workFlowExecutionId** | **UUID**|  | |
+
+### Return type
+
+[**WorkFlowExecutionResponseDTO**](WorkFlowExecutionResponseDTO.md)
+
+### Authorization
+
+No authorization required
+
+### HTTP request headers
+
+ - **Content-Type**: Not defined
+ - **Accept**: application/json, */*
+
+### HTTP response details
+| Status code | Description | Response headers |
+|-------------|-------------|------------------|
+| **202** | Accepted |  -  |
+| **400** | Bad Request |  -  |
+| **401** | Unauthorized |  -  |
+| **403** | Forbidden |  -  |
+| **404** | Not Found |  -  |
+| **409** | Conflict |  -  |
+
 <a name="updateWorkFlowCheckerTaskStatus"></a>
 # **updateWorkFlowCheckerTaskStatus**
 > String updateWorkFlowCheckerTaskStatus(workFlowExecutionId, workFlowCheckerTaskName, workFlowCheckerTaskRequestDTO)