replace abort mutation with call to delete in workflow pod (#304)

FLPATH-1676
https://issues.redhat.com/browse/FLPATH-1676

Signed-off-by: Yaron Dayagi <[email protected]>

Author: ydayagi

workspaces/orchestrator/.changeset/brown-rules-shop.md

@@ -0,0 +1,5 @@
+---
+'@red-hat-developer-hub/backstage-plugin-orchestrator-backend': patch
+---
+
+replace abort mutation with call to delete

workspaces/orchestrator/plugins/orchestrator-backend/src/service/DataIndexService.ts

@@ -128,31 +128,6 @@ export class DataIndexService {
     return pairs;
   }
 
-  public async abortWorkflowInstance(instanceId: string): Promise<void> {
-    this.logger.info(`Aborting workflow instance ${instanceId}`);
-    const ProcessInstanceAbortMutationDocument = gql`
-      mutation ProcessInstanceAbortMutation($id: String) {
-        ProcessInstanceAbort(id: $id)
-      }
-    `;
-
-    const result = await this.client.mutation(
-      ProcessInstanceAbortMutationDocument,
-      { id: instanceId },
-    );
-
-    this.logger.debug(
-      `Abort workflow instance result: ${JSON.stringify(result)}`,
-    );
-
-    if (result.error) {
-      throw new Error(
-        `Error aborting workflow instance ${instanceId}: ${result.error}`,
-      );
-    }
-    this.logger.debug(`Successfully aborted workflow instance ${instanceId}`);
-  }
-
   public async fetchWorkflowInfo(
     definitionId: string,
   ): Promise<WorkflowInfo | undefined> {

workspaces/orchestrator/plugins/orchestrator-backend/src/service/OrchestratorService.test.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 The Backstage Authors
+ * Copyright Red Hat, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 import {
   ProcessInstance,
   WorkflowDefinition,
@@ -88,64 +89,56 @@ describe('OrchestratorService', () => {
     });
 
     it('should execute the operation when the workflow is available', async () => {
-      dataIndexServiceMock.fetchDefinitionIdByInstanceId = jest
-        .fn()
-        .mockResolvedValue(definitionId);
       workflowCacheServiceMock.isAvailable = jest.fn().mockReturnValue(true);
-      dataIndexServiceMock.abortWorkflowInstance = jest.fn(
-        (_instanceId: string) => Promise.resolve(),
+      sonataFlowServiceMock.abortInstance = jest.fn(
+        (_args: {
+          definitionId: string;
+          instanceId: string;
+          serviceUrl: string;
+        }) => Promise.resolve(),
       );
 
       await orchestratorService.abortWorkflowInstance({
+        definitionId,
         instanceId,
+        serviceUrl,
         cacheHandler: 'skip',
       });
 
-      expect(
-        dataIndexServiceMock.fetchDefinitionIdByInstanceId,
-      ).toHaveBeenCalled();
-      expect(dataIndexServiceMock.abortWorkflowInstance).toHaveBeenCalled();
+      expect(sonataFlowServiceMock.abortInstance).toHaveBeenCalled();
     });
 
     it('should skip and not execute the operation when the workflow is not available', async () => {
-      dataIndexServiceMock.fetchDefinitionIdByInstanceId = jest
-        .fn()
-        .mockResolvedValue(definitionId);
       workflowCacheServiceMock.isAvailable = jest.fn().mockReturnValue(false);
 
       await orchestratorService.abortWorkflowInstance({
+        definitionId,
         instanceId,
+        serviceUrl,
         cacheHandler: 'skip',
       });
 
-      expect(
-        dataIndexServiceMock.fetchDefinitionIdByInstanceId,
-      ).toHaveBeenCalled();
-      expect(dataIndexServiceMock.abortWorkflowInstance).not.toHaveBeenCalled();
+      expect(sonataFlowServiceMock.abortInstance).not.toHaveBeenCalled();
     });
 
     it('should throw an error and not execute the operation when the workflow is not available', async () => {
-      dataIndexServiceMock.fetchDefinitionIdByInstanceId = jest
-        .fn()
-        .mockResolvedValue(definitionId);
       workflowCacheServiceMock.isAvailable = jest
         .fn()
         .mockImplementation(() => {
           throw new Error();
         });
 
       const promise = orchestratorService.abortWorkflowInstance({
+        definitionId,
         instanceId,
+        serviceUrl,
         cacheHandler: 'throw',
       });
 
       await expect(promise).rejects.toThrow();
 
-      expect(
-        dataIndexServiceMock.fetchDefinitionIdByInstanceId,
-      ).toHaveBeenCalled();
       expect(workflowCacheServiceMock.isAvailable).toHaveBeenCalled();
-      expect(dataIndexServiceMock.abortWorkflowInstance).not.toHaveBeenCalled();
+      expect(sonataFlowServiceMock.abortInstance).not.toHaveBeenCalled();
     });
   });
 

workspaces/orchestrator/plugins/orchestrator-backend/src/service/OrchestratorService.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 The Backstage Authors
+ * Copyright Red Hat, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 import {
   Filter,
   ProcessInstance,
@@ -41,18 +42,18 @@ export class OrchestratorService {
   }
 
   public async abortWorkflowInstance(args: {
+    definitionId: string;
     instanceId: string;
+    serviceUrl: string;
     cacheHandler?: CacheHandler;
   }): Promise<void> {
-    const { instanceId, cacheHandler } = args;
-    const definitionId =
-      await this.dataIndexService.fetchDefinitionIdByInstanceId(instanceId);
+    const { definitionId, cacheHandler } = args;
     const isWorkflowAvailable = this.workflowCacheService.isAvailable(
       definitionId,
       cacheHandler,
     );
     return isWorkflowAvailable
-      ? await this.dataIndexService.abortWorkflowInstance(instanceId)
+      ? await this.sonataFlowService.abortInstance(args)
       : undefined;
   }
 

workspaces/orchestrator/plugins/orchestrator-backend/src/service/SonataFlowService.ts

@@ -166,6 +166,31 @@ export class SonataFlowService {
     return true;
   }
 
+  public async abortInstance(args: {
+    definitionId: string;
+    instanceId: string;
+    serviceUrl: string;
+  }): Promise<void> {
+    const urlToFetch = `${args.serviceUrl}/management/processes/${args.definitionId}/instances/${args.instanceId}`;
+
+    const response = await fetch(urlToFetch, {
+      method: 'DELETE',
+    });
+
+    if (!response.ok) {
+      const json = await response.json();
+      this.logger.error(`Abort failed with: ${JSON.stringify(json)}`);
+      throw new Error(
+        `${await this.createPrefixFetchErrorMessage(
+          urlToFetch,
+          response,
+          json,
+          'DELETE',
+        )}`,
+      );
+    }
+  }
+
   public async fetchWorkflowOverview(
     definitionId: string,
   ): Promise<WorkflowOverview | undefined> {

workspaces/orchestrator/plugins/orchestrator-backend/src/service/api/v2.test.ts

@@ -548,7 +548,7 @@ describe('abortWorkflow', () => {
     const expectedResult = `Workflow instance ${workflowId} successfully aborted`;
 
     // Act
-    const actualResult: string = await v2.abortWorkflow(workflowId);
+    const actualResult: string = await v2.abortWorkflow('dummy', workflowId);
 
     // Assert
     expect(actualResult).toBeDefined();
@@ -562,7 +562,7 @@ describe('abortWorkflow', () => {
     ).mockRejectedValue(new Error('Simulated abort workflow error'));
 
     // Act
-    const promise = v2.abortWorkflow('instanceId');
+    const promise = v2.abortWorkflow('definitionId', 'instanceId');
 
     // Assert
     await expect(promise).rejects.toThrow('Simulated abort workflow error');

workspaces/orchestrator/plugins/orchestrator-backend/src/service/api/v2.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 The Backstage Authors
+ * Copyright Red Hat, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 import { ParsedRequest } from 'openapi-backend';
 
 import {
@@ -230,9 +231,24 @@ export class V2 {
     }
   }
 
-  public async abortWorkflow(instanceId: string): Promise<string> {
+  public async abortWorkflow(
+    workflowId: string,
+    instanceId: string,
+  ): Promise<string> {
+    const definition = await this.orchestratorService.fetchWorkflowInfo({
+      definitionId: workflowId,
+      cacheHandler: 'throw',
+    });
+    if (!definition) {
+      throw new Error(`Couldn't fetch workflow definition for ${workflowId}`);
+    }
+    if (!definition.serviceUrl) {
+      throw new Error(`ServiceURL is not defined for workflow ${workflowId}`);
+    }
     await this.orchestratorService.abortWorkflowInstance({
-      instanceId,
+      definitionId: workflowId,
+      instanceId: instanceId,
+      serviceUrl: definition.serviceUrl,
       cacheHandler: 'throw',
     });
     return `Workflow instance ${instanceId} successfully aborted`;

workspaces/orchestrator/plugins/orchestrator-backend/src/service/router.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 The Backstage Authors
+ * Copyright Red Hat, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 import { MiddlewareFactory } from '@backstage/backend-defaults/rootHttpRouter';
 import {
   HttpAuthService,
@@ -897,7 +898,7 @@ function setupInternalRoutes(
           manageDenyAuthorization(endpointName, endpoint, _req);
         }
 
-        const result = await routerApi.v2.abortWorkflow(instanceId);
+        const result = await routerApi.v2.abortWorkflow(workflowId, instanceId);
         res.status(200).json(result);
       } catch (error) {
         auditLogRequestError(error, endpointName, endpoint, _req);