[Observability:Streams] Manual ingest pipeline script validation

x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/validate_fields.ts

@@ -7,6 +7,8 @@
 
 import type { FieldDefinition, Streams } from '@kbn/streams-schema';
 import { isRoot, keepFields, namespacePrefixes } from '@kbn/streams-schema';
+import type { IScopedClusterClient } from '@kbn/core/server';
+import type { Script } from '@elastic/elasticsearch/lib/api/types';
 import { MalformedFieldsError } from '../errors/malformed_fields_error';
 import { baseMappings } from '../component_templates/logs_layer';
 
@@ -86,6 +88,30 @@ export function validateClassicFields(definition: Streams.ClassicStream.Definiti
   }
 }
 
+// TODO - potential issue with script here - ctx is not available in the script
+export async function validateManualIngestPipelineScripts(
+  definition: Streams.ClassicStream.Definition,
+  client: IScopedClusterClient
+) {
+  const ingestPipelineSteps = definition.ingest.processing?.steps ?? [];
+  for (const step of ingestPipelineSteps) {
+    if ('action' in step && step.action === 'manual_ingest_pipeline') {
+      for (const processor of step.processors) {
+        if ('script' in processor) {
+          const script = processor.script as Script;
+          try {
+            await client.asCurrentUser.scriptsPainlessExecute({
+              script,
+            });
+          } catch (error) {
+            throw new MalformedFieldsError(`${error.message}`);
+          }
+        }
+      }
+    }
+  }
+}
+
 export function validateDescendantFields({
   descendants,
   fields,

x-pack/platform/plugins/shared/streams/server/lib/streams/state_management/streams/classic_stream.ts

@@ -35,7 +35,10 @@ import type {
   ValidationResult,
 } from '../stream_active_record/stream_active_record';
 import { StreamActiveRecord } from '../stream_active_record/stream_active_record';
-import { validateClassicFields } from '../../helpers/validate_fields';
+import {
+  validateClassicFields,
+  validateManualIngestPipelineScripts,
+} from '../../helpers/validate_fields';
 import { validateBracketsInFieldNames, validateSettings } from '../../helpers/validate_stream';
 import type { DataStreamMappingsUpdateResponse } from '../../data_streams/manage_data_streams';
 import { formatSettings, settingsUpdateRequiresRollover } from './helpers';
@@ -228,6 +231,11 @@ export class ClassicStream extends StreamActiveRecord<Streams.ClassicStream.Defi
       }
     }
 
+    await validateManualIngestPipelineScripts(
+      this._definition,
+      this.dependencies.scopedClusterClient
+    );
+
     validateClassicFields(this._definition);
     validateBracketsInFieldNames(this._definition);
 

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/processing/simulation_handler.ts

@@ -328,13 +328,14 @@ const executePipelineSimulation = async (
       simulation: simulation as SuccessfulPipelineSimulateResponse,
     };
   } catch (error) {
+    // todo - why were we showing reason for this case rather than error message? error message is more descriptive
     if (error instanceof esErrors.ResponseError) {
-      const { processor_tag, reason } = error.body?.error;
+      const { processor_tag } = error.body?.error;
 
       return {
         status: 'failure',
         error: {
-          message: reason,
+          message: error.message,
           processor_id: processor_tag,
           type: 'generic_simulation_failure',
         },