[8.4] [ML] Explain Log Rate Spikes: Fix error handling. (#137947)

examples/response_stream/README.md

@@ -21,7 +21,7 @@ The request's headers get passed on to automatically identify if compression is
 On the client, the custom hook is used like this:
 
 ```ts
-const { error, start, cancel, data, isRunning } = useFetchStream<
+const { errors, start, cancel, data, isRunning } = useFetchStream<
     ApiSimpleStringStream, typeof basePath
 >(`${basePath}/internal/response_stream/simple_string_stream`);
 ```

examples/response_stream/public/containers/app/pages/page_reducer_stream/index.tsx

@@ -45,7 +45,7 @@ export const PageReducerStream: FC = () => {
 
   const [simulateErrors, setSimulateErrors] = useState(false);
 
-  const { dispatch, start, cancel, data, error, isCancelled, isRunning } = useFetchStream<
+  const { dispatch, start, cancel, data, errors, isCancelled, isRunning } = useFetchStream<
     ApiReducerStream,
     typeof basePath
   >(
@@ -65,13 +65,15 @@ export const PageReducerStream: FC = () => {
     }
   };
 
+  // TODO This approach needs to be adapted as it might miss when error messages arrive bulk.
   // This is for low level errors on the stream/HTTP level.
   useEffect(() => {
-    if (error) {
-      notifications.toasts.addDanger(error);
+    if (errors.length > 0) {
+      notifications.toasts.addDanger(errors[errors.length - 1]);
     }
-  }, [error, notifications.toasts]);
+  }, [errors, notifications.toasts]);
 
+  // TODO This approach needs to be adapted as it might miss when error messages arrive bulk.
   // This is for errors on the application level
   useEffect(() => {
     if (data.errors.length > 0) {

examples/response_stream/public/containers/app/pages/page_simple_string_stream/index.tsx

@@ -21,7 +21,7 @@ export const PageSimpleStringStream: FC = () => {
   const { core } = useDeps();
   const basePath = core.http?.basePath.get() ?? '';
 
-  const { dispatch, error, start, cancel, data, isRunning } = useFetchStream<
+  const { dispatch, errors, start, cancel, data, isRunning } = useFetchStream<
     ApiSimpleStringStream,
     typeof basePath
   >(`${basePath}/internal/response_stream/simple_string_stream`, { timeout: 500 });
@@ -61,9 +61,17 @@ export const PageSimpleStringStream: FC = () => {
       <EuiText>
         <p>{data}</p>
       </EuiText>
-      {error && (
+      {errors.length > 0 && (
         <EuiCallOut title="Sorry, there was an error" color="danger" iconType="alert">
-          <p>{error}</p>
+          {errors.length === 1 ? (
+            <p>{errors[0]}</p>
+          ) : (
+            <ul>
+              {errors.map((e, i) => (
+                <li key={i}>{e}</li>
+              ))}
+            </ul>
+          )}{' '}
         </EuiCallOut>
       )}
     </Page>

x-pack/packages/ml/aiops_utils/src/fetch_stream.ts

@@ -44,17 +44,24 @@ export async function* fetchStream<I extends UseFetchStreamParamsDefault, BasePa
 ): AsyncGenerator<
   [GeneratorError, ReducerAction<I['reducer']> | Array<ReducerAction<I['reducer']>> | undefined]
 > {
-  const stream = await fetch(endpoint, {
-    signal: abortCtrl.current.signal,
-    method: 'POST',
-    headers: {
-      // This refers to the format of the request body,
-      // not the response, which will be a uint8array Buffer.
-      'Content-Type': 'application/json',
-      'kbn-xsrf': 'stream',
-    },
-    ...(Object.keys(body).length > 0 ? { body: JSON.stringify(body) } : {}),
-  });
+  let stream: Response;
+
+  try {
+    stream = await fetch(endpoint, {
+      signal: abortCtrl.current.signal,
+      method: 'POST',
+      headers: {
+        // This refers to the format of the request body,
+        // not the response, which will be a uint8array Buffer.
+        'Content-Type': 'application/json',
+        'kbn-xsrf': 'stream',
+      },
+      ...(Object.keys(body).length > 0 ? { body: JSON.stringify(body) } : {}),
+    });
+  } catch (error) {
+    yield [error.toString(), undefined];
+    return;
+  }
 
   if (!stream.ok) {
     yield [`Error ${stream.status}: ${stream.statusText}`, undefined];

x-pack/packages/ml/aiops_utils/src/use_fetch_stream.ts

@@ -41,7 +41,7 @@ interface UseFetchStreamReturnType<Data, Action> {
   cancel: () => void;
   data: Data;
   dispatch: Dispatch<Action>;
-  error: string | undefined;
+  errors: string[];
   isCancelled: boolean;
   isRunning: boolean;
   start: () => Promise<void>;
@@ -76,7 +76,7 @@ export function useFetchStream<I extends UseFetchStreamParamsDefault, BasePath e
   body: I['body'],
   options?: { reducer: I['reducer']; initialState: ReducerState<I['reducer']> }
 ): UseFetchStreamReturnType<ReducerState<I['reducer']>, ReducerAction<I['reducer']>> {
-  const [error, setError] = useState<string | undefined>();
+  const [errors, setErrors] = useState<string[]>([]);
   const [isCancelled, setIsCancelled] = useState(false);
   const [isRunning, setIsRunning] = useState(false);
 
@@ -87,13 +87,17 @@ export function useFetchStream<I extends UseFetchStreamParamsDefault, BasePath e
 
   const abortCtrl = useRef(new AbortController());
 
+  const addError = (error: string) => {
+    setErrors((prevErrors) => [...prevErrors, error]);
+  };
+
   const start = async () => {
     if (isRunning) {
-      setError('Restart not supported yet.');
+      addError('Restart not supported yet.');
       return;
     }
 
-    setError(undefined);
+    setErrors([]);
     setIsRunning(true);
     setIsCancelled(false);
 
@@ -104,7 +108,7 @@ export function useFetchStream<I extends UseFetchStreamParamsDefault, BasePath e
       BasePath
     >(endpoint, abortCtrl, body, options !== undefined)) {
       if (fetchStreamError !== null) {
-        setError(fetchStreamError);
+        addError(fetchStreamError);
       } else if (actions.length > 0) {
         dispatch(actions as ReducerAction<I['reducer']>);
       }
@@ -128,7 +132,7 @@ export function useFetchStream<I extends UseFetchStreamParamsDefault, BasePath e
     cancel,
     data,
     dispatch,
-    error,
+    errors,
     isCancelled,
     isRunning,
     start,

x-pack/plugins/aiops/common/api/explain_log_rate_spikes/actions.ts

@@ -10,7 +10,7 @@ import type { ChangePoint, ChangePointHistogram } from '@kbn/ml-agg-utils';
 export const API_ACTION_NAME = {
   ADD_CHANGE_POINTS: 'add_change_points',
   ADD_CHANGE_POINTS_HISTOGRAM: 'add_change_points_histogram',
-  ERROR: 'error',
+  ADD_ERROR: 'add_error',
   RESET: 'reset',
   UPDATE_LOADING_STATE: 'update_loading_state',
 } as const;
@@ -44,14 +44,14 @@ export function addChangePointsHistogramAction(
   };
 }
 
-interface ApiActionError {
-  type: typeof API_ACTION_NAME.ERROR;
+interface ApiActionAddError {
+  type: typeof API_ACTION_NAME.ADD_ERROR;
   payload: string;
 }
 
-export function errorAction(payload: ApiActionError['payload']): ApiActionError {
+export function addErrorAction(payload: ApiActionAddError['payload']): ApiActionAddError {
   return {
-    type: API_ACTION_NAME.ERROR,
+    type: API_ACTION_NAME.ADD_ERROR,
     payload,
   };
 }
@@ -85,6 +85,6 @@ export function updateLoadingStateAction(
 export type AiopsExplainLogRateSpikesApiAction =
   | ApiActionAddChangePoints
   | ApiActionAddChangePointsHistogram
-  | ApiActionError
+  | ApiActionAddError
   | ApiActionReset
   | ApiActionUpdateLoadingState;

x-pack/plugins/aiops/common/api/explain_log_rate_spikes/index.ts

@@ -8,7 +8,7 @@
 export {
   addChangePointsAction,
   addChangePointsHistogramAction,
-  errorAction,
+  addErrorAction,
   resetAction,
   updateLoadingStateAction,
   API_ACTION_NAME,

x-pack/plugins/aiops/common/api/stream_reducer.test.ts

@@ -24,6 +24,7 @@ describe('streamReducer', () => {
       loaded: 50,
       loadingState: 'Loaded 50%',
       changePoints: [],
+      errors: [],
     });
   });
 

x-pack/plugins/aiops/common/api/stream_reducer.ts

@@ -12,13 +12,15 @@ import { API_ACTION_NAME, AiopsExplainLogRateSpikesApiAction } from './explain_l
 interface StreamState {
   ccsWarning: boolean;
   changePoints: ChangePoint[];
+  errors: string[];
   loaded: number;
   loadingState: string;
 }
 
 export const initialState: StreamState = {
   ccsWarning: false,
   changePoints: [],
+  errors: [],
   loaded: 0,
   loadingState: '',
 };
@@ -45,6 +47,8 @@ export function streamReducer(
         return cp;
       });
       return { ...state, changePoints };
+    case API_ACTION_NAME.ADD_ERROR:
+      return { ...state, errors: [...state.errors, action.payload] };
     case API_ACTION_NAME.RESET:
       return initialState;
     case API_ACTION_NAME.UPDATE_LOADING_STATE:

x-pack/plugins/aiops/public/components/explain_log_rate_spikes/explain_log_rate_spikes_analysis.tsx

@@ -8,12 +8,13 @@
 import React, { useEffect, useMemo, useState, FC } from 'react';
 import { isEqual } from 'lodash';
 
-import { EuiEmptyPrompt } from '@elastic/eui';
+import { EuiCallOut, EuiEmptyPrompt, EuiSpacer, EuiText } from '@elastic/eui';
 
 import type { DataView } from '@kbn/data-views-plugin/public';
 import { ProgressControls } from '@kbn/aiops-components';
 import { useFetchStream } from '@kbn/aiops-utils';
 import type { WindowParameters } from '@kbn/aiops-utils';
+import { i18n } from '@kbn/i18n';
 import { FormattedMessage } from '@kbn/i18n-react';
 import type { ChangePoint } from '@kbn/ml-agg-utils';
 import type { Query } from '@kbn/es-query';
@@ -59,10 +60,13 @@ export const ExplainLogRateSpikesAnalysis: FC<ExplainLogRateSpikesAnalysisProps>
     WindowParameters | undefined
   >();
 
-  const { cancel, start, data, isRunning, error } = useFetchStream<
-    ApiExplainLogRateSpikes,
-    typeof basePath
-  >(
+  const {
+    cancel,
+    start,
+    data,
+    isRunning,
+    errors: streamErrors,
+  } = useFetchStream<ApiExplainLogRateSpikes, typeof basePath>(
     `${basePath}/internal/aiops/explain_log_rate_spikes`,
     {
       start: earliest,
@@ -76,11 +80,7 @@ export const ExplainLogRateSpikesAnalysis: FC<ExplainLogRateSpikesAnalysisProps>
     { reducer: streamReducer, initialState }
   );
 
-  useEffect(() => {
-    setCurrentAnalysisWindowParameters(windowParameters);
-    start();
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, []);
+  const errors = useMemo(() => [...streamErrors, ...data.errors], [streamErrors, data.errors]);
 
   // Start handler clears possibly hovered or pinned
   // change points on analysis refresh.
@@ -96,6 +96,12 @@ export const ExplainLogRateSpikesAnalysis: FC<ExplainLogRateSpikesAnalysisProps>
     start();
   }
 
+  useEffect(() => {
+    setCurrentAnalysisWindowParameters(windowParameters);
+    start();
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, []);
+
   const shouldRerunAnalysis = useMemo(
     () =>
       currentAnalysisWindowParameters !== undefined &&
@@ -115,6 +121,7 @@ export const ExplainLogRateSpikesAnalysis: FC<ExplainLogRateSpikesAnalysisProps>
         onCancel={cancel}
         shouldRerunAnalysis={shouldRerunAnalysis}
       />
+      <EuiSpacer size="xs" />
       {!isRunning && !showSpikeAnalysisTable && (
         <EuiEmptyPrompt
           title={
@@ -136,11 +143,37 @@ export const ExplainLogRateSpikesAnalysis: FC<ExplainLogRateSpikesAnalysisProps>
           }
         />
       )}
+      {errors.length > 0 && (
+        <>
+          <EuiCallOut
+            title={i18n.translate('xpack.aiops.analysis.errorCallOutTitle', {
+              defaultMessage:
+                'The following {errorCount, plural, one {error} other {errors}} occurred running the analysis.',
+              values: { errorCount: errors.length },
+            })}
+            color="warning"
+            iconType="alert"
+            size="s"
+          >
+            <EuiText size="s">
+              {errors.length === 1 ? (
+                <p>{errors[0]}</p>
+              ) : (
+                <ul>
+                  {errors.map((e, i) => (
+                    <li key={i}>{e}</li>
+                  ))}
+                </ul>
+              )}
+            </EuiText>
+          </EuiCallOut>
+          <EuiSpacer size="xs" />
+        </>
+      )}
       {showSpikeAnalysisTable && (
         <SpikeAnalysisTable
           changePoints={data.changePoints}
           loading={isRunning}
-          error={error}
           onPinnedChangePoint={onPinnedChangePoint}
           onSelectedChangePoint={onSelectedChangePoint}
           selectedChangePoint={selectedChangePoint}

x-pack/plugins/aiops/public/components/spike_analysis_table/spike_analysis_table.tsx

@@ -34,7 +34,6 @@ const DEFAULT_SORT_DIRECTION = 'asc';
 
 interface SpikeAnalysisTableProps {
   changePoints: ChangePoint[];
-  error?: string;
   loading: boolean;
   onPinnedChangePoint?: (changePoint: ChangePoint | null) => void;
   onSelectedChangePoint?: (changePoint: ChangePoint | null) => void;
@@ -43,7 +42,6 @@ interface SpikeAnalysisTableProps {
 
 export const SpikeAnalysisTable: FC<SpikeAnalysisTableProps> = ({
   changePoints,
-  error,
   loading,
   onPinnedChangePoint,
   onSelectedChangePoint,
@@ -218,7 +216,6 @@ export const SpikeAnalysisTable: FC<SpikeAnalysisTableProps> = ({
       onChange={onChange}
       pagination={pagination}
       loading={false}
-      error={error}
       sorting={sorting as EuiTableSortingType<ChangePoint>}
       rowProps={(changePoint) => {
         return {

x-pack/plugins/aiops/server/routes/explain_log_rate_spikes.ts

@@ -21,7 +21,7 @@ import {
   addChangePointsAction,
   addChangePointsHistogramAction,
   aiopsExplainLogRateSpikesSchema,
-  errorAction,
+  addErrorAction,
   resetAction,
   updateLoadingStateAction,
   AiopsExplainLogRateSpikesApiAction,
@@ -112,7 +112,7 @@ export const defineExplainLogRateSpikesRoute = (
         try {
           fieldCandidates = await fetchFieldCandidates(client, request.body);
         } catch (e) {
-          push(errorAction(e.toString()));
+          push(addErrorAction(e.toString()));
           end();
           return;
         }
@@ -154,7 +154,7 @@ export const defineExplainLogRateSpikesRoute = (
           try {
             pValues = await fetchChangePointPValues(client, request.body, fieldCandidatesChunk);
           } catch (e) {
-            push(errorAction(e.toString()));
+            push(addErrorAction(e.toString()));
             end();
             return;
           }