change API

elastic · Mar 4, 2025 · 1117979 · 1117979
1 parent 1814c60
commit 1117979
Show file tree

Hide file tree

Showing 14 changed files with 91 additions and 78 deletions.
diff --git a/x-pack/platform/packages/shared/kbn-streams-schema/src/fixtures/ingest_stream_config.ts b/x-pack/platform/packages/shared/kbn-streams-schema/src/fixtures/ingest_stream_config.ts
@@ -22,15 +22,5 @@ export const ingestStreamConfig = {
         },
       },
     ],
-    routing: [
-      {
-        name: 'logs.errors',
-        condition: {
-          field: 'log.level',
-          operator: 'eq',
-          value: 'error',
-        },
-      },
-    ],
   },
 };
diff --git a/x-pack/platform/packages/shared/kbn-streams-schema/src/fixtures/wired_stream_config.ts b/x-pack/platform/packages/shared/kbn-streams-schema/src/fixtures/wired_stream_config.ts
@@ -23,22 +23,22 @@ export const wiredStreamConfig = {
         },
       },
     ],
-    routing: [
-      {
-        name: 'logs.errors',
-        condition: {
-          field: 'log.level',
-          operator: 'eq',
-          value: 'error',
-        },
-      },
-    ],
     wired: {
       fields: {
         new_field: {
           type: 'long',
         },
       },
+      routing: [
+        {
+          name: 'logs.errors',
+          condition: {
+            field: 'log.level',
+            operator: 'eq',
+            value: 'error',
+          },
+        },
+      ],
     },
   },
 };
diff --git a/x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/base.ts b/x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/base.ts
@@ -16,12 +16,12 @@ import { IngestStreamLifecycle, ingestStreamLifecycleSchema } from './lifecycle'
 interface IngestBase {
   lifecycle: IngestStreamLifecycle;
   processing: ProcessorDefinition[];
-  routing: RoutingDefinition[];
 }
 
 interface WiredIngest extends IngestBase {
   wired: {
     fields: FieldDefinition;
+    routing: RoutingDefinition[];
   };
 }
 
@@ -50,7 +50,6 @@ type IngestStreamDefinition = WiredStreamDefinition | UnwiredStreamDefinition;
 const ingestBaseSchema: z.Schema<IngestBase> = z.object({
   lifecycle: ingestStreamLifecycleSchema,
   processing: z.array(processorDefinitionSchema),
-  routing: z.array(routingDefinitionSchema),
 });
 
 const unwiredIngestSchema: z.Schema<UnwiredIngest> = z.intersection(
@@ -65,6 +64,7 @@ const wiredIngestSchema: z.Schema<WiredIngest> = z.intersection(
   z.object({
     wired: z.object({
       fields: fieldDefinitionSchema,
+      routing: z.array(routingDefinitionSchema),
     }),
   })
 );

diff --git a/x-pack/platform/plugins/shared/streams/server/lib/streams/client.ts b/x-pack/platform/plugins/shared/streams/server/lib/streams/client.ts
@@ -245,7 +245,7 @@ export class StreamsClient {
     });
 
     if (parentDefinition) {
-      const isRoutingToChild = parentDefinition.ingest.routing.find(
+      const isRoutingToChild = parentDefinition.ingest.wired.routing.find(
         (item) => item.destination === name
       );
 
@@ -255,7 +255,7 @@ export class StreamsClient {
         // The user can set the condition later on the parent
         await this.updateStreamRouting({
           definition: parentDefinition,
-          routing: parentDefinition.ingest.routing.concat({
+          routing: parentDefinition.ingest.wired.routing.concat({
             destination: name,
             if: { never: {} },
           }),
@@ -275,14 +275,14 @@ export class StreamsClient {
               ingest: {
                 lifecycle: { inherit: {} },
                 processing: [],
-                routing: [
-                  {
-                    destination: stream.name,
-                    if: { never: {} },
-                  },
-                ],
                 wired: {
                   fields: {},
+                  routing: [
+                    {
+                      destination: stream.name,
+                      if: { never: {} },
+                    },
+                  ],
                 },
               },
             },
@@ -471,7 +471,7 @@ export class StreamsClient {
       validateStreamChildrenChanges(existingDefinition, definition);
     }
 
-    for (const item of definition.ingest.routing) {
+    for (const item of definition.ingest.wired.routing) {
       if (descendantsById[item.destination]) {
         continue;
       }
@@ -486,9 +486,9 @@ export class StreamsClient {
           ingest: {
             lifecycle: { inherit: {} },
             processing: [],
-            routing: [],
             wired: {
               fields: {},
+              routing: [],
             },
           },
         },
@@ -543,14 +543,21 @@ export class StreamsClient {
     if: Condition;
   }): Promise<ForkStreamResponse> {
     const parentDefinition = asIngestStreamDefinition(await this.getStream(parent));
+    if (!isWiredStreamDefinition(parentDefinition)) {
+      throw new MalformedStreamIdError('Cannot fork a stream that is not managed');
+    }
 
     const childDefinition: WiredStreamDefinition = {
       name,
-      ingest: { lifecycle: { inherit: {} }, processing: [], routing: [], wired: { fields: {} } },
+      ingest: { lifecycle: { inherit: {} }, processing: [], wired: { fields: {}, routing: [] } },
     };
 
     // check whether root stream has a child of the given name already
-    if (parentDefinition.ingest.routing.some((item) => item.destination === childDefinition.name)) {
+    if (
+      parentDefinition.ingest.wired.routing.some(
+        (item) => item.destination === childDefinition.name
+      )
+    ) {
       throw new MalformedStreamIdError(
         `The stream with ID (${name}) already exists as a child of the parent stream`
       );
@@ -569,7 +576,7 @@ export class StreamsClient {
 
     await this.updateStreamRouting({
       definition: updatedParentDefinition!,
-      routing: parentDefinition.ingest.routing.concat({
+      routing: parentDefinition.ingest.wired.routing.concat({
         destination: name,
         if: condition,
       }),
@@ -696,7 +703,6 @@ export class StreamsClient {
       name: dataStream.name,
       ingest: {
         lifecycle: { inherit: {} },
-        routing: [],
         processing: [],
         unwired: {},
       },
@@ -820,15 +826,15 @@ export class StreamsClient {
 
         await this.updateStreamRouting({
           definition: parentDefinition,
-          routing: parentDefinition.ingest.routing.filter(
+          routing: parentDefinition.ingest.wired.routing.filter(
             (item) => item.destination !== definition.name
           ),
         });
       }
 
       // delete the children first, as this will update
       // the parent as well
-      for (const item of definition.ingest.routing) {
+      for (const item of definition.ingest.wired.routing) {
         await this.deleteStream(item.destination);
       }
 
@@ -856,10 +862,10 @@ export class StreamsClient {
     routing,
   }: {
     definition: WiredStreamDefinition;
-    routing: WiredStreamDefinition['ingest']['routing'];
+    routing: WiredStreamDefinition['ingest']['wired']['routing'];
   }) {
     const update = cloneDeep(definition);
-    update.ingest.routing = routing;
+    update.ingest.wired.routing = routing;
 
     await this.updateStoredStream(update);
 

diff --git a/x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/sync.ts b/x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/sync.ts
@@ -241,9 +241,6 @@ export async function syncUnwiredStreamDefinitionObjects({
   dataStream: IndicesDataStream;
   definition: UnwiredStreamDefinition;
 }) {
-  if (definition.ingest.routing.length) {
-    throw new Error('Unmanaged streams cannot have managed children, coming soon');
-  }
   const unmanagedAssets = await getUnmanagedElasticsearchAssets({
     dataStream,
     scopedClusterClient,

diff --git a/x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/validate_stream.ts b/x-pack/platform/plugins/shared/streams/server/lib/streams/helpers/validate_stream.ts
@@ -91,11 +91,11 @@ export function validateStreamChildrenChanges(
   currentStreamDefinition: WiredStreamDefinition,
   nextStreamDefinition: WiredStreamDefinition
 ) {
-  const existingChildren = currentStreamDefinition.ingest.routing.map(
+  const existingChildren = currentStreamDefinition.ingest.wired.routing.map(
     (routingDefinition) => routingDefinition.destination
   );
 
-  const nextChildren = nextStreamDefinition.ingest.routing.map(
+  const nextChildren = nextStreamDefinition.ingest.wired.routing.map(
     (routingDefinition) => routingDefinition.destination
   );
 

diff --git a/...m/plugins/shared/streams/server/lib/streams/ingest_pipelines/generate_reroute_pipeline.ts b/...m/plugins/shared/streams/server/lib/streams/ingest_pipelines/generate_reroute_pipeline.ts
@@ -5,19 +5,19 @@
  * 2.0.
  */
 
-import { IngestStreamDefinition } from '@kbn/streams-schema';
+import { WiredStreamDefinition } from '@kbn/streams-schema';
 import { ASSET_VERSION } from '../../../../common/constants';
 import { conditionToPainless } from '../helpers/condition_to_painless';
 import { getReroutePipelineName } from './name';
 
 interface GenerateReroutePipelineParams {
-  definition: IngestStreamDefinition;
+  definition: WiredStreamDefinition;
 }
 
 export function generateReroutePipeline({ definition }: GenerateReroutePipelineParams) {
   return {
     id: getReroutePipelineName(definition.name),
-    processors: definition.ingest.routing.map((child) => {
+    processors: definition.ingest.wired.routing.map((child) => {
       return {
         reroute: {
           destination: child.destination,

diff --git a/x-pack/platform/plugins/shared/streams/server/lib/streams/root_stream_definition.ts b/x-pack/platform/plugins/shared/streams/server/lib/streams/root_stream_definition.ts
@@ -14,8 +14,8 @@ export const rootStreamDefinition: WiredStreamDefinition = {
   ingest: {
     lifecycle: { dsl: {} },
     processing: [],
-    routing: [],
     wired: {
+      routing: [],
       fields: {
         '@timestamp': {
           type: 'date',

diff --git a/...ared/streams_app/public/components/data_management/schema_editor/flyout/field_summary.tsx b/...ared/streams_app/public/components/data_management/schema_editor/flyout/field_summary.tsx
@@ -203,9 +203,9 @@ export const FieldSummary = (props: FieldSummaryProps) => {
 
         <EuiHorizontalRule margin="xs" />
       </EuiFlexGroup>
-      {isEditing && stream.ingest.routing.length > 0 ? (
+      {isEditing && stream.ingest.wired.routing.length > 0 ? (
         <EuiFlexItem grow={false}>
-          <ChildrenAffectedCallout childStreams={stream.ingest.routing} />
+          <ChildrenAffectedCallout childStreams={stream.ingest.wired.routing} />
         </EuiFlexItem>
       ) : null}
     </>

diff --git a/...treams_app/public/components/data_management/stream_detail_routing/hooks/routing_state.ts b/...treams_app/public/components/data_management/stream_detail_routing/hooks/routing_state.ts
@@ -39,15 +39,15 @@ export function useRoutingState({
 
   // Child streams: either represents the child streams as they are, or the new order from drag and drop.
   const [childStreams, setChildStreams] = React.useState<
-    WiredStreamGetResponse['stream']['ingest']['routing']
-  >(definition?.stream.ingest.routing ?? []);
+    WiredStreamGetResponse['stream']['ingest']['wired']['routing']
+  >(definition?.stream.ingest.wired.routing ?? []);
 
   useEffect(() => {
-    setChildStreams(definition?.stream.ingest.routing ?? []);
+    setChildStreams(definition?.stream.ingest.wired.routing ?? []);
   }, [definition]);
 
   // Note: just uses reference equality to check if the order has changed as onChildStreamReorder will create a new array.
-  const hasChildStreamsOrderChanged = childStreams !== definition?.stream.ingest.routing;
+  const hasChildStreamsOrderChanged = childStreams !== definition?.stream.ingest.wired.routing;
 
   // Child stream currently being dragged
   const [draggingChildStream, setDraggingChildStream] = React.useState<string | undefined>();
@@ -73,8 +73,8 @@ export function useRoutingState({
 
   const cancelChanges = useCallback(() => {
     setChildUnderEdit(undefined);
-    setChildStreams(definition?.stream.ingest.routing ?? []);
-  }, [definition?.stream.ingest.routing]);
+    setChildStreams(definition?.stream.ingest.wired.routing ?? []);
+  }, [definition?.stream.ingest.wired.routing]);
 
   const debouncedChildUnderEdit = useDebounced(childUnderEdit, 300);
 

diff --git a/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/classic.ts b/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/classic.ts
@@ -50,7 +50,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
         ingest: {
           lifecycle: { inherit: {} },
           processing: [],
-          routing: [],
           unwired: {},
         },
       });
@@ -67,7 +66,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
             stream: {
               ingest: {
                 lifecycle: { inherit: {} },
-                routing: [],
                 processing: [
                   {
                     grok: {
@@ -122,7 +120,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
               },
             },
           ],
-          routing: [],
           unwired: {},
         },
       });
@@ -188,7 +185,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
               ingest: {
                 lifecycle: { inherit: {} },
                 processing: [],
-                routing: [],
                 unwired: {},
               },
             },
@@ -284,7 +280,6 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
               stream: {
                 ingest: {
                   lifecycle: { inherit: {} },
-                  routing: [],
                   processing: [
                     {
                       grok: {

diff --git a/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/enrichment.ts b/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/enrichment.ts
@@ -77,8 +77,8 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
                 },
               },
             ],
-            routing: [],
             wired: {
+              routing: [],
               fields: {
                 '@timestamp': {
                   type: 'date',

diff --git a/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/flush_config.ts b/x-pack/test/api_integration/deployment_agnostic/apis/observability/streams/flush_config.ts
@@ -8,6 +8,7 @@
 import expect from '@kbn/expect';
 import {
   isGroupStreamDefinitionBase,
+  isUnwiredStreamDefinition,
   StreamGetResponse,
   WiredStreamGetResponse,
 } from '@kbn/streams-schema';
@@ -40,8 +41,8 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
 
     it('checks whether deeply nested stream is created correctly', async () => {
       function getChildNames(stream: StreamGetResponse['stream']): string[] {
-        if (isGroupStreamDefinitionBase(stream)) return [];
-        return stream.ingest.routing.map((r) => r.destination);
+        if (isGroupStreamDefinitionBase(stream) || isUnwiredStreamDefinition(stream)) return [];
+        return stream.ingest.wired.routing.map((r) => r.destination);
       }
       const logs = await apiClient.fetch('GET /api/streams/{name}', {
         params: {