[Fleet] Add package policy upgrade API (#103017)

* Add Integrations page callout for package upgades

* Fix props

* Add missing file

* Add integrations upgrade callout message

* Add link to updates available tab

* Fix merge

* Upgrade ppolicies UI WIP

* Initial upgrade dry run API

* Add upgrade method

* Move overridePackageInputs and use for upgrade method

* Add new variables to dry run diff

* Revert UI changes to uto upgrade wizard

* Add vars and streams to error keys

* Type fix

* Fix jest

* Fix types

* Fix typecheck

* Fix types

* Add integration test for dry run API

* Flesh out test cases

* Clean up error responses for dry runs

* Fix failing tests

* WIP: Add (failing for now) test case for package upgrade w/ error

* Add compiled_stream to test API payload

* Fix failing test case for automatic upgrade

* Fix compiled stream in package policy upgrade

* Remove fleet and agent setup from integration test

* Unload esarchiver fixtures in api integration test

Co-authored-by: Kyle Pollich <kpollich1@gmail.com>
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

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

@@ -46,6 +46,7 @@ export const PACKAGE_POLICY_API_ROUTES = {
   CREATE_PATTERN: `${PACKAGE_POLICY_API_ROOT}`,
   UPDATE_PATTERN: `${PACKAGE_POLICY_API_ROOT}/{packagePolicyId}`,
   DELETE_PATTERN: `${PACKAGE_POLICY_API_ROOT}/delete`,
+  UPGRADE_PATTERN: `${PACKAGE_POLICY_API_ROOT}/upgrade`,
 };
 
 // Agent policy API routes

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

@@ -78,3 +78,7 @@ export interface PackagePolicy extends Omit<NewPackagePolicy, 'inputs'> {
 }
 
 export type PackagePolicySOAttributes = Omit<PackagePolicy, 'id' | 'version'>;
+
+export type DryRunPackagePolicy = NewPackagePolicy & {
+  errors?: Array<{ key: string | undefined; message: string }>;
+};

x-pack/plugins/fleet/common/types/rest_spec/package_policy.ts

@@ -5,7 +5,12 @@
  * 2.0.
  */
 
-import type { PackagePolicy, NewPackagePolicy, UpdatePackagePolicy } from '../models';
+import type {
+  PackagePolicy,
+  NewPackagePolicy,
+  UpdatePackagePolicy,
+  DryRunPackagePolicy,
+} from '../models';
 
 export interface GetPackagePoliciesRequest {
   query: {
@@ -57,3 +62,21 @@ export type DeletePackagePoliciesResponse = Array<{
   name?: string;
   success: boolean;
 }>;
+
+export interface UpgradePackagePolicyBaseResponse {
+  name?: string;
+}
+
+export interface UpgradePackagePolicyDryRunResponseItem extends UpgradePackagePolicyBaseResponse {
+  hasErrors: boolean;
+  diff?: [PackagePolicy, DryRunPackagePolicy];
+}
+
+export type UpgradePackagePolicyDryRunResponse = UpgradePackagePolicyDryRunResponseItem[];
+
+export interface UpgradePackagePolicyResponseItem extends UpgradePackagePolicyBaseResponse {
+  id: string;
+  success: boolean;
+}
+
+export type UpgradePackagePolicyResponse = UpgradePackagePolicyResponseItem[];

x-pack/plugins/fleet/public/hooks/use_package_installations.tsx

@@ -0,0 +1,84 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { useMemo } from 'react';
+import semverLt from 'semver/functions/lt';
+
+import { installationStatuses } from '../../common/constants';
+import type { PackagePolicy } from '../types';
+
+import { useGetPackages } from './use_request/epm';
+import { useGetAgentPolicies } from './use_request/agent_policy';
+
+export const usePackageInstallations = () => {
+  const { data: allPackages, isLoading: isLoadingPackages } = useGetPackages({
+    experimental: true,
+  });
+
+  const { data: agentPolicyData, isLoading: isLoadingPolicies } = useGetAgentPolicies({
+    full: true,
+  });
+
+  const allInstalledPackages = useMemo(
+    () =>
+      (allPackages?.response || []).filter((pkg) => pkg.status === installationStatuses.Installed),
+    [allPackages?.response]
+  );
+
+  const updatablePackages = useMemo(
+    () =>
+      allInstalledPackages.filter(
+        (item) =>
+          'savedObject' in item && semverLt(item.savedObject.attributes.version, item.version)
+      ),
+    [allInstalledPackages]
+  );
+
+  const updatableIntegrations = useMemo(
+    () =>
+      (agentPolicyData?.items || []).reduce((result, policy) => {
+        policy.package_policies.forEach((pkgPolicy: PackagePolicy | string) => {
+          if (typeof pkgPolicy === 'string' || !pkgPolicy.package) return false;
+          const { name, version } = pkgPolicy.package;
+          const installedPackage = allInstalledPackages.find(
+            (installedPkg) =>
+              'savedObject' in installedPkg && installedPkg.savedObject.attributes.name === name
+          );
+          if (
+            installedPackage &&
+            'savedObject' in installedPackage &&
+            semverLt(version, installedPackage.savedObject.attributes.version)
+          ) {
+            const packageData = result.get(name) ?? {
+              currentVersion: installedPackage.savedObject.attributes.version,
+              policiesToUpgrade: [],
+            };
+            packageData.policiesToUpgrade.push({
+              id: policy.id,
+              name: policy.name,
+              agentsCount: policy.agents,
+              pkgPolicyId: pkgPolicy.id,
+              pkgPolicyName: pkgPolicy.name,
+              pkgPolicyIntegrationVersion: version,
+            });
+            result.set(name, packageData);
+          }
+        });
+        return result;
+      }, new Map()),
+    [allInstalledPackages, agentPolicyData]
+  );
+
+  return {
+    allPackages,
+    allInstalledPackages,
+    updatablePackages,
+    updatableIntegrations,
+    isLoadingPackages,
+    isLoadingPolicies,
+  };
+};

x-pack/plugins/fleet/server/mocks/index.ts

@@ -75,6 +75,9 @@ export const createPackagePolicyServiceMock = () => {
     listIds: jest.fn(),
     update: jest.fn(),
     runExternalCallbacks: jest.fn(),
+    upgrade: jest.fn(),
+    getUpgradeDryRunDiff: jest.fn(),
+    getUpgradePackagePolicyInfo: jest.fn(),
   } as jest.Mocked<PackagePolicyServiceInterface>;
 };
 

x-pack/plugins/fleet/server/routes/package_policy/handlers.test.ts

@@ -17,10 +17,15 @@ import type { CreatePackagePolicyRequestSchema } from '../../types/rest_spec';
 
 import { registerRoutes } from './index';
 
+type PackagePolicyServicePublicInterface = Omit<
+  PackagePolicyServiceInterface,
+  'getUpgradePackagePolicyInfo'
+>;
+
 const packagePolicyServiceMock = packagePolicyService as jest.Mocked<PackagePolicyServiceInterface>;
 
 jest.mock('../../services/package_policy', (): {
-  packagePolicyService: jest.Mocked<PackagePolicyServiceInterface>;
+  packagePolicyService: jest.Mocked<PackagePolicyServicePublicInterface>;
 } => {
   return {
     packagePolicyService: {
@@ -56,6 +61,8 @@ jest.mock('../../services/package_policy', (): {
       runExternalCallbacks: jest.fn((callbackType, newPackagePolicy, context, request) =>
         Promise.resolve(newPackagePolicy)
       ),
+      upgrade: jest.fn(),
+      getUpgradeDryRunDiff: jest.fn(),
     },
   };
 });

x-pack/plugins/fleet/server/routes/package_policy/handlers.ts

@@ -17,8 +17,14 @@ import type {
   CreatePackagePolicyRequestSchema,
   UpdatePackagePolicyRequestSchema,
   DeletePackagePoliciesRequestSchema,
+  UpgradePackagePoliciesRequestSchema,
 } from '../../types';
-import type { CreatePackagePolicyResponse, DeletePackagePoliciesResponse } from '../../../common';
+import type {
+  CreatePackagePolicyResponse,
+  DeletePackagePoliciesResponse,
+  UpgradePackagePolicyDryRunResponse,
+  UpgradePackagePolicyResponse,
+} from '../../../common';
 import { defaultIngestErrorHandler } from '../../errors';
 
 export const getPackagePoliciesHandler: RequestHandler<
@@ -172,3 +178,38 @@ export const deletePackagePolicyHandler: RequestHandler<
     return defaultIngestErrorHandler({ error, response });
   }
 };
+
+export const upgradePackagePolicyHandler: RequestHandler<
+  unknown,
+  unknown,
+  TypeOf<typeof UpgradePackagePoliciesRequestSchema.body>
+> = async (context, request, response) => {
+  const soClient = context.core.savedObjects.client;
+  const esClient = context.core.elasticsearch.client.asCurrentUser;
+  const user = appContextService.getSecurity()?.authc.getCurrentUser(request) || undefined;
+  try {
+    if (request.body.dryRun) {
+      const body: UpgradePackagePolicyDryRunResponse = [];
+
+      for (const id of request.body.packagePolicyIds) {
+        const result = await packagePolicyService.getUpgradeDryRunDiff(soClient, id);
+        body.push(result);
+      }
+      return response.ok({
+        body,
+      });
+    } else {
+      const body: UpgradePackagePolicyResponse = await packagePolicyService.upgrade(
+        soClient,
+        esClient,
+        request.body.packagePolicyIds,
+        { user }
+      );
+      return response.ok({
+        body,
+      });
+    }
+  } catch (error) {
+    return defaultIngestErrorHandler({ error, response });
+  }
+};

x-pack/plugins/fleet/server/routes/package_policy/index.ts

@@ -14,6 +14,7 @@ import {
   CreatePackagePolicyRequestSchema,
   UpdatePackagePolicyRequestSchema,
   DeletePackagePoliciesRequestSchema,
+  UpgradePackagePoliciesRequestSchema,
 } from '../../types';
 
 import {
@@ -22,6 +23,7 @@ import {
   createPackagePolicyHandler,
   updatePackagePolicyHandler,
   deletePackagePolicyHandler,
+  upgradePackagePolicyHandler,
 } from './handlers';
 
 export const registerRoutes = (router: IRouter) => {
@@ -74,4 +76,14 @@ export const registerRoutes = (router: IRouter) => {
     },
     deletePackagePolicyHandler
   );
+
+  // Upgrade
+  router.post(
+    {
+      path: PACKAGE_POLICY_API_ROUTES.UPGRADE_PATTERN,
+      validate: UpgradePackagePoliciesRequestSchema,
+      options: { tags: [`access:${PLUGIN_ID}-all`] },
+    },
+    upgradePackagePolicyHandler
+  );
 };

x-pack/plugins/fleet/server/services/package_policy.ts

@@ -5,6 +5,8 @@
  * 2.0.
  */
 
+import { omit } from 'lodash';
+import { i18n } from '@kbn/i18n';
 import type { KibanaRequest } from 'src/core/server';
 import type {
   ElasticsearchClient,
@@ -16,18 +18,22 @@ import uuid from 'uuid';
 import type { AuthenticatedUser } from '../../../security/server';
 import {
   packageToPackagePolicy,
+  packageToPackagePolicyInputs,
   isPackageLimited,
   doesAgentPolicyAlreadyIncludePackage,
 } from '../../common';
 import type {
   DeletePackagePoliciesResponse,
+  UpgradePackagePolicyResponse,
   PackagePolicyInput,
   NewPackagePolicyInput,
+  NewPackagePolicyInputStream,
   PackagePolicyConfigRecordEntry,
   PackagePolicyInputStream,
   PackageInfo,
   ListWithKuery,
   ListResult,
+  UpgradePackagePolicyDryRunResponseItem,
 } from '../../common';
 import { PACKAGE_POLICY_SAVED_OBJECT_TYPE } from '../constants';
 import {
@@ -42,6 +48,7 @@ import type {
   PackagePolicy,
   PackagePolicySOAttributes,
   RegistryPackage,
+  DryRunPackagePolicy,
 } from '../types';
 import type { ExternalCallback } from '..';
 
@@ -54,6 +61,10 @@ import { compileTemplate } from './epm/agent/agent';
 import { normalizeKuery } from './saved_object';
 import { appContextService } from '.';
 
+export type InputsOverride = Partial<NewPackagePolicyInput> & {
+  vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
+};
+
 const SAVED_OBJECT_TYPE = PACKAGE_POLICY_SAVED_OBJECT_TYPE;
 
 class PackagePolicyService {
@@ -427,6 +438,146 @@ class PackagePolicyService {
     return result;
   }
 
+  public async getUpgradePackagePolicyInfo(soClient: SavedObjectsClientContract, id: string) {
+    const packagePolicy = await this.get(soClient, id);
+    if (!packagePolicy) {
+      throw new Error(
+        i18n.translate('xpack.fleet.packagePolicy.policyNotFoundError', {
+          defaultMessage: 'Package policy with id {id} not found',
+          values: { id },
+        })
+      );
+    }
+
+    if (!packagePolicy.package?.name) {
+      throw new Error(
+        i18n.translate('xpack.fleet.packagePolicy.packageNotFoundError', {
+          defaultMessage: 'Package policy with id {id} has no named package',
+          values: { id },
+        })
+      );
+    }
+
+    const installedPackage = await getInstallation({
+      savedObjectsClient: soClient,
+      pkgName: packagePolicy.package.name,
+    });
+    if (!installedPackage) {
+      throw new Error(
+        i18n.translate('xpack.fleet.packagePolicy.packageNotInstalledError', {
+          defaultMessage: 'Cannot upgrade package policy {id} because {pkgName} is not installed',
+          values: { id, pkgName: packagePolicy.package.name },
+        })
+      );
+    }
+
+    const installedPkgInfo = await getPackageInfo({
+      savedObjectsClient: soClient,
+      pkgName: packagePolicy.package.name,
+      pkgVersion: installedPackage.version,
+    });
+
+    return {
+      packagePolicy: packagePolicy as Required<PackagePolicy>,
+      installedPkgInfo,
+    };
+  }
+
+  public async upgrade(
+    soClient: SavedObjectsClientContract,
+    esClient: ElasticsearchClient,
+    ids: string[],
+    options?: { user?: AuthenticatedUser }
+  ): Promise<UpgradePackagePolicyResponse> {
+    const result: UpgradePackagePolicyResponse = [];
+
+    for (const id of ids) {
+      try {
+        const { packagePolicy, installedPkgInfo } = await this.getUpgradePackagePolicyInfo(
+          soClient,
+          id
+        );
+
+        const updatePackagePolicy = overridePackageInputs(
+          {
+            ...omit(packagePolicy, 'id'),
+            inputs: packageToPackagePolicyInputs(installedPkgInfo),
+            package: {
+              ...packagePolicy.package,
+              version: installedPkgInfo.version,
+            },
+          },
+          packagePolicy.inputs as InputsOverride[]
+        );
+
+        updatePackagePolicy.inputs = await this.compilePackagePolicyInputs(
+          installedPkgInfo,
+          updatePackagePolicy.vars || {},
+          updatePackagePolicy.inputs as PackagePolicyInput[]
+        );
+
+        await this.update(soClient, esClient, id, updatePackagePolicy, options);
+        result.push({
+          id,
+          name: packagePolicy.name,
+          success: true,
+        });
+      } catch (error) {
+        result.push({
+          id,
+          success: false,
+          ...ingestErrorToResponseOptions(error),
+        });
+      }
+    }
+
+    return result;
+  }
+
+  public async getUpgradeDryRunDiff(
+    soClient: SavedObjectsClientContract,
+    id: string
+  ): Promise<UpgradePackagePolicyDryRunResponseItem> {
+    try {
+      const { packagePolicy, installedPkgInfo } = await this.getUpgradePackagePolicyInfo(
+        soClient,
+        id
+      );
+
+      const updatedPackagePolicy = overridePackageInputs(
+        {
+          ...omit(packagePolicy, 'id'),
+          inputs: packageToPackagePolicyInputs(installedPkgInfo),
+          package: {
+            ...packagePolicy.package,
+            version: installedPkgInfo.version,
+          },
+        },
+        packagePolicy.inputs as InputsOverride[],
+        true
+      );
+
+      updatedPackagePolicy.inputs = await this.compilePackagePolicyInputs(
+        installedPkgInfo,
+        updatedPackagePolicy.vars || {},
+        updatedPackagePolicy.inputs as PackagePolicyInput[]
+      );
+
+      const hasErrors = 'errors' in updatedPackagePolicy;
+
+      return {
+        name: updatedPackagePolicy.name,
+        diff: [packagePolicy, updatedPackagePolicy],
+        hasErrors,
+      };
+    } catch (error) {
+      return {
+        hasErrors: true,
+        ...ingestErrorToResponseOptions(error),
+      };
+    }
+  }
+
   public async buildPackagePolicyFromPackage(
     soClient: SavedObjectsClientContract,
     pkgName: string
@@ -480,7 +631,6 @@ class PackagePolicyService {
     const externalCallbacks = appContextService.getExternalCallbacks(externalCallbackType);
     if (externalCallbacks && externalCallbacks.size > 0) {
       let updatedNewData: NewPackagePolicy = newData;
-
       for (const callback of externalCallbacks) {
         const result = await callback(updatedNewData, context, request);
         if (externalCallbackType === 'packagePolicyCreate') {
@@ -668,3 +818,158 @@ export type PackagePolicyServiceInterface = PackagePolicyService;
 export const packagePolicyService = new PackagePolicyService();
 
 export type { PackagePolicyService };
+
+export function overridePackageInputs(
+  basePackagePolicy: NewPackagePolicy,
+  inputsOverride?: InputsOverride[],
+  dryRun?: boolean
+): DryRunPackagePolicy {
+  if (!inputsOverride) return basePackagePolicy;
+
+  const inputs = [...basePackagePolicy.inputs];
+  const packageName = basePackagePolicy.package!.name;
+  const errors = [];
+
+  for (const override of inputsOverride) {
+    const originalInput = inputs.find((i) => i.type === override.type);
+    if (!originalInput) {
+      const e = {
+        error: new Error(
+          i18n.translate('xpack.fleet.packagePolicyInputOverrideError', {
+            defaultMessage: 'Input type {inputType} does not exist on package {packageName}',
+            values: {
+              inputType: override.type,
+              packageName,
+            },
+          })
+        ),
+        package: { name: packageName, version: basePackagePolicy.package!.version },
+      };
+      if (dryRun) {
+        errors.push({
+          key: override.type,
+          message: String(e.error),
+        });
+        continue;
+      } else throw e;
+    }
+
+    if (typeof override.enabled !== 'undefined') originalInput.enabled = override.enabled;
+    if (typeof override.keep_enabled !== 'undefined')
+      originalInput.keep_enabled = override.keep_enabled;
+
+    if (override.vars) {
+      try {
+        deepMergeVars(override, originalInput);
+      } catch (e) {
+        const varName = e.message;
+        const err = {
+          error: new Error(
+            i18n.translate('xpack.fleet.packagePolicyVarOverrideError', {
+              defaultMessage:
+                'Var {varName} does not exist on {inputType} of package {packageName}',
+              values: {
+                varName,
+                inputType: override.type,
+                packageName,
+              },
+            })
+          ),
+          package: { name: packageName, version: basePackagePolicy.package!.version },
+        };
+        if (dryRun) {
+          errors.push({
+            key: `${override.type}.vars.${varName}`,
+            message: String(err.error),
+          });
+        } else throw err;
+      }
+    }
+
+    if (override.streams) {
+      for (const stream of override.streams) {
+        const originalStream = originalInput.streams.find(
+          (s) => s.data_stream.dataset === stream.data_stream.dataset
+        );
+        if (!originalStream) {
+          const streamSet = stream.data_stream.dataset;
+          const e = {
+            error: new Error(
+              i18n.translate('xpack.fleet.packagePolicyStreamOverrideError', {
+                defaultMessage:
+                  'Data stream {streamSet} does not exist on {inputType} of package {packageName}',
+                values: {
+                  streamSet,
+                  inputType: override.type,
+                  packageName,
+                },
+              })
+            ),
+            package: { name: packageName, version: basePackagePolicy.package!.version },
+          };
+          if (dryRun) {
+            errors.push({
+              key: `${override.type}.streams.${streamSet}`,
+              message: String(e.error),
+            });
+            continue;
+          } else throw e;
+        }
+
+        if (typeof stream.enabled !== 'undefined') originalStream.enabled = stream.enabled;
+
+        if (stream.vars) {
+          try {
+            deepMergeVars(stream as InputsOverride, originalStream);
+          } catch (e) {
+            const varName = e.message;
+            const streamSet = stream.data_stream.dataset;
+            const err = {
+              error: new Error(
+                i18n.translate('xpack.fleet.packagePolicyStreamVarOverrideError', {
+                  defaultMessage:
+                    'Var {varName} does not exist on {streamSet} for {inputType} of package {packageName}',
+                  values: {
+                    varName,
+                    streamSet,
+                    inputType: override.type,
+                    packageName,
+                  },
+                })
+              ),
+              package: { name: packageName, version: basePackagePolicy.package!.version },
+            };
+            if (dryRun) {
+              errors.push({
+                key: `${override.type}.streams.${streamSet}.${varName}`,
+                message: String(err.error),
+              });
+            } else throw err;
+          }
+        }
+      }
+    }
+  }
+
+  if (dryRun && errors.length) return { ...basePackagePolicy, inputs, errors };
+  return { ...basePackagePolicy, inputs };
+}
+
+function deepMergeVars(
+  override: NewPackagePolicyInput | InputsOverride,
+  original: NewPackagePolicyInput | NewPackagePolicyInputStream
+) {
+  const overrideVars = Array.isArray(override.vars)
+    ? override.vars
+    : Object.entries(override.vars!).map(([key, rest]) => ({
+        name: key,
+        ...rest,
+      }));
+  for (const { name, ...val } of overrideVars) {
+    if (!original.vars || !Reflect.has(original.vars, name)) {
+      throw new Error(name);
+    }
+    const originalVar = original.vars[name];
+    Reflect.set(original.vars, name, { ...originalVar, ...val });
+  }
+}

x-pack/plugins/fleet/server/services/preconfiguration.test.ts

@@ -123,6 +123,7 @@ jest.mock('./epm/packages/get', () => ({
 }));
 
 jest.mock('./package_policy', () => ({
+  ...jest.requireActual('./package_policy'),
   packagePolicyService: {
     create(soClient: any, esClient: any, newPackagePolicy: NewPackagePolicy) {
       return {

x-pack/plugins/fleet/server/services/preconfiguration.ts

@@ -14,8 +14,6 @@ import type {
   AgentPolicy,
   Installation,
   Output,
-  NewPackagePolicyInput,
-  NewPackagePolicyInputStream,
   PreconfiguredAgentPolicy,
   PreconfiguredPackage,
   PreconfigurationError,
@@ -32,6 +30,8 @@ import { getInstallation } from './epm/packages';
 import { ensurePackagesCompletedInstall } from './epm/packages/install';
 import { bulkInstallPackages } from './epm/packages/bulk_install_packages';
 import { agentPolicyService, addPackageToAgentPolicy } from './agent_policy';
+import type { InputsOverride } from './package_policy';
+import { overridePackageInputs } from './package_policy';
 
 interface PreconfigurationResult {
   policies: Array<{ id: string; updated_at: string }>;
@@ -39,10 +39,6 @@ interface PreconfigurationResult {
   nonFatalErrors: PreconfigurationError[];
 }
 
-export type InputsOverride = Partial<NewPackagePolicyInput> & {
-  vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
-};
-
 export async function ensurePreconfiguredPackagesAndPolicies(
   soClient: SavedObjectsClientContract,
   esClient: ElasticsearchClient,
@@ -285,128 +281,3 @@ async function addPreconfiguredPolicyPackages(
     );
   }
 }
-
-function overridePackageInputs(
-  basePackagePolicy: NewPackagePolicy,
-  inputsOverride?: InputsOverride[]
-) {
-  if (!inputsOverride) return basePackagePolicy;
-
-  const inputs = [...basePackagePolicy.inputs];
-  const packageName = basePackagePolicy.package!.name;
-
-  for (const override of inputsOverride) {
-    const originalInput = inputs.find((i) => i.type === override.type);
-    if (!originalInput) {
-      const e = {
-        error: new Error(
-          i18n.translate('xpack.fleet.packagePolicyInputOverrideError', {
-            defaultMessage: 'Input type {inputType} does not exist on package {packageName}',
-            values: {
-              inputType: override.type,
-              packageName,
-            },
-          })
-        ),
-        package: { name: packageName, version: basePackagePolicy.package!.version },
-      };
-      throw e;
-    }
-
-    if (typeof override.enabled !== 'undefined') originalInput.enabled = override.enabled;
-    if (typeof override.keep_enabled !== 'undefined')
-      originalInput.keep_enabled = override.keep_enabled;
-
-    if (override.vars) {
-      try {
-        deepMergeVars(override, originalInput);
-      } catch (e) {
-        const err = {
-          error: new Error(
-            i18n.translate('xpack.fleet.packagePolicyVarOverrideError', {
-              defaultMessage:
-                'Var {varName} does not exist on {inputType} of package {packageName}',
-              values: {
-                varName: e.message,
-                inputType: override.type,
-                packageName,
-              },
-            })
-          ),
-          package: { name: packageName, version: basePackagePolicy.package!.version },
-        };
-        throw err;
-      }
-    }
-
-    if (override.streams) {
-      for (const stream of override.streams) {
-        const originalStream = originalInput.streams.find(
-          (s) => s.data_stream.dataset === stream.data_stream.dataset
-        );
-        if (!originalStream) {
-          const e = {
-            error: new Error(
-              i18n.translate('xpack.fleet.packagePolicyStreamOverrideError', {
-                defaultMessage:
-                  'Data stream {streamSet} does not exist on {inputType} of package {packageName}',
-                values: {
-                  streamSet: stream.data_stream.dataset,
-                  inputType: override.type,
-                  packageName,
-                },
-              })
-            ),
-            package: { name: packageName, version: basePackagePolicy.package!.version },
-          };
-          throw e;
-        }
-
-        if (typeof stream.enabled !== 'undefined') originalStream.enabled = stream.enabled;
-
-        if (stream.vars) {
-          try {
-            deepMergeVars(stream as InputsOverride, originalStream);
-          } catch (e) {
-            const err = {
-              error: new Error(
-                i18n.translate('xpack.fleet.packagePolicyStreamVarOverrideError', {
-                  defaultMessage:
-                    'Var {varName} does not exist on {streamSet} for {inputType} of package {packageName}',
-                  values: {
-                    varName: e.message,
-                    streamSet: stream.data_stream.dataset,
-                    inputType: override.type,
-                    packageName,
-                  },
-                })
-              ),
-              package: { name: packageName, version: basePackagePolicy.package!.version },
-            };
-            throw err;
-          }
-        }
-      }
-    }
-  }
-
-  return { ...basePackagePolicy, inputs };
-}
-
-function deepMergeVars(
-  override: InputsOverride,
-  original: NewPackagePolicyInput | NewPackagePolicyInputStream
-) {
-  for (const { name, ...val } of override.vars!) {
-    if (!original.vars || !Reflect.has(original.vars, name)) {
-      throw new Error(name);
-    }
-    const originalVar = original.vars[name];
-    const newVar =
-      // If a single value was passed in to a multi field, ensure it gets converted to a multi
-      Array.isArray(originalVar.value) && !Array.isArray(val.value)
-        ? { ...val, value: [val.value] }
-        : val;
-    Reflect.set(original.vars, name, { ...originalVar, ...newVar });
-  }
-}

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

@@ -23,6 +23,7 @@ export {
   PackagePolicyInputStream,
   NewPackagePolicy,
   UpdatePackagePolicy,
+  DryRunPackagePolicy,
   PackagePolicySOAttributes,
   FullAgentPolicyInput,
   FullAgentPolicy,

x-pack/plugins/fleet/server/types/rest_spec/package_policy.ts

@@ -36,3 +36,10 @@ export const DeletePackagePoliciesRequestSchema = {
     force: schema.maybe(schema.boolean()),
   }),
 };
+
+export const UpgradePackagePoliciesRequestSchema = {
+  body: schema.object({
+    packagePolicyIds: schema.arrayOf(schema.string()),
+    dryRun: schema.maybe(schema.boolean()),
+  }),
+};

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.1.0/data_stream/test_stream/agent/stream/test_stream.yml.hbs

@@ -0,0 +1 @@
+config.version: "2"

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.1.0/data_stream/test_stream/fields/fields.yml

@@ -0,0 +1,16 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: >
+    Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: >
+    Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: >
+    Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: >
+    Event timestamp.

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.1.0/data_stream/test_stream/manifest.yml

@@ -0,0 +1,4 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.1.0/docs/README.md

@@ -0,0 +1,3 @@
+# Test package
+
+This is a test package for testing automated upgrades for package policies

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.1.0/manifest.yml

@@ -0,0 +1,15 @@
+format_version: 1.0.0
+name: package_policy_upgrade
+title: Tests package policy upgrades
+description: This is a test package for upgrading package policies
+version: 0.1.0
+categories: []
+release: beta
+type: integration
+license: basic
+
+requirement:
+  elasticsearch:
+    versions: '>7.7.0'
+  kibana:
+    versions: '>7.7.0'

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.2.0/data_stream/test_stream/agent/stream/test_stream.yml.hbs

@@ -0,0 +1 @@
+config.version: "2"

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.2.0/data_stream/test_stream/fields/fields.yml

@@ -0,0 +1,16 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: >
+    Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: >
+    Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: >
+    Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: >
+    Event timestamp.

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.2.0/data_stream/test_stream/manifest.yml

@@ -0,0 +1,11 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input
+    vars:
+    - name: test_var
+      type: text
+      title: Test Var
+      required: true
+      show_user: true
+      default: Test Value

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.2.0/docs/README.md

@@ -0,0 +1,3 @@
+# Test package
+
+This is a test package for testing automated upgrades for package policies

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.2.0/manifest.yml

@@ -0,0 +1,23 @@
+format_version: 1.0.0
+name: package_policy_upgrade
+title: Tests package policy upgrades
+description: This is a test package for upgrading package policies
+version: 0.3.0
+categories: []
+release: beta
+type: integration
+license: basic
+requirement:
+  elasticsearch:
+    versions: '>7.7.0'
+  kibana:
+    versions: '>7.7.0'
+policy_templates:
+  - name: package_policy_upgrade
+    title: Package Policy Upgrade
+    description: Test Package for Upgrading Package Policies
+    inputs:
+      - type: test_input
+        title: Test Input
+        description: Test Input
+        enabled: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.3.0/data_stream/test_stream/agent/stream/test_stream.yml.hbs

@@ -0,0 +1 @@
+config.version: "2"

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.3.0/data_stream/test_stream/fields/fields.yml

@@ -0,0 +1,16 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: >
+    Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: >
+    Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: >
+    Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: >
+    Event timestamp.

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.3.0/data_stream/test_stream/manifest.yml

@@ -0,0 +1,11 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input
+    # vars:
+    # - name: test_var
+    #   type: text
+    #   title: Test Var
+    #   required: true
+    #   show_user: true
+    #   default: Test Value

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.3.0/docs/README.md

@@ -0,0 +1,3 @@
+# Test package
+
+This is a test package for testing automated upgrades for package policies

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.3.0/manifest.yml

@@ -0,0 +1,23 @@
+format_version: 1.0.0
+name: package_policy_upgrade
+title: Tests package policy upgrades
+description: This is a test package for upgrading package policies
+version: 0.3.0
+categories: []
+release: beta
+type: integration
+license: basic
+requirement:
+  elasticsearch:
+    versions: '>7.7.0'
+  kibana:
+    versions: '>7.7.0'
+policy_templates:
+  - name: package_policy_upgrade
+    title: Package Policy Upgrade
+    description: Test Package for Upgrading Package Policies
+    inputs:
+      - type: test_input
+        title: Test Input
+        description: Test Input
+        enabled: true

x-pack/test/fleet_api_integration/apis/index.js

@@ -30,6 +30,7 @@ export default function ({ loadTestFile }) {
     loadTestFile(require.resolve('./package_policy/update'));
     loadTestFile(require.resolve('./package_policy/get'));
     loadTestFile(require.resolve('./package_policy/delete'));
+    loadTestFile(require.resolve('./package_policy/upgrade'));
 
     // Agent policies
     loadTestFile(require.resolve('./agent_policy/index'));

x-pack/test/fleet_api_integration/apis/package_policy/upgrade.ts

@@ -0,0 +1,289 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+import expect from '@kbn/expect';
+import { FtrProviderContext } from '../../../api_integration/ftr_provider_context';
+import { skipIfNoDockerRegistry } from '../../helpers';
+
+import {
+  UpgradePackagePolicyDryRunResponse,
+  UpgradePackagePolicyResponse,
+} from '../../../../plugins/fleet/common';
+
+export default function (providerContext: FtrProviderContext) {
+  const { getService } = providerContext;
+  const supertest = getService('supertest');
+  const esArchiver = getService('esArchiver');
+
+  describe('Package Policy - upgrade', async function () {
+    skipIfNoDockerRegistry(providerContext);
+    let agentPolicyId: string;
+    let packagePolicyId: string;
+
+    before(async () => {
+      await esArchiver.load('x-pack/test/functional/es_archives/empty_kibana');
+      await esArchiver.load('x-pack/test/functional/es_archives/fleet/empty_fleet_server');
+    });
+
+    describe('when package is installed', function () {
+      before(async function () {
+        await supertest
+          .post(`/api/fleet/epm/packages/package_policy_upgrade-0.3.0`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ force: true })
+          .expect(200);
+      });
+
+      after(async function () {
+        await supertest
+          .delete(`/api/fleet/epm/packages/package_policy_upgrade-0.3.0`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ force: true })
+          .expect(200);
+      });
+
+      after(async () => {
+        await getService('esArchiver').unload('x-pack/test/functional/es_archives/empty_kibana');
+        await getService('esArchiver').unload(
+          'x-pack/test/functional/es_archives/fleet/empty_fleet_server'
+        );
+      });
+
+      beforeEach(async function () {
+        const { body: agentPolicyResponse } = await supertest
+          .post(`/api/fleet/agent_policies`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            name: 'Test policy',
+            namespace: 'default',
+          })
+          .expect(200);
+
+        agentPolicyId = agentPolicyResponse.item.id;
+
+        const { body: packagePolicyResponse } = await supertest
+          .post(`/api/fleet/package_policies`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            name: 'package_policy_upgrade_1',
+            description: '',
+            namespace: 'default',
+            policy_id: agentPolicyId,
+            enabled: true,
+            output_id: '',
+            inputs: [],
+            package: {
+              name: 'package_policy_upgrade',
+              title: 'This is a test package for upgrading package policies',
+              version: '0.1.0',
+            },
+          })
+          .expect(200);
+
+        packagePolicyId = packagePolicyResponse.item.id;
+      });
+
+      afterEach(async function () {
+        await supertest
+          .post(`/api/fleet/package_policies/delete`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ packagePolicyIds: [packagePolicyId] })
+          .expect(200);
+
+        await supertest
+          .post('/api/fleet/agent_policies/delete')
+          .set('kbn-xsrf', 'xxxx')
+          .send({ agentPolicyId })
+          .expect(200);
+      });
+
+      it('should return valid diff when "dryRun: true" is provided', async function () {
+        const { body }: { body: UpgradePackagePolicyDryRunResponse } = await supertest
+          .post(`/api/fleet/package_policies/upgrade`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            packagePolicyIds: [packagePolicyId],
+            dryRun: true,
+          })
+          .expect(200);
+
+        expect(body.length).to.be(1);
+        expect(body[0].diff?.length).to.be(2);
+        expect(body[0].hasErrors).to.be(false);
+
+        const [currentPackagePolicy, proposedPackagePolicy] = body[0].diff ?? [];
+
+        expect(currentPackagePolicy?.package?.version).to.be('0.1.0');
+        expect(proposedPackagePolicy?.package?.version).to.be('0.3.0');
+      });
+
+      it('should upgrade package policy when "dryRun: false" is provided', async function () {
+        const { body }: { body: UpgradePackagePolicyResponse } = await supertest
+          .post(`/api/fleet/package_policies/upgrade`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            packagePolicyIds: [packagePolicyId],
+            dryRun: false,
+          })
+          .expect(200);
+
+        expect(body.length).to.be(1);
+        expect(body[0].success).to.be(true);
+      });
+    });
+
+    describe('when upgrading to a version where an input has been removed', function () {
+      before(async function () {
+        await supertest
+          .post(`/api/fleet/epm/packages/package_policy_upgrade-0.3.0`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ force: true })
+          .expect(200);
+      });
+
+      after(async function () {
+        await supertest
+          .delete(`/api/fleet/epm/packages/package_policy_upgrade-0.3.0`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ force: true })
+          .expect(200);
+      });
+
+      beforeEach(async function () {
+        const { body: agentPolicyResponse } = await supertest
+          .post(`/api/fleet/agent_policies`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            name: 'Test policy',
+            namespace: 'default',
+          })
+          .expect(200);
+
+        agentPolicyId = agentPolicyResponse.item.id;
+
+        const { body: packagePolicyResponse } = await supertest
+          .post(`/api/fleet/package_policies`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            name: 'package_policy_upgrade_1',
+            description: '',
+            namespace: 'default',
+            policy_id: agentPolicyId,
+            enabled: true,
+            output_id: '',
+            inputs: [
+              {
+                policy_template: 'package_policy_upgrade',
+                type: 'test_input',
+                enabled: true,
+                streams: [
+                  {
+                    id: 'test-package_policy_upgrade-xxxx',
+                    enabled: true,
+                    data_stream: {
+                      type: 'test_stream',
+                      dataset: 'package_policy_upgrade.test_stream',
+                    },
+                    vars: {
+                      test_var: {
+                        value: 'Test Value',
+                      },
+                    },
+                  },
+                ],
+              },
+            ],
+            package: {
+              name: 'package_policy_upgrade',
+              title: 'This is a test package for upgrading package policies',
+              // The upgrade from `0.2.0` to `0.3.0` incurs an error state because a breaking
+              // change exists between these test package version
+              version: '0.2.0',
+            },
+          });
+
+        packagePolicyId = packagePolicyResponse.item.id;
+      });
+
+      afterEach(async function () {
+        await supertest
+          .post(`/api/fleet/package_policies/delete`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({ packagePolicyIds: [packagePolicyId] })
+          .expect(200);
+
+        await supertest
+          .post('/api/fleet/agent_policies/delete')
+          .set('kbn-xsrf', 'xxxx')
+          .send({ agentPolicyId })
+          .expect(200);
+      });
+
+      describe('when "dryRun: true" is provided', function () {
+        it('should return a diff with errors', async function () {
+          const { body }: { body: UpgradePackagePolicyDryRunResponse } = await supertest
+            .post(`/api/fleet/package_policies/upgrade`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              packagePolicyIds: [packagePolicyId],
+              dryRun: true,
+            })
+            .expect(200);
+
+          expect(body.length).to.be(1);
+          expect(body[0].diff?.length).to.be(2);
+          expect(body[0].hasErrors).to.be(true);
+        });
+      });
+
+      describe('when "dryRun: false" is provided', function () {
+        it('should respond with an error', async function () {
+          const { body }: { body: UpgradePackagePolicyResponse } = await supertest
+            .post(`/api/fleet/package_policies/upgrade`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              packagePolicyIds: [packagePolicyId],
+              dryRun: false,
+            })
+            .expect(200);
+
+          expect(body.length).to.be(1);
+          expect(body[0].success).to.be(false);
+        });
+      });
+    });
+
+    describe('when no package policy is not found', function () {
+      it('should return an 200 with errors when "dryRun:true" is provided', async function () {
+        const { body }: { body: UpgradePackagePolicyDryRunResponse } = await supertest
+          .post(`/api/fleet/package_policies/upgrade`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            packagePolicyIds: ['xxxx', 'yyyy'],
+            dryRun: true,
+          })
+          .expect(200);
+
+        expect(body[0].hasErrors).to.be(true);
+        expect(body[1].hasErrors).to.be(true);
+      });
+
+      it('should return a 200 with errors and "success:false" when "dryRun:false" is provided', async function () {
+        const { body }: { body: UpgradePackagePolicyResponse } = await supertest
+          .post(`/api/fleet/package_policies/upgrade`)
+          .set('kbn-xsrf', 'xxxx')
+          .send({
+            packagePolicyIds: ['xxxx', 'yyyy'],
+            dryRun: false,
+          })
+          .expect(200);
+
+        expect(body[0].success).to.be(false);
+        expect(body[1].success).to.be(false);
+      });
+    });
+  });
+}