[Fleet] Task to publish Agent metrics

x-pack/plugins/fleet/server/collectors/agent_collectors.ts

@@ -58,13 +58,13 @@ export const getAgentUsage = async (
   };
 };
 
+export interface AgentPerVersion {
+  version: string;
+  count: number;
+}
+
 export interface AgentData {
-  agents_per_version: Array<
-    {
-      version: string;
-      count: number;
-    } & AgentStatus
-  >;
+  agents_per_version: Array<AgentPerVersion & AgentStatus>;
   agent_checkin_status: {
     error: number;
     degraded: number;

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

@@ -84,6 +84,7 @@ export const createAppContextStartContractMock = (
     config$,
     kibanaVersion: '8.99.0', // Fake version :)
     kibanaBranch: 'main',
+    kibanaInstanceId: '1',
     telemetryEventsSender: createMockTelemetryEventsSender(),
     bulkActionsResolver: {} as any,
     messageSigningService: createMessageSigningServiceMock(),

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

@@ -130,6 +130,8 @@ import { FleetActionsClient, type FleetActionsClientInterface } from './services
 import type { FilesClientFactory } from './services/files/types';
 import { PolicyWatcher } from './services/agent_policy_watch';
 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';
 
 export interface FleetSetupDeps {
   security: SecurityPluginSetup;
@@ -167,6 +169,7 @@ export interface FleetAppContext {
   isProductionMode: PluginInitializerContext['env']['mode']['prod'];
   kibanaVersion: PluginInitializerContext['env']['packageInfo']['version'];
   kibanaBranch: PluginInitializerContext['env']['packageInfo']['branch'];
+  kibanaInstanceId: PluginInitializerContext['env']['instanceUuid'];
   cloud?: CloudSetup;
   logger?: Logger;
   httpSetup?: HttpServiceSetup;
@@ -251,6 +254,7 @@ export class FleetPlugin
   private isProductionMode: FleetAppContext['isProductionMode'];
   private kibanaVersion: FleetAppContext['kibanaVersion'];
   private kibanaBranch: FleetAppContext['kibanaBranch'];
+  private kibanaInstanceId: FleetAppContext['kibanaInstanceId'];
   private httpSetup?: HttpServiceSetup;
   private securitySetup!: SecurityPluginSetup;
   private encryptedSavedObjectsSetup?: EncryptedSavedObjectsPluginSetup;
@@ -259,6 +263,7 @@ export class FleetPlugin
   private bulkActionsResolver?: BulkActionsResolver;
   private fleetUsageSender?: FleetUsageSender;
   private checkDeletedFilesTask?: CheckDeletedFilesTask;
+  private fleetMetricsTask?: FleetMetricsTask;
 
   private agentService?: AgentService;
   private packageService?: PackageService;
@@ -270,6 +275,7 @@ export class FleetPlugin
     this.isProductionMode = this.initializerContext.env.mode.prod;
     this.kibanaVersion = this.initializerContext.env.packageInfo.version;
     this.kibanaBranch = this.initializerContext.env.packageInfo.branch;
+    this.kibanaInstanceId = this.initializerContext.env.instanceUuid;
     this.logger = this.initializerContext.logger.get();
     this.configInitialValue = this.initializerContext.config.get();
     this.telemetryEventsSender = new TelemetryEventsSender(this.logger.get('telemetry_events'));
@@ -440,6 +446,10 @@ export class FleetPlugin
     this.fleetUsageSender = new FleetUsageSender(deps.taskManager, core, fetch);
     registerFleetUsageLogger(deps.taskManager, async () => fetchAgentsUsage(core, config));
 
+    const fetchAgents = async (abortController: AbortController) =>
+      await fetchAgentMetrics(core, abortController);
+    this.fleetMetricsTask = new FleetMetricsTask(deps.taskManager, fetchAgents);
+
     const router: FleetRouter = core.http.createRouter<FleetRequestHandlerContext>();
     // Allow read-only users access to endpoints necessary for Integrations UI
     // Only some endpoints require superuser so we pass a raw IRouter here
@@ -490,6 +500,7 @@ export class FleetPlugin
       isProductionMode: this.isProductionMode,
       kibanaVersion: this.kibanaVersion,
       kibanaBranch: this.kibanaBranch,
+      kibanaInstanceId: this.kibanaInstanceId,
       httpSetup: this.httpSetup,
       cloud: this.cloud,
       logger: this.logger,
@@ -504,6 +515,7 @@ export class FleetPlugin
     this.fleetUsageSender?.start(plugins.taskManager);
     this.checkDeletedFilesTask?.start({ taskManager: plugins.taskManager });
     startFleetUsageLogger(plugins.taskManager);
+    this.fleetMetricsTask?.start(plugins.taskManager, core.elasticsearch.client.asInternalUser);
 
     const logger = appContextService.getLogger();
 

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

@@ -62,6 +62,7 @@ class AppContextService {
   private isProductionMode: FleetAppContext['isProductionMode'] = false;
   private kibanaVersion: FleetAppContext['kibanaVersion'] = kibanaPackageJson.version;
   private kibanaBranch: FleetAppContext['kibanaBranch'] = kibanaPackageJson.branch;
+  private kibanaInstanceId: FleetAppContext['kibanaInstanceId'] = '';
   private cloud?: CloudSetup;
   private logger: Logger | undefined;
   private httpSetup?: HttpServiceSetup;
@@ -86,6 +87,7 @@ class AppContextService {
     this.logger = appContext.logger;
     this.kibanaVersion = appContext.kibanaVersion;
     this.kibanaBranch = appContext.kibanaBranch;
+    this.kibanaInstanceId = appContext.kibanaInstanceId;
     this.httpSetup = appContext.httpSetup;
     this.telemetryEventsSender = appContext.telemetryEventsSender;
     this.savedObjectsTagging = appContext.savedObjectsTagging;
@@ -209,6 +211,10 @@ class AppContextService {
     return this.kibanaBranch;
   }
 
+  public getKibanaInstanceId() {
+    return this.kibanaInstanceId;
+  }
+
   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/metrics/fetch_agent_metrics.ts

@@ -0,0 +1,260 @@
+/*
+ * 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 { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import type { CoreSetup } from '@kbn/core/server';
+
+import { AGENTS_INDEX } from '../../../common';
+
+import type { AgentPerVersion, AgentUsage } from '../../collectors/agent_collectors';
+import { getAgentUsage } from '../../collectors/agent_collectors';
+import { getInternalClients } from '../../collectors/helpers';
+import { appContextService } from '../app_context';
+
+export interface AgentMetrics {
+  agents: AgentUsage;
+  agents_per_version: AgentPerVersion[];
+  upgrading_step: UpgradingSteps;
+  unhealthy_reason: UnhealthyReason;
+}
+
+export interface UpgradingSteps {
+  requested: number;
+  scheduled: number;
+  downloading: number;
+  extracting: number;
+  replacing: number;
+  restarting: number;
+  watching: number;
+  rollback: number;
+  failed: number;
+}
+
+export interface UnhealthyReason {
+  input: number;
+  output: number;
+  other: number;
+}
+
+export const fetchAgentMetrics = async (
+  core: CoreSetup,
+  abortController: AbortController
+): Promise<AgentMetrics | undefined> => {
+  const [soClient, esClient] = await getInternalClients(core);
+  if (!soClient || !esClient) {
+    return;
+  }
+  const usage = {
+    agents: await getAgentUsage(soClient, esClient),
+    agents_per_version: await getAgentsPerVersion(esClient, abortController),
+    upgrading_step: await getUpgradingSteps(esClient, abortController),
+    unhealthy_reason: await getUnhealthyReason(esClient, abortController),
+  };
+  return usage;
+};
+
+export const getAgentsPerVersion = async (
+  esClient: ElasticsearchClient,
+  abortController: AbortController
+): Promise<AgentPerVersion[]> => {
+  try {
+    const response = await esClient.search(
+      {
+        index: AGENTS_INDEX,
+        query: {
+          bool: {
+            filter: [
+              {
+                term: {
+                  active: 'true',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+        aggs: {
+          versions: {
+            terms: { field: 'agent.version' },
+          },
+        },
+      },
+      { signal: abortController.signal }
+    );
+    return ((response?.aggregations?.versions as any).buckets ?? []).map((bucket: any) => ({
+      version: bucket.key,
+      count: bucket.doc_count,
+    }));
+  } catch (error) {
+    if (error.statusCode === 404) {
+      appContextService.getLogger().debug('Index .fleet-agents does not exist yet.');
+    } else {
+      throw error;
+    }
+    return [];
+  }
+};
+
+export const getUpgradingSteps = async (
+  esClient: ElasticsearchClient,
+  abortController: AbortController
+): Promise<UpgradingSteps> => {
+  const upgradingSteps = {
+    requested: 0,
+    scheduled: 0,
+    downloading: 0,
+    extracting: 0,
+    replacing: 0,
+    restarting: 0,
+    watching: 0,
+    rollback: 0,
+    failed: 0,
+  };
+  try {
+    const response = await esClient.search(
+      {
+        index: AGENTS_INDEX,
+        query: {
+          bool: {
+            filter: [
+              {
+                term: {
+                  active: 'true',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+        aggs: {
+          upgrade_details: {
+            terms: { field: 'upgrade_details.state' },
+          },
+        },
+      },
+      { signal: abortController.signal }
+    );
+    ((response?.aggregations?.upgrade_details as any).buckets ?? []).forEach((bucket: any) => {
+      switch (bucket.key) {
+        case 'UPG_REQUESTED':
+          upgradingSteps.requested = bucket.doc_count;
+          break;
+        case 'UPG_SCHEDULED':
+          upgradingSteps.scheduled = bucket.doc_count;
+          break;
+        case 'UPG_DOWNLOADING':
+          upgradingSteps.downloading = bucket.doc_count;
+          break;
+        case 'UPG_EXTRACTING':
+          upgradingSteps.extracting = bucket.doc_count;
+          break;
+        case 'UPG_REPLACING':
+          upgradingSteps.replacing = bucket.doc_count;
+          break;
+        case 'UPG_RESTARTING':
+          upgradingSteps.restarting = bucket.doc_count;
+          break;
+        case 'UPG_WATCHING':
+          upgradingSteps.watching = bucket.doc_count;
+          break;
+        case 'UPG_ROLLBACK':
+          upgradingSteps.rollback = bucket.doc_count;
+          break;
+        case 'UPG_FAILED':
+          upgradingSteps.failed = bucket.doc_count;
+          break;
+        default:
+          break;
+      }
+    });
+    return upgradingSteps;
+  } catch (error) {
+    if (error.statusCode === 404) {
+      appContextService.getLogger().debug('Index .fleet-agents does not exist yet.');
+    } else {
+      throw error;
+    }
+    return upgradingSteps;
+  }
+};
+
+export const getUnhealthyReason = async (
+  esClient: ElasticsearchClient,
+  abortController: AbortController
+): Promise<UnhealthyReason> => {
+  const unhealthyReason = {
+    input: 0,
+    output: 0,
+    other: 0,
+  };
+  try {
+    const response = await esClient.search(
+      {
+        index: AGENTS_INDEX,
+        query: {
+          bool: {
+            filter: [
+              {
+                term: {
+                  active: 'true',
+                },
+              },
+              {
+                terms: {
+                  last_checkin_status: ['error', 'degraded'],
+                },
+              },
+            ],
+          },
+        },
+        size: 10000,
+        _source: ['components'],
+      },
+      { signal: abortController.signal }
+    );
+
+    response.hits.hits.forEach((hit: any) => {
+      // counting agent as other unhealthy reason if it doesn't have a component unit unhealthy
+      if (!hit._source.components || hit._source.components.length === 0) {
+        unhealthyReason.other++;
+        return;
+      }
+      // considering component unhealthy if not healthy
+      const hasUnhealthyUnit = (type: string) => {
+        const unhealthyComponent = hit._source.components.find(
+          (comp: any) => comp.status !== 'HEALTHY'
+        );
+        if (!unhealthyComponent) return false;
+        const unhealthyUnit = unhealthyComponent.units.find(
+          (unit: any) => unit.type === type && unit.status !== 'HEALTHY'
+        );
+        return unhealthyUnit !== undefined;
+      };
+      const hasUnhealthyInput = hasUnhealthyUnit('input');
+      // counting agents in both input and output unhealthy_reason if they have unhealthy component units in both
+      if (hasUnhealthyInput) {
+        unhealthyReason.input++;
+      }
+      const hasUnhealthyOutput = hasUnhealthyUnit('output');
+      if (hasUnhealthyOutput) {
+        unhealthyReason.output++;
+      }
+      if (!hasUnhealthyInput && !hasUnhealthyOutput) {
+        unhealthyReason.other++;
+      }
+    });
+
+    return unhealthyReason;
+  } catch (error) {
+    if (error.statusCode === 404) {
+      appContextService.getLogger().debug('Index .fleet-agents does not exist yet.');
+    } else {
+      throw error;
+    }
+    return unhealthyReason;
+  }
+};