[Ingest Manager] Refactor Package Installation (#71521)

* refactor installation to add/remove installed assets as they are added/removed

* update types

* uninstall assets when installation fails

* refactor installation to add/remove installed assets as they are added/removed

* update types

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

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

@@ -229,7 +229,8 @@ export type PackageInfo = Installable<
 >;
 
 export interface Installation extends SavedObjectAttributes {
-  installed: AssetReference[];
+  installed_kibana: KibanaAssetReference[];
+  installed_es: EsAssetReference[];
   es_index_patterns: Record<string, string>;
   name: string;
   version: string;
@@ -246,19 +247,14 @@ export type NotInstalled<T = {}> = T & {
   status: InstallationStatus.notInstalled;
 };
 
-export type AssetReference = Pick<SavedObjectReference, 'id'> & {
-  type: AssetType | IngestAssetType;
-};
+export type AssetReference = KibanaAssetReference | EsAssetReference;
 
-/**
- * Types of assets which can be installed/removed
- */
-export enum IngestAssetType {
-  IlmPolicy = 'ilm_policy',
-  IndexTemplate = 'index_template',
-  ComponentTemplate = 'component_template',
-  IngestPipeline = 'ingest_pipeline',
-}
+export type KibanaAssetReference = Pick<SavedObjectReference, 'id'> & {
+  type: KibanaAssetType;
+};
+export type EsAssetReference = Pick<SavedObjectReference, 'id'> & {
+  type: ElasticsearchAssetType;
+};
 
 export enum DefaultPackages {
   system = 'system',

x-pack/plugins/ingest_manager/server/routes/data_streams/handlers.ts

@@ -122,7 +122,7 @@ export const getListHandler: RequestHandler = async (context, request, response)
       if (pkg !== '' && pkgSavedObject.length > 0 && !packageMetadata[pkg]) {
         // then pick the dashboards from the package saved object
         const dashboards =
-          pkgSavedObject[0].attributes?.installed?.filter(
+          pkgSavedObject[0].attributes?.installed_kibana?.filter(
             (o) => o.type === KibanaAssetType.dashboard
           ) || [];
         // and then pick the human-readable titles from the dashboard saved objects

x-pack/plugins/ingest_manager/server/routes/epm/handlers.ts

@@ -5,6 +5,7 @@
  */
 import { TypeOf } from '@kbn/config-schema';
 import { RequestHandler, CustomHttpResponseOptions } from 'src/core/server';
+import { appContextService } from '../../services';
 import {
   GetInfoResponse,
   InstallPackageResponse,
@@ -29,6 +30,7 @@ import {
   installPackage,
   removeInstallation,
   getLimitedPackages,
+  getInstallationObject,
 } from '../../services/epm/packages';
 
 export const getCategoriesHandler: RequestHandler<
@@ -146,10 +148,12 @@ export const getInfoHandler: RequestHandler<TypeOf<typeof GetInfoRequestSchema.p
 export const installPackageHandler: RequestHandler<TypeOf<
   typeof InstallPackageRequestSchema.params
 >> = async (context, request, response) => {
+  const logger = appContextService.getLogger();
+  const savedObjectsClient = context.core.savedObjects.client;
+  const callCluster = context.core.elasticsearch.legacy.client.callAsCurrentUser;
+  const { pkgkey } = request.params;
+  const [pkgName, pkgVersion] = pkgkey.split('-');
   try {
-    const { pkgkey } = request.params;
-    const savedObjectsClient = context.core.savedObjects.client;
-    const callCluster = context.core.elasticsearch.legacy.client.callAsCurrentUser;
     const res = await installPackage({
       savedObjectsClient,
       pkgkey,
@@ -161,6 +165,17 @@ export const installPackageHandler: RequestHandler<TypeOf<
     };
     return response.ok({ body });
   } catch (e) {
+    try {
+      const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
+      const isUpdate = installedPkg && installedPkg.attributes.version < pkgVersion ? true : false;
+      // if this is a failed install, remove any assets installed
+      if (!isUpdate) {
+        await removeInstallation({ savedObjectsClient, pkgkey, callCluster });
+      }
+    } catch (error) {
+      logger.error(`could not remove assets from failed installation attempt for ${pkgkey}`);
+    }
+
     if (e.isBoom) {
       return response.customError({
         statusCode: e.output.statusCode,

x-pack/plugins/ingest_manager/server/saved_objects/index.ts

@@ -249,7 +249,14 @@ const savedObjectTypes: { [key: string]: SavedObjectsType } = {
           enabled: false,
           type: 'object',
         },
-        installed: {
+        installed_es: {
+          type: 'nested',
+          properties: {
+            id: { type: 'keyword' },
+            type: { type: 'keyword' },
+          },
+        },
+        installed_kibana: {
           type: 'nested',
           properties: {
             id: { type: 'keyword' },

x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/index.ts

@@ -0,0 +1,9 @@
+/*
+ * 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.
+ */
+
+export { installPipelines } from './install';
+
+export { deletePipelines, deletePipeline } from './remove';

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

@@ -4,15 +4,16 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { SavedObjectsClientContract } from 'src/core/server';
 import {
-  AssetReference,
+  EsAssetReference,
   Dataset,
   ElasticsearchAssetType,
-  IngestAssetType,
   RegistryPackage,
 } from '../../../../types';
 import * as Registry from '../../registry';
 import { CallESAsCurrentUser } from '../../../../types';
+import { saveInstalledEsRefs } from '../../packages/install';
 
 interface RewriteSubstitution {
   source: string;
@@ -23,12 +24,16 @@ interface RewriteSubstitution {
 export const installPipelines = async (
   registryPackage: RegistryPackage,
   paths: string[],
-  callCluster: CallESAsCurrentUser
+  callCluster: CallESAsCurrentUser,
+  savedObjectsClient: SavedObjectsClientContract
 ) => {
+  // unlike other ES assets, pipeline names are versioned so after a template is updated
+  // it can be created pointing to the new template, without removing the old one and effecting data
+  // so do not remove the currently installed pipelines here
   const datasets = registryPackage.datasets;
   const pipelinePaths = paths.filter((path) => isPipeline(path));
   if (datasets) {
-    const pipelines = datasets.reduce<Array<Promise<AssetReference[]>>>((acc, dataset) => {
+    const pipelines = datasets.reduce<Array<Promise<EsAssetReference[]>>>((acc, dataset) => {
       if (dataset.ingest_pipeline) {
         acc.push(
           installPipelinesForDataset({
@@ -41,7 +46,8 @@ export const installPipelines = async (
       }
       return acc;
     }, []);
-    return Promise.all(pipelines).then((results) => results.flat());
+    const pipelinesToSave = await Promise.all(pipelines).then((results) => results.flat());
+    return saveInstalledEsRefs(savedObjectsClient, registryPackage.name, pipelinesToSave);
   }
   return [];
 };
@@ -77,7 +83,7 @@ export async function installPipelinesForDataset({
   pkgVersion: string;
   paths: string[];
   dataset: Dataset;
-}): Promise<AssetReference[]> {
+}): Promise<EsAssetReference[]> {
   const pipelinePaths = paths.filter((path) => isDatasetPipeline(path, dataset.path));
   let pipelines: any[] = [];
   const substitutions: RewriteSubstitution[] = [];
@@ -123,7 +129,7 @@ async function installPipeline({
 }: {
   callCluster: CallESAsCurrentUser;
   pipeline: any;
-}): Promise<AssetReference> {
+}): Promise<EsAssetReference> {
   const callClusterParams: {
     method: string;
     path: string;
@@ -146,7 +152,7 @@ async function installPipeline({
   // which we could otherwise use.
   // See src/core/server/elasticsearch/api_types.ts for available endpoints.
   await callCluster('transport.request', callClusterParams);
-  return { id: pipeline.nameForInstallation, type: IngestAssetType.IngestPipeline };
+  return { id: pipeline.nameForInstallation, type: ElasticsearchAssetType.ingestPipeline };
 }
 
 const isDirectory = ({ path }: Registry.ArchiveEntry) => path.endsWith('/');

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

@@ -0,0 +1,60 @@
+/*
+ * 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 'src/core/server';
+import { appContextService } from '../../../';
+import { CallESAsCurrentUser, ElasticsearchAssetType } from '../../../../types';
+import { getInstallation } from '../../packages/get';
+import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../../../common';
+
+export const deletePipelines = async (
+  callCluster: CallESAsCurrentUser,
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  pkgVersion: string
+) => {
+  const logger = appContextService.getLogger();
+  const previousPipelinesPattern = `*-${pkgName}.*-${pkgVersion}`;
+
+  try {
+    await deletePipeline(callCluster, previousPipelinesPattern);
+  } catch (e) {
+    logger.error(e);
+  }
+  try {
+    await deletePipelineRefs(savedObjectsClient, pkgName, pkgVersion);
+  } catch (e) {
+    logger.error(e);
+  }
+};
+
+export const deletePipelineRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  pkgVersion: string
+) => {
+  const installation = await getInstallation({ savedObjectsClient, pkgName });
+  if (!installation) return;
+  const installedEsAssets = installation.installed_es;
+  const filteredAssets = installedEsAssets.filter(({ type, id }) => {
+    if (type !== ElasticsearchAssetType.ingestPipeline) return true;
+    if (!id.includes(pkgVersion)) return true;
+    return false;
+  });
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_es: filteredAssets,
+  });
+};
+export async function deletePipeline(callCluster: CallESAsCurrentUser, id: string): Promise<void> {
+  // '*' shouldn't ever appear here, but it still would delete all ingest pipelines
+  if (id && id !== '*') {
+    try {
+      await callCluster('ingest.deletePipeline', { id });
+    } catch (err) {
+      throw new Error(`error deleting pipeline ${id}`);
+    }
+  }
+}

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

@@ -5,6 +5,7 @@
  */
 
 import Boom from 'boom';
+import { SavedObjectsClientContract } from 'src/core/server';
 import {
   Dataset,
   RegistryPackage,
@@ -17,13 +18,14 @@ import { Field, loadFieldsFromYaml, processFields } from '../../fields/field';
 import { getPipelineNameForInstallation } from '../ingest_pipeline/install';
 import { generateMappings, generateTemplateName, getTemplate } from './template';
 import * as Registry from '../../registry';
+import { removeAssetsFromInstalledEsByType, saveInstalledEsRefs } from '../../packages/install';
 
 export const installTemplates = async (
   registryPackage: RegistryPackage,
+  isUpdate: boolean,
   callCluster: CallESAsCurrentUser,
-  pkgName: string,
-  pkgVersion: string,
-  paths: string[]
+  paths: string[],
+  savedObjectsClient: SavedObjectsClientContract
 ): Promise<TemplateRef[]> => {
   // install any pre-built index template assets,
   // atm, this is only the base package's global index templates
@@ -31,6 +33,12 @@ export const installTemplates = async (
   await installPreBuiltComponentTemplates(paths, callCluster);
   await installPreBuiltTemplates(paths, callCluster);
 
+  // remove package installation's references to index templates
+  await removeAssetsFromInstalledEsByType(
+    savedObjectsClient,
+    registryPackage.name,
+    ElasticsearchAssetType.indexTemplate
+  );
   // build templates per dataset from yml files
   const datasets = registryPackage.datasets;
   if (datasets) {
@@ -46,7 +54,17 @@ export const installTemplates = async (
     }, []);
 
     const res = await Promise.all(installTemplatePromises);
-    return res.flat();
+    const installedTemplates = res.flat();
+    // get template refs to save
+    const installedTemplateRefs = installedTemplates.map((template) => ({
+      id: template.templateName,
+      type: ElasticsearchAssetType.indexTemplate,
+    }));
+
+    // add package installation's references to index templates
+    await saveInstalledEsRefs(savedObjectsClient, registryPackage.name, installedTemplateRefs);
+
+    return installedTemplates;
   }
   return [];
 };

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

@@ -326,9 +326,10 @@ export const updateCurrentWriteIndices = async (
   callCluster: CallESAsCurrentUser,
   templates: TemplateRef[]
 ): Promise<void> => {
-  if (!templates) return;
+  if (!templates.length) return;
 
   const allIndices = await queryIndicesFromTemplates(callCluster, templates);
+  if (!allIndices.length) return;
   return updateAllIndices(allIndices, callCluster);
 };
 
@@ -358,12 +359,12 @@ const getIndices = async (
     method: 'GET',
     path: `/_data_stream/${templateName}-*`,
   });
-  if (res.length) {
-    return res.map((datastream: any) => ({
-      indexName: datastream.indices[datastream.indices.length - 1].index_name,
-      indexTemplate,
-    }));
-  }
+  const dataStreams = res.data_streams;
+  if (!dataStreams.length) return;
+  return dataStreams.map((dataStream: any) => ({
+    indexName: dataStream.indices[dataStream.indices.length - 1].index_name,
+    indexTemplate,
+  }));
 };
 
 const updateAllIndices = async (

x-pack/plugins/ingest_manager/server/services/epm/kibana/assets/install.ts

@@ -0,0 +1,126 @@
+/*
+ * 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 {
+  SavedObject,
+  SavedObjectsBulkCreateObject,
+  SavedObjectsClientContract,
+} from 'src/core/server';
+import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../../../common';
+import * as Registry from '../../registry';
+import { AssetType, KibanaAssetType, AssetReference } from '../../../../types';
+import { deleteKibanaSavedObjectsAssets } from '../../packages/remove';
+import { getInstallationObject, savedObjectTypes } from '../../packages';
+import { saveInstalledKibanaRefs } from '../../packages/install';
+
+type SavedObjectToBe = Required<SavedObjectsBulkCreateObject> & { type: AssetType };
+export type ArchiveAsset = Pick<
+  SavedObject,
+  'id' | 'attributes' | 'migrationVersion' | 'references'
+> & {
+  type: AssetType;
+};
+
+export async function getKibanaAsset(key: string) {
+  const buffer = Registry.getAsset(key);
+
+  // cache values are buffers. convert to string / JSON
+  return JSON.parse(buffer.toString('utf8'));
+}
+
+export function createSavedObjectKibanaAsset(asset: ArchiveAsset): SavedObjectToBe {
+  // convert that to an object
+  return {
+    type: asset.type,
+    id: asset.id,
+    attributes: asset.attributes,
+    references: asset.references || [],
+    migrationVersion: asset.migrationVersion || {},
+  };
+}
+
+// TODO: make it an exhaustive list
+// e.g. switch statement with cases for each enum key returning `never` for default case
+export async function installKibanaAssets(options: {
+  savedObjectsClient: SavedObjectsClientContract;
+  pkgName: string;
+  paths: string[];
+  isUpdate: boolean;
+}): Promise<AssetReference[]> {
+  const { savedObjectsClient, paths, pkgName, isUpdate } = options;
+
+  if (isUpdate) {
+    // delete currently installed kibana saved objects and installation references
+    const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
+    const installedKibanaRefs = installedPkg?.attributes.installed_kibana;
+
+    if (installedKibanaRefs?.length) {
+      await deleteKibanaSavedObjectsAssets(savedObjectsClient, installedKibanaRefs);
+      await deleteKibanaInstalledRefs(savedObjectsClient, pkgName, installedKibanaRefs);
+    }
+  }
+
+  // install the new assets and save installation references
+  const kibanaAssetTypes = Object.values(KibanaAssetType);
+  const installationPromises = kibanaAssetTypes.map((assetType) =>
+    installKibanaSavedObjects({ savedObjectsClient, assetType, paths })
+  );
+  // installKibanaSavedObjects returns AssetReference[], so .map creates AssetReference[][]
+  // call .flat to flatten into one dimensional array
+  const newInstalledKibanaAssets = await Promise.all(installationPromises).then((results) =>
+    results.flat()
+  );
+  await saveInstalledKibanaRefs(savedObjectsClient, pkgName, newInstalledKibanaAssets);
+  return newInstalledKibanaAssets;
+}
+export const deleteKibanaInstalledRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  installedKibanaRefs: AssetReference[]
+) => {
+  const installedAssetsToSave = installedKibanaRefs.filter(({ id, type }) => {
+    const assetType = type as AssetType;
+    return !savedObjectTypes.includes(assetType);
+  });
+
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_kibana: installedAssetsToSave,
+  });
+};
+
+async function installKibanaSavedObjects({
+  savedObjectsClient,
+  assetType,
+  paths,
+}: {
+  savedObjectsClient: SavedObjectsClientContract;
+  assetType: KibanaAssetType;
+  paths: string[];
+}) {
+  const isSameType = (path: string) => assetType === Registry.pathParts(path).type;
+  const pathsOfType = paths.filter((path) => isSameType(path));
+  const kibanaAssets = await Promise.all(pathsOfType.map((path) => getKibanaAsset(path)));
+  const toBeSavedObjects = await Promise.all(
+    kibanaAssets.map((asset) => createSavedObjectKibanaAsset(asset))
+  );
+
+  if (toBeSavedObjects.length === 0) {
+    return [];
+  } else {
+    const createResults = await savedObjectsClient.bulkCreate(toBeSavedObjects, {
+      overwrite: true,
+    });
+    const createdObjects = createResults.saved_objects;
+    const installed = createdObjects.map(toAssetReference);
+    return installed;
+  }
+}
+
+function toAssetReference({ id, type }: SavedObject) {
+  const reference: AssetReference = { id, type: type as KibanaAssetType };
+
+  return reference;
+}

x-pack/plugins/ingest_manager/server/services/epm/packages/get_objects.ts

@@ -1,32 +0,0 @@
-/*
- * 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 { SavedObject, SavedObjectsBulkCreateObject } from 'src/core/server';
-import { AssetType } from '../../../types';
-import * as Registry from '../registry';
-
-type ArchiveAsset = Pick<SavedObject, 'attributes' | 'migrationVersion' | 'references'>;
-type SavedObjectToBe = Required<SavedObjectsBulkCreateObject> & { type: AssetType };
-
-export async function getObject(key: string) {
-  const buffer = Registry.getAsset(key);
-
-  // cache values are buffers. convert to string / JSON
-  const json = buffer.toString('utf8');
-  // convert that to an object
-  const asset: ArchiveAsset = JSON.parse(json);
-
-  const { type, file } = Registry.pathParts(key);
-  const savedObject: SavedObjectToBe = {
-    type,
-    id: file.replace('.json', ''),
-    attributes: asset.attributes,
-    references: asset.references || [],
-    migrationVersion: asset.migrationVersion || {},
-  };
-
-  return savedObject;
-}

x-pack/plugins/ingest_manager/server/services/epm/packages/index.ts

@@ -23,7 +23,7 @@ export {
   SearchParams,
 } from './get';
 
-export { installKibanaAssets, installPackage, ensureInstalledPackage } from './install';
+export { installPackage, ensureInstalledPackage } from './install';
 export { removeInstallation } from './remove';
 
 type RequiredPackage = 'system' | 'endpoint';

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

@@ -4,27 +4,27 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { SavedObject, SavedObjectsClientContract } from 'src/core/server';
+import { SavedObjectsClientContract } from 'src/core/server';
 import Boom from 'boom';
 import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../constants';
 import {
   AssetReference,
   Installation,
-  KibanaAssetType,
   CallESAsCurrentUser,
   DefaultPackages,
+  AssetType,
+  KibanaAssetReference,
+  EsAssetReference,
   ElasticsearchAssetType,
-  IngestAssetType,
 } from '../../../types';
 import { installIndexPatterns } from '../kibana/index_pattern/install';
 import * as Registry from '../registry';
-import { getObject } from './get_objects';
 import { getInstallation, getInstallationObject, isRequiredPackage } from './index';
 import { installTemplates } from '../elasticsearch/template/install';
 import { generateESIndexPatterns } from '../elasticsearch/template/template';
-import { installPipelines } from '../elasticsearch/ingest_pipeline/install';
+import { installPipelines, deletePipelines } from '../elasticsearch/ingest_pipeline/';
 import { installILMPolicy } from '../elasticsearch/ilm/install';
-import { deleteAssetsByType, deleteKibanaSavedObjectsAssets } from './remove';
+import { installKibanaAssets } from '../kibana/assets/install';
 import { updateCurrentWriteIndices } from '../elasticsearch/template/template';
 
 export async function installLatestPackage(options: {
@@ -92,127 +92,113 @@ export async function installPackage(options: {
   const { savedObjectsClient, pkgkey, callCluster } = options;
   // TODO: change epm API to /packageName/version so we don't need to do this
   const [pkgName, pkgVersion] = pkgkey.split('-');
-  const paths = await Registry.getArchiveInfo(pkgName, pkgVersion);
-  // see if some version of this package is already installed
   // TODO: calls to getInstallationObject, Registry.fetchInfo, and Registry.fetchFindLatestPackge
   // and be replaced by getPackageInfo after adjusting for it to not group/use archive assets
-  const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
-  const registryPackageInfo = await Registry.fetchInfo(pkgName, pkgVersion);
   const latestPackage = await Registry.fetchFindLatestPackage(pkgName);
 
   if (pkgVersion < latestPackage.version)
     throw Boom.badRequest('Cannot install or update to an out-of-date package');
 
+  const paths = await Registry.getArchiveInfo(pkgName, pkgVersion);
+  const registryPackageInfo = await Registry.fetchInfo(pkgName, pkgVersion);
+
+  // get the currently installed package
+  const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
+  const isUpdate = installedPkg && installedPkg.attributes.version < pkgVersion ? true : false;
+
   const reinstall = pkgVersion === installedPkg?.attributes.version;
   const removable = !isRequiredPackage(pkgName);
   const { internal = false } = registryPackageInfo;
+  const toSaveESIndexPatterns = generateESIndexPatterns(registryPackageInfo.datasets);
 
-  // delete the previous version's installation's SO kibana assets before installing new ones
-  // in case some assets were removed in the new version
-  if (installedPkg) {
-    try {
-      await deleteKibanaSavedObjectsAssets(savedObjectsClient, installedPkg.attributes.installed);
-    } catch (err) {
-      // log these errors, some assets may not exist if deleted during a failed update
-    }
-  }
-
-  const [installedKibanaAssets, installedPipelines] = await Promise.all([
-    installKibanaAssets({
+  // add the package installation to the saved object
+  if (!installedPkg) {
+    await createInstallation({
       savedObjectsClient,
       pkgName,
       pkgVersion,
-      paths,
-    }),
-    installPipelines(registryPackageInfo, paths, callCluster),
-    // index patterns and ilm policies are not currently associated with a particular package
-    // so we do not save them in the package saved object state.
-    installIndexPatterns(savedObjectsClient, pkgName, pkgVersion),
-    // currenly only the base package has an ILM policy
-    // at some point ILM policies can be installed/modified
-    // per dataset and we should then save them
-    installILMPolicy(paths, callCluster),
-  ]);
+      internal,
+      removable,
+      installed_kibana: [],
+      installed_es: [],
+      toSaveESIndexPatterns,
+    });
+  }
 
-  // install or update the templates
+  const installIndexPatternPromise = installIndexPatterns(savedObjectsClient, pkgName, pkgVersion);
+  const installKibanaAssetsPromise = installKibanaAssets({
+    savedObjectsClient,
+    pkgName,
+    paths,
+    isUpdate,
+  });
+
+  // the rest of the installation must happen in sequential order
+
+  // currently only the base package has an ILM policy
+  // at some point ILM policies can be installed/modified
+  // per dataset and we should then save them
+  await installILMPolicy(paths, callCluster);
+
+  // installs versionized pipelines without removing currently installed ones
+  const installedPipelines = await installPipelines(
+    registryPackageInfo,
+    paths,
+    callCluster,
+    savedObjectsClient
+  );
+  // install or update the templates referencing the newly installed pipelines
   const installedTemplates = await installTemplates(
     registryPackageInfo,
+    isUpdate,
     callCluster,
-    pkgName,
-    pkgVersion,
-    paths
+    paths,
+    savedObjectsClient
   );
-  const toSaveESIndexPatterns = generateESIndexPatterns(registryPackageInfo.datasets);
 
+  // update current backing indices of each data stream
+  await updateCurrentWriteIndices(callCluster, installedTemplates);
+
+  // if this is an update, delete the previous version's pipelines
+  if (installedPkg && !reinstall) {
+    await deletePipelines(
+      callCluster,
+      savedObjectsClient,
+      pkgName,
+      installedPkg.attributes.version
+    );
+  }
+
+  // update to newly installed version when all assets are successfully installed
+  if (isUpdate) await updateVersion(savedObjectsClient, pkgName, pkgVersion);
   // get template refs to save
   const installedTemplateRefs = installedTemplates.map((template) => ({
     id: template.templateName,
-    type: IngestAssetType.IndexTemplate,
+    type: ElasticsearchAssetType.indexTemplate,
   }));
-
-  if (installedPkg) {
-    // update current index for every index template created
-    await updateCurrentWriteIndices(callCluster, installedTemplates);
-    if (!reinstall) {
-      try {
-        // delete the previous version's installation's pipelines
-        // this must happen after the template is updated
-        await deleteAssetsByType({
-          savedObjectsClient,
-          callCluster,
-          installedObjects: installedPkg.attributes.installed,
-          assetType: ElasticsearchAssetType.ingestPipeline,
-        });
-      } catch (err) {
-        throw new Error(err.message);
-      }
-    }
-  }
-  const toSaveAssetRefs: AssetReference[] = [
-    ...installedKibanaAssets,
-    ...installedPipelines,
-    ...installedTemplateRefs,
-  ];
-  // Save references to installed assets in the package's saved object state
-  return saveInstallationReferences({
-    savedObjectsClient,
-    pkgName,
-    pkgVersion,
-    internal,
-    removable,
-    toSaveAssetRefs,
-    toSaveESIndexPatterns,
+  const [installedKibanaAssets] = await Promise.all([
+    installKibanaAssetsPromise,
+    installIndexPatternPromise,
+  ]);
+  return [...installedKibanaAssets, ...installedPipelines, ...installedTemplateRefs];
+}
+const updateVersion = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  pkgVersion: string
+) => {
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    version: pkgVersion,
   });
-}
-
-// TODO: make it an exhaustive list
-// e.g. switch statement with cases for each enum key returning `never` for default case
-export async function installKibanaAssets(options: {
-  savedObjectsClient: SavedObjectsClientContract;
-  pkgName: string;
-  pkgVersion: string;
-  paths: string[];
-}) {
-  const { savedObjectsClient, paths } = options;
-
-  // Only install Kibana assets during package installation.
-  const kibanaAssetTypes = Object.values(KibanaAssetType);
-  const installationPromises = kibanaAssetTypes.map(async (assetType) =>
-    installKibanaSavedObjects({ savedObjectsClient, assetType, paths })
-  );
-
-  // installKibanaSavedObjects returns AssetReference[], so .map creates AssetReference[][]
-  // call .flat to flatten into one dimensional array
-  return Promise.all(installationPromises).then((results) => results.flat());
-}
-
-export async function saveInstallationReferences(options: {
+};
+export async function createInstallation(options: {
   savedObjectsClient: SavedObjectsClientContract;
   pkgName: string;
   pkgVersion: string;
   internal: boolean;
   removable: boolean;
-  toSaveAssetRefs: AssetReference[];
+  installed_kibana: KibanaAssetReference[];
+  installed_es: EsAssetReference[];
   toSaveESIndexPatterns: Record<string, string>;
 }) {
   const {
@@ -221,14 +207,15 @@ export async function saveInstallationReferences(options: {
     pkgVersion,
     internal,
     removable,
-    toSaveAssetRefs,
+    installed_kibana: installedKibana,
+    installed_es: installedEs,
     toSaveESIndexPatterns,
   } = options;
-
   await savedObjectsClient.create<Installation>(
     PACKAGES_SAVED_OBJECT_TYPE,
     {
-      installed: toSaveAssetRefs,
+      installed_kibana: installedKibana,
+      installed_es: installedEs,
       es_index_patterns: toSaveESIndexPatterns,
       name: pkgName,
       version: pkgVersion,
@@ -237,37 +224,46 @@ export async function saveInstallationReferences(options: {
     },
     { id: pkgName, overwrite: true }
   );
-
-  return toSaveAssetRefs;
+  return [...installedKibana, ...installedEs];
 }
 
-async function installKibanaSavedObjects({
-  savedObjectsClient,
-  assetType,
-  paths,
-}: {
-  savedObjectsClient: SavedObjectsClientContract;
-  assetType: KibanaAssetType;
-  paths: string[];
-}) {
-  const isSameType = (path: string) => assetType === Registry.pathParts(path).type;
-  const pathsOfType = paths.filter((path) => isSameType(path));
-  const toBeSavedObjects = await Promise.all(pathsOfType.map(getObject));
+export const saveInstalledKibanaRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  installedAssets: AssetReference[]
+) => {
+  await savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_kibana: installedAssets,
+  });
+  return installedAssets;
+};
 
-  if (toBeSavedObjects.length === 0) {
-    return [];
-  } else {
-    const createResults = await savedObjectsClient.bulkCreate(toBeSavedObjects, {
-      overwrite: true,
-    });
-    const createdObjects = createResults.saved_objects;
-    const installed = createdObjects.map(toAssetReference);
-    return installed;
-  }
-}
+export const saveInstalledEsRefs = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  installedAssets: EsAssetReference[]
+) => {
+  const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
+  const installedAssetsToSave = installedPkg?.attributes.installed_es.concat(installedAssets);
+  await savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_es: installedAssetsToSave,
+  });
+  return installedAssets;
+};
 
-function toAssetReference({ id, type }: SavedObject) {
-  const reference: AssetReference = { id, type: type as KibanaAssetType };
+export const removeAssetsFromInstalledEsByType = async (
+  savedObjectsClient: SavedObjectsClientContract,
+  pkgName: string,
+  assetType: AssetType
+) => {
+  const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
+  const installedAssets = installedPkg?.attributes.installed_es;
+  if (!installedAssets?.length) return;
+  const installedAssetsToSave = installedAssets?.filter(({ id, type }) => {
+    return type !== assetType;
+  });
 
-  return reference;
-}
+  return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
+    installed_es: installedAssetsToSave,
+  });
+};

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

@@ -10,8 +10,9 @@ import { PACKAGES_SAVED_OBJECT_TYPE, PACKAGE_CONFIG_SAVED_OBJECT_TYPE } from '..
 import { AssetReference, AssetType, ElasticsearchAssetType } from '../../../types';
 import { CallESAsCurrentUser } from '../../../types';
 import { getInstallation, savedObjectTypes } from './index';
+import { deletePipeline } from '../elasticsearch/ingest_pipeline/';
 import { installIndexPatterns } from '../kibana/index_pattern/install';
-import { packageConfigService } from '../..';
+import { packageConfigService, appContextService } from '../..';
 
 export async function removeInstallation(options: {
   savedObjectsClient: SavedObjectsClientContract;
@@ -25,7 +26,6 @@ export async function removeInstallation(options: {
   if (!installation) throw Boom.badRequest(`${pkgName} is not installed`);
   if (installation.removable === false)
     throw Boom.badRequest(`${pkgName} is installed by default and cannot be removed`);
-  const installedObjects = installation.installed || [];
 
   const { total } = await packageConfigService.list(savedObjectsClient, {
     kuery: `${PACKAGE_CONFIG_SAVED_OBJECT_TYPE}.package.name:${pkgName}`,
@@ -38,48 +38,40 @@ export async function removeInstallation(options: {
       `unable to remove package with existing package config(s) in use by agent(s)`
     );
 
+  // recreate or delete index patterns when a package is uninstalled
+  await installIndexPatterns(savedObjectsClient);
+
+  // Delete the installed assets
+  const installedAssets = [...installation.installed_kibana, ...installation.installed_es];
+  await deleteAssets(installedAssets, savedObjectsClient, callCluster);
+
   // Delete the manager saved object with references to the asset objects
   // could also update with [] or some other state
   await savedObjectsClient.delete(PACKAGES_SAVED_OBJECT_TYPE, pkgName);
 
-  // recreate or delete index patterns when a package is uninstalled
-  await installIndexPatterns(savedObjectsClient);
-
-  // Delete the installed asset
-  await deleteAssets(installedObjects, savedObjectsClient, callCluster);
-
   // successful delete's in SO client return {}. return something more useful
-  return installedObjects;
+  return installedAssets;
 }
 async function deleteAssets(
   installedObjects: AssetReference[],
   savedObjectsClient: SavedObjectsClientContract,
   callCluster: CallESAsCurrentUser
 ) {
+  const logger = appContextService.getLogger();
   const deletePromises = installedObjects.map(async ({ id, type }) => {
     const assetType = type as AssetType;
     if (savedObjectTypes.includes(assetType)) {
-      savedObjectsClient.delete(assetType, id);
+      return savedObjectsClient.delete(assetType, id);
     } else if (assetType === ElasticsearchAssetType.ingestPipeline) {
-      deletePipeline(callCluster, id);
+      return deletePipeline(callCluster, id);
     } else if (assetType === ElasticsearchAssetType.indexTemplate) {
-      deleteTemplate(callCluster, id);
+      return deleteTemplate(callCluster, id);
     }
   });
   try {
     await Promise.all([...deletePromises]);
   } catch (err) {
-    throw new Error(err.message);
-  }
-}
-async function deletePipeline(callCluster: CallESAsCurrentUser, id: string): Promise<void> {
-  // '*' shouldn't ever appear here, but it still would delete all ingest pipelines
-  if (id && id !== '*') {
-    try {
-      await callCluster('ingest.deletePipeline', { id });
-    } catch (err) {
-      throw new Error(`error deleting pipeline ${id}`);
-    }
+    logger.error(err);
   }
 }
 
@@ -108,31 +100,14 @@ async function deleteTemplate(callCluster: CallESAsCurrentUser, name: string): P
   }
 }
 
-export async function deleteAssetsByType({
-  savedObjectsClient,
-  callCluster,
-  installedObjects,
-  assetType,
-}: {
-  savedObjectsClient: SavedObjectsClientContract;
-  callCluster: CallESAsCurrentUser;
-  installedObjects: AssetReference[];
-  assetType: ElasticsearchAssetType;
-}) {
-  const toDelete = installedObjects.filter((asset) => asset.type === assetType);
-  try {
-    await deleteAssets(toDelete, savedObjectsClient, callCluster);
-  } catch (err) {
-    throw new Error(err.message);
-  }
-}
-
 export async function deleteKibanaSavedObjectsAssets(
   savedObjectsClient: SavedObjectsClientContract,
   installedObjects: AssetReference[]
 ) {
+  const logger = appContextService.getLogger();
   const deletePromises = installedObjects.map(({ id, type }) => {
     const assetType = type as AssetType;
+
     if (savedObjectTypes.includes(assetType)) {
       return savedObjectsClient.delete(assetType, id);
     }
@@ -140,6 +115,6 @@ export async function deleteKibanaSavedObjectsAssets(
   try {
     await Promise.all(deletePromises);
   } catch (err) {
-    throw new Error('error deleting saved object asset');
+    logger.warn(err);
   }
 }

x-pack/plugins/ingest_manager/server/types/index.tsx

@@ -43,8 +43,9 @@ export {
   Dataset,
   RegistryElasticsearch,
   AssetReference,
+  EsAssetReference,
+  KibanaAssetReference,
   ElasticsearchAssetType,
-  IngestAssetType,
   RegistryPackage,
   AssetType,
   Installable,

x-pack/plugins/security_solution/public/management/pages/policy/store/policy_list/test_mock_utils.ts

@@ -9,7 +9,8 @@ import { INGEST_API_PACKAGE_CONFIGS, INGEST_API_EPM_PACKAGES } from './services/
 import { EndpointDocGenerator } from '../../../../../../common/endpoint/generate_data';
 import { GetPolicyListResponse } from '../../types';
 import {
-  AssetReference,
+  KibanaAssetReference,
+  EsAssetReference,
   GetPackagesResponse,
   InstallationStatus,
 } from '../../../../../../../ingest_manager/common';
@@ -43,26 +44,28 @@ export const apiPathMockResponseProviders = {
             type: 'epm-packages',
             id: 'endpoint',
             attributes: {
-              installed: [
+              installed_kibana: [
                 { id: '826759f0-7074-11ea-9bc8-6b38f4d29a16', type: 'dashboard' },
                 { id: '1cfceda0-728b-11ea-9bc8-6b38f4d29a16', type: 'visualization' },
                 { id: '1e525190-7074-11ea-9bc8-6b38f4d29a16', type: 'visualization' },
                 { id: '55387750-729c-11ea-9bc8-6b38f4d29a16', type: 'visualization' },
                 { id: '92b1edc0-706a-11ea-9bc8-6b38f4d29a16', type: 'visualization' },
                 { id: 'a3a3bd10-706b-11ea-9bc8-6b38f4d29a16', type: 'map' },
-                { id: 'logs-endpoint.alerts', type: 'index-template' },
-                { id: 'events-endpoint', type: 'index-template' },
-                { id: 'logs-endpoint.events.file', type: 'index-template' },
-                { id: 'logs-endpoint.events.library', type: 'index-template' },
-                { id: 'metrics-endpoint.metadata', type: 'index-template' },
-                { id: 'metrics-endpoint.metadata_mirror', type: 'index-template' },
-                { id: 'logs-endpoint.events.network', type: 'index-template' },
-                { id: 'metrics-endpoint.policy', type: 'index-template' },
-                { id: 'logs-endpoint.events.process', type: 'index-template' },
-                { id: 'logs-endpoint.events.registry', type: 'index-template' },
-                { id: 'logs-endpoint.events.security', type: 'index-template' },
-                { id: 'metrics-endpoint.telemetry', type: 'index-template' },
-              ] as AssetReference[],
+              ] as KibanaAssetReference[],
+              installed_es: [
+                { id: 'logs-endpoint.alerts', type: 'index_template' },
+                { id: 'events-endpoint', type: 'index_template' },
+                { id: 'logs-endpoint.events.file', type: 'index_template' },
+                { id: 'logs-endpoint.events.library', type: 'index_template' },
+                { id: 'metrics-endpoint.metadata', type: 'index_template' },
+                { id: 'metrics-endpoint.metadata_mirror', type: 'index_template' },
+                { id: 'logs-endpoint.events.network', type: 'index_template' },
+                { id: 'metrics-endpoint.policy', type: 'index_template' },
+                { id: 'logs-endpoint.events.process', type: 'index_template' },
+                { id: 'logs-endpoint.events.registry', type: 'index_template' },
+                { id: 'logs-endpoint.events.security', type: 'index_template' },
+                { id: 'metrics-endpoint.telemetry', type: 'index_template' },
+              ] as EsAssetReference[],
               es_index_patterns: {
                 alerts: 'logs-endpoint.alerts-*',
                 events: 'events-endpoint-*',