[Fleet] Fix upgrades for packages with restructured inputs (#109887)

* Fix upgrades for packages with restructured inputs

Addresses errors surfaced when testing upgrades from AWS 0.6.1 to 0.10.4.
Namely, when inputs are removed from a package between versions,we were
initially throwing errors for each input in the new package that didn't
exist on the outdated package version. Now, we instead simply skip over
cases like this in which an input no longer exists on the new package version.

* Add basic test cases for restructured packages

x-pack/plugins/fleet/common/services/validate_package_policy.ts

@@ -97,12 +97,14 @@ export const validatePackagePolicy = (
   >((varDefs, policyTemplate) => {
     (policyTemplate.inputs || []).forEach((input) => {
       const varDefKey = hasIntegrations ? `${policyTemplate.name}-${input.type}` : input.type;
+
       if ((input.vars || []).length) {
         varDefs[varDefKey] = keyBy(input.vars || [], 'name');
       }
     });
     return varDefs;
   }, {});
+
   const streamsByDatasetAndInput = (packageInfo.data_streams || []).reduce<
     Record<string, RegistryStream>
   >((streams, dataStream) => {
@@ -149,6 +151,7 @@ export const validatePackagePolicy = (
     if (input.streams.length) {
       input.streams.forEach((stream) => {
         const streamValidationResults: PackagePolicyConfigValidationResults = {};
+
         const streamVarDefs =
           streamVarDefsByDatasetAndInput[`${stream.data_stream.dataset}-${input.type}`];
 
@@ -157,7 +160,7 @@ export const validatePackagePolicy = (
           streamValidationResults.vars = Object.entries(stream.vars).reduce(
             (results, [name, configEntry]) => {
               results[name] =
-                streamVarDefs[name] && input.enabled && stream.enabled
+                streamVarDefs && streamVarDefs[name] && input.enabled && stream.enabled
                   ? validatePackagePolicyConfig(configEntry, streamVarDefs[name])
                   : null;
               return results;

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

@@ -911,41 +911,51 @@ export function overridePackageInputs(
 ): DryRunPackagePolicy {
   if (!inputsOverride) return basePackagePolicy;
 
-  const inputs = [...basePackagePolicy.inputs];
-  const packageName = basePackagePolicy.package!.name;
-  let errors = [];
+  const availablePolicyTemplates = packageInfo.policy_templates ?? [];
+
+  const inputs = [
+    ...basePackagePolicy.inputs.filter((input) => {
+      if (!input.policy_template) {
+        return true;
+      }
+
+      const policyTemplate = availablePolicyTemplates.find(
+        ({ name }) => name === input.policy_template
+      );
+
+      // Ignore any policy template removes in the new package version
+      if (!policyTemplate) {
+        return false;
+      }
+
+      // Ignore any inputs removed from this policy template in the new package version
+      const policyTemplateStillIncludesInput =
+        policyTemplate.inputs?.some(
+          (policyTemplateInput) => policyTemplateInput.type === input.type
+        ) ?? false;
+
+      return policyTemplateStillIncludesInput;
+    }),
+  ];
 
   for (const override of inputsOverride) {
     let originalInput = inputs.find((i) => i.type === override.type);
 
+    // If there's no corresponding input on the original package policy, just
+    // take the override value from the new package as-is. This case typically
+    // occurs when inputs or package policies are added/removed between versions.
     if (!originalInput) {
-      const e = {
-        error: new IngestManagerError(
-          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;
-      }
+      inputs.push(override as NewPackagePolicyInput);
+      continue;
     }
 
-    if (typeof override.enabled !== 'undefined') originalInput.enabled = override.enabled;
-    if (typeof override.keep_enabled !== 'undefined')
+    if (typeof override.enabled !== 'undefined') {
+      originalInput.enabled = override.enabled;
+    }
+
+    if (typeof override.keep_enabled !== 'undefined') {
       originalInput.keep_enabled = override.keep_enabled;
+    }
 
     if (override.vars) {
       originalInput = deepMergeVars(originalInput, override);
@@ -957,36 +967,7 @@ export function overridePackageInputs(
           (s) => s.data_stream.dataset === stream.data_stream.dataset
         );
 
-        if (!originalStream) {
-          const streamSet = stream.data_stream.dataset;
-          const e = {
-            error: new IngestManagerError(
-              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') {
+        if (typeof stream.enabled !== 'undefined' && originalStream) {
           originalStream.enabled = stream.enabled;
         }
 
@@ -1012,20 +993,22 @@ export function overridePackageInputs(
       }))
       .filter(({ message }) => !!message);
 
-    errors = [...errors, ...responseFormattedValidationErrors];
-  }
+    if (responseFormattedValidationErrors.length) {
+      if (dryRun) {
+        return { ...resultingPackagePolicy, errors: responseFormattedValidationErrors };
+      }
 
-  if (errors.length) {
-    if (dryRun) {
-      return { ...resultingPackagePolicy, errors };
+      throw new IngestManagerError(
+        i18n.translate('xpack.fleet.packagePolicyInvalidError', {
+          defaultMessage: 'Package policy is invalid: {errors}',
+          values: {
+            errors: responseFormattedValidationErrors
+              .map(({ key, message }) => `${key}: ${message}`)
+              .join('\n'),
+          },
+        })
+      );
     }
-
-    throw new IngestManagerError(
-      i18n.translate('xpack.fleet.packagePolicyInvalidError', {
-        defaultMessage: 'Package policy is invalid: {errors}',
-        values: { errors: errors.map(({ key, message }) => `${key}: ${message}`).join('\n') },
-      })
-    );
   }
 
   return resultingPackagePolicy;

x-pack/plugins/translations/translations/ja-JP.json

@@ -9663,8 +9663,6 @@
     "xpack.fleet.oldAppTitle": "Ingest Manager",
     "xpack.fleet.overviewPageSubtitle": "ElasticElasticエージェントの集中管理",
     "xpack.fleet.overviewPageTitle": "Fleet",
-    "xpack.fleet.packagePolicyInputOverrideError": "パッケージ{packageName}には入力タイプ{inputType}が存在しません。",
-    "xpack.fleet.packagePolicyStreamOverrideError": "パッケージ{packageName}の{inputType}にはデータストリーム{streamSet}が存在しません",
     "xpack.fleet.packagePolicyValidation.invalidArrayErrorMessage": "無効なフォーマット",
     "xpack.fleet.packagePolicyValidation.invalidYamlFormatErrorMessage": "YAML形式が無効です",
     "xpack.fleet.packagePolicyValidation.nameRequiredErrorMessage": "名前が必要です",

x-pack/plugins/translations/translations/zh-CN.json

@@ -9936,8 +9936,6 @@
     "xpack.fleet.oldAppTitle": "采集管理器",
     "xpack.fleet.overviewPageSubtitle": "Elastic 代理的集中管理",
     "xpack.fleet.overviewPageTitle": "Fleet",
-    "xpack.fleet.packagePolicyInputOverrideError": "输入类型 {inputType} 在软件包 {packageName} 上不存在",
-    "xpack.fleet.packagePolicyStreamOverrideError": "数据流 {streamSet} 在软件包 {packageName} 的 {inputType} 上不存在",
     "xpack.fleet.packagePolicyValidation.invalidArrayErrorMessage": "格式无效",
     "xpack.fleet.packagePolicyValidation.invalidYamlFormatErrorMessage": "YAML 格式无效",
     "xpack.fleet.packagePolicyValidation.nameRequiredErrorMessage": "“名称”必填",

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.5.0-restructure-inputs/data_stream/test_stream_new/agent/stream/stream.yml.hbs

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

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.5.0-restructure-inputs/data_stream/test_stream_new/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.5.0-restructure-inputs/data_stream/test_stream_new/manifest.yml

@@ -0,0 +1,17 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input_new
+    vars:
+    - name: test_var_new
+      type: text
+      title: Test Var New
+      default: Test Var New
+      required: true
+      show_user: true
+    - name: test_var_new_2
+      type: text
+      title: Test Var New 2
+      default: Test Var New 2
+      required: true
+      show_user: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.5.0-restructure-inputs/data_stream/test_stream_new_2/agent/stream/stream.yml.hbs

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

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.5.0-restructure-inputs/data_stream/test_stream_new_2/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.5.0-restructure-inputs/data_stream/test_stream_new_2/manifest.yml

@@ -0,0 +1,17 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input_new_2
+    vars:
+    - name: test_input_new_2_var_1
+      type: text
+      title: Test Input New 2 Var 1
+      default: Test Input New 2 Var 1
+      required: true
+      show_user: true
+    - name: test_input_new_2_var_2
+      type: text
+      title: Test Input New 2 Var 2
+      default: Test Input New 2 Var 2
+      required: true
+      show_user: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.5.0-restructure-inputs/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.5.0-restructure-inputs/manifest.yml

@@ -0,0 +1,27 @@
+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.5.0-restructure-inputs
+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_new
+        title: Test Input New
+        description: Test Input New
+        enabled: true
+      - type: test_input_new_2
+        title: Test Input New 2
+        description: Test Input New 2
+        enabled: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.6.0-restructure-policy-templates/data_stream/test_stream_new/agent/stream/stream.yml.hbs

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

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.6.0-restructure-policy-templates/data_stream/test_stream_new/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.6.0-restructure-policy-templates/data_stream/test_stream_new/manifest.yml

@@ -0,0 +1,17 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input_new
+    vars:
+    - name: test_var_new
+      type: text
+      title: Test Var New
+      default: Test Var New
+      required: true
+      show_user: true
+    - name: test_var_new_2
+      type: text
+      title: Test Var New 2
+      default: Test Var New 2
+      required: true
+      show_user: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.6.0-restructure-policy-templates/data_stream/test_stream_new_2/agent/stream/stream.yml.hbs

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

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.6.0-restructure-policy-templates/data_stream/test_stream_new_2/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.6.0-restructure-policy-templates/data_stream/test_stream_new_2/manifest.yml

@@ -0,0 +1,17 @@
+title: Test stream
+type: logs
+streams:
+  - input: test_input_new_2
+    vars:
+    - name: test_input_new_2_var_1
+      type: text
+      title: Test Input New 2 Var 1
+      default: Test Input New 2 Var 1
+      required: true
+      show_user: true
+    - name: test_input_new_2_var_2
+      type: text
+      title: Test Input New 2 Var 2
+      default: Test Input New 2 Var 2
+      required: true
+      show_user: true

x-pack/test/fleet_api_integration/apis/fixtures/test_packages/package_policy_upgrade/0.6.0-restructure-policy-templates/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.6.0-restructure-policy-templates/manifest.yml

@@ -0,0 +1,27 @@
+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.6.0-restructure-policy-templates
+categories: []
+release: beta
+type: integration
+license: basic
+requirement:
+  elasticsearch:
+    versions: '>7.7.0'
+  kibana:
+    versions: '>7.7.0'
+policy_templates:
+  - name: package_policy_upgrade_new
+    title: Package Policy Upgrade New
+    description: Test Package for Upgrading Package Policies
+    inputs:
+      - type: test_input_new
+        title: Test Input New
+        description: Test Input New
+        enabled: true
+      - type: test_input_new_2
+        title: Test Input New 2
+        description: Test Input New 2
+        enabled: true

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

@@ -357,7 +357,7 @@ export default function (providerContext: FtrProviderContext) {
       });
 
       describe('upgrade', function () {
-        it('successfully upgrade package policy', async function () {
+        it('successfully upgrades package policy', async function () {
           const { body }: { body: UpgradePackagePolicyResponse } = await supertest
             .post(`/api/fleet/package_policies/upgrade`)
             .set('kbn-xsrf', 'xxxx')
@@ -573,6 +573,236 @@ export default function (providerContext: FtrProviderContext) {
       });
     });
 
+    describe('when upgrading to a version where inputs have been restructured', function () {
+      withTestPackageVersion('0.5.0-restructure-inputs');
+
+      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',
+              version: '0.2.0-add-non-required-test-var',
+            },
+          });
+
+        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('dry run', function () {
+        it('returns a valid diff', 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[0].hasErrors).to.be(false);
+        });
+      });
+
+      describe('upgrade', function () {
+        it('successfully upgrades package policy', 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[0].success).to.be(true);
+        });
+      });
+    });
+
+    describe('when upgrading to a version where policy templates have been restructured', function () {
+      withTestPackageVersion('0.6.0-restructure-policy-templates');
+
+      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_new',
+                enabled: true,
+                streams: [
+                  {
+                    id: 'test-package_policy_upgrade-xxxx',
+                    enabled: true,
+                    data_stream: {
+                      type: 'test_stream_new',
+                      dataset: 'package_policy_upgrade.test_stream_new',
+                    },
+                    vars: {
+                      test_var_new: {
+                        value: 'Test value 1',
+                      },
+                      test_var_new_2: {
+                        value: 'Test value 2',
+                      },
+                    },
+                  },
+                ],
+              },
+              {
+                policy_template: 'package_policy_upgrade',
+                type: 'test_input_new_2',
+                enabled: true,
+                streams: [
+                  {
+                    id: 'test-package_policy_upgrade-xxxx',
+                    enabled: true,
+                    data_stream: {
+                      type: 'test_stream_new_2',
+                      dataset: 'package_policy_upgrade.test_stream_new_2',
+                    },
+                    vars: {
+                      test_var_new_2_var_1: {
+                        value: 'Test value 1',
+                      },
+                      test_var_new_2_var_2: {
+                        value: 'Test value 2',
+                      },
+                    },
+                  },
+                ],
+              },
+            ],
+            package: {
+              name: 'package_policy_upgrade',
+              title: 'This is a test package for upgrading package policies',
+              version: '0.5.0-restructure-inputs',
+            },
+          });
+
+        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('dry run', function () {
+        it('returns a valid diff', 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[0].hasErrors).to.be(false);
+        });
+      });
+
+      describe('upgrade', function () {
+        it('successfully upgrades package policy', 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[0].success).to.be(true);
+        });
+      });
+    });
+
     describe('when 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