Create edit ILM flow

x-pack/platform/packages/shared/kbn-streams-schema/src/helpers/converters.test.ts

@@ -103,6 +103,7 @@ describe('Converter Helpers', () => {
           read_failure_store: true,
           manage_failure_store: true,
           view_index_metadata: true,
+          create_snapshot_repository: true,
         },
         effective_failure_store: {
           disabled: {},
@@ -142,6 +143,7 @@ describe('Converter Helpers', () => {
           read_failure_store: true,
           manage_failure_store: true,
           view_index_metadata: true,
+          create_snapshot_repository: true,
         },
         effective_lifecycle: {
           dsl: {},

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/base.ts

@@ -37,6 +37,8 @@ interface IngestStreamPrivileges {
   read_failure_store: boolean;
   // User can manage failure store information
   manage_failure_store: boolean;
+  // User can create snapshot repositories (needed for frozen phase searchable snapshots)
+  create_snapshot_repository: boolean;
 }
 
 const ingestStreamPrivilegesSchema: z.Schema<IngestStreamPrivileges> = z.object({
@@ -48,6 +50,7 @@ const ingestStreamPrivilegesSchema: z.Schema<IngestStreamPrivileges> = z.object(
   text_structure: z.boolean(),
   read_failure_store: z.boolean(),
   manage_failure_store: z.boolean(),
+  create_snapshot_repository: z.boolean(),
 });
 
 export interface IngestBase {

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/classic.test.ts

@@ -120,6 +120,7 @@ describe('ClassicStream', () => {
           text_structure: true,
           read_failure_store: true,
           manage_failure_store: true,
+          create_snapshot_repository: true,
           view_index_metadata: true,
         },
         data_stream_exists: true,
@@ -157,6 +158,7 @@ describe('ClassicStream', () => {
           text_structure: true,
           failure_store: true,
           view_index_metadata: true,
+          create_snapshot_repository: true,
         },
         data_stream_exists: true,
         dashboards: [],

x-pack/platform/packages/shared/kbn-streams-schema/src/models/ingest/wired.test.ts

@@ -119,6 +119,7 @@ describe('WiredStream', () => {
           read_failure_store: true,
           manage_failure_store: true,
           view_index_metadata: true,
+          create_snapshot_repository: true,
         },
         effective_lifecycle: {
           dsl: {},
@@ -165,6 +166,7 @@ describe('WiredStream', () => {
           text_structure: true,
           failure_store: true,
           view_index_metadata: true,
+          create_snapshot_repository: true,
         },
         dashboards: [],
         queries: [],

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

@@ -546,6 +546,8 @@ export class StreamsClient {
       REQUIRED_MANAGE_PRIVILEGES.push('monitor_text_structure');
     }
 
+    const CREATE_SNAPSHOT_REPOSITORY_CLUSTER_PRIVILEGE = 'cluster:admin/repository/put';
+
     const REQUIRED_INDEX_PRIVILEGES = [
       'read',
       'write',
@@ -563,7 +565,7 @@ export class StreamsClient {
 
     const privileges =
       await this.dependencies.scopedClusterClient.asCurrentUser.security.hasPrivileges({
-        cluster: REQUIRED_MANAGE_PRIVILEGES,
+        cluster: [...REQUIRED_MANAGE_PRIVILEGES, CREATE_SNAPSHOT_REPOSITORY_CLUSTER_PRIVILEGE],
         index: [
           {
             names,
@@ -594,6 +596,8 @@ export class StreamsClient {
       text_structure: isServerless ? true : privileges.cluster.monitor_text_structure,
       read_failure_store: names.every((name) => privileges.index[name].read_failure_store),
       manage_failure_store: names.every((name) => privileges.index[name].manage_failure_store),
+      create_snapshot_repository:
+        privileges.cluster[CREATE_SNAPSHOT_REPOSITORY_CLUSTER_PRIVILEGE] === true,
     };
   }
 

x-pack/platform/plugins/shared/streams/server/lib/streams/lifecycle/ilm_policies.test.ts

@@ -7,6 +7,15 @@
 
 import { buildPolicyUsage, normalizeIlmPhases } from './ilm_policies';
 
+const isRecord = (value: unknown): value is Record<string, unknown> =>
+  value !== null && typeof value === 'object';
+
+const getActions = (phase: unknown): Record<string, unknown> | undefined => {
+  if (!isRecord(phase)) return undefined;
+  const actions = phase.actions;
+  return isRecord(actions) ? actions : undefined;
+};
+
 describe('lifecycle helpers', () => {
   describe('buildPolicyUsage', () => {
     it('derives data streams from backing indices', () => {
@@ -67,17 +76,40 @@ describe('lifecycle helpers', () => {
       expect(normalizeIlmPhases(undefined)).toEqual({});
     });
 
-    it('drops undefined phases', () => {
+    it('normalizes ES ILM phases into Streams ILM phases', () => {
       const normalized = normalizeIlmPhases({
-        hot: { actions: { rollover: { max_age: '1d' } } },
-        warm: { min_age: '2d' },
+        hot: {
+          actions: {
+            rollover: { max_age: '1d' },
+            set_priority: { priority: 100 },
+          },
+        },
+        warm: {
+          min_age: '2d',
+          actions: { readonly: {} },
+        },
         frozen: undefined,
       } as unknown as Parameters<typeof normalizeIlmPhases>[0]);
 
-      expect(normalized).toEqual({
-        hot: { actions: { rollover: { max_age: '1d' } } },
-        warm: { min_age: '2d' },
+      expect(normalized.hot).toMatchObject({
+        name: 'hot',
+        size_in_bytes: 0,
+        rollover: { max_age: '1d' },
       });
+      expect(getActions(normalized.hot)).toEqual({
+        rollover: { max_age: '1d' },
+        set_priority: { priority: 100 },
+      });
+
+      expect(normalized.warm).toMatchObject({
+        name: 'warm',
+        size_in_bytes: 0,
+        min_age: '2d',
+        readonly: true,
+      });
+      expect(getActions(normalized.warm)).toEqual({ readonly: {} });
+
+      expect(normalized.frozen).toBeUndefined();
     });
   });
 });

x-pack/platform/plugins/shared/streams/server/lib/streams/lifecycle/ilm_policies.ts

@@ -5,16 +5,22 @@
  * 2.0.
  */
 
+import type {
+  IlmPhases as EsIlmPhases,
+  IlmPhase as EsIlmPhase,
+} from '@elastic/elasticsearch/lib/api/types';
 import type {
   IlmPolicy,
+  IlmPolicyDeletePhase,
+  IlmPolicyHotPhase,
   IlmPolicyPhase,
-  IlmPolicyPhases,
   IlmPolicyUsage,
+  PhaseName,
 } from '@kbn/streams-schema';
 
 interface IlmPolicyEntry {
   policy?: {
-    phases?: IlmPolicyPhases;
+    phases?: EsIlmPhases;
     _meta?: Record<string, unknown>;
     deprecated?: boolean;
   };
@@ -30,6 +36,141 @@ export interface IlmPoliciesResponse {
   [policyName: string]: IlmPolicyEntry;
 }
 
+const isRecord = (value: unknown): value is Record<string, unknown> =>
+  value !== null && typeof value === 'object';
+
+const asRecord = (value: unknown): Record<string, unknown> => (isRecord(value) ? value : {});
+
+const durationToString = (value: unknown): string | undefined => {
+  if (typeof value === 'string') return value;
+  if (typeof value === 'number' && Number.isFinite(value)) return String(value);
+  if (value == null) return undefined;
+  return String(value);
+};
+
+type PhaseActions = Record<string, unknown>;
+type IlmPolicyPhaseWithActions = IlmPolicyPhase & { actions: PhaseActions };
+type IlmPolicyHotPhaseWithActions = IlmPolicyHotPhase & { actions: PhaseActions };
+type IlmPolicyDeletePhaseWithActions = IlmPolicyDeletePhase & { actions: PhaseActions };
+
+const toStreamsDownsample = (
+  phaseName: Exclude<PhaseName, 'delete'>,
+  phaseMinAge: string | undefined,
+  actions: PhaseActions
+): IlmPolicyPhase['downsample'] | undefined => {
+  const downsampleAction = asRecord(actions.downsample);
+  const fixedInterval = downsampleAction.fixed_interval;
+  if (typeof fixedInterval !== 'string' || fixedInterval.trim() === '') {
+    return undefined;
+  }
+
+  const after = phaseName === 'hot' ? '0ms' : phaseMinAge ?? '0ms';
+  return { after, fixed_interval: fixedInterval };
+};
+
+const toStreamsSearchableSnapshot = (actions: PhaseActions): string | undefined => {
+  const searchableSnapshotAction = asRecord(actions.searchable_snapshot);
+  const repo = searchableSnapshotAction.snapshot_repository;
+  return typeof repo === 'string' && repo.trim() !== '' ? repo : undefined;
+};
+
+const toStreamsHotRollover = (actions: PhaseActions): IlmPolicyHotPhase['rollover'] => {
+  const rolloverAction = asRecord(actions.rollover);
+
+  // ES can send `-1` to indicate "unset" for max_age.
+  const maxAgeRaw = rolloverAction.max_age;
+  const maxAge =
+    typeof maxAgeRaw === 'number' && maxAgeRaw === -1 ? undefined : durationToString(maxAgeRaw);
+
+  const rollover: IlmPolicyHotPhase['rollover'] = {};
+
+  const maxSize = rolloverAction.max_size;
+  if (typeof maxSize === 'string' || typeof maxSize === 'number') {
+    rollover.max_size = maxSize;
+  }
+
+  const maxPrimaryShardSize = rolloverAction.max_primary_shard_size;
+  if (typeof maxPrimaryShardSize === 'string' || typeof maxPrimaryShardSize === 'number') {
+    rollover.max_primary_shard_size = maxPrimaryShardSize;
+  }
+
+  if (maxAge != null) rollover.max_age = maxAge;
+
+  const maxDocs = rolloverAction.max_docs;
+  if (typeof maxDocs === 'number') {
+    rollover.max_docs = maxDocs;
+  }
+
+  const maxPrimaryShardDocs = rolloverAction.max_primary_shard_docs;
+  if (typeof maxPrimaryShardDocs === 'number') {
+    rollover.max_primary_shard_docs = maxPrimaryShardDocs;
+  }
+
+  return rollover;
+};
+
+function toStreamsNonDeletePhase(
+  phaseName: 'hot',
+  phase: EsIlmPhase | undefined
+): IlmPolicyHotPhaseWithActions | undefined;
+function toStreamsNonDeletePhase(
+  phaseName: Exclude<PhaseName, 'delete' | 'hot'>,
+  phase: EsIlmPhase | undefined
+): IlmPolicyPhaseWithActions | undefined;
+function toStreamsNonDeletePhase(
+  phaseName: Exclude<PhaseName, 'delete'>,
+  phase: EsIlmPhase | undefined
+): IlmPolicyHotPhaseWithActions | IlmPolicyPhaseWithActions | undefined {
+  if (!phase) return undefined;
+
+  const actions = { ...asRecord(phase.actions) };
+  const minAge = durationToString(phase.min_age);
+
+  const readonlyEnabled = Object.prototype.hasOwnProperty.call(actions, 'readonly');
+  const searchableSnapshot = toStreamsSearchableSnapshot(actions);
+  const downsample = toStreamsDownsample(phaseName, minAge, actions);
+
+  const base = {
+    name: phaseName,
+    size_in_bytes: 0,
+    ...(minAge ? { min_age: minAge } : {}),
+    ...(downsample ? { downsample } : {}),
+    ...(readonlyEnabled ? { readonly: true } : {}),
+    ...(searchableSnapshot ? { searchable_snapshot: searchableSnapshot } : {}),
+    actions,
+  };
+
+  if (phaseName === 'hot') {
+    return {
+      ...base,
+      name: 'hot',
+      rollover: toStreamsHotRollover(actions),
+    };
+  }
+
+  return base;
+}
+
+const toStreamsDeletePhase = (
+  phase: EsIlmPhase | undefined
+): IlmPolicyDeletePhaseWithActions | undefined => {
+  if (!phase) return undefined;
+
+  const actions = { ...asRecord(phase.actions) };
+  const minAge = durationToString(phase.min_age) ?? '';
+  const deleteAction = asRecord(actions.delete);
+  const deleteSearchableSnapshot = deleteAction.delete_searchable_snapshot;
+
+  return {
+    name: 'delete',
+    min_age: minAge,
+    ...(typeof deleteSearchableSnapshot === 'boolean'
+      ? { delete_searchable_snapshot: deleteSearchableSnapshot }
+      : {}),
+    actions,
+  };
+};
+
 export const buildPolicyUsage = (
   policyEntry: IlmPolicyEntry,
   dataStreamByBackingIndices: Record<string, string> = {}
@@ -45,13 +186,24 @@ export const buildPolicyUsage = (
   return { in_use_by: { data_streams: dataStreams, indices } };
 };
 
-export const normalizeIlmPhases = (phases?: IlmPolicyPhases): IlmPolicy['phases'] => {
+export const normalizeIlmPhases = (phases?: EsIlmPhases): IlmPolicy['phases'] => {
   if (!phases) {
     return {};
   }
 
-  const entries = Object.entries(phases) as Array<[string, IlmPolicyPhase]>;
-  return Object.fromEntries(
-    entries.filter(([, phase]) => phase !== undefined)
-  ) as IlmPolicy['phases'];
+  const hotPhase = toStreamsNonDeletePhase('hot', phases.hot);
+  const warmPhase = toStreamsNonDeletePhase('warm', phases.warm);
+  const coldPhase = toStreamsNonDeletePhase('cold', phases.cold);
+  const frozenPhase = toStreamsNonDeletePhase('frozen', phases.frozen);
+  const deletePhase = toStreamsDeletePhase(phases.delete);
+
+  // Keep output stable: omit missing phases entirely.
+  const normalized: IlmPolicy['phases'] = {};
+  if (hotPhase) normalized.hot = hotPhase;
+  if (warmPhase) normalized.warm = warmPhase;
+  if (coldPhase) normalized.cold = coldPhase;
+  if (frozenPhase) normalized.frozen = frozenPhase;
+  if (deletePhase) normalized.delete = deletePhase;
+
+  return normalized;
 };

x-pack/platform/plugins/shared/streams/server/routes/internal/streams/lifecycle/route.ts

@@ -223,9 +223,39 @@ const lifecycleIlmPoliciesUpdateRoute = createServerRoute({
   },
 });
 
+const lifecycleSnapshotRepositoriesRoute = createServerRoute({
+  endpoint: 'GET /internal/streams/lifecycle/_snapshot_repositories',
+  options: {
+    access: 'internal',
+  },
+  security: {
+    authz: {
+      requiredPrivileges: [STREAMS_API_PRIVILEGES.read],
+    },
+  },
+  params: z.object({}),
+  handler: async ({
+    request,
+    getScopedClients,
+  }): Promise<{ repositories: Array<{ name: string; type: string }> }> => {
+    const { scopedClusterClient } = await getScopedClients({ request });
+    const repositoriesByName = await scopedClusterClient.asCurrentUser.snapshot.getRepository({
+      name: '*',
+    });
+
+    const repositories = Object.entries(repositoriesByName).map(([name, { type }]) => ({
+      name,
+      type: type ?? '',
+    }));
+
+    return { repositories };
+  },
+});
+
 export const internalLifecycleRoutes = {
   ...lifecycleStatsRoute,
   ...lifecycleIlmExplainRoute,
   ...lifecycleIlmPoliciesRoute,
   ...lifecycleIlmPoliciesUpdateRoute,
+  ...lifecycleSnapshotRepositoriesRoute,
 };

x-pack/platform/plugins/shared/streams_app/public/components/data_management/schema_editor/utils.test.ts

@@ -99,6 +99,7 @@ const privileges = {
   text_structure: true,
   read_failure_store: true,
   manage_failure_store: true,
+  create_snapshot_repository: true,
 };
 
 const buildWiredDefinition = (): Streams.WiredStream.GetResponse => ({