From 0d1353d94025504ac916a461a62539250599a771 Mon Sep 17 00:00:00 2001
From: Tyler Ohlsen <ohltyler@amazon.com>
Date: Thu, 29 Feb 2024 10:28:13 -0800
Subject: [PATCH] Fetch workflow state; align data models with backend
Signed-off-by: Tyler Ohlsen <ohltyler@amazon.com>
---
common/constants.ts | 1 +
common/interfaces.ts | 10 +++--
.../workflow_detail/launches/launch_list.tsx | 2 +-
.../pages/workflows/workflow_list/columns.tsx | 19 ++++++----
public/utils/utils.ts | 8 ++--
server/cluster/flow_framework_plugin.ts | 10 +++++
.../routes/flow_framework_routes_service.ts | 23 ++++++++----
server/routes/helpers.ts | 37 +++++++++++++++++--
8 files changed, 82 insertions(+), 28 deletions(-)
diff --git a/common/constants.ts b/common/constants.ts
index 033e8a0c..0903f38c 100644
--- a/common/constants.ts
+++ b/common/constants.ts
@@ -11,6 +11,7 @@ export const PLUGIN_ID = 'flow-framework';
export const FLOW_FRAMEWORK_API_ROUTE_PREFIX = '/_plugins/_flow_framework';
export const FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX = `${FLOW_FRAMEWORK_API_ROUTE_PREFIX}/workflow`;
export const FLOW_FRAMEWORK_SEARCH_WORKFLOWS_ROUTE = `${FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX}/_search`;
+export const FLOW_FRAMEWORK_SEARCH_WORKFLOW_STATE_ROUTE = `${FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX}/state/_search`;
/**
* NODE APIs
diff --git a/common/interfaces.ts b/common/interfaces.ts
index 19c8011a..a493d650 100644
--- a/common/interfaces.ts
+++ b/common/interfaces.ts
@@ -71,11 +71,13 @@ export type UseCaseTemplate = {
export type Workflow = {
id: string;
name: string;
+ useCase: string;
description?: string;
// ReactFlow state may not exist if a workflow is created via API/backend-only.
workspaceFlowState?: WorkspaceFlowState;
template: UseCaseTemplate;
lastUpdated: number;
+ lastLaunched: number;
state: WORKFLOW_STATE;
};
@@ -95,12 +97,12 @@ export type WorkflowLaunch = {
lastUpdated: number;
};
-// TODO: finalize list of possible workflow states from backend
+// Based off of https://github.com/opensearch-project/flow-framework/blob/main/src/main/java/org/opensearch/flowframework/model/State.java
export enum WORKFLOW_STATE {
- SUCCEEDED = 'Succeeded',
- FAILED = 'Failed',
- IN_PROGRESS = 'In progress',
NOT_STARTED = 'Not started',
+ PROVISIONING = 'Provisioning',
+ FAILED = 'Failed',
+ COMPLETED = 'Completed',
}
export type WorkflowDict = {
diff --git a/public/pages/workflow_detail/launches/launch_list.tsx b/public/pages/workflow_detail/launches/launch_list.tsx
index 451e0591..3062887f 100644
--- a/public/pages/workflow_detail/launches/launch_list.tsx
+++ b/public/pages/workflow_detail/launches/launch_list.tsx
@@ -30,7 +30,7 @@ export function LaunchList(props: LaunchListProps) {
const workflowLaunches = [
{
id: 'Launch_1',
- state: WORKFLOW_STATE.IN_PROGRESS,
+ state: WORKFLOW_STATE.PROVISIONING,
lastUpdated: 12345678,
},
{
diff --git a/public/pages/workflows/workflow_list/columns.tsx b/public/pages/workflows/workflow_list/columns.tsx
index 88aa7daf..6de07fb6 100644
--- a/public/pages/workflows/workflow_list/columns.tsx
+++ b/public/pages/workflows/workflow_list/columns.tsx
@@ -17,18 +17,23 @@ export const columns = [
),
},
{
- field: 'id',
- name: 'ID',
+ field: 'state',
+ name: 'Status',
sortable: true,
},
{
- field: 'description',
- name: 'Description',
- sortable: false,
+ field: 'useCase',
+ name: 'Type',
+ sortable: true,
},
{
- field: 'state',
- name: 'Status',
+ field: 'lastUpdated',
+ name: 'Last updated',
+ sortable: true,
+ },
+ {
+ field: 'lastLaunched',
+ name: 'Last launched',
sortable: true,
},
];
diff --git a/public/utils/utils.ts b/public/utils/utils.ts
index a36be3d5..5eaa77c1 100644
--- a/public/utils/utils.ts
+++ b/public/utils/utils.ts
@@ -131,22 +131,22 @@ export function getStateOptions(): EuiFilterSelectItem[] {
return [
// @ts-ignore
{
- name: WORKFLOW_STATE.SUCCEEDED,
+ name: WORKFLOW_STATE.NOT_STARTED,
checked: 'on',
} as EuiFilterSelectItem,
// @ts-ignore
{
- name: WORKFLOW_STATE.NOT_STARTED,
+ name: WORKFLOW_STATE.PROVISIONING,
checked: 'on',
} as EuiFilterSelectItem,
// @ts-ignore
{
- name: WORKFLOW_STATE.IN_PROGRESS,
+ name: WORKFLOW_STATE.FAILED,
checked: 'on',
} as EuiFilterSelectItem,
// @ts-ignore
{
- name: WORKFLOW_STATE.FAILED,
+ name: WORKFLOW_STATE.COMPLETED,
checked: 'on',
} as EuiFilterSelectItem,
];
diff --git a/server/cluster/flow_framework_plugin.ts b/server/cluster/flow_framework_plugin.ts
index 084d3bbf..3b7616dc 100644
--- a/server/cluster/flow_framework_plugin.ts
+++ b/server/cluster/flow_framework_plugin.ts
@@ -5,6 +5,7 @@
import {
FLOW_FRAMEWORK_SEARCH_WORKFLOWS_ROUTE,
+ FLOW_FRAMEWORK_SEARCH_WORKFLOW_STATE_ROUTE,
FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX,
} from '../../common';
@@ -55,6 +56,15 @@ export function flowFrameworkPlugin(Client: any, config: any, components: any) {
method: 'GET',
});
+ flowFramework.searchWorkflowState = ca({
+ url: {
+ fmt: FLOW_FRAMEWORK_SEARCH_WORKFLOW_STATE_ROUTE,
+ },
+ needBody: true,
+ // Exposed client rejects making GET requests with a body. So, we use POST
+ method: 'POST',
+ });
+
flowFramework.createWorkflow = ca({
url: {
fmt: FLOW_FRAMEWORK_WORKFLOW_ROUTE_PREFIX,
diff --git a/server/routes/flow_framework_routes_service.ts b/server/routes/flow_framework_routes_service.ts
index be98978b..149997d5 100644
--- a/server/routes/flow_framework_routes_service.ts
+++ b/server/routes/flow_framework_routes_service.ts
@@ -17,9 +17,8 @@ import {
GET_WORKFLOW_NODE_API_PATH,
GET_WORKFLOW_STATE_NODE_API_PATH,
SEARCH_WORKFLOWS_NODE_API_PATH,
- WorkflowDict,
} from '../../common';
-import { generateCustomError, toWorkflowObj } from './helpers';
+import { generateCustomError, getWorkflowsFromResponses } from './helpers';
/**
* Server-side routes to process flow-framework-related node API calls and execute the
@@ -112,6 +111,9 @@ export class FlowFrameworkRoutesService {
}
};
+ // TODO: can remove or simplify if we can fetch all data from a single API call. Tracking issue:
+ // https://github.com/opensearch-project/flow-framework/issues/171
+ // Current implementation is making two calls and combining results via helper fn
searchWorkflows = async (
context: RequestHandlerContext,
req: OpenSearchDashboardsRequest,
@@ -119,15 +121,20 @@ export class FlowFrameworkRoutesService {
): Promise<IOpenSearchDashboardsResponse<any>> => {
const body = req.body;
try {
- const response = await this.client
+ const workflowsResponse = await this.client
.asScoped(req)
.callAsCurrentUser('flowFramework.searchWorkflows', { body });
- const workflowHits = response.hits.hits as any[];
- const workflowDict = {} as WorkflowDict;
- workflowHits.forEach((workflowHit: any) => {
- workflowDict[workflowHit._id] = toWorkflowObj(workflowHit);
- });
+ const workflowHits = workflowsResponse.hits.hits as any[];
+
+ const workflowStatesResponse = await this.client
+ .asScoped(req)
+ .callAsCurrentUser('flowFramework.searchWorkflowState', { body });
+ const workflowStateHits = workflowStatesResponse.hits.hits as any[];
+ const workflowDict = getWorkflowsFromResponses(
+ workflowHits,
+ workflowStateHits
+ );
return res.ok({ body: { workflows: workflowDict } });
} catch (err: any) {
return generateCustomError(res, err);
diff --git a/server/routes/helpers.ts b/server/routes/helpers.ts
index e67d90f4..9a47fb82 100644
--- a/server/routes/helpers.ts
+++ b/server/routes/helpers.ts
@@ -3,7 +3,7 @@
* SPDX-License-Identifier: Apache-2.0
*/
-import { WORKFLOW_STATE, Workflow } from '../../common';
+import { WORKFLOW_STATE, Workflow, WorkflowDict } from '../../common';
// OSD does not provide an interface for this response, but this is following the suggested
// implementations. To prevent typescript complaining, leaving as loosely-typed 'any'
@@ -19,18 +19,47 @@ export function generateCustomError(res: any, err: any) {
});
}
-export function toWorkflowObj(workflowHit: any): Workflow {
+function toWorkflowObj(workflowHit: any): Workflow {
// TODO: update schema parsing after hit schema has been updated.
// https://github.com/opensearch-project/flow-framework/issues/546
const hitSource = workflowHit.fields.filter[0];
- // const hitSource = workflowHit._source;
return {
id: workflowHit._id,
name: hitSource.name,
+ useCase: hitSource.use_case,
description: hitSource.description || '',
// TODO: update below values after frontend Workflow interface is finalized
template: {},
+ // TODO: this needs to be persisted by backend. Tracking issue:
+ // https://github.com/opensearch-project/flow-framework/issues/548
lastUpdated: 1234,
- state: WORKFLOW_STATE.SUCCEEDED,
} as Workflow;
}
+
+// TODO: can remove or simplify if we can fetch all data from a single API call. Tracking issue:
+// https://github.com/opensearch-project/flow-framework/issues/171
+// Current implementation combines 2 search responses to create a single set of workflows with
+// static information + state information
+export function getWorkflowsFromResponses(
+ workflowHits: any[],
+ workflowStateHits: any[]
+): WorkflowDict {
+ const workflowDict = {} as WorkflowDict;
+ workflowHits.forEach((workflowHit: any) => {
+ workflowDict[workflowHit._id] = toWorkflowObj(workflowHit);
+ const workflowStateHit = workflowStateHits.find(
+ (workflowStateHit) => workflowStateHit._id === workflowHit._id
+ );
+ const workflowState = workflowStateHit._source
+ .state as typeof WORKFLOW_STATE;
+ workflowDict[workflowHit._id] = {
+ ...workflowDict[workflowHit._id],
+ // @ts-ignore
+ state: WORKFLOW_STATE[workflowState],
+ // TODO: this needs to be persisted by backend. Tracking issue:
+ // https://github.com/opensearch-project/flow-framework/issues/548
+ lastLaunched: 1234,
+ };
+ });
+ return workflowDict;
+}