Get update working; executing has some parsing issue

Signed-off-by: Tyler Ohlsen <[email protected]>
opensearch-project · Apr 5, 2024 · 8bf8897 · 8bf8897
1 parent ce28197
commit 8bf8897
Show file tree

Hide file tree

Showing 7 changed files with 154 additions and 7 deletions.
diff --git a/common/constants.ts b/common/constants.ts
@@ -37,6 +37,7 @@ export const GET_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}`;
 export const SEARCH_WORKFLOWS_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/search`;
 export const GET_WORKFLOW_STATE_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/state`;
 export const CREATE_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/create`;
+export const UPDATE_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/update`;
 export const PROVISION_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/provision`;
 export const DEPROVISION_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/deprovision`;
 export const DELETE_WORKFLOW_NODE_API_PATH = `${BASE_WORKFLOW_NODE_API_PATH}/delete`;

diff --git a/public/pages/workflow_detail/workspace/resizable_workspace.tsx b/public/pages/workflow_detail/workspace/resizable_workspace.tsx
@@ -34,7 +34,7 @@ import {
   DEFAULT_NEW_WORKFLOW_DESCRIPTION,
   USE_CASE,
   WORKFLOW_STATE,
-  processNodes,
+  reduceToTemplate,
 } from '../../../../common';
 import {
   AppState,
@@ -44,6 +44,7 @@ import {
   provisionWorkflow,
   removeDirty,
   setDirty,
+  updateWorkflow,
   useAppDispatch,
 } from '../../../store';
 import { Workspace } from './workspace';
@@ -390,8 +391,21 @@ export function ResizableWorkspace(props: ResizableWorkspaceProps) {
                     // The callback fn to run if everything is valid.
                     (updatedWorkflow) => {
                       if (updatedWorkflow.id) {
-                        // TODO: add update workflow API
-                        // make sure to set isSaving to false in catch block
+                        dispatch(
+                          updateWorkflow({
+                            workflowId: updatedWorkflow.id,
+                            workflowTemplate: reduceToTemplate(updatedWorkflow),
+                          })
+                        )
+                          .unwrap()
+                          .then((result) => {
+                            setIsSaving(false);
+                          })
+                          .catch((error: any) => {
+                            // TODO: process error (toast msg?)
+                            console.log('error: ', error);
+                            setIsSaving(false);
+                          });
                       } else {
                         dispatch(createWorkflow(updatedWorkflow))
                           .unwrap()

diff --git a/public/route_service.ts b/public/route_service.ts
@@ -15,6 +15,8 @@ import {
   SEARCH_MODELS_NODE_API_PATH,
   PROVISION_WORKFLOW_NODE_API_PATH,
   DEPROVISION_WORKFLOW_NODE_API_PATH,
+  UPDATE_WORKFLOW_NODE_API_PATH,
+  WorkflowTemplate,
 } from '../common';
 
 /**
@@ -28,9 +30,10 @@ export interface RouteService {
   getWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
   searchWorkflows: (body: {}) => Promise<any | HttpFetchError>;
   getWorkflowState: (workflowId: string) => Promise<any | HttpFetchError>;
-  createWorkflow: (
-    body: {},
-    provision?: boolean
+  createWorkflow: (body: {}) => Promise<any | HttpFetchError>;
+  updateWorkflow: (
+    workflowId: string,
+    workflowTemplate: WorkflowTemplate
   ) => Promise<any | HttpFetchError>;
   provisionWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
   deprovisionWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
@@ -88,6 +91,22 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
+    updateWorkflow: async (
+      workflowId: string,
+      workflowTemplate: WorkflowTemplate
+    ) => {
+      try {
+        const response = await core.http.put<{ respString: string }>(
+          `${UPDATE_WORKFLOW_NODE_API_PATH}/${workflowId}`,
+          {
+            body: JSON.stringify(workflowTemplate),
+          }
+        );
+        return response;
+      } catch (e: any) {
+        return e as HttpFetchError;
+      }
+    },
     provisionWorkflow: async (workflowId: string) => {
       try {
         const response = await core.http.post<{ respString: string }>(

diff --git a/public/store/reducers/workflows_reducer.ts b/public/store/reducers/workflows_reducer.ts
@@ -4,7 +4,7 @@
  */
 
 import { createAsyncThunk, createSlice } from '@reduxjs/toolkit';
-import { Workflow, WorkflowDict } from '../../../common';
+import { Workflow, WorkflowDict, WorkflowTemplate } from '../../../common';
 import { HttpFetchError } from '../../../../../src/core/public';
 import { getRouteService } from '../../services';
 
@@ -20,6 +20,7 @@ const GET_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/get`;
 const SEARCH_WORKFLOWS_ACTION = `${WORKFLOWS_ACTION_PREFIX}/search`;
 const GET_WORKFLOW_STATE_ACTION = `${WORKFLOWS_ACTION_PREFIX}/getState`;
 const CREATE_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/create`;
+const UPDATE_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/update`;
 const PROVISION_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/provision`;
 const DEPROVISION_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/deprovision`;
 const DELETE_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/delete`;
@@ -90,6 +91,29 @@ export const createWorkflow = createAsyncThunk(
   }
 );
 
+export const updateWorkflow = createAsyncThunk(
+  UPDATE_WORKFLOW_ACTION,
+  async (
+    workflowInfo: { workflowId: string; workflowTemplate: WorkflowTemplate },
+    { rejectWithValue }
+  ) => {
+    const { workflowId, workflowTemplate } = workflowInfo;
+    const response:
+      | any
+      | HttpFetchError = await getRouteService().updateWorkflow(
+      workflowId,
+      workflowTemplate
+    );
+    if (response instanceof HttpFetchError) {
+      return rejectWithValue(
+        'Error updating workflow: ' + response.body.message
+      );
+    } else {
+      return response;
+    }
+  }
+);
+
 export const provisionWorkflow = createAsyncThunk(
   PROVISION_WORKFLOW_ACTION,
   async (workflowId: string, { rejectWithValue }) => {
@@ -173,6 +197,10 @@ const workflowsSlice = createSlice({
         state.loading = true;
         state.errorMessage = '';
       })
+      .addCase(updateWorkflow.pending, (state, action) => {
+        state.loading = true;
+        state.errorMessage = '';
+      })
       .addCase(provisionWorkflow.pending, (state, action) => {
         state.loading = true;
         state.errorMessage = '';
@@ -228,6 +256,22 @@ const workflowsSlice = createSlice({
         state.loading = false;
         state.errorMessage = '';
       })
+      .addCase(updateWorkflow.fulfilled, (state, action) => {
+        const { workflowId, workflowTemplate } = action.payload as {
+          workflowId: string;
+          workflowTemplate: WorkflowTemplate;
+        };
+        state.workflows = {
+          ...state.workflows,
+          [workflowId]: {
+            // only overwrite the stateless / template fields. persist any existing state (e.g., lastUpdated, lastProvisioned)
+            ...state.workflows[workflowId],
+            ...workflowTemplate,
+          },
+        };
+        state.loading = false;
+        state.errorMessage = '';
+      })
       .addCase(provisionWorkflow.fulfilled, (state, action) => {
         state.loading = false;
         state.errorMessage = '';
@@ -266,6 +310,10 @@ const workflowsSlice = createSlice({
         state.errorMessage = action.payload as string;
         state.loading = false;
       })
+      .addCase(updateWorkflow.rejected, (state, action) => {
+        state.errorMessage = action.payload as string;
+        state.loading = false;
+      })
       .addCase(provisionWorkflow.rejected, (state, action) => {
         state.errorMessage = action.payload as string;
         state.loading = false;

diff --git a/public/utils/utils.ts b/public/utils/utils.ts
@@ -16,6 +16,8 @@ import {
   WorkspaceFormValues,
   WORKFLOW_STATE,
   ReactFlowComponent,
+  Workflow,
+  WorkflowTemplate,
 } from '../../common';
 
 // Append 16 random characters
@@ -72,6 +74,19 @@ export function formikToComponentData(
   } as IComponentData;
 }
 
+// Helper fn to remove state-related fields from a workflow and have a stateless template
+// to export and/or pass around, use when updating, etc.
+export function reduceToTemplate(workflow: Workflow): WorkflowTemplate {
+  const {
+    id,
+    lastUpdated,
+    lastLaunched,
+    state,
+    ...workflowTemplate
+  } = workflow;
+  return workflowTemplate;
+}
+
 // Helper fn to get an initial value based on the field type
 export function getInitialValue(fieldType: FieldType): FieldValue {
   switch (fieldType) {

diff --git a/server/cluster/flow_framework_plugin.ts b/server/cluster/flow_framework_plugin.ts
@@ -73,6 +73,20 @@ export function flowFrameworkPlugin(Client: any, config: any, components: any) {
     method: 'POST',
   });
 
+  flowFramework.updateWorkflow = ca({
+    url: {
+      fmt: `${FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX}/<%=workflow_id%>`,
+      req: {
+        workflow_id: {
+          type: 'string',
+          required: true,
+        },
+      },
+    },
+    needBody: true,
+    method: 'PUT',
+  });
+
   flowFramework.provisionWorkflow = ca({
     url: {
       fmt: `${FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX}/<%=workflow_id%>/_provision`,

diff --git a/server/routes/flow_framework_routes_service.ts b/server/routes/flow_framework_routes_service.ts
@@ -22,6 +22,7 @@ import {
   GET_WORKFLOW_STATE_NODE_API_PATH,
   PROVISION_WORKFLOW_NODE_API_PATH,
   SEARCH_WORKFLOWS_NODE_API_PATH,
+  UPDATE_WORKFLOW_NODE_API_PATH,
   WORKFLOW_STATE,
   Workflow,
   WorkflowTemplate,
@@ -85,6 +86,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.createWorkflow
   );
 
+  router.put(
+    {
+      path: `${UPDATE_WORKFLOW_NODE_API_PATH}/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    flowFrameworkRoutesService.updateWorkflow
+  );
+
   router.post(
     {
       path: `${PROVISION_WORKFLOW_NODE_API_PATH}/{workflow_id}`,
@@ -227,6 +241,28 @@ export class FlowFrameworkRoutesService {
     }
   };
 
+  updateWorkflow = async (
+    context: RequestHandlerContext,
+    req: OpenSearchDashboardsRequest,
+    res: OpenSearchDashboardsResponseFactory
+  ): Promise<IOpenSearchDashboardsResponse<any>> => {
+    const { workflow_id } = req.params as { workflow_id: string };
+    const workflowTemplate = req.body as WorkflowTemplate;
+
+    try {
+      await this.client
+        .asScoped(req)
+        .callAsCurrentUser('flowFramework.updateWorkflow', {
+          workflow_id,
+          body: workflowTemplate,
+        });
+
+      return res.ok({ body: { workflowId: workflow_id, workflowTemplate } });
+    } catch (err: any) {
+      return generateCustomError(res, err);
+    }
+  };
+
   provisionWorkflow = async (
     context: RequestHandlerContext,
     req: OpenSearchDashboardsRequest,