🌊 Disallow manual_ingest_pipeline processor for wired streams (#233902)

As discussed, we don't want to offer manual_ingest_pipeline for wired
streams. This PR makes sure it can only be used for classic streams
(disallowing on the API and hiding from the UI)

* Tests for `manual_ingest_pipeline` were currently done on wired
streams, moved that over to classic streams.
* The test for testing a half-created wired stream was done with a
broken `manual_ingest_pipeline`, which is now not possible anymore -
this PR changes the test to manually delete the streams definition while
leaving the actual data stream in place, simulating the situation of a
restored data stream.

Author: flash1293

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

@@ -8,6 +8,7 @@
 import type { Streams } from '@kbn/streams-schema';
 import { isInheritLifecycle } from '@kbn/streams-schema';
 import { isEqual } from 'lodash';
+import type { StreamlangStep } from '@kbn/streamlang/types/streamlang';
 import { MalformedStreamError } from '../errors/malformed_stream_error';
 import { RootStreamImmutabilityError } from '../errors/root_stream_immutability_error';
 
@@ -42,3 +43,14 @@ export function validateRootStreamChanges(
     throw new MalformedStreamError('Root stream cannot inherit lifecycle');
   }
 }
+
+export function validateNoManualIngestPipelineUsage(steps: StreamlangStep[]) {
+  for (const step of steps) {
+    if ('action' in step && step.action === 'manual_ingest_pipeline') {
+      throw new MalformedStreamError('Manual ingest pipelines are not allowed');
+    }
+    if ('where' in step && step.where && 'steps' in step.where) {
+      validateNoManualIngestPipelineUsage(step.where.steps);
+    }
+  }
+}

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

@@ -34,7 +34,10 @@ import {
   validateDescendantFields,
   validateSystemFields,
 } from '../../helpers/validate_fields';
-import { validateRootStreamChanges } from '../../helpers/validate_stream';
+import {
+  validateNoManualIngestPipelineUsage,
+  validateRootStreamChanges,
+} from '../../helpers/validate_stream';
 import { generateIndexTemplate } from '../../index_templates/generate_index_template';
 import { getIndexTemplateName } from '../../index_templates/name';
 import { generateIngestPipeline } from '../../ingest_pipelines/generate_ingest_pipeline';
@@ -299,6 +302,11 @@ export class WiredStream extends StreamActiveRecord<Streams.WiredStream.Definiti
 
     const existsInStartingState = startingState.has(this._definition.name);
 
+    if (this._changes.processing && this._definition.ingest.processing.steps.length > 0) {
+      // recursively go through all steps to make sure it's not using manual_ingest_pipeline
+      validateNoManualIngestPipelineUsage(this._definition.ingest.processing.steps);
+    }
+
     if (!existsInStartingState) {
       // Check for conflicts
       const { existsAsIndex, existsAsManagedDataStream, existsAsDataStream } =

x-pack/platform/plugins/shared/streams_app/public/components/data_management/stream_detail_enrichment/processors/processor_type_selector.tsx

@@ -13,6 +13,7 @@ import { FormattedMessage } from '@kbn/i18n-react';
 import { useController, useFormContext, useWatch } from 'react-hook-form';
 import type { DocLinksStart } from '@kbn/core/public';
 import type { ProcessorType } from '@kbn/streamlang';
+import { Streams } from '@kbn/streams-schema';
 import { useKibana } from '../../../../hooks/use_kibana';
 import { getDefaultFormStateByType } from '../utils';
 import type { ProcessorFormState } from '../types';
@@ -42,6 +43,9 @@ export const ProcessorTypeSelector = ({
   });
 
   const processorType = useWatch<{ action: ProcessorType }>({ name: 'action' });
+  const isWired = useStreamEnrichmentSelector((snapshot) =>
+    Streams.WiredStream.GetResponse.is(snapshot.context.definition)
+  );
 
   const grokCollection = useStreamEnrichmentSelector((state) => state.context.grokCollection);
 
@@ -54,19 +58,21 @@ export const ProcessorTypeSelector = ({
     reset(formState);
   };
 
+  const selectorOptions = React.useMemo(() => getProcessorTypeSelectorOptions(isWired), [isWired]);
+
   return (
     <EuiFormRow
       fullWidth
       label={i18n.translate(
         'xpack.streams.streamDetailView.managementTab.enrichment.processor.typeSelectorLabel',
         { defaultMessage: 'Processor' }
       )}
-      helpText={getProcessorDescription(core.docLinks)(processorType)}
+      helpText={getProcessorDescription(core.docLinks, isWired)(processorType)}
     >
       <EuiSuperSelect
         data-test-subj="streamsAppProcessorTypeSelector"
         disabled={disabled}
-        options={processorTypeSelectorOptions}
+        options={selectorOptions}
         isInvalid={fieldState.invalid}
         valueOfSelected={field.value}
         onChange={handleChange}
@@ -80,7 +86,7 @@ export const ProcessorTypeSelector = ({
   );
 };
 
-const availableProcessors: TAvailableProcessors = {
+const getAvailableProcessors: (isWired: boolean) => Partial<TAvailableProcessors> = (isWired) => ({
   date: {
     type: 'date',
     inputDisplay: 'Date',
@@ -171,22 +177,29 @@ const availableProcessors: TAvailableProcessors = {
     },
   },
   ...configDrivenProcessors,
-  manual_ingest_pipeline: {
-    type: 'manual_ingest_pipeline',
-    inputDisplay: 'Manual pipeline configuration',
-    getDocUrl: () => (
-      <FormattedMessage
-        id="xpack.streams.streamDetailView.managementTab.enrichment.processor.manualIngestPipelineHelpText"
-        defaultMessage="Specify an array of ingest pipeline processors using JSON."
-      />
-    ),
-  },
-};
+  ...(isWired
+    ? {}
+    : {
+        manual_ingest_pipeline: {
+          type: 'manual_ingest_pipeline',
+          inputDisplay: 'Manual pipeline configuration',
+          getDocUrl: () => (
+            <FormattedMessage
+              id="xpack.streams.streamDetailView.managementTab.enrichment.processor.manualIngestPipelineHelpText"
+              defaultMessage="Specify an array of ingest pipeline processors using JSON."
+            />
+          ),
+        },
+      }),
+});
 
-const getProcessorDescription = (docLinks: DocLinksStart) => (type: ProcessorType) => {
-  return availableProcessors[type].getDocUrl(docLinks);
-};
+const getProcessorDescription =
+  (docLinks: DocLinksStart, isWired: boolean) => (type: ProcessorType) => {
+    return getAvailableProcessors(isWired)[type]?.getDocUrl(docLinks);
+  };
 
-const processorTypeSelectorOptions = Object.values(availableProcessors).map(
-  ({ type, inputDisplay }) => ({ value: type, inputDisplay })
-);
+const getProcessorTypeSelectorOptions = (isWired: boolean) =>
+  Object.values(getAvailableProcessors(isWired)).map(({ type, inputDisplay }) => ({
+    value: type,
+    inputDisplay,
+  }));

x-pack/platform/test/api_integration_deployment_agnostic/apis/streams/basic.ts

@@ -595,12 +595,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
 
       after(async () => {
         await disableStreams(apiClient);
-
-        await esClient.indices.deleteDataStream({ name: 'logs.invalid_pipeline' });
-        await esClient.indices.deleteIndexTemplate({ name: 'logs.invalid_pipeline@stream' });
-        await esClient.cluster.deleteComponentTemplate({
-          name: '[email protected]',
-        });
       });
 
       it('inherit fields', async () => {
@@ -678,47 +672,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         );
       });
 
-      it('reports when it fails to create a stream', async () => {
-        const body: Streams.WiredStream.UpsertRequest = {
-          dashboards: [],
-          queries: [],
-          rules: [],
-          stream: {
-            description: 'Should cause a failure due to invalid ingest pipeline',
-            ingest: {
-              lifecycle: { inherit: {} },
-              processing: {
-                steps: [
-                  {
-                    action: 'manual_ingest_pipeline',
-                    processors: [
-                      {
-                        set: {
-                          field: 'fails',
-                          value: 'whatever',
-                          fail: 'because this property is not valid',
-                        },
-                      },
-                    ],
-                  },
-                ],
-              },
-              wired: {
-                fields: {},
-                routing: [],
-              },
-            },
-          },
-        };
-
-        const response = await putStream(apiClient, 'logs.invalid_pipeline', body, 500);
-
-        expect((response as any).message).to.contain('Failed to change state:');
-        expect((response as any).message).to.contain(
-          `The cluster state may be inconsistent. If you experience issues, please use the resync API to restore a consistent state.`
-        );
-      });
-
       it('does not allow super deeply nested streams', async () => {
         const body: Streams.WiredStream.UpsertRequest = {
           dashboards: [],