🌊 Streams: Actually disable routing when status == 'disabled' (elastic#234545)

I noticed that with elastic#231992 , all
the hard work of passing the status of a routing item around is done,
but it's not actually reflected in the ingest pipeline.

This PR is fixing this problem by filtering out all disabled routing
items when constructing the pipeline and adding a test that disabled
routing is actually disabled in practice.

Author: flash1293

x-pack/platform/plugins/shared/streams/server/lib/streams/ingest_pipelines/generate_reroute_pipeline.ts

@@ -17,14 +17,16 @@ interface GenerateReroutePipelineParams {
 export function generateReroutePipeline({ definition }: GenerateReroutePipelineParams) {
   return {
     id: getReroutePipelineName(definition.name),
-    processors: definition.ingest.wired.routing.map((child) => {
-      return {
-        reroute: {
-          destination: child.destination,
-          if: conditionToPainless(child.where),
-        },
-      };
-    }),
+    processors: definition.ingest.wired.routing
+      .filter((child) => child.status !== 'disabled')
+      .map((child) => {
+        return {
+          reroute: {
+            destination: child.destination,
+            if: conditionToPainless(child.where),
+          },
+        };
+      }),
     _meta: {
       description: `Reroute pipeline for the ${definition.name} stream`,
       managed: true,

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

@@ -277,16 +277,21 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         expect(response).to.have.property('acknowledged', true);
       });
 
+      const accessLogDoc = {
+        '@timestamp': '2024-01-01T00:00:20.000Z',
+        message: JSON.stringify({
+          'log.level': 'info',
+          'log.logger': 'nginx',
+          message: 'test',
+        }),
+      };
+
       it('Index an Nginx access log message, should goto logs.nginx.access', async () => {
-        const doc = {
-          '@timestamp': '2024-01-01T00:00:20.000Z',
-          message: JSON.stringify({
-            'log.level': 'info',
-            'log.logger': 'nginx',
-            message: 'test',
-          }),
-        };
-        const result = await indexAndAssertTargetStream(esClient, 'logs.nginx.access', doc);
+        const result = await indexAndAssertTargetStream(
+          esClient,
+          'logs.nginx.access',
+          accessLogDoc
+        );
         expect(result._source).to.eql({
           '@timestamp': '2024-01-01T00:00:20.000Z',
           body: { text: 'test' },
@@ -298,6 +303,33 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         });
       });
 
+      it('Does not index to logs.nginx.access if routing is disabled', async () => {
+        await putStream(apiClient, 'logs.nginx', {
+          ...emptyAssets,
+          stream: {
+            description: '',
+            ingest: {
+              lifecycle: { inherit: {} },
+              processing: {
+                steps: [],
+              },
+              wired: {
+                fields: {},
+                routing: [
+                  {
+                    destination: 'logs.nginx.access',
+                    where: { field: 'severity_text', eq: 'info' },
+                    status: 'disabled',
+                  },
+                ],
+              },
+            },
+          },
+        });
+
+        await indexAndAssertTargetStream(esClient, 'logs.nginx', accessLogDoc);
+      });
+
       it('Fork logs to logs.nginx.error with invalid condition', async () => {
         const body = {
           stream: {