Add verbose simulate ingest pipeline

Signed-off-by: Tyler Ohlsen <ohltyler@amazon.com>
opensearch-project · Feb 3, 2025 · 5e883c3 · 5e883c3
1 parent bdaa464
commit 5e883c3
Show file tree

Hide file tree

Showing 4 changed files with 62 additions and 4 deletions.
diff --git a/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx b/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx
@@ -23,6 +23,9 @@ import {
 import {
   CONFIG_STEP,
   CachedFormikState,
+  IngestPipelineConfig,
+  PROCESSOR_CONTEXT,
+  SimulateIngestPipelineResponse,
   TemplateNode,
   WORKFLOW_STEP_TYPE,
   Workflow,
@@ -38,6 +41,7 @@ import {
   deprovisionWorkflow,
   getWorkflow,
   provisionWorkflow,
+  simulatePipeline,
   updateWorkflow,
   useAppDispatch,
 } from '../../../store';
@@ -51,6 +55,9 @@ import {
   generateId,
   getResourcesToBeForceDeleted,
   getDataSourceId,
+  prepareDocsForSimulate,
+  formikToPartialPipeline,
+  unwrapTransformedDocs,
 } from '../../../utils';
 import { BooleanField } from './input_fields';
 import '../workspace/workspace-styles.scss';
@@ -524,6 +531,48 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
       if (ingestDocsObjs.length > 0 && !isEmpty(ingestDocsObjs[0])) {
         success = await validateAndUpdateWorkflow(false, true, false);
         if (success) {
+          console.log('values: ', values);
+          console.log('uiconfig: ', props.uiConfig);
+
+          if (
+            !isEmpty(values?.ingest?.enrich) &&
+            values?.ingest?.pipelineName !== undefined &&
+            values?.ingest?.pipelineName !== ''
+          ) {
+            const curDocs = prepareDocsForSimulate(
+              values?.ingest?.docs,
+              values?.ingest?.index?.name
+            );
+            await dispatch(
+              simulatePipeline({
+                apiBody: {
+                  pipeline: {},
+                  docs: curDocs,
+                },
+                dataSourceId,
+              })
+            )
+              .unwrap()
+              .then((resp: SimulateIngestPipelineResponse) => {
+                console.log('resp: ', resp);
+                const docObjs = unwrapTransformedDocs(resp);
+                //if (docObjs.length > 0) {
+                //setSourceInput(customStringify(docObjs[0]));
+                //}
+              })
+              .catch((error: any) => {
+                console.log('error: ', error);
+                getCore().notifications.toasts.addDanger(
+                  `Failed to fetch input data`
+                );
+              })
+              .finally(() => {
+                //setIsFetching(false);
+              });
+          } else {
+            console.log('enrich config is empty or pipeline name not found');
+          }
+
           const bulkBody = prepareBulkBody(
             values.ingest.index.name,
             ingestDocsObjs

diff --git a/public/pages/workflows/new_workflow/quick_configure_modal.tsx b/public/pages/workflows/new_workflow/quick_configure_modal.tsx
@@ -72,7 +72,8 @@ export function QuickConfigureModal(props: QuickConfigureModalProps) {
   const dataSourceId = getDataSourceId();
   const history = useHistory();
   const { models } = useSelector((state: AppState) => state.ml);
-  const { workflows } = useSelector((state: AppState) => state.workflows);
+  //const { workflows } = useSelector((state: AppState) => state.workflows);
+  const workflows = undefined;
 
   // model interface states
   const [embeddingModelInterface, setEmbeddingModelInterface] = useState<
@@ -87,7 +88,7 @@ export function QuickConfigureModal(props: QuickConfigureModalProps) {
   const [workflowNameTouched, setWorkflowNameTouched] = useState<boolean>(
     false
   );
-  const workflowNameExists = Object.values(workflows)
+  const workflowNameExists = Object.values(workflows || {})
     .map((workflow) => workflow.name)
     .includes(workflowName);
 

diff --git a/public/store/reducers/workflows_reducer.ts b/public/store/reducers/workflows_reducer.ts
@@ -273,7 +273,7 @@ const workflowsSlice = createSlice({
       })
       .addCase(searchWorkflows.fulfilled, (state, action) => {
         const { workflows } = action.payload as { workflows: WorkflowDict };
-        state.workflows = workflows;
+        state.workflows = workflows || {};
         state.loading = false;
         state.errorMessage = '';
       })

diff --git a/server/routes/opensearch_routes_service.ts b/server/routes/opensearch_routes_service.ts
@@ -525,10 +525,18 @@ export class OpenSearchRoutesService {
         this.client
       );
 
+      const pipelineName = 'ingest_pipeline_5e0b9785e4ade15c';
+      console.log('pipeline name: ', pipelineName);
+      console.log('docs: ', docs);
+
       const response = await callWithRequest('ingest.simulate', {
-        body: { pipeline, docs },
+        body: { docs },
+        id: pipelineName,
+        verbose: true,
       });
 
+      console.log('response: ', response);
+
       return res.ok({
         body: { docs: response.docs } as SimulateIngestPipelineResponse,
       });