[Fleet] Upgrade managed package policies in a background task (#191097)

x-pack/plugins/fleet/server/plugin.ts

@@ -130,6 +130,7 @@ import { getPackageSpecTagId } from './services/epm/kibana/assets/tag_assets';
 import { FleetMetricsTask } from './services/metrics/fleet_metrics_task';
 import { fetchAgentMetrics } from './services/metrics/fetch_agent_metrics';
 import { registerIntegrationFieldsExtractor } from './services/register_integration_fields_extractor';
+import { registerUpgradeManagedPackagePoliciesTask } from './services/setup/managed_package_policies';
 
 export interface FleetSetupDeps {
   security: SecurityPluginSetup;
@@ -180,6 +181,7 @@ export interface FleetAppContext {
   auditLogger?: AuditLogger;
   uninstallTokenService: UninstallTokenServiceInterface;
   unenrollInactiveAgentsTask: UnenrollInactiveAgentsTask;
+  taskManagerStart?: TaskManagerStartContract;
 }
 
 export type FleetSetupContract = void;
@@ -596,6 +598,8 @@ export class FleetPlugin
     registerRoutes(fleetAuthzRouter, config);
 
     this.telemetryEventsSender.setup(deps.telemetry);
+    // Register task
+    registerUpgradeManagedPackagePoliciesTask(deps.taskManager);
     this.bulkActionsResolver = new BulkActionsResolver(deps.taskManager, core);
     this.checkDeletedFilesTask = new CheckDeletedFilesTask({
       core,
@@ -653,6 +657,7 @@ export class FleetPlugin
       messageSigningService,
       uninstallTokenService,
       unenrollInactiveAgentsTask: this.unenrollInactiveAgentsTask!,
+      taskManagerStart: plugins.taskManager,
     });
     licenseService.start(plugins.licensing.license$);
     this.telemetryEventsSender.start(plugins.telemetry, core).catch(() => {});

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

@@ -25,14 +25,12 @@ import type {
   EncryptedSavedObjectsPluginSetup,
   EncryptedSavedObjectsPluginStart,
 } from '@kbn/encrypted-saved-objects-plugin/server';
-
 import type { SecurityPluginStart, SecurityPluginSetup } from '@kbn/security-plugin/server';
-
 import type { CloudSetup } from '@kbn/cloud-plugin/server';
 import { DEFAULT_SPACE_ID } from '@kbn/spaces-plugin/common';
 import type { SavedObjectTaggingStart } from '@kbn/saved-objects-tagging-plugin/server';
-
 import { SECURITY_EXTENSION_ID, SPACES_EXTENSION_ID } from '@kbn/core-saved-objects-server';
+import type { TaskManagerStartContract } from '@kbn/task-manager-plugin/server';
 
 import type { FleetConfigType } from '../../common/types';
 import {
@@ -84,6 +82,7 @@ class AppContextService {
   private bulkActionsResolver: BulkActionsResolver | undefined;
   private messageSigningService: MessageSigningServiceInterface | undefined;
   private uninstallTokenService: UninstallTokenServiceInterface | undefined;
+  private taskManagerStart: TaskManagerStartContract | undefined;
 
   public start(appContext: FleetAppContext) {
     this.data = appContext.data;
@@ -108,6 +107,7 @@ class AppContextService {
     this.bulkActionsResolver = appContext.bulkActionsResolver;
     this.messageSigningService = appContext.messageSigningService;
     this.uninstallTokenService = appContext.uninstallTokenService;
+    this.taskManagerStart = appContext.taskManagerStart;
 
     if (appContext.config$) {
       this.config$ = appContext.config$;
@@ -282,6 +282,10 @@ class AppContextService {
     return this.kibanaInstanceId;
   }
 
+  public getTaskManagerStart() {
+    return this.taskManagerStart;
+  }
+
   public addExternalCallback(type: ExternalCallback[0], callback: ExternalCallback[1]) {
     if (!this.externalCallbacks.has(type)) {
       this.externalCallbacks.set(type, new Set());

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

@@ -0,0 +1,72 @@
+/*
+ * 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 {
+  getPackageAssetsMapCache,
+  getPackageInfoCache,
+  runWithCache,
+  setPackageAssetsMapCache,
+  setPackageInfoCache,
+} from './cache';
+
+const PKG_NAME = 'test';
+const PKG_VERSION = '1.0.0';
+
+describe('EPM CacheSession', () => {
+  describe('outside of a cache session', () => {
+    it('should not cache package info', () => {
+      setPackageInfoCache(PKG_NAME, PKG_VERSION, {
+        name: 'test',
+      } as any);
+      const cache = getPackageInfoCache(PKG_NAME, PKG_VERSION);
+      expect(cache).toBeUndefined();
+    });
+
+    it('should not cache assetsMap', () => {
+      setPackageAssetsMapCache(PKG_NAME, PKG_VERSION, new Map());
+      const cache = getPackageAssetsMapCache(PKG_NAME, PKG_VERSION);
+      expect(cache).toBeUndefined();
+    });
+  });
+
+  describe('in of a cache session', () => {
+    it('should cache package info', async () => {
+      function setCache() {
+        setPackageInfoCache(PKG_NAME, PKG_VERSION, {
+          name: 'test',
+        } as any);
+      }
+      function getCache() {
+        const cache = getPackageInfoCache(PKG_NAME, PKG_VERSION);
+        expect(cache).toEqual({ name: 'test' });
+      }
+
+      await runWithCache(async () => {
+        setCache();
+        getCache();
+      });
+    });
+
+    it('should cache assetsMap', async () => {
+      function setCache() {
+        const map = new Map();
+        map.set('test.yaml', Buffer.from('name: test'));
+        setPackageAssetsMapCache(PKG_NAME, PKG_VERSION, map);
+      }
+      function getCache() {
+        const cache = getPackageAssetsMapCache(PKG_NAME, PKG_VERSION);
+        expect(cache).not.toBeUndefined();
+        expect(cache?.get('test.yaml')?.toString()).toEqual('name: test');
+      }
+
+      await runWithCache(async () => {
+        setCache();
+        getCache();
+      });
+    });
+  });
+});

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

@@ -0,0 +1,72 @@
+/*
+ * 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 { AsyncLocalStorage } from 'async_hooks';
+
+import LRUCache from 'lru-cache';
+
+import type { AssetsMap } from '../../../../common/types';
+
+import type { PackageInfo } from '../../../../common';
+
+const cacheStore = new AsyncLocalStorage<CacheSession>();
+
+const PACKAGE_INFO_CACHE_SIZE = 20;
+const PACKAGE_ASSETS_MAP_CACHE_SIZE = 1;
+
+class CacheSession {
+  private _packageInfoCache?: LRUCache<string, PackageInfo>;
+
+  private _packageAssetsMap?: LRUCache<string, AssetsMap>;
+
+  getPackageInfoCache() {
+    if (!this._packageInfoCache) {
+      this._packageInfoCache = new LRUCache<string, PackageInfo>({
+        max: PACKAGE_INFO_CACHE_SIZE,
+      });
+    }
+    return this._packageInfoCache;
+  }
+
+  getPackageAssetsMapCache() {
+    if (!this._packageAssetsMap) {
+      this._packageAssetsMap = new LRUCache<string, AssetsMap>({
+        max: PACKAGE_ASSETS_MAP_CACHE_SIZE,
+      });
+    }
+    return this._packageAssetsMap;
+  }
+}
+
+export function getPackageInfoCache(pkgName: string, pkgVersion: string) {
+  return cacheStore.getStore()?.getPackageInfoCache()?.get(`${pkgName}:${pkgVersion}`);
+}
+
+export function setPackageInfoCache(pkgName: string, pkgVersion: string, packageInfo: PackageInfo) {
+  return cacheStore.getStore()?.getPackageInfoCache()?.set(`${pkgName}:${pkgVersion}`, packageInfo);
+}
+
+export function getPackageAssetsMapCache(pkgName: string, pkgVersion: string) {
+  return cacheStore.getStore()?.getPackageAssetsMapCache()?.get(`${pkgName}:${pkgVersion}`);
+}
+
+export function setPackageAssetsMapCache(
+  pkgName: string,
+  pkgVersion: string,
+  assetsMap: AssetsMap
+) {
+  return cacheStore
+    .getStore()
+    ?.getPackageAssetsMapCache()
+    ?.set(`${pkgName}:${pkgVersion}`, assetsMap);
+}
+
+export async function runWithCache<T = any>(cb: () => Promise<T>): Promise<T> {
+  const cache = new CacheSession();
+
+  return cacheStore.run(cache, cb);
+}

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

@@ -60,6 +60,12 @@ import { auditLoggingService } from '../../audit_logging';
 import { getFilteredSearchPackages } from '../filtered_packages';
 
 import { createInstallableFrom } from '.';
+import {
+  getPackageAssetsMapCache,
+  setPackageAssetsMapCache,
+  getPackageInfoCache,
+  setPackageInfoCache,
+} from './cache';
 
 export { getFile } from '../registry';
 
@@ -415,6 +421,10 @@ export async function getPackageInfo({
   ignoreUnverified?: boolean;
   prerelease?: boolean;
 }): Promise<PackageInfo> {
+  const cacheResult = getPackageInfoCache(pkgName, pkgVersion);
+  if (cacheResult) {
+    return cacheResult;
+  }
   const [savedObject, latestPackage] = await Promise.all([
     getInstallationObject({ savedObjectsClient, pkgName }),
     Registry.fetchFindLatestPackageOrUndefined(pkgName, { prerelease }),
@@ -468,7 +478,10 @@ export async function getPackageInfo({
   };
   const updated = { ...packageInfo, ...additions };
 
-  return createInstallableFrom(updated, savedObject);
+  const installable = createInstallableFrom(updated, savedObject);
+  setPackageInfoCache(pkgName, pkgVersion, installable);
+
+  return installable;
 }
 
 export const getPackageUsageStats = async ({
@@ -720,6 +733,10 @@ export async function getPackageAssetsMap({
   logger: Logger;
   ignoreUnverified?: boolean;
 }) {
+  const cache = getPackageAssetsMapCache(packageInfo.name, packageInfo.version);
+  if (cache) {
+    return cache;
+  }
   const installedPackageWithAssets = await getInstalledPackageWithAssets({
     savedObjectsClient,
     pkgName: packageInfo.name,
@@ -736,6 +753,7 @@ export async function getPackageAssetsMap({
   } else {
     assetsMap = installedPackageWithAssets.assetsMap;
   }
+  setPackageAssetsMapCache(packageInfo.name, packageInfo.version, assetsMap);
 
   return assetsMap;
 }

x-pack/plugins/fleet/server/services/package_policies/utils.ts

@@ -20,55 +20,16 @@ import { outputService } from '../output';
 import { appContextService } from '../app_context';
 
 export const mapPackagePolicySavedObjectToPackagePolicy = ({
-  /* eslint-disable @typescript-eslint/naming-convention */
   id,
   version,
-  attributes: {
-    name,
-    description,
-    namespace,
-    enabled,
-    is_managed,
-    policy_id,
-    policy_ids,
-    output_id,
-    // `package` is a reserved keyword
-    package: packageInfo,
-    inputs,
-    vars,
-    elasticsearch,
-    agents,
-    revision,
-    secret_references,
-    updated_at,
-    updated_by,
-    created_at,
-    created_by,
-    /* eslint-enable @typescript-eslint/naming-convention */
-  },
+  attributes,
+  namespaces,
 }: SavedObject<PackagePolicySOAttributes>): PackagePolicy => {
   return {
     id,
-    name,
-    description,
-    namespace,
-    enabled,
-    is_managed,
-    policy_id,
-    policy_ids,
-    output_id,
-    package: packageInfo,
-    inputs,
-    vars,
-    elasticsearch,
     version,
-    agents,
-    revision,
-    secret_references,
-    updated_at,
-    updated_by,
-    created_at,
-    created_by,
+    spaceIds: namespaces,
+    ...attributes,
   };
 };
 

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

@@ -118,7 +118,6 @@ import {
   mapPackagePolicySavedObjectToPackagePolicy,
   preflightCheckPackagePolicy,
 } from './package_policies';
-import { updateDatastreamExperimentalFeatures } from './epm/packages/update';
 import type {
   PackagePolicyClient,
   PackagePolicyClientFetchAllItemsOptions,
@@ -1651,13 +1650,6 @@ class PackagePolicyClientImpl implements PackagePolicyClient {
 
     await this.update(soClient, esClient, id, updatePackagePolicy, updateOptions);
 
-    // Persist any experimental feature opt-ins that come through the upgrade process to the Installation SO
-    await updateDatastreamExperimentalFeatures(
-      soClient,
-      packagePolicy.package!.name,
-      experimentalDataStreamFeatures
-    );
-
     result.push({
       id,
       name: packagePolicy.name,

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

@@ -42,7 +42,7 @@ import { agentPolicyService, addPackageToAgentPolicy } from './agent_policy';
 import { type InputsOverride, packagePolicyService } from './package_policy';
 import { preconfigurePackageInputs } from './package_policy';
 import { appContextService } from './app_context';
-import type { UpgradeManagedPackagePoliciesResult } from './managed_package_policies';
+import type { UpgradeManagedPackagePoliciesResult } from './setup/managed_package_policies';
 import { isDefaultAgentlessPolicyEnabled } from './utils/agentless';
 
 interface PreconfigurationResult {