[Fleet] add ilm policy per data stream (#85492)

Co-authored-by: kevinlog <kevin.logan@elastic.co>
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/fleet/common/types/models/epm.ts

@@ -65,6 +65,7 @@ export enum ElasticsearchAssetType {
   indexTemplate = 'index_template',
   ilmPolicy = 'ilm_policy',
   transform = 'transform',
+  dataStreamIlmPolicy = 'data_stream_ilm_policy',
 }
 
 export type DataType = typeof dataTypes;
@@ -207,6 +208,7 @@ export type ElasticsearchAssetTypeToParts = Record<
 
 export interface RegistryDataStream {
   type: string;
+  ilm_policy?: string;
   hidden?: boolean;
   dataset: string;
   title: string;

x-pack/plugins/fleet/public/applications/fleet/sections/epm/constants.tsx

@@ -27,6 +27,7 @@ export const AssetTitleMap: Record<AssetType, string> = {
   visualization: 'Visualization',
   input: 'Agent input',
   map: 'Map',
+  data_stream_ilm_policy: 'Data Stream ILM Policy',
 };
 
 export const ServiceTitleMap: Record<Extract<ServiceName, 'kibana'>, string> = {

x-pack/plugins/fleet/server/services/epm/elasticsearch/datastream_ilm/install.ts

@@ -0,0 +1,137 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { SavedObjectsClientContract } from 'kibana/server';
+import {
+  ElasticsearchAssetType,
+  EsAssetReference,
+  InstallablePackage,
+  RegistryDataStream,
+} from '../../../../../common/types/models';
+import { CallESAsCurrentUser } from '../../../../types';
+import { getInstallation } from '../../packages';
+import { deleteIlmRefs, deleteIlms } from './remove';
+import { saveInstalledEsRefs } from '../../packages/install';
+import { getAsset } from '../transform/common';
+
+interface IlmInstallation {
+  installationName: string;
+  content: string;
+}
+
+interface IlmPathDataset {
+  path: string;
+  dataStream: RegistryDataStream;
+}
+
+export const installIlmForDataStream = async (
+  registryPackage: InstallablePackage,
+  paths: string[],
+  callCluster: CallESAsCurrentUser,
+  savedObjectsClient: SavedObjectsClientContract
+) => {
+  const installation = await getInstallation({ savedObjectsClient, pkgName: registryPackage.name });
+  let previousInstalledIlmEsAssets: EsAssetReference[] = [];
+  if (installation) {
+    previousInstalledIlmEsAssets = installation.installed_es.filter(
+      ({ type, id }) => type === ElasticsearchAssetType.dataStreamIlmPolicy
+    );
+  }
+
+  // delete all previous ilm
+  await deleteIlms(
+    callCluster,
+    previousInstalledIlmEsAssets.map((asset) => asset.id)
+  );
+  // install the latest dataset
+  const dataStreams = registryPackage.data_streams;
+  if (!dataStreams?.length) return [];
+  const dataStreamIlmPaths = paths.filter((path) => isDataStreamIlm(path));
+  let installedIlms: EsAssetReference[] = [];
+  if (dataStreamIlmPaths.length > 0) {
+    const ilmPathDatasets = dataStreams.reduce<IlmPathDataset[]>((acc, dataStream) => {
+      dataStreamIlmPaths.forEach((path) => {
+        if (isDatasetIlm(path, dataStream.path)) {
+          acc.push({ path, dataStream });
+        }
+      });
+      return acc;
+    }, []);
+
+    const ilmRefs = ilmPathDatasets.reduce<EsAssetReference[]>((acc, ilmPathDataset) => {
+      if (ilmPathDataset) {
+        acc.push({
+          id: getIlmNameForInstallation(ilmPathDataset),
+          type: ElasticsearchAssetType.dataStreamIlmPolicy,
+        });
+      }
+      return acc;
+    }, []);
+
+    await saveInstalledEsRefs(savedObjectsClient, registryPackage.name, ilmRefs);
+
+    const ilmInstallations: IlmInstallation[] = ilmPathDatasets.map(
+      (ilmPathDataset: IlmPathDataset) => {
+        return {
+          installationName: getIlmNameForInstallation(ilmPathDataset),
+          content: getAsset(ilmPathDataset.path).toString('utf-8'),
+        };
+      }
+    );
+
+    const installationPromises = ilmInstallations.map(async (ilmInstallation) => {
+      return handleIlmInstall({ callCluster, ilmInstallation });
+    });
+
+    installedIlms = await Promise.all(installationPromises).then((results) => results.flat());
+  }
+
+  if (previousInstalledIlmEsAssets.length > 0) {
+    const currentInstallation = await getInstallation({
+      savedObjectsClient,
+      pkgName: registryPackage.name,
+    });
+
+    // remove the saved object reference
+    await deleteIlmRefs(
+      savedObjectsClient,
+      currentInstallation?.installed_es || [],
+      registryPackage.name,
+      previousInstalledIlmEsAssets.map((asset) => asset.id),
+      installedIlms.map((installed) => installed.id)
+    );
+  }
+  return installedIlms;
+};
+
+async function handleIlmInstall({
+  callCluster,
+  ilmInstallation,
+}: {
+  callCluster: CallESAsCurrentUser;
+  ilmInstallation: IlmInstallation;
+}): Promise<EsAssetReference> {
+  await callCluster('transport.request', {
+    method: 'PUT',
+    path: `/_ilm/policy/${ilmInstallation.installationName}`,
+    body: ilmInstallation.content,
+  });
+
+  return { id: ilmInstallation.installationName, type: ElasticsearchAssetType.dataStreamIlmPolicy };
+}
+
+const isDataStreamIlm = (path: string) => {
+  return new RegExp('(?<package>.*)/data_stream/(?<dataset>.*)/elasticsearch/ilm/*.*').test(path);
+};
+
+const isDatasetIlm = (path: string, datasetName: string) => {
+  return new RegExp(`(?<package>.*)/data_stream\\/${datasetName}/elasticsearch/ilm/*.*`).test(path);
+};
+
+const getIlmNameForInstallation = (ilmPathDataset: IlmPathDataset) => {
+  const filename = ilmPathDataset?.path.split('/')?.pop()?.split('.')[0];
+  return `${ilmPathDataset.dataStream.type}-${ilmPathDataset.dataStream.package}.${ilmPathDataset.dataStream.path}-${filename}`;
+};

x-pack/plugins/fleet/server/services/epm/elasticsearch/datastream_ilm/remove.ts

@@ -0,0 +1,42 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { SavedObjectsClientContract } from 'kibana/server';
+import { CallESAsCurrentUser, ElasticsearchAssetType, EsAssetReference } from '../../../../types';
+import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../../../common/constants';
+
+export const deleteIlms = async (callCluster: CallESAsCurrentUser, ilmPolicyIds: string[]) => {
+  await Promise.all(
+    ilmPolicyIds.map(async (ilmPolicyId) => {
+      await callCluster('transport.request', {
+        method: 'DELETE',
+        path: `_ilm/policy/${ilmPolicyId}`,
+        ignore: [404, 400],
+      });
+    })
+  );
+};
+
+export const deleteIlmRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  installedEsAssets: EsAssetReference[],
+  pkgName: string,
+  installedEsIdToRemove: string[],
+  currentInstalledEsIlmIds: string[]
+) => {
+  const seen = new Set<string>();
+  const filteredAssets = installedEsAssets.filter(({ type, id }) => {
+    if (type !== ElasticsearchAssetType.dataStreamIlmPolicy) return true;
+    const add =
+      (currentInstalledEsIlmIds.includes(id) || !installedEsIdToRemove.includes(id)) &&
+      !seen.has(id);
+    seen.add(id);
+    return add;
+  });
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_es: filteredAssets,
+  });
+};

x-pack/plugins/fleet/server/services/epm/elasticsearch/template/install.ts

@@ -314,6 +314,7 @@ export async function installTemplate({
     pipelineName,
     packageName,
     composedOfTemplates,
+    ilmPolicy: dataStream.ilm_policy,
     hidden: dataStream.hidden,
   });
 

x-pack/plugins/fleet/server/services/epm/elasticsearch/template/template.ts

@@ -45,6 +45,7 @@ export function getTemplate({
   pipelineName,
   packageName,
   composedOfTemplates,
+  ilmPolicy,
   hidden,
 }: {
   type: string;
@@ -53,6 +54,7 @@ export function getTemplate({
   pipelineName?: string | undefined;
   packageName: string;
   composedOfTemplates: string[];
+  ilmPolicy?: string | undefined;
   hidden?: boolean;
 }): IndexTemplate {
   const template = getBaseTemplate(
@@ -61,6 +63,7 @@ export function getTemplate({
     mappings,
     packageName,
     composedOfTemplates,
+    ilmPolicy,
     hidden
   );
   if (pipelineName) {
@@ -263,6 +266,7 @@ function getBaseTemplate(
   mappings: IndexTemplateMappings,
   packageName: string,
   composedOfTemplates: string[],
+  ilmPolicy?: string | undefined,
   hidden?: boolean
 ): IndexTemplate {
   // Meta information to identify Ingest Manager's managed templates and indices
@@ -287,7 +291,7 @@ function getBaseTemplate(
         index: {
           // ILM Policy must be added here, for now point to the default global ILM policy name
           lifecycle: {
-            name: type,
+            name: ilmPolicy ? ilmPolicy : type,
           },
           // What should be our default for the compression?
           codec: 'best_compression',

x-pack/plugins/fleet/server/services/epm/packages/_install_package.ts

@@ -29,6 +29,7 @@ import { updateCurrentWriteIndices } from '../elasticsearch/template/template';
 import { deleteKibanaSavedObjectsAssets } from './remove';
 import { installTransform } from '../elasticsearch/transform/install';
 import { createInstallation, saveKibanaAssetsRefs, updateVersion } from './install';
+import { installIlmForDataStream } from '../elasticsearch/datastream_ilm/install';
 import { saveArchiveEntries } from '../archive/storage';
 import { ConcurrentInstallOperationError } from '../../../errors';
 
@@ -134,6 +135,13 @@ export async function _installPackage({
     // per data stream and we should then save them
     await installILMPolicy(paths, callCluster);
 
+    const installedDataStreamIlm = await installIlmForDataStream(
+      packageInfo,
+      paths,
+      callCluster,
+      savedObjectsClient
+    );
+
     // installs versionized pipelines without removing currently installed ones
     const installedPipelines = await installPipelines(
       packageInfo,
@@ -212,6 +220,7 @@ export async function _installPackage({
     return [
       ...installedKibanaAssetsRefs,
       ...installedPipelines,
+      ...installedDataStreamIlm,
       ...installedTemplateRefs,
       ...installedTransforms,
     ];

x-pack/plugins/fleet/server/services/epm/packages/remove.ts

@@ -23,6 +23,7 @@ import { deleteTransforms } from '../elasticsearch/transform/remove';
 import { packagePolicyService, appContextService } from '../..';
 import { splitPkgKey } from '../registry';
 import { deletePackageCache } from '../archive';
+import { deleteIlms } from '../elasticsearch/datastream_ilm/remove';
 import { removeArchiveEntries } from '../archive/storage';
 
 export async function removeInstallation(options: {
@@ -93,6 +94,8 @@ function deleteESAssets(installedObjects: EsAssetReference[], callCluster: CallE
       return deleteTemplate(callCluster, id);
     } else if (assetType === ElasticsearchAssetType.transform) {
       return deleteTransforms(callCluster, [id]);
+    } else if (assetType === ElasticsearchAssetType.dataStreamIlmPolicy) {
+      return deleteIlms(callCluster, [id]);
     }
   });
 }

x-pack/test/fleet_api_integration/apis/epm/install_remove_assets.ts

@@ -459,6 +459,14 @@ const expectAssetsInstalled = ({
         },
       ],
       installed_es: [
+        {
+          id: 'logs-all_assets.test_logs-all_assets',
+          type: 'data_stream_ilm_policy',
+        },
+        {
+          id: 'metrics-all_assets.test_metrics-all_assets',
+          type: 'data_stream_ilm_policy',
+        },
         {
           id: 'logs-all_assets.test_logs',
           type: 'index_template',
@@ -496,6 +504,7 @@ const expectAssetsInstalled = ({
         { id: '96c6eb85-fe2e-56c6-84be-5fda976796db', type: 'epm-packages-assets' },
         { id: '2d73a161-fa69-52d0-aa09-1bdc691b95bb', type: 'epm-packages-assets' },
         { id: '0a00c2d2-ce63-5b9c-9aa0-0cf1938f7362', type: 'epm-packages-assets' },
+        { id: '691f0505-18c5-57a6-9f40-06e8affbdf7a', type: 'epm-packages-assets' },
         { id: 'b36e6dd0-58f7-5dd0-a286-8187e4019274', type: 'epm-packages-assets' },
         { id: 'f839c76e-d194-555a-90a1-3265a45789e4', type: 'epm-packages-assets' },
         { id: '9af7bbb3-7d8a-50fa-acc9-9dde6f5efca2', type: 'epm-packages-assets' },

x-pack/test/fleet_api_integration/apis/epm/update_assets.ts

@@ -293,6 +293,10 @@ export default function (providerContext: FtrProviderContext) {
           },
         ],
         installed_es: [
+          {
+            id: 'logs-all_assets.test_logs-all_assets',
+            type: 'data_stream_ilm_policy',
+          },
           {
             id: 'logs-all_assets.test_logs-0.2.0',
             type: 'ingest_pipeline',

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/all_assets/0.1.0/data_stream/test_metrics/elasticsearch/ilm_policy/all_assets.json

@@ -0,0 +1,15 @@
+{
+    "policy": {
+        "phases": {
+            "hot": {
+                "min_age": "0ms",
+                "actions": {
+                    "rollover": {
+                        "max_size": "50gb",
+                        "max_age": "30d"
+                    }
+                }
+            }
+        }
+    }
+}