Add fine-grained error handling; misc bug fixes

common/interfaces.ts

@@ -568,6 +568,10 @@ export type CachedFormikState = {
   touched?: {};
 };
 
+export type IngestPipelineErrors = {
+  [idx: number]: { processorType: string; errorMsg: string };
+};
+
 /**
  ********** OPENSEARCH TYPES/INTERFACES ************
  */
@@ -596,6 +600,24 @@ export type SimulateIngestPipelineResponse = {
   docs: SimulateIngestPipelineDocResponse[];
 };
 
+// verbose mode
+// from https://opensearch.org/docs/latest/ingest-pipelines/simulate-ingest/#query-parameters
+export type SimulateIngestPipelineDocResponseVerbose = SimulateIngestPipelineDocResponse & {
+  processor_type: string;
+  status: 'success' | 'error';
+  description?: string;
+};
+
+// verbose mode
+// from https://opensearch.org/docs/latest/ingest-pipelines/simulate-ingest/#query-parameters
+export type SimulateIngestPipelineResponseVerbose = {
+  docs: [
+    {
+      processor_results: SimulateIngestPipelineDocResponseVerbose[];
+    }
+  ];
+};
+
 export type SearchHit = SimulateIngestPipelineDoc;
 
 export type SearchResponse = {
@@ -619,6 +641,29 @@ export type SearchResponse = {
   ext?: {};
 };
 
+export type SearchProcessorInputData = {
+  _index: string;
+  _id: string;
+  _score: number;
+  _source: {};
+};
+export type SearchProcessorOutputData = SearchProcessorInputData;
+
+export type SearchProcessorResult = {
+  processor_name: string;
+  duration_millis: number;
+  status: 'success' | 'fail';
+  error?: string;
+  input_data: SearchProcessorInputData[] | null;
+  output_data: SearchProcessorOutputData[] | null;
+};
+
+export type SearchResponseVerbose = SearchResponse & {
+  processor_results: SearchProcessorResult[];
+};
+
+export type SearchPipelineErrors = IngestPipelineErrors;
+
 export type IndexResponse = {
   indexName: string;
   indexDetails: IndexConfiguration;

public/configs/ingest_processors/split_ingest_processor.ts

@@ -13,5 +13,13 @@ export class SplitIngestProcessor extends SplitProcessor {
   constructor() {
     super();
     this.id = generateId('split_processor_ingest');
+    this.optionalFields = [
+      ...(this.optionalFields || []),
+      {
+        id: 'ignore_missing',
+        type: 'boolean',
+        value: false,
+      },
+    ];
   }
 }

public/configs/ingest_processors/text_chunking_ingest_processor.ts

@@ -70,6 +70,11 @@ export class TextChunkingIngestProcessor extends Processor {
         id: 'tag',
         type: 'string',
       },
+      {
+        id: 'ignore_missing',
+        type: 'boolean',
+        value: false,
+      },
     ];
   }
 }

public/configs/sort_processor.ts

@@ -40,6 +40,11 @@ export abstract class SortProcessor extends Processor {
         id: 'tag',
         type: 'string',
       },
+      {
+        id: 'ignore_failure',
+        type: 'boolean',
+        value: false,
+      },
     ];
   }
 }

public/configs/split_processor.ts

@@ -45,6 +45,11 @@ export abstract class SplitProcessor extends Processor {
         id: 'tag',
         type: 'string',
       },
+      {
+        id: 'ignore_failure',
+        type: 'boolean',
+        value: false,
+      },
     ];
   }
 }

public/pages/workflow_detail/tools/query/query.tsx

@@ -23,14 +23,23 @@ import {
   FETCH_ALL_QUERY,
   QueryParam,
   SearchResponse,
+  SearchResponseVerbose,
   WorkflowFormValues,
 } from '../../../../../common';
-import { AppState, searchIndex, useAppDispatch } from '../../../../store';
+import {
+  AppState,
+  searchIndex,
+  setOpenSearchError,
+  setSearchPipelineErrors,
+  useAppDispatch,
+} from '../../../../store';
 import {
   containsEmptyValues,
   containsSameValues,
+  formatSearchPipelineErrors,
   getDataSourceId,
   getPlaceholdersFromQuery,
+  getSearchPipelineErrors,
   injectParameters,
 } from '../../../../utils';
 import { QueryParamsList, Results } from '../../../../general_components';
@@ -166,9 +175,7 @@ export function Query(props: QueryProps) {
                               : [SEARCH_OPTIONS[1]]
                           }
                           selectedOptions={
-                            props.hasSearchPipeline &&
-                            includePipeline &&
-                            props.selectedStep === CONFIG_STEP.SEARCH
+                            includePipeline
                               ? [SEARCH_OPTIONS[0]]
                               : [SEARCH_OPTIONS[1]]
                           }
@@ -194,23 +201,49 @@ export function Query(props: QueryProps) {
                             apiBody: {
                               index: indexToSearch,
                               body: injectParameters(queryParams, tempRequest),
-                              searchPipeline:
-                                props.hasSearchPipeline &&
-                                includePipeline &&
-                                props.selectedStep === CONFIG_STEP.SEARCH &&
-                                !isEmpty(values?.search?.pipelineName)
-                                  ? values?.search?.pipelineName
-                                  : '_none',
+                              searchPipeline: includePipeline
+                                ? values?.search?.pipelineName
+                                : '_none',
                             },
                             dataSourceId,
+                            verbose: includePipeline,
                           })
                         )
                           .unwrap()
-                          .then(async (resp: SearchResponse) => {
-                            setQueryResponse(resp);
-                          })
+                          .then(
+                            async (
+                              resp: SearchResponse | SearchResponseVerbose
+                            ) => {
+                              if (includePipeline) {
+                                const searchPipelineErrors = getSearchPipelineErrors(
+                                  resp as SearchResponseVerbose
+                                );
+                                // The errors map may be empty; in which case, this dispatch will clear
+                                // any older errors.
+                                dispatch(
+                                  setSearchPipelineErrors({
+                                    errors: searchPipelineErrors,
+                                  })
+                                );
+                                if (!isEmpty(searchPipelineErrors)) {
+                                  dispatch(
+                                    setOpenSearchError({
+                                      error: `Error running search pipeline. ${formatSearchPipelineErrors(
+                                        searchPipelineErrors
+                                      )}`,
+                                    })
+                                  );
+                                }
+                              } else {
+                                setSearchPipelineErrors({ errors: {} });
+                              }
+
+                              setQueryResponse(resp);
+                            }
+                          )
                           .catch((error: any) => {
                             setQueryResponse(undefined);
+                            setSearchPipelineErrors({ errors: {} });
                             console.error('Error running query: ', error);
                           });
                       }}

public/pages/workflow_detail/workflow_inputs/processors_list.tsx

@@ -4,8 +4,8 @@
  */
 
 import React, { useEffect, useState } from 'react';
+import { useSelector } from 'react-redux';
 import semver from 'semver';
-import { getEffectiveVersion } from '../../../pages/workflows/new_workflow/new_workflow';
 import {
   EuiSmallButtonEmpty,
   EuiSmallButtonIcon,
@@ -29,7 +29,7 @@ import {
   WorkflowFormValues,
 } from '../../../../common';
 import { formikToUiConfig, getDataSourceFromURL } from '../../../utils';
-
+import { getEffectiveVersion } from '../../../pages/workflows/new_workflow/new_workflow';
 import {
   CollapseProcessor,
   CopyIngestProcessor,
@@ -53,6 +53,12 @@ import {
   MIN_SUPPORTED_VERSION,
   MINIMUM_FULL_SUPPORTED_VERSION,
 } from '../../../../common';
+import {
+  AppState,
+  setIngestPipelineErrors,
+  setSearchPipelineErrors,
+  useAppDispatch,
+} from '../../../store';
 
 interface ProcessorsListProps {
   uiConfig: WorkflowConfig;
@@ -67,13 +73,26 @@ const PANEL_ID = 0;
  * General component for configuring pipeline processors (ingest / search request / search response)
  */
 export function ProcessorsList(props: ProcessorsListProps) {
+  const dispatch = useAppDispatch();
+  const {
+    ingestPipeline: ingestPipelineErrors,
+    searchPipeline: searchPipelineErrors,
+  } = useSelector((state: AppState) => state.errors);
   const { values, errors, touched } = useFormikContext<WorkflowFormValues>();
   const [version, setVersion] = useState<string>('');
   const location = useLocation();
   const [processorAdded, setProcessorAdded] = useState<boolean>(false);
   const [isPopoverOpen, setPopover] = useState(false);
   const [processors, setProcessors] = useState<IProcessorConfig[]>([]);
 
+  function clearProcessorErrors(): void {
+    if (props.context === PROCESSOR_CONTEXT.INGEST) {
+      dispatch(setIngestPipelineErrors({ errors: {} }));
+    } else {
+      dispatch(setSearchPipelineErrors({ errors: {} }));
+    }
+  }
+
   const closePopover = () => {
     setPopover(false);
   };
@@ -258,6 +277,7 @@ export function ProcessorsList(props: ProcessorsListProps) {
   // the list of processors. Additionally, persist any current form state
   // (touched, errors) so they are re-initialized when the form is reset.
   function addProcessor(processor: IProcessorConfig): void {
+    clearProcessorErrors();
     props.setCachedFormikState({
       errors,
       touched,
@@ -295,6 +315,7 @@ export function ProcessorsList(props: ProcessorsListProps) {
   // (getting any updated/interim values along the way) delete
   // the specified processor from the list of processors
   function deleteProcessor(processorIdToDelete: string): void {
+    clearProcessorErrors();
     const existingConfig = cloneDeep(props.uiConfig as WorkflowConfig);
     let newConfig = formikToUiConfig(values, existingConfig);
     switch (props.context) {
@@ -341,6 +362,23 @@ export function ProcessorsList(props: ProcessorsListProps) {
           }
         } catch (e) {}
 
+        const processorFormError =
+          hasErrors && allTouched
+            ? 'Invalid or missing fields detected'
+            : undefined;
+        const processorRuntimeError =
+          props.context === PROCESSOR_CONTEXT.INGEST
+            ? getIn(
+                ingestPipelineErrors,
+                `${processorIndex}.errorMsg`,
+                undefined
+              )
+            : getIn(
+                searchPipelineErrors,
+                `${processorIndex}.errorMsg`,
+                undefined
+              );
+
         return (
           <EuiFlexItem key={processorIndex}>
             <EuiPanel paddingSize="s">
@@ -354,14 +392,15 @@ export function ProcessorsList(props: ProcessorsListProps) {
                     <EuiFlexItem grow={false}>
                       <EuiText>{`${processor.name || 'Processor'}`}</EuiText>
                     </EuiFlexItem>
-                    {hasErrors && allTouched && (
+                    {(processorFormError !== undefined ||
+                      processorRuntimeError !== undefined) && (
                       <EuiFlexItem grow={false}>
                         <EuiIconTip
                           aria-label="Warning"
                           size="m"
                           type="alert"
                           color="danger"
-                          content="Invalid or missing fields detected"
+                          content={processorFormError || processorRuntimeError}
                           position="right"
                         />
                       </EuiFlexItem>