[8.x] 🌊 Add support for existing data streams (#202057) (#202771)

# Backport This will backport the following commits from `main` to `8.x`: - [🌊 Add support for existing data streams (#202057)](#202057)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Chris Cowan <[email protected]>
elastic · Dec 3, 2024 · 60c300e · 60c300e
1 parent f3da0dc
commit 60c300e
Show file tree

Hide file tree

Showing 14 changed files with 258 additions and 18 deletions.
diff --git a/x-pack/plugins/streams/common/types.ts b/x-pack/plugins/streams/common/types.ts
@@ -94,6 +94,15 @@ export type StreamWithoutIdDefinition = z.infer<typeof streamDefinitonSchema>;
 
 export const streamDefinitonSchema = streamWithoutIdDefinitonSchema.extend({
   id: z.string(),
+  managed: z.boolean().default(true),
+  unmanaged_elasticsearch_assets: z.optional(
+    z.array(
+      z.object({
+        type: z.enum(['ingest_pipeline', 'component_template', 'index_template', 'data_stream']),
+        id: z.string(),
+      })
+    )
+  ),
 });
 
 export type StreamDefinition = z.infer<typeof streamDefinitonSchema>;

diff --git a/x-pack/plugins/streams/server/lib/streams/internal_stream_mapping.ts b/x-pack/plugins/streams/server/lib/streams/internal_stream_mapping.ts
@@ -29,6 +29,9 @@ export function createStreamsIndex(scopedClusterClient: IScopedClusterClient) {
         id: {
           type: 'keyword',
         },
+        managed: {
+          type: 'boolean',
+        },
       },
     },
   });

diff --git a/x-pack/plugins/streams/server/lib/streams/root_stream_definition.ts b/x-pack/plugins/streams/server/lib/streams/root_stream_definition.ts
@@ -9,6 +9,7 @@ import { StreamDefinition } from '../../../common/types';
 
 export const rootStreamDefinition: StreamDefinition = {
   id: 'logs',
+  managed: true,
   processing: [],
   children: [],
   fields: [

diff --git a/x-pack/plugins/streams/server/lib/streams/stream_crud.ts b/x-pack/plugins/streams/server/lib/streams/stream_crud.ts
@@ -81,7 +81,7 @@ async function upsertInternalStream({ definition, scopedClusterClient }: BasePar
   return scopedClusterClient.asInternalUser.index({
     id: definition.id,
     index: STREAMS_INDEX,
-    document: definition,
+    document: { ...definition, managed: true },
     refresh: 'wait_for',
   });
 }
@@ -101,15 +101,32 @@ export async function listStreams({
     size: 10000,
     sort: [{ id: 'asc' }],
   });
-  const definitions = response.hits.hits.map((hit) => hit._source!);
+
+  const dataStreams = await listDataStreamsAsStreams({ scopedClusterClient });
+  const definitions = response.hits.hits.map((hit) => ({ ...hit._source!, managed: true }));
   const total = response.hits.total!;
 
   return {
-    definitions,
-    total: typeof total === 'number' ? total : total.value,
+    definitions: [...definitions, ...dataStreams],
+    total: (typeof total === 'number' ? total : total.value) + dataStreams.length,
   };
 }
 
+export async function listDataStreamsAsStreams({
+  scopedClusterClient,
+}: ListStreamsParams): Promise<StreamDefinition[]> {
+  const response = await scopedClusterClient.asInternalUser.indices.getDataStream();
+  return response.data_streams
+    .filter((dataStream) => dataStream.template.endsWith('@stream') === false)
+    .map((dataStream) => ({
+      id: dataStream.name,
+      managed: false,
+      children: [],
+      fields: [],
+      processing: [],
+    }));
+}
+
 interface ReadStreamParams extends BaseParams {
   id: string;
   skipAccessCheck?: boolean;
@@ -137,16 +154,80 @@ export async function readStream({
       }
     }
     return {
-      definition,
+      definition: {
+        ...definition,
+        managed: true,
+      },
     };
   } catch (e) {
     if (e.meta?.statusCode === 404) {
-      throw new DefinitionNotFound(`Stream definition for ${id} not found.`);
+      return readDataStreamAsStream({ id, scopedClusterClient, skipAccessCheck });
     }
     throw e;
   }
 }
 
+export async function readDataStreamAsStream({
+  id,
+  scopedClusterClient,
+  skipAccessCheck,
+}: ReadStreamParams) {
+  const response = await scopedClusterClient.asInternalUser.indices.getDataStream({ name: id });
+  if (response.data_streams.length === 1) {
+    const definition: StreamDefinition = {
+      id,
+      managed: false,
+      children: [],
+      fields: [],
+      processing: [],
+    };
+    if (!skipAccessCheck) {
+      const hasAccess = await checkReadAccess({ id, scopedClusterClient });
+      if (!hasAccess) {
+        throw new DefinitionNotFound(`Stream definition for ${id} not found.`);
+      }
+    }
+    // retrieve linked index template, component template and ingest pipeline
+    const templateName = response.data_streams[0].template;
+    const componentTemplates: string[] = [];
+    const template = await scopedClusterClient.asInternalUser.indices.getIndexTemplate({
+      name: templateName,
+    });
+    if (template.index_templates.length) {
+      template.index_templates[0].index_template.composed_of.forEach((componentTemplateName) => {
+        componentTemplates.push(componentTemplateName);
+      });
+    }
+    const writeIndexName = response.data_streams[0].indices.at(-1)?.index_name!;
+    const currentIndex = await scopedClusterClient.asInternalUser.indices.get({
+      index: writeIndexName,
+    });
+    const ingestPipelineId = currentIndex[writeIndexName].settings?.index?.default_pipeline!;
+
+    definition.unmanaged_elasticsearch_assets = [
+      {
+        type: 'ingest_pipeline',
+        id: ingestPipelineId,
+      },
+      ...componentTemplates.map((componentTemplateName) => ({
+        type: 'component_template' as const,
+        id: componentTemplateName,
+      })),
+      {
+        type: 'index_template',
+        id: templateName,
+      },
+      {
+        type: 'data_stream',
+        id,
+      },
+    ];
+
+    return { definition };
+  }
+  throw new DefinitionNotFound(`Stream definition for ${id} not found.`);
+}
+
 interface ReadAncestorsParams extends BaseParams {
   id: string;
 }
@@ -285,6 +366,10 @@ export async function syncStream({
   rootDefinition,
   logger,
 }: SyncStreamParams) {
+  if (!definition.managed) {
+    // TODO For now, we just don't allow reads at all - later on we will relax this to allow certain operations, but they will use a completely different syncing logic
+    throw new Error('Cannot sync an unmanaged stream');
+  }
   const componentTemplate = generateLayer(definition.id, definition);
   await upsertComponent({
     esClient: scopedClusterClient.asCurrentUser,

diff --git a/x-pack/plugins/streams/server/routes/streams/edit.ts b/x-pack/plugins/streams/server/routes/streams/edit.ts
@@ -76,6 +76,7 @@ export const editStreamRoute = createServerRoute({
           children: [],
           fields: [],
           processing: [],
+          managed: true,
         };
 
         await syncStream({
@@ -87,7 +88,7 @@ export const editStreamRoute = createServerRoute({
 
       await syncStream({
         scopedClusterClient,
-        definition: { ...streamDefinition, id: params.path.id },
+        definition: { ...streamDefinition, id: params.path.id, managed: true },
         rootDefinition: parentDefinition,
         logger,
       });

diff --git a/x-pack/plugins/streams/server/routes/streams/fork.ts b/x-pack/plugins/streams/server/routes/streams/fork.ts
@@ -55,7 +55,7 @@ export const forkStreamsRoute = createServerRoute({
         id: params.path.id,
       });
 
-      const childDefinition = { ...params.body.stream, children: [] };
+      const childDefinition = { ...params.body.stream, children: [], managed: true };
 
       // check whether root stream has a child of the given name already
       if (rootDefinition.children.some((child) => child.id === childDefinition.id)) {

diff --git a/x-pack/plugins/streams/server/routes/streams/list.ts b/x-pack/plugins/streams/server/routes/streams/list.ts
@@ -52,9 +52,11 @@ export interface StreamTree {
   children: StreamTree[];
 }
 
-function asTrees(definitions: Array<{ id: string }>) {
+function asTrees(definitions: Array<{ id: string; managed?: boolean }>) {
   const trees: StreamTree[] = [];
-  const ids = definitions.map((definition) => definition.id);
+  const ids = definitions
+    .filter((definition) => definition.managed)
+    .map((definition) => definition.id);
 
   ids.sort((a, b) => a.split('.').length - b.split('.').length);
 

diff --git a/x-pack/plugins/streams/server/routes/streams/read.ts b/x-pack/plugins/streams/server/routes/streams/read.ts
@@ -45,6 +45,13 @@ export const readStreamRoute = createServerRoute({
         id: params.path.id,
       });
 
+      if (streamEntity.definition.managed === false) {
+        return {
+          ...streamEntity.definition,
+          inheritedFields: [],
+        };
+      }
+
       const { ancestors } = await readAncestors({
         id: streamEntity.definition.id,
         scopedClusterClient,

diff --git a/x-pack/plugins/streams_app/public/components/entity_detail_view/index.tsx b/x-pack/plugins/streams_app/public/components/entity_detail_view/index.tsx
@@ -4,9 +4,10 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-import { EuiFlexGroup, EuiFlexItem, EuiIcon, EuiLink, EuiPanel } from '@elastic/eui';
+import { EuiFlexGroup, EuiFlexItem, EuiIcon, EuiLink, EuiPanel, EuiBadge } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
 import React from 'react';
+import { StreamDefinition } from '@kbn/streams-plugin/common';
 import { useStreamsAppBreadcrumbs } from '../../hooks/use_streams_app_breadcrumbs';
 import { useStreamsAppRouter } from '../../hooks/use_streams_app_router';
 import { EntityOverviewTabList } from '../entity_overview_tab_list';
@@ -25,13 +26,15 @@ export function EntityDetailViewWithoutParams({
   selectedTab,
   tabs,
   entity,
+  definition,
 }: {
   selectedTab: string;
   tabs: EntityViewTab[];
   entity: {
     displayName?: string;
     id: string;
   };
+  definition?: StreamDefinition;
 }) {
   const router = useStreamsAppRouter();
   useStreamsAppBreadcrumbs(() => {
@@ -86,7 +89,26 @@ export function EntityDetailViewWithoutParams({
       <EuiFlexItem grow={false}>
         <StreamsAppPageHeader
           verticalPaddingSize="none"
-          title={<StreamsAppPageHeaderTitle title={entity.displayName} />}
+          title={
+            <StreamsAppPageHeaderTitle
+              title={
+                <>
+                  {entity.displayName}
+                  {definition && !definition.managed ? (
+                    <>
+                      {' '}
+                      <EuiBadge>
+                        {i18n.translate(
+                          'xpack.streams.entityDetailViewWithoutParams.unmanagedBadgeLabel',
+                          { defaultMessage: 'Unmanaged' }
+                        )}
+                      </EuiBadge>
+                    </>
+                  ) : null}
+                </>
+              }
+            />
+          }
         >
           <EntityOverviewTabList
             tabs={Object.entries(tabMap).map(([tabKey, { label, href }]) => {
-Original file line number
+Diff line change
@@ Expand Up @@
             id: {
               type: 'keyword',
             },
+            managed: {
+              type: 'boolean',
+            },
           },
         },
       });
@@ Expand Down @@