🌊 [Group streams] Promote unmanaged Classic streams when added as member (#233896)

## Summary

If a user submits a Group stream that has an unmanaged Classic stream as
a member, we would fail since during validation we check to find all
members in the desired state.
Since we do not load all unmanaged Classic streams into the state, this
would fail for a request that is otherwise valid.

In this PR, I'm adding a cascading effect that "promotes" any unmanaged
Classic stream to a managed Classic stream if it is added as a member of
a Group stream.

Author: miltonhultgren

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

@@ -45,7 +45,51 @@ export class GroupStream extends StreamActiveRecord<Streams.GroupStream.Definiti
 
     this._definition = definition;
 
-    return { cascadingChanges: [], changeStatus: 'upserted' };
+    const missingMembers: string[] = [];
+    for (const member of this._definition.group.members) {
+      if (!desiredState.has(member)) {
+        missingMembers.push(member);
+      }
+    }
+
+    const existsAsDataStream = await Promise.all(
+      missingMembers.map(async (member) => {
+        try {
+          const dataStreamResult =
+            await this.dependencies.scopedClusterClient.asCurrentUser.indices.getDataStream({
+              name: member,
+            });
+
+          return dataStreamResult.data_streams.length > 0 ? member : null;
+        } catch (error) {
+          if (!isNotFoundError(error)) {
+            throw error;
+          }
+          return null;
+        }
+      })
+    );
+
+    const cascadingChanges: StreamChange[] = existsAsDataStream
+      .filter((member): member is string => member !== null)
+      .map((member) => ({
+        type: 'upsert',
+        definition: {
+          name: member,
+          description: '',
+          ingest: {
+            classic: {},
+            lifecycle: {
+              inherit: {},
+            },
+            processing: {
+              steps: [],
+            },
+          },
+        },
+      }));
+
+    return { cascadingChanges, changeStatus: 'upserted' };
   }
 
   protected async doHandleDeleteChange(
@@ -115,17 +159,25 @@ export class GroupStream extends StreamActiveRecord<Streams.GroupStream.Definiti
       };
     }
 
+    const missingMembers: string[] = [];
     for (const member of this._definition.group.members) {
       const relatedStream = desiredState.get(member);
       if (!relatedStream || relatedStream.isDeleted()) {
-        return {
-          isValid: false,
-          errors: [
-            new Error(`Group stream ${this.name} has ${member} as a member which was not found`),
-          ],
-        };
+        missingMembers.push(member);
       }
     }
+    if (missingMembers.length > 0) {
+      return {
+        isValid: false,
+        errors: [
+          new Error(
+            `Group stream ${
+              this.name
+            } has the following members which were not found: ${missingMembers.join(', ')}`
+          ),
+        ],
+      };
+    }
 
     const duplicates = this._definition.group.members.filter(
       (name, index) => this._definition.group.members.indexOf(name) !== index

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

@@ -19,6 +19,8 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
   const esClient = getService('es');
 
   let apiClient: StreamsSupertestRepositoryClient;
+  const existingDataStreamName = 'logs-existing-datastream';
+  const unmanagedStreamName = 'logs-test-unmanaged';
 
   describe('Group streams', () => {
     before(async () => {
@@ -29,7 +31,7 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
     after(async () => {
       await disableStreams(apiClient);
       await esClient.indices.deleteDataStream({
-        name: 'logs-existing-datastream',
+        name: [existingDataStreamName, unmanagedStreamName],
       });
     });
 
@@ -310,6 +312,38 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
             .expect(400);
         });
 
+        it('promotes unmanaged Classic streams to managed Classic streams when added as a member', async () => {
+          await esClient.indices.createDataStream({
+            name: unmanagedStreamName,
+          });
+
+          await apiClient
+            .fetch('PUT /api/streams/{name} 2023-10-31', {
+              params: {
+                path: { name: 'test-group' },
+                body: {
+                  stream: {
+                    description: 'A Group stream',
+                    group: {
+                      tags: [],
+                      members: [unmanagedStreamName],
+                    },
+                  },
+                  dashboards: [],
+                  rules: [],
+                  queries: [],
+                },
+              },
+            })
+            .expect(200);
+
+          const { found } = await esClient.get({
+            index: '.kibana_streams',
+            id: unmanagedStreamName,
+          });
+          expect(found).to.be(true);
+        });
+
         it('cannot create a Group stream with duplicated relationships', async () => {
           await apiClient
             .fetch('PUT /api/streams/{name} 2023-10-31', {
@@ -335,13 +369,13 @@ export default function ({ getService }: DeploymentAgnosticFtrProviderContext) {
 
         it('cannot overwrite an existing data stream', async () => {
           await esClient.indices.createDataStream({
-            name: 'logs-existing-datastream',
+            name: existingDataStreamName,
           });
 
           await apiClient
             .fetch('PUT /api/streams/{name} 2023-10-31', {
               params: {
-                path: { name: 'logs-existing-datastream' },
+                path: { name: existingDataStreamName },
                 body: {
                   stream: {
                     description: 'A Group stream',