[Fleet] rollback input package install on failure

x-pack/plugins/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/single_page_layout/hooks/form.tsx

@@ -272,7 +272,8 @@ export function useOnSubmit({
           setFormState('LOADING');
           if ((withSysMonitoring || newAgentPolicy.monitoring_enabled?.length) ?? 0 > 0) {
             const packagesToPreinstall: Array<string | { name: string; version: string }> = [];
-            if (packageInfo) {
+            // skip preinstall of input package, to be able to rollback when package policy creation fails
+            if (packageInfo && packageInfo.type !== 'input') {
               packagesToPreinstall.push({ name: packageInfo.name, version: packageInfo.version });
             }
             if (withSysMonitoring) {

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

@@ -48,7 +48,12 @@ import {
   PackagePolicyNotFoundError,
   PackagePolicyRequestError,
 } from '../../errors';
-import { getInstallations, getPackageInfo } from '../../services/epm/packages';
+import {
+  getInstallation,
+  getInstallations,
+  getPackageInfo,
+  removeInstallation,
+} from '../../services/epm/packages';
 import { PACKAGES_SAVED_OBJECT_TYPE, SO_SEARCH_LIMIT } from '../../constants';
 import {
   simplifiedPackagePolicytoNewPackagePolicy,
@@ -229,6 +234,7 @@ export const createPackagePolicyHandler: FleetRequestHandler<
   const user = appContextService.getSecurity()?.authc.getCurrentUser(request) || undefined;
   const { force, id, package: pkg, ...newPolicy } = request.body;
   const authorizationHeader = HTTPAuthorizationHeader.parseFromRequest(request, user?.username);
+  let wasPackageAlreadyInstalled = false;
 
   if ('output_id' in newPolicy) {
     // TODO Remove deprecated APIs https://github.com/elastic/kibana/issues/121485
@@ -258,6 +264,12 @@ export const createPackagePolicyHandler: FleetRequestHandler<
       } as NewPackagePolicy);
     }
 
+    const installation = await getInstallation({
+      savedObjectsClient: soClient,
+      pkgName: pkg!.name,
+    });
+    wasPackageAlreadyInstalled = installation?.install_status === 'installed';
+
     // Create package policy
     const packagePolicy = await fleetContext.packagePolicyService.asCurrentUser.create(
       soClient,
@@ -282,6 +294,27 @@ export const createPackagePolicyHandler: FleetRequestHandler<
       },
     });
   } catch (error) {
+    appContextService
+      .getLogger()
+      .error(`Error while creating package policy due to error: ${error.message}`);
+    if (!wasPackageAlreadyInstalled) {
+      const installation = await getInstallation({
+        savedObjectsClient: soClient,
+        pkgName: pkg!.name,
+      });
+      if (installation) {
+        appContextService
+          .getLogger()
+          .info(`rollback ${pkg!.name}-${pkg!.version} package installation after error`);
+        await removeInstallation({
+          savedObjectsClient: soClient,
+          pkgName: pkg!.name,
+          pkgVersion: pkg!.version,
+          esClient,
+        });
+      }
+    }
+
     if (error.statusCode) {
       return response.customError({
         statusCode: error.statusCode,

x-pack/plugins/fleet/server/services/epm/packages/install_index_template_pipeline.ts

@@ -68,16 +68,7 @@ export async function installIndexTemplatesAndPipelines({
   // cleanup in the case that a single asset fails to install.
   let newEsReferences: EsAssetReference[] = [];
 
-  if (onlyForDataStreams) {
-    // if onlyForDataStreams is present that means we are in create package policy flow
-    // not install flow, meaning we do not have a lock on the installation SO
-    // so we need to use optimistic concurrency control
-    newEsReferences = await optimisticallyAddEsAssetReferences(
-      savedObjectsClient,
-      packageInstallContext.packageInfo.name,
-      [...preparedIngestPipelines.assetsToAdd, ...preparedIndexTemplates.assetsToAdd]
-    );
-  } else {
+  if (!onlyForDataStreams) {
     newEsReferences = await updateEsAssetReferences(
       savedObjectsClient,
       packageInstallContext.packageInfo.name,
@@ -103,6 +94,18 @@ export async function installIndexTemplatesAndPipelines({
     ),
   ]);
 
+  // only add ES references if templates and pipelines were installed successfully, to prevent upgrade issues for referencing invalid template name
+  if (onlyForDataStreams) {
+    // if onlyForDataStreams is present that means we are in create package policy flow
+    // not install flow, meaning we do not have a lock on the installation SO
+    // so we need to use optimistic concurrency control
+    newEsReferences = await optimisticallyAddEsAssetReferences(
+      savedObjectsClient,
+      packageInstallContext.packageInfo.name,
+      [...preparedIngestPipelines.assetsToAdd, ...preparedIndexTemplates.assetsToAdd]
+    );
+  }
+
   return {
     esReferences: newEsReferences,
     installedTemplates,

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

@@ -19,5 +19,6 @@ export default function loadTests({ loadTestFile, getService }) {
     loadTestFile(require.resolve('./delete'));
     loadTestFile(require.resolve('./upgrade'));
     loadTestFile(require.resolve('./input_package_create_upgrade'));
+    loadTestFile(require.resolve('./input_package_rollback'));
   });
 }

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

@@ -0,0 +1,136 @@
+/*
+ * 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 { sortBy } from 'lodash';
+import { FtrProviderContext } from '../../../api_integration/ftr_provider_context';
+import { skipIfNoDockerRegistry } from '../../helpers';
+import { setupFleetAndAgents } from '../agents/services';
+const PACKAGE_NAME = 'input_package_upgrade';
+const START_VERSION = '1.0.0';
+
+const expectIdArraysEqual = (arr1: any[], arr2: any[]) => {
+  expect(sortBy(arr1, 'id')).to.eql(sortBy(arr2, 'id'));
+};
+export default function (providerContext: FtrProviderContext) {
+  const { getService } = providerContext;
+  const supertest = getService('supertest');
+  const uninstallPackage = async (name: string, version: string) => {
+    await supertest.delete(`/api/fleet/epm/packages/${name}/${version}`).set('kbn-xsrf', 'xxxx');
+  };
+
+  const installPackage = async (name: string, version: string) => {
+    return await supertest
+      .post(`/api/fleet/epm/packages/${name}/${version}`)
+      .set('kbn-xsrf', 'xxxx')
+      .send({ force: true })
+      .expect(200);
+  };
+
+  const getInstallationSavedObject = async (name: string, version: string) => {
+    const res = await supertest.get(`/api/fleet/epm/packages/${name}/${version}`).expect(200);
+    return res.body.item.savedObject.attributes;
+  };
+
+  const getPackage = async (name: string, version: string) => {
+    const res = await supertest.get(`/api/fleet/epm/packages/${name}/${version}`).expect(200);
+    return res.body.item;
+  };
+
+  const createPackagePolicyWithDataset = async (
+    agentPolicyId: string,
+    dataset: string,
+    expectStatusCode = 200,
+    force = false
+  ) => {
+    const policy = {
+      force,
+      policy_id: agentPolicyId,
+      package: {
+        name: PACKAGE_NAME,
+        version: START_VERSION,
+      },
+      name: 'test-policy-' + dataset,
+      description: '',
+      namespace: 'default',
+      inputs: {
+        'logs-logfile': {
+          enabled: true,
+          streams: {
+            'input_package_upgrade.logs': {
+              enabled: true,
+              vars: {
+                paths: ['/tmp/test/log'],
+                tags: ['tag1'],
+                ignore_older: '72h',
+                'data_stream.dataset': dataset,
+              },
+            },
+          },
+        },
+      },
+    };
+    const res = await supertest
+      .post(`/api/fleet/package_policies`)
+      .set('kbn-xsrf', 'xxxx')
+      .send(policy)
+      .expect(expectStatusCode);
+
+    return res.body.item;
+  };
+
+  const createAgentPolicy = async (name = 'Input Package Test 3') => {
+    const res = await supertest
+      .post(`/api/fleet/agent_policies`)
+      .set('kbn-xsrf', 'xxxx')
+      .send({
+        name,
+        namespace: 'default',
+      })
+      .expect(200);
+    return res.body.item;
+  };
+
+  const deleteAgentPolicy = async (agentPolicyId: string) => {
+    if (!agentPolicyId) return;
+    return supertest
+      .post(`/api/fleet/agent_policies/delete`)
+      .set('kbn-xsrf', 'xxxx')
+      .send({ agentPolicyId });
+  };
+
+  describe('input package policy rollback', async function () {
+    skipIfNoDockerRegistry(providerContext);
+
+    let agentPolicyId: string;
+    before(async () => {
+      const agentPolicy = await createAgentPolicy();
+      agentPolicyId = agentPolicy.id;
+    });
+
+    after(async () => {
+      await deleteAgentPolicy(agentPolicyId);
+    });
+    setupFleetAndAgents(providerContext);
+
+    it('should rollback package install on package policy create failure', async () => {
+      await createPackagePolicyWithDataset(agentPolicyId, 'test*', 400);
+
+      const pkg = await getPackage(PACKAGE_NAME, START_VERSION);
+      expect(pkg?.status).to.eql('not_installed');
+    });
+
+    it('should not add es references on package policy create failure when package is already installed', async () => {
+      await installPackage(PACKAGE_NAME, START_VERSION);
+      await createPackagePolicyWithDataset(agentPolicyId, 'test*', 400);
+
+      const installation = await getInstallationSavedObject(PACKAGE_NAME, START_VERSION);
+      expectIdArraysEqual(installation.installed_es, []);
+
+      await uninstallPackage(PACKAGE_NAME, START_VERSION);
+    });
+  });
+}