[7.x] [Ingest Manager] Adding bulk packages upgrade api (#77827)

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

@@ -15,9 +15,11 @@ export const LIMITED_CONCURRENCY_ROUTE_TAG = 'ingest:limited-concurrency';
 
 // EPM API routes
 const EPM_PACKAGES_MANY = `${EPM_API_ROOT}/packages`;
+const EPM_PACKAGES_BULK = `${EPM_PACKAGES_MANY}/_bulk`;
 const EPM_PACKAGES_ONE = `${EPM_PACKAGES_MANY}/{pkgkey}`;
 const EPM_PACKAGES_FILE = `${EPM_PACKAGES_MANY}/{pkgName}/{pkgVersion}`;
 export const EPM_API_ROUTES = {
+  BULK_INSTALL_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/services/routes.ts

@@ -46,6 +46,10 @@ export const epmRouteService = {
     ); // trim trailing slash
   },
 
+  getBulkInstallPath: () => {
+    return EPM_API_ROUTES.BULK_INSTALL_PATTERN;
+  },
+
   getRemovePath: (pkgkey: string) => {
     return EPM_API_ROUTES.DELETE_PATTERN.replace('{pkgkey}', pkgkey).replace(/\/$/, ''); // trim trailing slash
   },

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

@@ -71,6 +71,30 @@ export interface InstallPackageResponse {
   response: AssetReference[];
 }
 
+export interface IBulkInstallPackageError {
+  name: string;
+  statusCode: number;
+  error: string | Error;
+}
+
+export interface BulkInstallPackageInfo {
+  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 BulkInstallPackagesResponse {
+  response: Array<BulkInstallPackageInfo | IBulkInstallPackageError>;
+}
+
+export interface BulkInstallPackagesRequest {
+  body: {
+    packages: string[];
+  };
+}
+
 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 ingestErrorToResponseOptions(error: IngestErrorHandlerParams['error']) {
   const logger = appContextService.getLogger();
   if (isLegacyESClientError(error)) {
     // there was a problem communicating with ES (e.g. via `callCluster`)
@@ -72,36 +69,44 @@ 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> => {
+  const options = ingestErrorToResponseOptions(error);
+  return response.customError(options);
 };

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

@@ -5,7 +5,7 @@
  */
 
 /* eslint-disable max-classes-per-file */
-export { defaultIngestErrorHandler } from './handlers';
+export { defaultIngestErrorHandler, ingestErrorToResponseOptions } from './handlers';
 
 export class IngestManagerError extends Error {
   constructor(message?: string) {

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,
+  BulkInstallPackagesResponse,
 } from '../../../common';
 import {
   GetCategoriesRequestSchema,
@@ -23,6 +23,7 @@ import {
   InstallPackageFromRegistryRequestSchema,
   InstallPackageByUploadRequestSchema,
   DeletePackageRequestSchema,
+  BulkUpgradePackagesFromRegistryRequestSchema,
 } from '../../types';
 import {
   getCategories,
@@ -34,9 +35,12 @@ 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,
+  bulkInstallPackages,
+} from '../../services/epm/packages/install';
 
 export const getCategoriesHandler: RequestHandler<
   undefined,
@@ -136,13 +140,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 +157,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,
+      error: 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 bulkInstallPackagesFromRegistryHandler: 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 bulkInstallPackages({
+    savedObjectsClient,
+    callCluster,
+    packagesToUpgrade: request.body.packages,
+  });
+  const body: BulkInstallPackagesResponse = {
+    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,
+  bulkInstallPackagesFromRegistryHandler,
 } 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_INSTALL_PATTERN,
+      validate: BulkUpgradePackagesFromRegistryRequestSchema,
+      options: { tags: [`access:${PLUGIN_ID}-all`] },
+    },
+    bulkInstallPackagesFromRegistryHandler
+  );
+
   router.post(
     {
       path: EPM_API_ROUTES.INSTALL_BY_UPLOAD_PATTERN,