feat(streams): improve error handling for processors using simulation

elastic · tonyghiani · Jan 15, 2025 · Jan 13, 2025 · Jan 13, 2025 · Jan 13, 2025
commit 1a0af3fd0db7f2e389662673892d6c8f0221bfa3
@@ -19,7 +19,7 @@ import { DangerZone } from './danger_zone';
 import { DissectProcessorForm } from './dissect';
 import { GrokProcessorForm } from './grok';
 import { convertFormStateToProcessing, getDefaultFormState } from '../utils';
-import { useProcessorSimulationGate } from '../hooks/use_processor_simulation_gate';
+import { useProcessingSimulator } from '../hooks/use_processing_simulator';
 
 const ProcessorOutcomePreview = dynamic(() =>
   import(/* webpackChunkName: "management_processor_outcome" */ './processor_outcome_preview').then(
@@ -59,12 +59,18 @@ export function AddProcessorFlyout({
     [defaultValues, formFields]
   );
 
-  const simulate = useProcessorSimulationGate({ definition });
+  const { error, isLoading, refreshSamples, simulation, samples, simulate } =
+    useProcessingSimulator({
+      definition,
+      condition: { field: formFields.field, operator: 'exists' },
+    });
 
   const handleSubmit: SubmitHandler<ProcessorFormState> = async (data) => {
     const processingDefinition = convertFormStateToProcessing(data);
 
-    simulate(processingDefinition).then(() => {
+    simulate(processingDefinition).then((responseBody) => {
+      if (responseBody instanceof Error) return;
+
       onAddProcessor(processingDefinition, data.detected_fields);
       onClose();
     });
@@ -98,7 +104,16 @@ export function AddProcessorFlyout({
           {formFields.type === 'dissect' && <DissectProcessorForm />}
         </EuiForm>
         <EuiHorizontalRule />
-        <ProcessorOutcomePreview definition={definition} formFields={formFields} />
+        <ProcessorOutcomePreview
+          definition={definition}
+          formFields={formFields}
+          simulation={simulation}
+          samples={samples}
+          onSimulate={simulate}
+          onRefreshSamples={refreshSamples}
+          simulationError={error}
+          isLoading={isLoading}
+        />
       </FormProvider>
     </ProcessorFlyoutTemplate>
   );
@@ -125,12 +140,17 @@ export function EditProcessorFlyout({
     [defaultValues, formFields]
   );
 
-  const simulate = useProcessorSimulationGate({ definition });
+  const { simulate, error } = useProcessingSimulator({
+    definition,
+    condition: { field: formFields.field, operator: 'exists' },
+  });
 
   const handleSubmit: SubmitHandler<ProcessorFormState> = (data) => {
     const processingDefinition = convertFormStateToProcessing(data);
 
-    simulate(processingDefinition).then(() => {
+    simulate(processingDefinition).then((responseBody) => {
+      if (responseBody instanceof Error) return;
+
       onUpdateProcessor(processor.id, { id: processor.id, ...processingDefinition });
       onClose();
     });
@@ -161,7 +181,7 @@ export function EditProcessorFlyout({
       confirmButton={
         <EuiButton
           onClick={methods.handleSubmit(handleSubmit)}
-          disabled={!hasChanges || !methods.formState.isValid}
+          disabled={!methods.formState.isValid}
         >
           {i18n.translate(
             'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.confirmEditProcessor',
@@ -170,6 +190,7 @@ export function EditProcessorFlyout({
         </EuiButton>
       }
     >
+      {error && <InvalidProcessorConfig message={error.body?.message} />}
       <FormProvider {...methods}>
         <EuiForm component="form" fullWidth onSubmit={methods.handleSubmit(handleSubmit)}>
           <ProcessorTypeSelector disabled />
@@ -183,3 +204,21 @@ export function EditProcessorFlyout({
     </ProcessorFlyoutTemplate>
   );
 }
+
+const InvalidProcessorConfig = ({ message }: { message?: string }) => {
+  return (
+    <>
+      <EuiCallOut
+        title={i18n.translate(
+          'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.calloutEditError',
+          { defaultMessage: 'The processor configuration is invalid.' }
+        )}
+        iconType="alert"
+        color="danger"
+      >
+        {message}
+      </EuiCallOut>
+      <EuiSpacer size="m" />
+    </>
+  );
+};
@@ -35,107 +35,45 @@ import { useController, useFieldArray } from 'react-hook-form';
 import { css } from '@emotion/react';
 import { flattenObject } from '@kbn/object-utils';
 import { IHttpFetchError, ResponseErrorBody } from '@kbn/core/public';
-import { useStreamsAppFetch } from '../../../hooks/use_streams_app_fetch';
 import { useKibana } from '../../../hooks/use_kibana';
 import { StreamsAppSearchBar, StreamsAppSearchBarProps } from '../../streams_app_search_bar';
 import { PreviewTable } from '../../preview_table';
-import { convertFormStateToProcessing, isCompleteProcessingDefinition } from '../utils';
+import { convertFormStateToProcessing } from '../utils';
 import { DetectedField, ProcessorFormState } from '../types';
+import { UseProcessingSimulatorReturnType } from '../hooks/use_processing_simulator';
 
 interface ProcessorOutcomePreviewProps {
   definition: ReadStreamDefinition;
   formFields: ProcessorFormState;
+  isLoading: UseProcessingSimulatorReturnType['isLoading'];
+  simulation: UseProcessingSimulatorReturnType['simulation'];
+  samples: UseProcessingSimulatorReturnType['samples'];
+  onRefreshSamples: UseProcessingSimulatorReturnType['refreshSamples'];
+  onSimulate: UseProcessingSimulatorReturnType['simulate'];
+  simulationError: UseProcessingSimulatorReturnType['error'];
 }
 
 export const ProcessorOutcomePreview = ({
   definition,
   formFields,
+  isLoading,
+  simulation,
+  samples,
+  onRefreshSamples,
+  onSimulate,
+  simulationError,
 }: ProcessorOutcomePreviewProps) => {
   const { dependencies } = useKibana();
-  const {
-    data,
-    streams: { streamsRepositoryClient },
-  } = dependencies.start;
+  const { data } = dependencies.start;
 
-  const {
-    timeRange,
-    absoluteTimeRange: { start, end },
-    setTimeRange,
-  } = useDateRange({ data });
+  const { timeRange, setTimeRange } = useDateRange({ data });
 
   const [selectedDocsFilter, setSelectedDocsFilter] =
     useState<DocsFilterOption>('outcome_filter_all');
 
-  const {
-    value: samples,
-    loading: isLoadingSamples,
-    refresh: refreshSamples,
-  } = useStreamsAppFetch(
-    ({ signal }) => {
-      if (!definition || !formFields.field) {
-        return { documents: [] };
-      }
-
-      return streamsRepositoryClient.fetch('POST /api/streams/{id}/_sample', {
-        signal,
-        params: {
-          path: { id: definition.name },
-          body: {
-            condition: { field: formFields.field, operator: 'exists' },
-            start: start?.valueOf(),
-            end: end?.valueOf(),
-            number: 100,
-          },
-        },
-      });
-    },
-    [definition, formFields.field, streamsRepositoryClient, start, end],
-    { disableToastOnError: true }
-  );
-
-  const {
-    value: simulation,
-    loading: isLoadingSimulation,
-    error,
-    refresh: refreshSimulation,
-  } = useStreamsAppFetch(
-    async ({ signal }) => {
-      if (!definition || !samples || isEmpty(samples.documents)) {
-        return Promise.resolve(null);
-      }
-
-      const processingDefinition = convertFormStateToProcessing(formFields);
-
-      if (!isCompleteProcessingDefinition(processingDefinition)) {
-        return Promise.resolve(null);
-      }
-
-      const simulationResult = await streamsRepositoryClient.fetch(
-        'POST /api/streams/{id}/processing/_simulate',
-        {
-          signal,
-          params: {
-            path: { id: definition.name },
-            body: {
-              documents: samples.documents as Array<Record<PropertyKey, unknown>>,
-              processing: [processingDefinition],
-            },
-          },
-        }
-      );
-
-      return simulationResult;
-    },
-    [definition, samples, streamsRepositoryClient],
-    { disableToastOnError: true }
-  );
-
-  const simulationError = error as IHttpFetchError<ResponseErrorBody> | undefined;
-
   const simulationDocuments = useMemo(() => {
     if (!simulation?.documents) {
-      const docs = (samples?.documents ?? []) as Array<Record<PropertyKey, unknown>>;
-      return docs.map((doc) => flattenObject(doc));
+      return samples.map((doc) => flattenObject(doc));
     }
 
     const filterDocuments = (filter: DocsFilterOption) => {
@@ -151,7 +89,7 @@ export const ProcessorOutcomePreview = ({
     };
 
     return filterDocuments(selectedDocsFilter).map((doc) => doc.value);
-  }, [samples?.documents, simulation?.documents, selectedDocsFilter]);
+  }, [samples, simulation?.documents, selectedDocsFilter]);
 
   const detectedFieldsColumns = simulation?.detected_fields
     ? simulation.detected_fields.map((field) => field.name)
@@ -175,8 +113,8 @@ export const ProcessorOutcomePreview = ({
           iconType="play"
           color="accentSecondary"
           size="s"
-          onClick={refreshSimulation}
-          isLoading={isLoadingSimulation}
+          onClick={() => onSimulate(convertFormStateToProcessing(formFields))}
+          isLoading={isLoading}
         >
           {i18n.translate(
             'xpack.streams.streamDetailView.managementTab.enrichment.processorFlyout.runSimulation',
@@ -191,7 +129,7 @@ export const ProcessorOutcomePreview = ({
         onDocsFilterChange={setSelectedDocsFilter}
         timeRange={timeRange}
         onTimeRangeChange={setTimeRange}
-        onTimeRangeRefresh={refreshSamples}
+        onTimeRangeRefresh={onRefreshSamples}
         simulationFailureRate={simulation?.failure_rate}
         simulationSuccessRate={simulation?.success_rate}
       />
@@ -200,7 +138,7 @@ export const ProcessorOutcomePreview = ({
         documents={simulationDocuments}
         columns={[formFields.field, ...detectedFieldsColumns]}
         error={simulationError}
-        isLoading={isLoadingSamples || isLoadingSimulation}
+        isLoading={isLoading}
       />
     </EuiPanel>
   );

@@ -0,0 +1,106 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { useAbortController } from '@kbn/observability-utils-browser/hooks/use_abort_controller';
+import { ReadStreamDefinition, ProcessingDefinition, Condition } from '@kbn/streams-schema';
+import useAsyncFn from 'react-use/lib/useAsyncFn';
+import { IHttpFetchError, ResponseErrorBody } from '@kbn/core/public';
+import { useDateRange } from '@kbn/observability-utils-browser/hooks/use_date_range';
+import { APIReturnType } from '@kbn/streams-plugin/public/api';
+import { useStreamsAppFetch } from '../../../hooks/use_streams_app_fetch';
+import { useKibana } from '../../../hooks/use_kibana';
+
+type Simulation = APIReturnType<'POST /api/streams/{id}/processing/_simulate'>;
+
+export interface UseProcessingSimulatorReturnType {
+  error?: IHttpFetchError<ResponseErrorBody>;
+  isLoading: boolean;
+  refreshSamples: () => void;
+  samples: Array<Record<PropertyKey, unknown>>;
+  simulate: (processing: ProcessingDefinition) => Promise<Simulation | null>;
+  simulation?: Simulation | null;
+}
+
+export const useProcessingSimulator = ({
+  definition,
+  condition,
+}: {
+  definition: ReadStreamDefinition;
+  condition?: Condition;
+}): UseProcessingSimulatorReturnType => {
+  const { dependencies } = useKibana();
+  const {
+    data,
+    streams: { streamsRepositoryClient },
+  } = dependencies.start;
+
+  const {
+    absoluteTimeRange: { start, end },
+  } = useDateRange({ data });
+
+  const abortController = useAbortController();
+
+  const serializedCondition = JSON.stringify(condition);
+
+  const {
+    loading: isLoadingSamples,
+    value: samples,
+    refresh: refreshSamples,
+  } = useStreamsAppFetch(
+    ({ signal }) => {
+      if (!definition) {
+        return { documents: [] };
+      }
+
+      return streamsRepositoryClient.fetch('POST /api/streams/{id}/_sample', {
+        signal,
+        params: {
+          path: { id: definition.name },
+          body: {
+            condition,
+            start: start?.valueOf(),
+            end: end?.valueOf(),
+            number: 100,
+          },
+        },
+      });
+    },
+    [definition, streamsRepositoryClient, start, end, serializedCondition],
+    { disableToastOnError: true }
+  );
+
+  const sampleDocs = (samples?.documents ?? []) as Array<Record<PropertyKey, unknown>>;
+
+  const [{ loading: isLoadingSimulation, error, value }, simulate] = useAsyncFn(
+    (processingDefinition: ProcessingDefinition) => {
+      if (!definition) {
+        return Promise.resolve(null);
+      }
+
+      return streamsRepositoryClient.fetch('POST /api/streams/{id}/processing/_simulate', {
+        signal: abortController.signal,
+        params: {
+          path: { id: definition.name },
+          body: {
+            documents: sampleDocs,
+            processing: [processingDefinition],
+          },
+        },
+      });
+    },
+    [definition, sampleDocs]
+  );
+
+  return {
+    isLoading: isLoadingSamples || isLoadingSimulation,
+    error: error as IHttpFetchError<ResponseErrorBody> | undefined,
+    refreshSamples,
+    simulate,
+    simulation: value,
+    samples: sampleDocs,
+  };
+};