[Fleet] cleanup old package assets (#112644)

* cleanup on server side

* cleanup all older versions

* fixed type errors

* added unit tests

Co-authored-by: Kibana Machine <[email protected]>

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

@@ -0,0 +1,110 @@
+/*
+ * 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 type { SavedObjectsClientContract } from 'kibana/server';
+import { savedObjectsClientMock } from 'src/core/server/mocks';
+
+import type { PackagePolicyServiceInterface } from '../../package_policy';
+import * as storage from '../archive/storage';
+import { packagePolicyService } from '../../package_policy';
+
+import { removeOldAssets } from './cleanup';
+
+jest.mock('../..', () => ({
+  appContextService: {
+    getLogger: () => ({
+      info: jest.fn(),
+    }),
+  },
+}));
+
+jest.mock('../../package_policy');
+
+describe(' Cleanup old assets', () => {
+  let soClient: jest.Mocked<SavedObjectsClientContract>;
+  const packagePolicyServiceMock =
+    packagePolicyService as jest.Mocked<PackagePolicyServiceInterface>;
+  let removeArchiveEntriesMock: jest.MockedFunction<typeof storage.removeArchiveEntries>;
+
+  function mockFindVersions(versions: string[]) {
+    soClient.find.mockResolvedValue({
+      page: 0,
+      per_page: 0,
+      total: 0,
+      saved_objects: [],
+      aggregations: {
+        versions: {
+          buckets: versions.map((v) => ({ key: '0.3.3' })),
+        },
+      },
+    });
+  }
+
+  beforeEach(() => {
+    soClient = savedObjectsClientMock.create();
+    packagePolicyServiceMock.list.mockClear();
+    removeArchiveEntriesMock = jest.spyOn(storage, 'removeArchiveEntries') as any;
+    removeArchiveEntriesMock.mockClear();
+  });
+  it('should remove old assets from 2 versions if none of the policies are using it', async () => {
+    mockFindVersions(['0.3.3', '0.3.4']);
+    packagePolicyServiceMock.list.mockResolvedValue({ total: 0, items: [], page: 0, perPage: 0 });
+    soClient.createPointInTimeFinder = jest.fn().mockResolvedValue({
+      close: jest.fn(),
+      find: function* asyncGenerator() {
+        yield { saved_objects: [{ id: '1' }, { id: '2' }] };
+      },
+    });
+
+    await removeOldAssets({ soClient, pkgName: 'apache', currentVersion: '1.0.0' });
+
+    expect(removeArchiveEntriesMock).toHaveBeenCalledWith({
+      savedObjectsClient: soClient,
+      refs: [
+        { id: '1', type: 'epm-packages-assets' },
+        { id: '2', type: 'epm-packages-assets' },
+      ],
+    });
+    expect(removeArchiveEntriesMock).toHaveBeenCalledTimes(2);
+  });
+
+  it('should not remove old assets if used by policies', async () => {
+    mockFindVersions(['0.3.3']);
+    packagePolicyServiceMock.list.mockResolvedValue({ total: 1, items: [], page: 0, perPage: 0 });
+
+    await removeOldAssets({ soClient, pkgName: 'apache', currentVersion: '1.0.0' });
+
+    expect(removeArchiveEntriesMock).not.toHaveBeenCalled();
+  });
+
+  it('should remove old assets from all pages', async () => {
+    mockFindVersions(['0.3.3']);
+    packagePolicyServiceMock.list.mockResolvedValue({ total: 0, items: [], page: 0, perPage: 0 });
+    soClient.createPointInTimeFinder = jest.fn().mockResolvedValue({
+      close: jest.fn(),
+      find: function* asyncGenerator() {
+        yield { saved_objects: [{ id: '1' }, { id: '2' }] };
+        yield { saved_objects: [{ id: '3' }] };
+      },
+    });
+
+    await removeOldAssets({ soClient, pkgName: 'apache', currentVersion: '1.0.0' });
+
+    expect(removeArchiveEntriesMock).toHaveBeenCalledWith({
+      savedObjectsClient: soClient,
+      refs: [
+        { id: '1', type: 'epm-packages-assets' },
+        { id: '2', type: 'epm-packages-assets' },
+      ],
+    });
+    expect(removeArchiveEntriesMock).toHaveBeenCalledWith({
+      savedObjectsClient: soClient,
+      refs: [{ id: '3', type: 'epm-packages-assets' }],
+    });
+    expect(removeArchiveEntriesMock).toHaveBeenCalledTimes(2);
+  });
+});

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

@@ -0,0 +1,80 @@
+/*
+ * 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 type { SavedObjectsClientContract } from 'src/core/server';
+
+import { removeArchiveEntries } from '../archive/storage';
+
+import { ASSETS_SAVED_OBJECT_TYPE, PACKAGE_POLICY_SAVED_OBJECT_TYPE } from '../../../../common';
+import type { PackageAssetReference } from '../../../../common';
+import { packagePolicyService } from '../../package_policy';
+import { appContextService } from '../..';
+
+export async function removeOldAssets(options: {
+  soClient: SavedObjectsClientContract;
+  pkgName: string;
+  currentVersion: string;
+}) {
+  const { soClient, pkgName, currentVersion } = options;
+
+  // find all assets of older versions
+  const aggs = {
+    versions: { terms: { field: `${ASSETS_SAVED_OBJECT_TYPE}.attributes.package_version` } },
+  };
+  const oldVersionsAgg = await soClient.find<any, any>({
+    type: ASSETS_SAVED_OBJECT_TYPE,
+    filter: `${ASSETS_SAVED_OBJECT_TYPE}.attributes.package_name:${pkgName} AND ${ASSETS_SAVED_OBJECT_TYPE}.attributes.package_version<${currentVersion}`,
+    aggs,
+    page: 0,
+    perPage: 0,
+  });
+
+  const oldVersions = oldVersionsAgg.aggregations.versions.buckets.map(
+    (obj: { key: string }) => obj.key
+  );
+
+  for (const oldVersion of oldVersions) {
+    await removeAssetsFromVersion(soClient, pkgName, oldVersion);
+  }
+}
+
+async function removeAssetsFromVersion(
+  soClient: SavedObjectsClientContract,
+  pkgName: string,
+  oldVersion: string
+) {
+  // check if any policies are using this package version
+  const { total } = await packagePolicyService.list(soClient, {
+    kuery: `${PACKAGE_POLICY_SAVED_OBJECT_TYPE}.package.name:${pkgName} AND ${PACKAGE_POLICY_SAVED_OBJECT_TYPE}.package.version:${oldVersion}`,
+    page: 0,
+    perPage: 0,
+  });
+  // don't delete if still being used
+  if (total > 0) {
+    appContextService
+      .getLogger()
+      .info(`Package "${pkgName}-${oldVersion}" still being used by policies`);
+    return;
+  }
+
+  // check if old version has assets
+  const finder = await soClient.createPointInTimeFinder({
+    type: ASSETS_SAVED_OBJECT_TYPE,
+    filter: `${ASSETS_SAVED_OBJECT_TYPE}.attributes.package_name:${pkgName} AND ${ASSETS_SAVED_OBJECT_TYPE}.attributes.package_version:${oldVersion}`,
+    perPage: 1000,
+    fields: ['id'],
+  });
+
+  for await (const assets of finder.find()) {
+    const refs = assets.saved_objects.map(
+      (obj) => ({ id: obj.id, type: ASSETS_SAVED_OBJECT_TYPE } as PackageAssetReference)
+    );
+
+    await removeArchiveEntries({ savedObjectsClient: soClient, refs });
+  }
+  await finder.close();
+}

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

@@ -38,6 +38,7 @@ import { isUnremovablePackage, getInstallation, getInstallationObject } from './
 import { removeInstallation } from './remove';
 import { getPackageSavedObjects } from './get';
 import { _installPackage } from './_install_package';
+import { removeOldAssets } from './cleanup';
 
 export async function isPackageInstalled(options: {
   savedObjectsClient: SavedObjectsClientContract;
@@ -267,7 +268,12 @@ async function installPackageFromRegistry({
       installType,
       installSource: 'registry',
     })
-      .then((assets) => {
+      .then(async (assets) => {
+        await removeOldAssets({
+          soClient: savedObjectsClient,
+          pkgName: packageInfo.name,
+          currentVersion: packageInfo.version,
+        });
         return { assets, status: 'installed', installType };
       })
       .catch(async (err: Error) => {

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

@@ -65,6 +65,7 @@ import { getAssetsData } from './epm/packages/assets';
 import { compileTemplate } from './epm/agent/agent';
 import { normalizeKuery } from './saved_object';
 import { appContextService } from '.';
+import { removeOldAssets } from './epm/packages/cleanup';
 
 export type InputsOverride = Partial<NewPackagePolicyInput> & {
   vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
@@ -575,6 +576,11 @@ class PackagePolicyService {
           name: packagePolicy.name,
           success: true,
         });
+        await removeOldAssets({
+          soClient,
+          pkgName: packageInfo.name,
+          currentVersion: packageInfo.version,
+        });
       } catch (error) {
         result.push({
           id,