[Fleet] Support migrating from integration type package to input type…

… package (#141472)

* it works

* neaten

* test failing

* fix test pkgs

* fix input to policy template matching

* move to zip packages

* never cache registry info for input packages

* remove .only

* fix dataset name in unit test

x-pack/plugins/fleet/common/services/policy_template.test.ts

@@ -228,7 +228,7 @@ describe('getNormalizedDataStreams', () => {
     ).toEqual([
       {
         type: 'foo',
-        dataset: 'nginx.foo',
+        dataset: 'nginx.bar',
         title: expect.any(String),
         release: 'ga',
         package: 'nginx',

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

@@ -106,4 +106,4 @@ const addDatasetVarIfNotPresent = (vars?: RegistryVarsEntry[]): RegistryVarsEntr
 const createDefaultDatasetName = (
   packageInfo: PackageInfo,
   policyTemplate: RegistryPolicyInputOnlyTemplate
-): string => packageInfo.name + '.' + policyTemplate.type;
+): string => packageInfo.name + '.' + policyTemplate.name;

x-pack/plugins/fleet/server/services/epm/registry/index.ts

@@ -237,13 +237,14 @@ export async function fetchCategories(
   return fetchUrl(url.toString()).then(JSON.parse);
 }
 
-export async function getInfo(name: string, version: string, options: { cache?: boolean } = {}) {
-  const cache = options.cache ?? true;
+export async function getInfo(name: string, version: string) {
   return withPackageSpan('Fetch package info', async () => {
     let packageInfo = getPackageInfo({ name, version });
     if (!packageInfo) {
       packageInfo = await fetchInfo(name, version);
-      if (cache) setPackageInfo({ name, version, packageInfo });
+      // only cache registry pkg info for integration pkgs because
+      // input type packages must get their pkg info from the archive
+      if (packageInfo.type === 'integration') setPackageInfo({ name, version, packageInfo });
     }
     return packageInfo as RegistryPackage;
   });
@@ -324,7 +325,7 @@ export async function fetchArchiveBuffer({
   verificationResult?: PackageVerificationResult;
 }> {
   const logger = appContextService.getLogger();
-  const { download: archivePath } = await getInfo(pkgName, pkgVersion, { cache: false });
+  const { download: archivePath } = await getInfo(pkgName, pkgVersion);
   const archiveUrl = `${getRegistryUrl()}${archivePath}`;
   const archiveBuffer = await getResponseStream(archiveUrl).then(streamToBuffer);
   if (shouldVerify) {
@@ -352,9 +353,7 @@ export async function getPackageArchiveSignatureOrUndefined({
   pkgVersion: string;
   logger: Logger;
 }): Promise<string | undefined> {
-  const { signature_path: signaturePath } = await getInfo(pkgName, pkgVersion, {
-    cache: false,
-  });
+  const { signature_path: signaturePath } = await getInfo(pkgName, pkgVersion);
 
   if (!signaturePath) {
     logger.debug(

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

@@ -801,6 +801,7 @@ class PackagePolicyClientImpl implements PackagePolicyClient {
         savedObjectsClient: soClient,
         pkgName: packagePolicy!.package!.name,
         pkgVersion: pkgVersion ?? '',
+        skipArchive: true,
       });
     }
 
@@ -1625,7 +1626,7 @@ export function updatePackageInputs(
 
       // Ignore any inputs removed from this policy template in the new package version
       const policyTemplateStillIncludesInput = isInputOnlyPolicyTemplate(policyTemplate)
-        ? policyTemplate.type === input.type
+        ? policyTemplate.input === input.type
         : policyTemplate.inputs?.some(
             (policyTemplateInput) => policyTemplateInput.type === input.type
           ) ?? false;
@@ -1688,11 +1689,25 @@ export function updatePackageInputs(
     }
 
     if (update.streams) {
+      const isInputPkgUpdate =
+        packageInfo.type === 'input' &&
+        update.streams.length === 1 &&
+        originalInput?.streams.length === 1;
+
       for (const stream of update.streams) {
         let originalStream = originalInput?.streams.find(
           (s) => s.data_stream.dataset === stream.data_stream.dataset
         );
 
+        // this handles the input only pkg case where the new stream cannot have a dataset name
+        // so will never match. Input only packages only ever have one stream.
+        if (!originalStream && isInputPkgUpdate) {
+          originalStream = {
+            ...update.streams[0],
+            vars: originalInput?.streams[0].vars,
+          };
+        }
+
         if (originalStream === undefined) {
           originalInput.streams.push(stream);
           continue;
@@ -1703,7 +1718,10 @@ export function updatePackageInputs(
         }
 
         if (stream.vars) {
-          const indexOfStream = originalInput.streams.indexOf(originalStream);
+          // streams wont match for input pkgs
+          const indexOfStream = isInputPkgUpdate
+            ? 0
+            : originalInput.streams.indexOf(originalStream);
           originalInput.streams[indexOfStream] = deepMergeVars(
             originalStream,
             stream as InputsOverride,

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

@@ -19,21 +19,14 @@ export default function (providerContext: FtrProviderContext) {
   const esArchiver = getService('esArchiver');
   const kibanaServer = getService('kibanaServer');
 
-  function withTestPackageVersion(version: string) {
+  function withTestPackage(name: string, version: string) {
+    const pkgRoute = `/api/fleet/epm/packages/${name}/${version}`;
     before(async function () {
-      await supertest
-        .post(`/api/fleet/epm/packages/package_policy_upgrade/${version}`)
-        .set('kbn-xsrf', 'xxxx')
-        .send({ force: true })
-        .expect(200);
+      await supertest.post(pkgRoute).set('kbn-xsrf', 'xxxx').send({ force: true }).expect(200);
     });
 
     after(async function () {
-      await supertest
-        .delete(`/api/fleet/epm/packages/package_policy_upgrade/${version}`)
-        .set('kbn-xsrf', 'xxxx')
-        .send({ force: true })
-        .expect(200);
+      await supertest.delete(pkgRoute).set('kbn-xsrf', 'xxxx').send({ force: true }).expect(200);
     });
   }
 
@@ -121,7 +114,7 @@ export default function (providerContext: FtrProviderContext) {
       });
 
       describe('dry run', function () {
-        withTestPackageVersion('0.2.0-add-non-required-test-var');
+        withTestPackage('package_policy_upgrade', '0.2.0-add-non-required-test-var');
         it('returns a valid diff', async function () {
           const { body }: { body: UpgradePackagePolicyDryRunResponse } = await supertest
             .post(`/api/fleet/package_policies/upgrade/dryrun`)
@@ -145,7 +138,7 @@ export default function (providerContext: FtrProviderContext) {
       });
 
       describe('upgrade', function () {
-        withTestPackageVersion('0.2.0-add-non-required-test-var');
+        withTestPackage('package_policy_upgrade', '0.2.0-add-non-required-test-var');
         it('should respond with an error', async function () {
           // upgrade policy to 0.2.0
           await supertest
@@ -176,7 +169,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version with no breaking changes', function () {
-      withTestPackageVersion('0.2.5-non-breaking-change');
+      withTestPackage('package_policy_upgrade', '0.2.5-non-breaking-change');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -290,7 +283,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where a non-required variable has been added', function () {
-      withTestPackageVersion('0.2.0-add-non-required-test-var');
+      withTestPackage('package_policy_upgrade', '0.2.0-add-non-required-test-var');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -394,7 +387,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where a variable has been removed', function () {
-      withTestPackageVersion('0.3.0-remove-test-var');
+      withTestPackage('package_policy_upgrade', '0.3.0-remove-test-var');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -497,7 +490,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where a required variable has been added', function () {
-      withTestPackageVersion('0.4.0-add-test-var-as-bool');
+      withTestPackage('package_policy_upgrade', '0.4.0-add-test-var-as-bool');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -590,7 +583,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where a variable has changed types', function () {
-      withTestPackageVersion('0.4.0-add-test-var-as-bool');
+      withTestPackage('package_policy_upgrade', '0.4.0-add-test-var-as-bool');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -687,7 +680,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where inputs have been restructured', function () {
-      withTestPackageVersion('0.5.0-restructure-inputs');
+      withTestPackage('package_policy_upgrade', '0.5.0-restructure-inputs');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -786,7 +779,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where policy templates have been restructured', function () {
-      withTestPackageVersion('0.6.0-restructure-policy-templates');
+      withTestPackage('package_policy_upgrade', '0.6.0-restructure-policy-templates');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -917,7 +910,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe('when upgrading to a version where an input with no variables has variables added', function () {
-      withTestPackageVersion('0.8.0-add-vars-to-stream-with-no-vars');
+      withTestPackage('package_policy_upgrade', '0.8.0-add-vars-to-stream-with-no-vars');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -1056,7 +1049,7 @@ export default function (providerContext: FtrProviderContext) {
     });
 
     describe("when policy's package version is up to date", function () {
-      withTestPackageVersion('0.1.0');
+      withTestPackage('package_policy_upgrade', '0.1.0');
 
       beforeEach(async function () {
         const { body: agentPolicyResponse } = await supertest
@@ -1146,5 +1139,92 @@ export default function (providerContext: FtrProviderContext) {
         });
       });
     });
+
+    describe('when upgrading from an integration package to an input package where a required variable has been added', function () {
+      withTestPackage('integration_to_input', '0.9.1');
+
+      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({
+            policy_id: agentPolicyId,
+            package: {
+              name: 'integration_to_input',
+              version: '0.9.0',
+            },
+            name: 'integration_to_input-1',
+            description: '',
+            namespace: 'default',
+            inputs: {
+              'logs-logfile': {
+                enabled: true,
+                streams: {
+                  'integration_to_input.log': {
+                    enabled: true,
+                    vars: {
+                      paths: ['/tmp/test.log'],
+                      'data_stream.dataset': 'generic',
+                      custom: '',
+                    },
+                  },
+                },
+              },
+            },
+          });
+
+        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 diff with errors', async function () {
+          const { body }: { body: UpgradePackagePolicyDryRunResponse } = await supertest
+            .post(`/api/fleet/package_policies/upgrade/dryrun`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              packagePolicyIds: [packagePolicyId],
+            })
+            .expect(200);
+          expect(body[0].hasErrors).to.be(true);
+        });
+      });
+
+      describe('upgrade', function () {
+        it('fails to upgrade package policy', async function () {
+          await supertest
+            .post(`/api/fleet/package_policies/upgrade`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              packagePolicyIds: [packagePolicyId],
+            })
+            .expect(400);
+        });
+      });
+    });
   });
 }