[Ingest Manager] Adding bulk packages upgrade api

x-pack/plugins/ingest_manager/common/constants/routes.ts

@@ -14,10 +14,12 @@ export const FLEET_API_ROOT = `${API_ROOT}/fleet`;
 export const LIMITED_CONCURRENCY_ROUTE_TAG = 'ingest:limited-concurrency';
 
 // EPM API routes
+const EPM_PACKAGES_BULK = `${EPM_API_ROOT}/packages_bulk`;
 const EPM_PACKAGES_MANY = `${EPM_API_ROOT}/packages`;
 const EPM_PACKAGES_ONE = `${EPM_PACKAGES_MANY}/{pkgkey}`;
 const EPM_PACKAGES_FILE = `${EPM_PACKAGES_MANY}/{pkgName}/{pkgVersion}`;
 export const EPM_API_ROUTES = {
+  BULK_UPGRADE_PATTERN: EPM_PACKAGES_BULK,
   LIST_PATTERN: EPM_PACKAGES_MANY,
   LIMITED_LIST_PATTERN: `${EPM_PACKAGES_MANY}/limited`,
   INFO_PATTERN: EPM_PACKAGES_ONE,

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

@@ -71,6 +71,26 @@ export interface InstallPackageResponse {
   response: AssetReference[];
 }
 
+export interface UpgradePackageError {
+  name: string;
+  statusCode: number;
+  body: {
+    message: string;
+  };
+}
+
+export interface UpgradePackageInfo {
+  name: string;
+  newVersion: string;
+  // this will be null if no package was present before the upgrade (aka it was an install)
+  oldVersion: string | null;
+  assets: AssetReference[];
+}
+
+export interface UpgradePackagesResponse {
+  response: Array<UpgradePackageInfo | UpgradePackageError>;
+}
+
 export interface MessageResponse {
   response: string;
 }

x-pack/plugins/ingest_manager/server/errors/handlers.ts

@@ -56,10 +56,7 @@ const getHTTPResponseCode = (error: IngestManagerError): number => {
   return 400; // Bad Request
 };
 
-export const defaultIngestErrorHandler: IngestErrorHandler = async ({
-  error,
-  response,
-}: IngestErrorHandlerParams): Promise<IKibanaResponse> => {
+export function handleIngestError(error: IngestManagerError | Boom | Error) {
   const logger = appContextService.getLogger();
   if (isLegacyESClientError(error)) {
     // there was a problem communicating with ES (e.g. via `callCluster`)
@@ -72,36 +69,43 @@ export const defaultIngestErrorHandler: IngestErrorHandler = async ({
 
     logger.error(message);
 
-    return response.customError({
+    return {
       statusCode: error?.statusCode || error.status,
       body: { message },
-    });
+    };
   }
 
   // our "expected" errors
   if (error instanceof IngestManagerError) {
     // only log the message
     logger.error(error.message);
-    return response.customError({
+    return {
       statusCode: getHTTPResponseCode(error),
       body: { message: error.message },
-    });
+    };
   }
 
   // handle any older Boom-based errors or the few places our app uses them
   if (isBoom(error)) {
     // only log the message
     logger.error(error.output.payload.message);
-    return response.customError({
+    return {
       statusCode: error.output.statusCode,
       body: { message: error.output.payload.message },
-    });
+    };
   }
 
   // not sure what type of error this is. log as much as possible
   logger.error(error);
-  return response.customError({
+  return {
     statusCode: 500,
     body: { message: error.message },
-  });
+  };
+}
+
+export const defaultIngestErrorHandler: IngestErrorHandler = async ({
+  error,
+  response,
+}: IngestErrorHandlerParams): Promise<IKibanaResponse> => {
+  return response.customError(handleIngestError(error));
 };

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

@@ -5,7 +5,6 @@
  */
 import { TypeOf } from '@kbn/config-schema';
 import { RequestHandler, CustomHttpResponseOptions } from 'src/core/server';
-import { appContextService } from '../../services';
 import {
   GetInfoResponse,
   InstallPackageResponse,
@@ -14,6 +13,7 @@ import {
   GetCategoriesResponse,
   GetPackagesResponse,
   GetLimitedPackagesResponse,
+  UpgradePackagesResponse,
 } from '../../../common';
 import {
   GetCategoriesRequestSchema,
@@ -23,6 +23,7 @@ import {
   InstallPackageFromRegistryRequestSchema,
   InstallPackageByUploadRequestSchema,
   DeletePackageRequestSchema,
+  BulkUpgradePackagesFromRegistryRequestSchema,
 } from '../../types';
 import {
   getCategories,
@@ -34,9 +35,9 @@ import {
   getLimitedPackages,
   getInstallationObject,
 } from '../../services/epm/packages';
-import { IngestManagerError, defaultIngestErrorHandler } from '../../errors';
+import { defaultIngestErrorHandler } from '../../errors';
 import { splitPkgKey } from '../../services/epm/registry';
-import { getInstallType } from '../../services/epm/packages/install';
+import { handleInstallPackageFailure, upgradePackages } from '../../services/epm/packages/install';
 
 export const getCategoriesHandler: RequestHandler<
   undefined,
@@ -136,13 +137,11 @@ export const installPackageFromRegistryHandler: RequestHandler<
   undefined,
   TypeOf<typeof InstallPackageFromRegistryRequestSchema.body>
 > = 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 } = splitPkgKey(pkgkey);
   const installedPkg = await getInstallationObject({ savedObjectsClient, pkgName });
-  const installType = getInstallType({ pkgVersion, installedPkg });
   try {
     const res = await installPackage({
       savedObjectsClient,
@@ -155,36 +154,38 @@ export const installPackageFromRegistryHandler: RequestHandler<
     };
     return response.ok({ body });
   } catch (e) {
-    // could have also done `return defaultIngestErrorHandler({ error: e, response })` at each of the returns,
-    // but doing it this way will log the outer/install errors before any inner/rollback errors
     const defaultResult = await defaultIngestErrorHandler({ error: e, response });
-    if (e instanceof IngestManagerError) {
-      return defaultResult;
-    }
+    await handleInstallPackageFailure(
+      savedObjectsClient,
+      e,
+      pkgName,
+      pkgVersion,
+      installedPkg,
+      callCluster
+    );
 
-    // if there is an unknown server error, uninstall any package assets or reinstall the previous version if update
-    try {
-      if (installType === 'install' || installType === 'reinstall') {
-        logger.error(`uninstalling ${pkgkey} after error installing`);
-        await removeInstallation({ savedObjectsClient, pkgkey, callCluster });
-      }
-      if (installType === 'update') {
-        // @ts-ignore getInstallType ensures we have installedPkg
-        const prevVersion = `${pkgName}-${installedPkg.attributes.version}`;
-        logger.error(`rolling back to ${prevVersion} after error installing ${pkgkey}`);
-        await installPackage({
-          savedObjectsClient,
-          pkgkey: prevVersion,
-          callCluster,
-        });
-      }
-    } catch (error) {
-      logger.error(`failed to uninstall or rollback package after installation error ${error}`);
-    }
     return defaultResult;
   }
 };
 
+export const upgradePackagesFromRegistryHandler: RequestHandler<
+  undefined,
+  undefined,
+  TypeOf<typeof BulkUpgradePackagesFromRegistryRequestSchema.body>
+> = async (context, request, response) => {
+  const savedObjectsClient = context.core.savedObjects.client;
+  const callCluster = context.core.elasticsearch.legacy.client.callAsCurrentUser;
+  const res = await upgradePackages({
+    savedObjectsClient,
+    callCluster,
+    packagesToUpgrade: request.body.upgrade,
+  });
+  const body: UpgradePackagesResponse = {
+    response: res,
+  };
+  return response.ok({ body });
+};
+
 export const installPackageByUploadHandler: RequestHandler<
   undefined,
   undefined,

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

@@ -14,6 +14,7 @@ import {
   installPackageFromRegistryHandler,
   installPackageByUploadHandler,
   deletePackageHandler,
+  upgradePackagesFromRegistryHandler,
 } from './handlers';
 import {
   GetCategoriesRequestSchema,
@@ -23,6 +24,7 @@ import {
   InstallPackageFromRegistryRequestSchema,
   InstallPackageByUploadRequestSchema,
   DeletePackageRequestSchema,
+  BulkUpgradePackagesFromRegistryRequestSchema,
 } from '../../types';
 
 const MAX_FILE_SIZE_BYTES = 104857600; // 100MB
@@ -82,6 +84,15 @@ export const registerRoutes = (router: IRouter) => {
     installPackageFromRegistryHandler
   );
 
+  router.post(
+    {
+      path: EPM_API_ROUTES.BULK_UPGRADE_PATTERN,
+      validate: BulkUpgradePackagesFromRegistryRequestSchema,
+      options: { tags: [`access:${PLUGIN_ID}-all`] },
+    },
+    upgradePackagesFromRegistryHandler
+  );
+
   router.post(
     {
       path: EPM_API_ROUTES.INSTALL_BY_UPLOAD_PATTERN,

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

@@ -6,6 +6,8 @@
 
 import { SavedObject, SavedObjectsClientContract } from 'src/core/server';
 import semver from 'semver';
+import Boom from 'boom';
+import { UpgradePackageError, UpgradePackageInfo } from '../../../../common';
 import { PACKAGES_SAVED_OBJECT_TYPE, MAX_TIME_COMPLETE_INSTALL } from '../../../constants';
 import {
   AssetReference,
@@ -17,6 +19,7 @@ import {
   EsAssetReference,
   ElasticsearchAssetType,
   InstallType,
+  RegistrySearchResult,
 } from '../../../types';
 import { installIndexPatterns } from '../kibana/index_pattern/install';
 import * as Registry from '../registry';
@@ -32,11 +35,12 @@ import {
   ArchiveAsset,
 } from '../kibana/assets/install';
 import { updateCurrentWriteIndices } from '../elasticsearch/template/template';
-import { deleteKibanaSavedObjectsAssets } from './remove';
-import { PackageOutdatedError } from '../../../errors';
+import { deleteKibanaSavedObjectsAssets, removeInstallation } from './remove';
+import { IngestManagerError, PackageOutdatedError } from '../../../errors';
 import { getPackageSavedObjects } from './get';
 import { installTransformForDataset } from '../elasticsearch/transform/install';
 import { appContextService } from '../../app_context';
+import { handleIngestError } from '../../../errors/handlers';
 
 export async function installLatestPackage(options: {
   savedObjectsClient: SavedObjectsClientContract;
@@ -95,6 +99,117 @@ export async function ensureInstalledPackage(options: {
   return installation;
 }
 
+export async function handleInstallPackageFailure(
+  savedObjectsClient: SavedObjectsClientContract,
+  error: IngestManagerError | Boom | Error,
+  pkgName: string,
+  pkgVersion: string,
+  installedPkg: SavedObject<Installation> | undefined,
+  callCluster: CallESAsCurrentUser
+) {
+  if (error instanceof IngestManagerError) {
+    return;
+  }
+  const logger = appContextService.getLogger();
+  const pkgkey = Registry.pkgToPkgKey({
+    name: pkgName,
+    version: pkgVersion,
+  });
+
+  // if there is an unknown server error, uninstall any package assets or reinstall the previous version if update
+  try {
+    const installType = getInstallType({ pkgVersion, installedPkg });
+    if (installType === 'install' || installType === 'reinstall') {
+      logger.error(`uninstalling ${pkgkey} after error installing`);
+      await removeInstallation({ savedObjectsClient, pkgkey, callCluster });
+    }
+
+    if (installType === 'update') {
+      if (!installedPkg) {
+        logger.error(
+          `failed to rollback package after installation error ${error} because saved object was undefined`
+        );
+        return;
+      }
+      const prevVersion = `${pkgName}-${installedPkg.attributes.version}`;
+      logger.error(`rolling back to ${prevVersion} after error installing ${pkgkey}`);
+      await installPackage({
+        savedObjectsClient,
+        pkgkey: prevVersion,
+        callCluster,
+      });
+    }
+  } catch (e) {
+    logger.error(`failed to uninstall or rollback package after installation error ${e}`);
+  }
+}
+
+export async function upgradePackages({
+  savedObjectsClient,
+  packagesToUpgrade,
+  callCluster,
+}: {
+  savedObjectsClient: SavedObjectsClientContract;
+  packagesToUpgrade: string[];
+  callCluster: CallESAsCurrentUser;
+}): Promise<Array<UpgradePackageInfo | UpgradePackageError>> {
+  const res: Array<UpgradePackageInfo | UpgradePackageError> = [];
+
+  for (const pkgToUpgrade of packagesToUpgrade) {
+    let installedPkg: SavedObject<Installation> | undefined;
+    let latestPackage: RegistrySearchResult | undefined;
+    try {
+      [installedPkg, latestPackage] = await Promise.all([
+        getInstallationObject({ savedObjectsClient, pkgName: pkgToUpgrade }),
+        Registry.fetchFindLatestPackage(pkgToUpgrade),
+      ]);
+    } catch (e) {
+      res.push({ name: pkgToUpgrade, ...handleIngestError(e) });
+      continue;
+    }
+
+    // TODO I think we want version here and not `install_version`?
+    if (!installedPkg || semver.gt(latestPackage.version, installedPkg.attributes.version)) {
+      const pkgkey = Registry.pkgToPkgKey({
+        name: latestPackage.name,
+        version: latestPackage.version,
+      });
+
+      try {
+        const assets = await installPackage({ savedObjectsClient, pkgkey, callCluster });
+        res.push({
+          name: pkgToUpgrade,
+          newVersion: latestPackage.version,
+          oldVersion: installedPkg?.attributes.version ?? null,
+          assets,
+        });
+      } catch (e) {
+        res.push({ name: pkgToUpgrade, ...handleIngestError(e) });
+        await handleInstallPackageFailure(
+          savedObjectsClient,
+          e,
+          latestPackage.name,
+          latestPackage.version,
+          installedPkg,
+          callCluster
+        );
+      }
+    } else {
+      // package was already at the latest version
+      res.push({
+        name: pkgToUpgrade,
+        newVersion: latestPackage.version,
+        oldVersion: latestPackage.version,
+        assets: [
+          ...installedPkg.attributes.installed_es,
+          ...installedPkg.attributes.installed_kibana,
+        ],
+      });
+    }
+  }
+  return res;
+}
+
 export async function installPackage({
   savedObjectsClient,
   pkgkey,