[ML] Fix response_stream plugin error handling.

examples/response_stream/common/api/reducer_stream/index.ts

@@ -14,6 +14,7 @@ export const API_ACTION_NAME = {
   UPDATE_PROGRESS: 'update_progress',
   ADD_TO_ENTITY: 'add_to_entity',
   DELETE_ENTITY: 'delete_entity',
+  ERROR: 'error',
 } as const;
 export type ApiActionName = typeof API_ACTION_NAME[keyof typeof API_ACTION_NAME];
 
@@ -59,7 +60,20 @@ export function deleteEntityAction(payload: string): ApiActionDeleteEntity {
   };
 }
 
+interface ApiActionError {
+  type: typeof API_ACTION_NAME.ERROR;
+  payload: string;
+}
+
+export function errorAction(payload: string): ApiActionError {
+  return {
+    type: API_ACTION_NAME.ERROR,
+    payload,
+  };
+}
+
 export type ReducerStreamApiAction =
   | ApiActionUpdateProgress
   | ApiActionAddToEntity
-  | ApiActionDeleteEntity;
+  | ApiActionDeleteEntity
+  | ApiActionError;

examples/response_stream/common/api/reducer_stream/reducer.ts

@@ -14,10 +14,12 @@ export const UI_ACTION_NAME = {
 export type UiActionName = typeof UI_ACTION_NAME[keyof typeof UI_ACTION_NAME];
 
 export interface StreamState {
+  errors: string[];
   progress: number;
   entities: Record<string, number>;
 }
 export const initialState: StreamState = {
+  errors: [],
   progress: 0,
   entities: {},
 };
@@ -64,6 +66,11 @@ export function reducerStreamReducer(
         ...state,
         entities: addToEntities,
       };
+    case API_ACTION_NAME.ERROR:
+      return {
+        ...state,
+        errors: [...state.errors, action.payload],
+      };
     case UI_ACTION_NAME.RESET:
       return initialState;
     default:

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

@@ -65,12 +65,20 @@ export const PageReducerStream: FC = () => {
     }
   };
 
+  // This is for low level errors on the stream/HTTP level.
   useEffect(() => {
     if (error) {
       notifications.toasts.addDanger(error);
     }
   }, [error, notifications.toasts]);
 
+  // This is for errors on the application level
+  useEffect(() => {
+    if (data.errors.length > 0) {
+      notifications.toasts.addDanger(data.errors[data.errors.length - 1]);
+    }
+  }, [data.errors, notifications.toasts]);
+
   const buttonLabel = isRunning ? 'Stop development' : 'Start development';
 
   return (

examples/response_stream/server/routes/reducer_stream.ts

@@ -10,6 +10,7 @@ import type { IRouter, Logger } from '@kbn/core/server';
 import { streamFactory } from '@kbn/aiops-utils';
 
 import {
+  errorAction,
   reducerStreamRequestBodySchema,
   updateProgressAction,
   addToEntityAction,
@@ -38,8 +39,9 @@ export const defineReducerStreamRoute = (router: IRouter, logger: Logger) => {
         shouldStop = true;
       });
 
-      const { end, error, push, responseWithHeaders } = streamFactory<ReducerStreamApiAction>(
-        request.headers
+      const { end, push, responseWithHeaders } = streamFactory<ReducerStreamApiAction>(
+        request.headers,
+        logger
       );
 
       const entities = [
@@ -84,18 +86,17 @@ export const defineReducerStreamRoute = (router: IRouter, logger: Logger) => {
               push(deleteEntityAction(randomEntity));
             } else if (randomAction === 'throw-error') {
               // Throw an error. It should not crash Kibana!
-              // It should be caught, logged and passed on as a stream error.
+              // It should be caught and logged to the Kibana server console.
               throw new Error('There was a (simulated) server side error!');
             } else if (randomAction === 'emit-error') {
-              // Directly emit an error to the stream, this will not be logged.
-              error('Error pushed to the stream');
+              // Emit an error as a stream action.
+              push(errorAction('(Simulated) error pushed to the stream'));
               return;
             }
 
             pushStreamUpdate();
           } catch (e) {
             logger.error(e);
-            error(e);
           }
         }, Math.floor(Math.random() * maxTimeoutMs));
       }