🌊 Streams: Show data retention on stream

x-pack/packages/kbn-streams-schema/src/models/common.ts

@@ -116,3 +116,10 @@ export const elasticsearchAssetSchema = z.array(
 );
 
 export type ElasticsearchAsset = z.infer<typeof elasticsearchAssetSchema>;
+
+export const lifecycleSchema = z.discriminatedUnion('type', [
+  z.object({ type: z.literal('dlm'), data_retention: z.optional(z.string()) }),
+  z.object({ type: z.literal('ilm'), policy: z.string() }),
+]);
+
+export type StreamLifecycle = z.infer<typeof lifecycleSchema>;

x-pack/packages/kbn-streams-schema/src/models/read_streams/ingest_read_stream.ts

@@ -7,11 +7,12 @@
 
 import { z } from '@kbn/zod';
 import { ingestStreamDefinitonSchema } from '../streams';
-import { inheritedFieldDefinitionSchema } from '../common';
+import { inheritedFieldDefinitionSchema, lifecycleSchema } from '../common';
 
 export const ingestReadStreamDefinitonSchema = ingestStreamDefinitonSchema
   .extend({
     inherited_fields: inheritedFieldDefinitionSchema.default({}),
+    lifecycle: lifecycleSchema,
   })
   .strict();
 

x-pack/packages/kbn-streams-schema/src/models/read_streams/wired_read_stream.ts

@@ -7,11 +7,12 @@
 
 import { z } from '@kbn/zod';
 import { wiredStreamDefinitonSchema } from '../streams';
-import { inheritedFieldDefinitionSchema } from '../common';
+import { inheritedFieldDefinitionSchema, lifecycleSchema } from '../common';
 
 export const wiredReadStreamDefinitonSchema = wiredStreamDefinitonSchema
   .extend({
     inherited_fields: inheritedFieldDefinitionSchema.default({}),
+    lifecycle: lifecycleSchema,
   })
   .strict();
 

x-pack/platform/packages/shared/index-lifecycle-management/index_lifecycle_management_common_shared/index.ts

@@ -7,3 +7,4 @@
 
 export const ILM_LOCATOR_ID = 'ILM_LOCATOR_ID';
 export * from './src/policies';
+export * from './src/locator';

x-pack/platform/packages/shared/index-lifecycle-management/index_lifecycle_management_common_shared/src/locator.ts

@@ -0,0 +1,12 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+import type { SerializableRecord } from '@kbn/utility-types';
+
+export interface IlmLocatorParams extends SerializableRecord {
+  page: 'policies_list' | 'policy_edit' | 'policy_create';
+  policyName?: string;
+}

x-pack/platform/packages/shared/index-lifecycle-management/index_lifecycle_management_common_shared/tsconfig.json

@@ -16,5 +16,6 @@
     "target/**/*"
   ],
   "kbn_references": [
+    "@kbn/utility-types",
   ]
 }

x-pack/platform/plugins/private/index_lifecycle_management/public/index.ts

@@ -13,6 +13,3 @@ import { IndexLifecycleManagementPlugin } from './plugin';
 export const plugin = (initializerContext: PluginInitializerContext) => {
   return new IndexLifecycleManagementPlugin(initializerContext);
 };
-
-export type { IlmLocatorParams } from './locator';
-export { ILM_LOCATOR_ID } from './locator';

x-pack/platform/plugins/private/index_lifecycle_management/public/locator.ts

@@ -5,10 +5,9 @@
  * 2.0.
  */
 
-import type { SerializableRecord } from '@kbn/utility-types';
 import { ManagementAppLocator } from '@kbn/management-plugin/common';
 import { LocatorDefinition } from '@kbn/share-plugin/public';
-import { ILM_LOCATOR_ID } from '@kbn/index-lifecycle-management-common-shared';
+import { ILM_LOCATOR_ID, IlmLocatorParams } from '@kbn/index-lifecycle-management-common-shared';
 import {
   getPoliciesListPath,
   getPolicyCreatePath,
@@ -18,11 +17,6 @@ import { PLUGIN } from '../common/constants';
 
 export { ILM_LOCATOR_ID };
 
-export interface IlmLocatorParams extends SerializableRecord {
-  page: 'policies_list' | 'policy_edit' | 'policy_create';
-  policyName?: string;
-}
-
 export interface IlmLocatorDefinitionDependencies {
   managementAppLocator: ManagementAppLocator;
 }

x-pack/platform/plugins/private/index_lifecycle_management/tsconfig.json

@@ -28,7 +28,6 @@
     "@kbn/test-jest-helpers",
     "@kbn/core-http-browser-mocks",
     "@kbn/i18n",
-    "@kbn/utility-types",
     "@kbn/analytics",
     "@kbn/es-ui-shared-plugin",
     "@kbn/i18n-react",

x-pack/solutions/observability/plugins/streams/server/lib/streams/component_templates/generate_layer.ts

@@ -12,7 +12,7 @@ import {
 } from '@elastic/elasticsearch/lib/api/types';
 import { WiredStreamDefinition } from '@kbn/streams-schema';
 import { ASSET_VERSION } from '../../../../common/constants';
-import { logsSettings } from './logs_layer';
+import { logsSettings, logsLifecycle } from './logs_layer';
 import { isRoot } from '../helpers/hierarchy';
 import { getComponentTemplateName } from './name';
 
@@ -38,6 +38,7 @@ export function generateLayer(
     name: getComponentTemplateName(id),
     template: {
       settings: isRoot(definition.name) ? logsSettings : {},
+      lifecycle: isRoot(definition.name) ? logsLifecycle : undefined,
       mappings: {
         subobjects: false,
         dynamic: false,

x-pack/solutions/observability/plugins/streams/server/lib/streams/component_templates/logs_layer.ts

@@ -5,13 +5,14 @@
  * 2.0.
  */
 
-import { IndicesIndexSettings } from '@elastic/elasticsearch/lib/api/types';
+import {
+  IndicesIndexSettings,
+  IndicesDataStreamLifecycle,
+} from '@elastic/elasticsearch/lib/api/types';
 
 export const logsSettings: IndicesIndexSettings = {
   index: {
-    lifecycle: {
-      name: 'logs',
-    },
+    mode: 'logsdb',
     codec: 'best_compression',
     mapping: {
       total_fields: {
@@ -21,3 +22,5 @@ export const logsSettings: IndicesIndexSettings = {
     },
   },
 };
+
+export const logsLifecycle: IndicesDataStreamLifecycle = {};

x-pack/solutions/observability/plugins/streams/server/lib/streams/stream_crud.ts

@@ -17,6 +17,9 @@ import {
   ListStreamsResponse,
   isWiredStream,
   FieldDefinition,
+  StreamLifecycle,
+  ReadStreamDefinition,
+  IngestReadStreamDefinition,
 } from '@kbn/streams-schema';
 import { omit } from 'lodash';
 import { STREAMS_INDEX } from '../../../common/constants';
@@ -63,8 +66,9 @@ export async function deleteUnmanagedStreamObjects({
   scopedClusterClient,
   logger,
 }: DeleteStreamParams) {
+  const dataStream = await getDataStream({ name: id, scopedClusterClient });
   const unmanagedAssets = await getUnmanagedElasticsearchAssets({
-    name: id,
+    dataStream,
     scopedClusterClient,
   });
   const pipelineName = unmanagedAssets.find((asset) => asset.type === 'ingest_pipeline')?.id;
@@ -152,7 +156,7 @@ async function upsertInternalStream({ definition, scopedClusterClient }: BasePar
   return scopedClusterClient.asInternalUser.index({
     id: definition.name,
     index: STREAMS_INDEX,
-    document: { ...omit(definition, 'elasticsearch_assets') },
+    document: { ...omit(definition, 'elasticsearch_assets', 'inherited_fields') },
     refresh: 'wait_for',
   });
 }
@@ -169,7 +173,9 @@ export async function listStreams({
   });
 
   const dataStreams = await listDataStreamsAsStreams({ scopedClusterClient });
-  let definitions = response.hits.hits.map((hit) => ({ ...hit._source! }));
+  let definitions: StreamDefinition[] = response.hits.hits.map((hit) => ({
+    ...hit._source!,
+  }));
   const hasAccess = await Promise.all(
     definitions.map((definition) => checkReadAccess({ id: definition.name, scopedClusterClient }))
   );
@@ -188,10 +194,28 @@ export async function listStreams({
   };
 }
 
+function getDataStreamLifecycle(dataStream: IndicesDataStream): StreamLifecycle {
+  if (
+    dataStream.ilm_policy &&
+    (!dataStream.lifecycle || typeof dataStream.prefer_ilm === 'undefined' || dataStream.prefer_ilm)
+  ) {
+    return {
+      type: 'ilm',
+      policy: dataStream.ilm_policy,
+    };
+  }
+  return {
+    type: 'dlm',
+    data_retention: dataStream.lifecycle?.data_retention
+      ? String(dataStream.lifecycle.data_retention)
+      : undefined,
+  };
+}
+
 export async function listDataStreamsAsStreams({
   scopedClusterClient,
 }: ListStreamsParams): Promise<IngestStreamDefinition[]> {
-  const response = await scopedClusterClient.asInternalUser.indices.getDataStream();
+  const response = await scopedClusterClient.asCurrentUser.indices.getDataStream();
   return response.data_streams
     .filter((dataStream) => dataStream.template.endsWith('@stream') === false)
     .map((dataStream) => ({
@@ -214,7 +238,7 @@ export async function readStream({
   id,
   scopedClusterClient,
   skipAccessCheck,
-}: ReadStreamParams): Promise<StreamDefinition> {
+}: ReadStreamParams): Promise<ReadStreamDefinition> {
   try {
     const response = await scopedClusterClient.asInternalUser.get<StreamDefinition>({
       id,
@@ -227,7 +251,12 @@ export async function readStream({
         throw new DefinitionNotFound(`Stream definition for ${id} not found.`);
       }
     }
-    return definition;
+    const dataStream = await getDataStream({ name: id, scopedClusterClient });
+    return {
+      ...definition,
+      inherited_fields: {},
+      lifecycle: getDataStreamLifecycle(dataStream),
+    };
   } catch (e) {
     if (e.meta?.statusCode === 404) {
       return readDataStreamAsStream({ id, scopedClusterClient, skipAccessCheck });
@@ -237,8 +266,11 @@ export async function readStream({
 }
 
 export async function readDataStreamAsStream({ id, scopedClusterClient }: ReadStreamParams) {
-  const definition: IngestStreamDefinition = {
+  const dataStream = await getDataStream({ name: id, scopedClusterClient });
+  const definition: IngestReadStreamDefinition = {
     name: id,
+    lifecycle: getDataStreamLifecycle(dataStream),
+    inherited_fields: {},
     stream: {
       ingest: {
         routing: [],
@@ -248,36 +280,43 @@ export async function readDataStreamAsStream({ id, scopedClusterClient }: ReadSt
   };
 
   definition.elasticsearch_assets = await getUnmanagedElasticsearchAssets({
-    name: id,
+    dataStream,
     scopedClusterClient,
   });
 
   return definition;
 }
 
 interface ReadUnmanagedAssetsParams extends BaseParams {
-  name: string;
+  dataStream: IndicesDataStream;
 }
 
-async function getUnmanagedElasticsearchAssets({
+async function getDataStream({
   name,
   scopedClusterClient,
-}: ReadUnmanagedAssetsParams) {
-  let dataStream: IndicesDataStream;
+}: {
+  name: string;
+  scopedClusterClient: IScopedClusterClient;
+}) {
   try {
-    const response = await scopedClusterClient.asInternalUser.indices.getDataStream({ name });
-    dataStream = response.data_streams[0];
+    const response = await scopedClusterClient.asCurrentUser.indices.getDataStream({ name });
+    return response.data_streams[0];
   } catch (e) {
     if (e.meta?.statusCode === 404) {
       throw new DefinitionNotFound(`Stream definition for ${name} not found.`);
     }
     throw e;
   }
+}
 
+async function getUnmanagedElasticsearchAssets({
+  dataStream,
+  scopedClusterClient,
+}: ReadUnmanagedAssetsParams) {
   // retrieve linked index template, component template and ingest pipeline
   const templateName = dataStream.template;
   const componentTemplates: string[] = [];
-  const template = await scopedClusterClient.asInternalUser.indices.getIndexTemplate({
+  const template = await scopedClusterClient.asCurrentUser.indices.getIndexTemplate({
     name: templateName,
   });
   if (template.index_templates.length) {
@@ -286,7 +325,7 @@ async function getUnmanagedElasticsearchAssets({
     });
   }
   const writeIndexName = dataStream.indices.at(-1)?.index_name!;
-  const currentIndex = await scopedClusterClient.asInternalUser.indices.get({
+  const currentIndex = await scopedClusterClient.asCurrentUser.indices.get({
     index: writeIndexName,
   });
   const ingestPipelineId = currentIndex[writeIndexName].settings?.index?.default_pipeline!;
@@ -306,7 +345,7 @@ async function getUnmanagedElasticsearchAssets({
     },
     {
       type: 'data_stream' as const,
-      id: name,
+      id: dataStream.name,
     },
   ];
 }
@@ -533,8 +572,9 @@ async function syncUnmanagedStream({ scopedClusterClient, definition }: SyncStre
   if (definition.stream.ingest.routing.length) {
     throw new Error('Unmanaged streams cannot have managed children, coming soon');
   }
+  const dataStream = await getDataStream({ name: definition.name, scopedClusterClient });
   const unmanagedAssets = await getUnmanagedElasticsearchAssets({
-    name: definition.name,
+    dataStream,
     scopedClusterClient,
   });
   const executionPlan: ExecutionPlanStep[] = [];

x-pack/solutions/observability/plugins/streams/server/routes/streams/delete.ts

@@ -9,7 +9,7 @@ import { z } from '@kbn/zod';
 import { IScopedClusterClient } from '@kbn/core-elasticsearch-server';
 import { Logger } from '@kbn/logging';
 import { badRequest, internal, notFound } from '@hapi/boom';
-import { isWiredStream } from '@kbn/streams-schema';
+import { isWiredReadStream } from '@kbn/streams-schema';
 import {
   DefinitionNotFound,
   ForkConditionMissing,
@@ -80,7 +80,7 @@ export async function deleteStream(
 ) {
   try {
     const definition = await readStream({ scopedClusterClient, id });
-    if (!isWiredStream(definition)) {
+    if (!isWiredReadStream(definition)) {
       await deleteUnmanagedStreamObjects({ scopedClusterClient, id, logger });
       return;
     }

x-pack/solutions/observability/plugins/streams/server/routes/streams/fork.ts

@@ -7,7 +7,7 @@
 
 import { z } from '@kbn/zod';
 import { badRequest, internal, notFound } from '@hapi/boom';
-import { conditionSchema, isWiredStream, WiredStreamDefinition } from '@kbn/streams-schema';
+import { conditionSchema, isWiredReadStream, WiredStreamDefinition } from '@kbn/streams-schema';
 import {
   DefinitionNotFound,
   ForkConditionMissing,
@@ -58,7 +58,7 @@ export const forkStreamsRoute = createServerRoute({
         id: params.path.id,
       });
 
-      if (!isWiredStream(rootDefinition)) {
+      if (!isWiredReadStream(rootDefinition)) {
         throw new MalformedStreamId('Cannot fork a stream that is not managed');
       }
 

x-pack/solutions/observability/plugins/streams/server/routes/streams/read.ts

@@ -9,8 +9,8 @@ import { z } from '@kbn/zod';
 import { notFound, internal } from '@hapi/boom';
 import {
   FieldDefinitionConfig,
-  isIngestStream,
-  isWiredStream,
+  isIngestReadStream,
+  isWiredReadStream,
   ReadStreamDefinition,
 } from '@kbn/streams-schema';
 import { createServerRoute } from '../create_server_route';
@@ -42,7 +42,7 @@ export const readStreamRoute = createServerRoute({
 
       // TODO: I have no idea why I can just do `isIngestStream` here but when I do,
       // streamEntity becomes `streamEntity: never` in the statements afterwards
-      if (!isWiredStream(streamEntity) && isIngestStream(streamEntity)) {
+      if (!isWiredReadStream(streamEntity) && isIngestReadStream(streamEntity)) {
         return {
           ...streamEntity,
           inherited_fields: {},