EMT-661: use new metadata current

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

@@ -38,6 +38,7 @@ export enum ElasticsearchAssetType {
   ingestPipeline = 'ingest_pipeline',
   indexTemplate = 'index_template',
   ilmPolicy = 'ilm_policy',
+  transform = 'transform',
 }
 
 export enum AgentAssetType {

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

@@ -19,6 +19,7 @@ export const AssetTitleMap: Record<AssetType, string> = {
   dashboard: 'Dashboard',
   ilm_policy: 'ILM Policy',
   ingest_pipeline: 'Ingest Pipeline',
+  transform: 'Transform',
   'index-pattern': 'Index Pattern',
   index_template: 'Index Template',
   component_template: 'Component Template',

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/transform/common.ts

@@ -0,0 +1,11 @@
+/*
+ * 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 * as Registry from '../../registry';
+
+export const getAsset = (path: string): Buffer => {
+  return Registry.getAsset(path);
+};

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/transform/install.ts

@@ -0,0 +1,164 @@
+/*
+ * 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 { saveInstalledEsRefs } from '../../packages/install';
+import * as Registry from '../../registry';
+import {
+  Dataset,
+  ElasticsearchAssetType,
+  EsAssetReference,
+  RegistryPackage,
+} from '../../../../../common/types/models';
+import { CallESAsCurrentUser } from '../../../../types';
+import { getInstallation } from '../../packages';
+import { deleteTransforms, deleteTransformRefs } from './remove';
+import { getAsset } from './common';
+
+interface TransformInstallation {
+  installationName: string;
+  content: string;
+}
+
+interface TransformPathDataset {
+  path: string;
+  dataset: Dataset;
+}
+
+export const installTransformForDataset = async (
+  registryPackage: RegistryPackage,
+  paths: string[],
+  callCluster: CallESAsCurrentUser,
+  savedObjectsClient: SavedObjectsClientContract
+) => {
+  const installation = await getInstallation({ savedObjectsClient, pkgName: registryPackage.name });
+  let previousInstalledTransformEsAssets: EsAssetReference[] = [];
+  if (installation) {
+    previousInstalledTransformEsAssets = installation.installed_es.filter(
+      ({ type, id }) => type === ElasticsearchAssetType.transform
+    );
+  }
+
+  // delete all previous transform
+  await deleteTransforms(
+    callCluster,
+    previousInstalledTransformEsAssets.map((asset) => asset.id)
+  );
+  // install the latest dataset
+  const datasets = registryPackage.datasets;
+  if (!datasets?.length) return [];
+  const installNameSuffix = `${registryPackage.version}`;
+
+  const transformPaths = paths.filter((path) => isTransform(path));
+  let installedTransforms: EsAssetReference[] = [];
+  if (transformPaths.length > 0) {
+    const transformPathDatasets = datasets.reduce<TransformPathDataset[]>((acc, dataset) => {
+      transformPaths.forEach((path) => {
+        if (isDatasetTransform(path, dataset.path)) {
+          acc.push({ path, dataset });
+        }
+      });
+      return acc;
+    }, []);
+
+    const transformRefs = transformPathDatasets.reduce<EsAssetReference[]>(
+      (acc, transformPathDataset) => {
+        if (transformPathDataset) {
+          acc.push({
+            id: getTransformNameForInstallation(transformPathDataset, installNameSuffix),
+            type: ElasticsearchAssetType.transform,
+          });
+        }
+        return acc;
+      },
+      []
+    );
+
+    // get and save transform refs before installing transforms
+    await saveInstalledEsRefs(savedObjectsClient, registryPackage.name, transformRefs);
+
+    const transforms: TransformInstallation[] = transformPathDatasets.map(
+      (transformPathDataset: TransformPathDataset) => {
+        return {
+          installationName: getTransformNameForInstallation(
+            transformPathDataset,
+            installNameSuffix
+          ),
+          content: getAsset(transformPathDataset.path).toString('utf-8'),
+        };
+      }
+    );
+
+    const installationPromises = transforms.map(async (transform) => {
+      return installTransform({ callCluster, transform });
+    });
+
+    installedTransforms = await Promise.all(installationPromises).then((results) => results.flat());
+  }
+
+  const currentInstallation = await getInstallation({
+    savedObjectsClient,
+    pkgName: registryPackage.name,
+  });
+
+  // remove the saved object reference
+  await deleteTransformRefs(
+    savedObjectsClient,
+    currentInstallation?.installed_es || [],
+    registryPackage.name,
+    previousInstalledTransformEsAssets.map((asset) => asset.id),
+    installedTransforms.map((installed) => installed.id)
+  );
+
+  return installedTransforms;
+};
+
+const isTransform = (path: string) => {
+  const pathParts = Registry.pathParts(path);
+  return pathParts.type === ElasticsearchAssetType.transform;
+};
+
+const isDatasetTransform = (path: string, datasetName: string) => {
+  const pathParts = Registry.pathParts(path);
+  return (
+    !path.endsWith('/') &&
+    pathParts.type === ElasticsearchAssetType.transform &&
+    pathParts.dataset !== undefined &&
+    datasetName === pathParts.dataset
+  );
+};
+
+async function installTransform({
+  callCluster,
+  transform,
+}: {
+  callCluster: CallESAsCurrentUser;
+  transform: TransformInstallation;
+}): Promise<EsAssetReference> {
+  // defer validation on put if the source index is not available
+  await callCluster('transport.request', {
+    method: 'PUT',
+    path: `_transform/${transform.installationName}`,
+    query: 'defer_validation=true',
+    body: transform.content,
+  });
+
+  await callCluster('transport.request', {
+    method: 'POST',
+    path: `_transform/${transform.installationName}/_start`,
+  });
+
+  return { id: transform.installationName, type: ElasticsearchAssetType.transform };
+}
+
+const getTransformNameForInstallation = (
+  transformDataset: TransformPathDataset,
+  suffix: string
+) => {
+  const filename = transformDataset?.path.split('/')?.pop()?.split('.')[0];
+  return `${transformDataset.dataset.type}-${transformDataset.dataset.name}-${filename}-${suffix}`;
+};

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/transform/remove.test.ts

@@ -0,0 +1,69 @@
+/*
+ * 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';
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { savedObjectsClientMock } from '../../../../../../../../src/core/server/saved_objects/service/saved_objects_client.mock';
+import { deleteTransformRefs } from './remove';
+import { EsAssetReference } from '../../../../../common/types/models';
+
+describe('test transform install', () => {
+  let savedObjectsClient: jest.Mocked<SavedObjectsClientContract>;
+  beforeEach(() => {
+    savedObjectsClient = savedObjectsClientMock.create();
+  });
+
+  test('can delete transform ref and handle duplicate when previous version and current version are the same', async () => {
+    await deleteTransformRefs(
+      savedObjectsClient,
+      [
+        { id: 'metrics-endpoint.policy-0.16.0-dev.0', type: 'ingest_pipeline' },
+        { id: 'metrics-endpoint.metadata-current-default-0.16.0-dev.0', type: 'transform' },
+      ] as EsAssetReference[],
+      'endpoint',
+      ['metrics-endpoint.metadata-current-default-0.16.0-dev.0'],
+      ['metrics-endpoint.metadata-current-default-0.16.0-dev.0']
+    );
+    expect(savedObjectsClient.update.mock.calls).toEqual([
+      [
+        'epm-packages',
+        'endpoint',
+        {
+          installed_es: [
+            { id: 'metrics-endpoint.policy-0.16.0-dev.0', type: 'ingest_pipeline' },
+            { id: 'metrics-endpoint.metadata-current-default-0.16.0-dev.0', type: 'transform' },
+          ],
+        },
+      ],
+    ]);
+  });
+
+  test('can delete transform ref when previous version and current version are not the same', async () => {
+    await deleteTransformRefs(
+      savedObjectsClient,
+      [
+        { id: 'metrics-endpoint.policy-0.16.0-dev.0', type: 'ingest_pipeline' },
+        { id: 'metrics-endpoint.metadata-current-default-0.16.0-dev.0', type: 'transform' },
+      ] as EsAssetReference[],
+      'endpoint',
+      ['metrics-endpoint.metadata-current-default-0.15.0-dev.0'],
+      ['metrics-endpoint.metadata-current-default-0.16.0-dev.0']
+    );
+
+    expect(savedObjectsClient.update.mock.calls).toEqual([
+      [
+        'epm-packages',
+        'endpoint',
+        {
+          installed_es: [
+            { id: 'metrics-endpoint.policy-0.16.0-dev.0', type: 'ingest_pipeline' },
+            { id: 'metrics-endpoint.metadata-current-default-0.16.0-dev.0', type: 'transform' },
+          ],
+        },
+      ],
+    ]);
+  });
+});

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/transform/remove.ts

@@ -0,0 +1,58 @@
+/*
+ * 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 stopTransforms = async (transformIds: string[], callCluster: CallESAsCurrentUser) => {
+  for (const transformId of transformIds) {
+    await callCluster('transport.request', {
+      method: 'POST',
+      path: `_transform/${transformId}/_stop`,
+      query: 'force=true',
+      ignore: [404],
+    });
+  }
+};
+
+export const deleteTransforms = async (
+  callCluster: CallESAsCurrentUser,
+  transformIds: string[]
+) => {
+  await Promise.all(
+    transformIds.map(async (transformId) => {
+      await stopTransforms([transformId], callCluster);
+      await callCluster('transport.request', {
+        method: 'DELETE',
+        query: 'force=true',
+        path: `_transform/${transformId}`,
+        ignore: [404],
+      });
+    })
+  );
+};
+
+export const deleteTransformRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  installedEsAssets: EsAssetReference[],
+  pkgName: string,
+  installedEsIdToRemove: string[],
+  currentInstalledEsTransformIds: string[]
+) => {
+  const seen = new Set<string>();
+  const filteredAssets = installedEsAssets.filter(({ type, id }) => {
+    if (type !== ElasticsearchAssetType.transform) return true;
+    const add =
+      (currentInstalledEsTransformIds.includes(id) || !installedEsIdToRemove.includes(id)) &&
+      !seen.has(id);
+    seen.add(id);
+    return add;
+  });
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_es: filteredAssets,
+  });
+};