[Fleet] Package policy upgrade telemetry with sender (#115180)

* draft of upgrade usage collector

* telemetry sender service

* fixed tests and types

* cleanup

* type fix

* removed collector

* made required field message generic, added test

* cleanup

* cleanup

* cleanup

* removed v3-dev as outdated

* removed conditional from telemetry url creation

* supporting multiple channels in sender

* fix types

* refactor

* using json content type

* fix test

* simplified telemetry url

* fixed type

* added back ndjson

* moved telemetry to update, added dryrun

* fix types

* fix prettier

* updated after review

* fix imports

* added error_message field

* review fixes

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/fleet/kibana.json

@@ -9,7 +9,7 @@
   "ui": true,
   "configPath": ["xpack", "fleet"],
   "requiredPlugins": ["licensing", "data", "encryptedSavedObjects", "navigation", "customIntegrations", "share"],
-  "optionalPlugins": ["security", "features", "cloud", "usageCollection", "home", "globalSearch"],
+  "optionalPlugins": ["security", "features", "cloud", "usageCollection", "home", "globalSearch", "telemetry"],
   "extraPublicDirs": ["common"],
   "requiredBundles": ["kibanaReact", "esUiShared", "home", "infra", "kibanaUtils", "usageCollection"]
 }

x-pack/plugins/fleet/server/mocks/index.ts

@@ -19,6 +19,7 @@ import { securityMock } from '../../../security/server/mocks';
 import type { PackagePolicyServiceInterface } from '../services/package_policy';
 import type { AgentPolicyServiceInterface, AgentService } from '../services';
 import type { FleetAppContext } from '../plugin';
+import { createMockTelemetryEventsSender } from '../telemetry/__mocks__';
 
 // Export all mocks from artifacts
 export * from '../services/artifacts/mocks';
@@ -59,6 +60,7 @@ export const createAppContextStartContractMock = (): MockedFleetAppContext => {
     config$,
     kibanaVersion: '8.99.0', // Fake version :)
     kibanaBranch: 'main',
+    telemetryEventsSender: createMockTelemetryEventsSender(),
   };
 };
 

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

@@ -18,6 +18,8 @@ import type {
 } from 'kibana/server';
 import type { UsageCollectionSetup } from 'src/plugins/usage_collection/server';
 
+import type { TelemetryPluginSetup, TelemetryPluginStart } from 'src/plugins/telemetry/server';
+
 import { DEFAULT_APP_CATEGORIES } from '../../../../src/core/server';
 import type { PluginStart as DataPluginStart } from '../../../../src/plugins/data/server';
 import type { LicensingPluginSetup, ILicense } from '../../licensing/server';
@@ -83,6 +85,7 @@ import { RouterWrappers } from './routes/security';
 import { startFleetServerSetup } from './services/fleet_server';
 import { FleetArtifactsClient } from './services/artifacts';
 import type { FleetRouter } from './types/request_context';
+import { TelemetryEventsSender } from './telemetry/sender';
 
 export interface FleetSetupDeps {
   licensing: LicensingPluginSetup;
@@ -91,12 +94,14 @@ export interface FleetSetupDeps {
   encryptedSavedObjects: EncryptedSavedObjectsPluginSetup;
   cloud?: CloudSetup;
   usageCollection?: UsageCollectionSetup;
+  telemetry?: TelemetryPluginSetup;
 }
 
 export interface FleetStartDeps {
   data: DataPluginStart;
   encryptedSavedObjects: EncryptedSavedObjectsPluginStart;
   security?: SecurityPluginStart;
+  telemetry?: TelemetryPluginStart;
 }
 
 export interface FleetAppContext {
@@ -115,6 +120,7 @@ export interface FleetAppContext {
   cloud?: CloudSetup;
   logger?: Logger;
   httpSetup?: HttpServiceSetup;
+  telemetryEventsSender: TelemetryEventsSender;
 }
 
 export type FleetSetupContract = void;
@@ -176,6 +182,7 @@ export class FleetPlugin
   private httpSetup?: HttpServiceSetup;
   private securitySetup?: SecurityPluginSetup;
   private encryptedSavedObjectsSetup?: EncryptedSavedObjectsPluginSetup;
+  private readonly telemetryEventsSender: TelemetryEventsSender;
 
   constructor(private readonly initializerContext: PluginInitializerContext) {
     this.config$ = this.initializerContext.config.create<FleetConfigType>();
@@ -184,6 +191,7 @@ export class FleetPlugin
     this.kibanaBranch = this.initializerContext.env.packageInfo.branch;
     this.logger = this.initializerContext.logger.get();
     this.configInitialValue = this.initializerContext.config.get();
+    this.telemetryEventsSender = new TelemetryEventsSender(this.logger.get('telemetry_events'));
   }
 
   public setup(core: CoreSetup, deps: FleetSetupDeps) {
@@ -302,6 +310,8 @@ export class FleetPlugin
         });
       }
     }
+
+    this.telemetryEventsSender.setup(deps.telemetry);
   }
 
   public start(core: CoreStart, plugins: FleetStartDeps): FleetStartContract {
@@ -321,11 +331,14 @@ export class FleetPlugin
       httpSetup: this.httpSetup,
       cloud: this.cloud,
       logger: this.logger,
+      telemetryEventsSender: this.telemetryEventsSender,
     });
     licenseService.start(this.licensing$);
 
     const fleetServerSetup = startFleetServerSetup();
 
+    this.telemetryEventsSender.start(plugins.telemetry, core);
+
     return {
       fleetSetupCompleted: () =>
         new Promise<void>((resolve) => {
@@ -362,5 +375,6 @@ export class FleetPlugin
   public async stop() {
     appContextService.stop();
     licenseService.stop();
+    this.telemetryEventsSender.stop();
   }
 }

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

@@ -146,7 +146,8 @@ export const updatePackagePolicyHandler: RequestHandler<
       esClient,
       request.params.packagePolicyId,
       { ...newData, package: pkg, inputs },
-      { user }
+      { user },
+      packagePolicy.package?.version
     );
     return response.ok({
       body: { item: updatedPackagePolicy },

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

@@ -33,6 +33,7 @@ import type {
 } from '../types';
 import type { FleetAppContext } from '../plugin';
 import type { CloudSetup } from '../../../cloud/server';
+import type { TelemetryEventsSender } from '../telemetry/sender';
 
 class AppContextService {
   private encryptedSavedObjects: EncryptedSavedObjectsClient | undefined;
@@ -51,6 +52,7 @@ class AppContextService {
   private logger: Logger | undefined;
   private httpSetup?: HttpServiceSetup;
   private externalCallbacks: ExternalCallbacksStorage = new Map();
+  private telemetryEventsSender: TelemetryEventsSender | undefined;
 
   public start(appContext: FleetAppContext) {
     this.data = appContext.data;
@@ -66,6 +68,7 @@ class AppContextService {
     this.kibanaVersion = appContext.kibanaVersion;
     this.kibanaBranch = appContext.kibanaBranch;
     this.httpSetup = appContext.httpSetup;
+    this.telemetryEventsSender = appContext.telemetryEventsSender;
 
     if (appContext.config$) {
       this.config$ = appContext.config$;
@@ -203,6 +206,10 @@ class AppContextService {
       >;
     }
   }
+
+  public getTelemetryEventsSender() {
+    return this.telemetryEventsSender;
+  }
 }
 
 export const appContextService = new AppContextService();

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

@@ -134,6 +134,12 @@ jest.mock('./epm/packages/cleanup', () => {
   };
 });
 
+jest.mock('./upgrade_usage', () => {
+  return {
+    sendTelemetryEvents: jest.fn(),
+  };
+});
+
 const mockedFetchInfo = fetchInfo as jest.Mock<ReturnType<typeof fetchInfo>>;
 
 type CombinedExternalCallback = PutPackagePolicyUpdateCallback | PostPackagePolicyCreateCallback;

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

@@ -9,7 +9,7 @@ import { omit, partition } from 'lodash';
 import { i18n } from '@kbn/i18n';
 import semverLte from 'semver/functions/lte';
 import { getFlattenedObject } from '@kbn/std';
-import type { KibanaRequest, LogMeta } from 'src/core/server';
+import type { KibanaRequest } from 'src/core/server';
 import type {
   ElasticsearchClient,
   RequestHandlerContext,
@@ -68,6 +68,8 @@ import { compileTemplate } from './epm/agent/agent';
 import { normalizeKuery } from './saved_object';
 import { appContextService } from '.';
 import { removeOldAssets } from './epm/packages/cleanup';
+import type { PackagePolicyUpgradeUsage } from './upgrade_usage';
+import { sendTelemetryEvents } from './upgrade_usage';
 
 export type InputsOverride = Partial<NewPackagePolicyInput> & {
   vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
@@ -84,17 +86,6 @@ export const DATA_STREAM_ALLOWED_INDEX_PRIVILEGES = new Set([
   'read_cross_cluster',
 ]);
 
-interface PackagePolicyUpgradeLogMeta extends LogMeta {
-  package_policy_upgrade: {
-    package_name: string;
-    current_version: string;
-    new_version: string;
-    status: 'success' | 'failure';
-    error?: any[];
-    dryRun?: boolean;
-  };
-}
-
 class PackagePolicyService {
   public async create(
     soClient: SavedObjectsClientContract,
@@ -369,7 +360,8 @@ class PackagePolicyService {
     esClient: ElasticsearchClient,
     id: string,
     packagePolicy: UpdatePackagePolicy,
-    options?: { user?: AuthenticatedUser }
+    options?: { user?: AuthenticatedUser },
+    currentVersion?: string
   ): Promise<PackagePolicy> {
     const oldPackagePolicy = await this.get(soClient, id);
     const { version, ...restOfPackagePolicy } = packagePolicy;
@@ -445,22 +437,22 @@ class PackagePolicyService {
         currentVersion: packagePolicy.package.version,
       });
 
-      const upgradeMeta: PackagePolicyUpgradeLogMeta = {
-        package_policy_upgrade: {
+      if (packagePolicy.package.version !== currentVersion) {
+        const upgradeTelemetry: PackagePolicyUpgradeUsage = {
           package_name: packagePolicy.package.name,
+          current_version: currentVersion || 'unknown',
           new_version: packagePolicy.package.version,
-          current_version: 'unknown',
           status: 'success',
           dryRun: false,
-        },
-      };
-
-      appContextService
-        .getLogger()
-        .info<PackagePolicyUpgradeLogMeta>(
-          `Package policy successfully upgraded ${JSON.stringify(upgradeMeta)}`,
-          upgradeMeta
+        };
+        sendTelemetryEvents(
+          appContextService.getLogger(),
+          appContextService.getTelemetryEventsSender(),
+          upgradeTelemetry
         );
+        appContextService.getLogger().info(`Package policy upgraded successfully`);
+        appContextService.getLogger().debug(JSON.stringify(upgradeTelemetry));
+      }
     }
 
     return newPolicy;
@@ -629,7 +621,14 @@ class PackagePolicyService {
         );
         updatePackagePolicy.elasticsearch = registryPkgInfo.elasticsearch;
 
-        await this.update(soClient, esClient, id, updatePackagePolicy, options);
+        await this.update(
+          soClient,
+          esClient,
+          id,
+          updatePackagePolicy,
+          options,
+          packagePolicy.package.version
+        );
         result.push({
           id,
           name: packagePolicy.name,
@@ -691,24 +690,27 @@ class PackagePolicyService {
       const hasErrors = 'errors' in updatedPackagePolicy;
 
       if (packagePolicy.package.version !== packageInfo.version) {
-        const upgradeMeta: PackagePolicyUpgradeLogMeta = {
-          package_policy_upgrade: {
-            package_name: packageInfo.name,
-            current_version: packagePolicy.package.version,
-            new_version: packageInfo.version,
-            status: hasErrors ? 'failure' : 'success',
-            error: hasErrors ? updatedPackagePolicy.errors : undefined,
-            dryRun: true,
-          },
+        const upgradeTelemetry: PackagePolicyUpgradeUsage = {
+          package_name: packageInfo.name,
+          current_version: packagePolicy.package.version,
+          new_version: packageInfo.version,
+          status: hasErrors ? 'failure' : 'success',
+          error: hasErrors ? updatedPackagePolicy.errors : undefined,
+          dryRun: true,
         };
+        sendTelemetryEvents(
+          appContextService.getLogger(),
+          appContextService.getTelemetryEventsSender(),
+          upgradeTelemetry
+        );
         appContextService
           .getLogger()
-          .info<PackagePolicyUpgradeLogMeta>(
+          .info(
             `Package policy upgrade dry run ${
               hasErrors ? 'resulted in errors' : 'ran successfully'
-            } ${JSON.stringify(upgradeMeta)}`,
-            upgradeMeta
+            }`
           );
+        appContextService.getLogger().debug(JSON.stringify(upgradeTelemetry));
       }
 
       return {

x-pack/plugins/fleet/server/services/upgrade_usage.test.ts

@@ -0,0 +1,70 @@
+/*
+ * 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 { Logger } from 'src/core/server';
+import { loggingSystemMock } from 'src/core/server/mocks';
+
+import type { TelemetryEventsSender } from '../telemetry/sender';
+import { createMockTelemetryEventsSender } from '../telemetry/__mocks__';
+
+import { sendTelemetryEvents, capErrorSize } from './upgrade_usage';
+import type { PackagePolicyUpgradeUsage } from './upgrade_usage';
+
+describe('sendTelemetryEvents', () => {
+  let eventsTelemetryMock: jest.Mocked<TelemetryEventsSender>;
+  let loggerMock: jest.Mocked<Logger>;
+
+  beforeEach(() => {
+    eventsTelemetryMock = createMockTelemetryEventsSender();
+    loggerMock = loggingSystemMock.createLogger();
+  });
+
+  it('should queue telemetry events with generic error', () => {
+    const upgardeMessage: PackagePolicyUpgradeUsage = {
+      package_name: 'aws',
+      current_version: '0.6.1',
+      new_version: '1.3.0',
+      status: 'failure',
+      error: [
+        { key: 'queueUrl', message: ['Queue URL is required'] },
+        { message: 'Invalid format' },
+      ],
+      dryRun: true,
+    };
+
+    sendTelemetryEvents(loggerMock, eventsTelemetryMock, upgardeMessage);
+
+    expect(eventsTelemetryMock.queueTelemetryEvents).toHaveBeenCalledWith('fleet-upgrades', [
+      {
+        current_version: '0.6.1',
+        error: [
+          {
+            key: 'queueUrl',
+            message: ['Queue URL is required'],
+          },
+          {
+            message: 'Invalid format',
+          },
+        ],
+        error_message: ['Field is required', 'Invalid format'],
+        new_version: '1.3.0',
+        package_name: 'aws',
+        status: 'failure',
+        dryRun: true,
+      },
+    ]);
+  });
+
+  it('should cap error size', () => {
+    const maxSize = 2;
+    const errors = [{ message: '1' }, { message: '2' }, { message: '3' }];
+
+    const result = capErrorSize(errors, maxSize);
+
+    expect(result.length).toEqual(maxSize);
+  });
+});