Fleet Usage telemetry extension

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

@@ -68,10 +68,10 @@ const DEFAULT_AGENT_DATA = {
   agent_checkin_status_last_1h: { error: 0, degraded: 0 },
 };
 
-export const getAgentData = async (esClient?: ElasticsearchClient): Promise<AgentData> => {
-  if (!esClient) {
-    return DEFAULT_AGENT_DATA;
-  }
+export const getAgentData = async (
+  esClient: ElasticsearchClient,
+  abortController: AbortController
+): Promise<AgentData> => {
   try {
     const transformLastCheckinStatusBuckets = (resp: any) =>
       ((resp?.aggregations?.last_checkin_status as any).buckets ?? []).reduce(
@@ -81,52 +81,58 @@ export const getAgentData = async (esClient?: ElasticsearchClient): Promise<Agen
         },
         { error: 0, degraded: 0 }
       );
-    const response = await esClient.search({
-      index: AGENTS_INDEX,
-      size: 0,
-      aggs: {
-        versions: {
-          terms: { field: 'agent.version' },
-        },
-        last_checkin_status: {
-          terms: { field: 'last_checkin_status' },
+    const response = await esClient.search(
+      {
+        index: AGENTS_INDEX,
+        size: 0,
+        aggs: {
+          versions: {
+            terms: { field: 'agent.version' },
+          },
+          last_checkin_status: {
+            terms: { field: 'last_checkin_status' },
+          },
         },
       },
-    });
+      { signal: abortController.signal }
+    );
     const versions = ((response?.aggregations?.versions as any).buckets ?? []).map(
       (bucket: any) => bucket.key
     );
     const statuses = transformLastCheckinStatusBuckets(response);
 
-    const responseLast1h = await esClient.search({
-      index: AGENTS_INDEX,
-      size: 0,
-      query: {
-        bool: {
-          filter: [
-            {
-              bool: {
-                must: [
-                  {
-                    range: {
-                      last_checkin: {
-                        gte: 'now-1h/h',
-                        lt: 'now/h',
+    const responseLast1h = await esClient.search(
+      {
+        index: AGENTS_INDEX,
+        size: 0,
+        query: {
+          bool: {
+            filter: [
+              {
+                bool: {
+                  must: [
+                    {
+                      range: {
+                        last_checkin: {
+                          gte: 'now-1h/h',
+                          lt: 'now/h',
+                        },
                       },
                     },
-                  },
-                ],
+                  ],
+                },
               },
-            },
-          ],
+            ],
+          },
         },
-      },
-      aggs: {
-        last_checkin_status: {
-          terms: { field: 'last_checkin_status' },
+        aggs: {
+          last_checkin_status: {
+            terms: { field: 'last_checkin_status' },
+          },
         },
       },
-    });
+      { signal: abortController.signal }
+    );
     const statusesLast1h = transformLastCheckinStatusBuckets(responseLast1h);
 
     return {

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

@@ -5,25 +5,24 @@
  * 2.0.
  */
 
-import type { SavedObjectsClient, ElasticsearchClient } from '@kbn/core/server';
+import type { ElasticsearchClient } from '@kbn/core/server';
 
 import { AGENT_POLICY_INDEX } from '../../common';
 import { ES_SEARCH_LIMIT } from '../../common/constants';
 
 export const getAgentPoliciesUsage = async (
-  soClient?: SavedObjectsClient,
-  esClient?: ElasticsearchClient
+  esClient: ElasticsearchClient,
+  abortController: AbortController
 ): Promise<any> => {
-  if (!soClient || !esClient) {
-    return {};
-  }
-
-  const res = await esClient.search({
-    index: AGENT_POLICY_INDEX,
-    size: ES_SEARCH_LIMIT,
-    track_total_hits: true,
-    rest_total_hits_as_int: true,
-  });
+  const res = await esClient.search(
+    {
+      index: AGENT_POLICY_INDEX,
+      size: ES_SEARCH_LIMIT,
+      track_total_hits: true,
+      rest_total_hits_as_int: true,
+    },
+    { signal: abortController.signal }
+  );
 
   const agentPolicies = res.hits.hits;
 

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

@@ -87,13 +87,7 @@ export const getFleetServerUsage = async (
   };
 };
 
-export const getFleetServerConfig = async (
-  soClient?: SavedObjectsClient,
-  esClient?: ElasticsearchClient
-): Promise<any> => {
-  if (!soClient || !esClient) {
-    return {};
-  }
+export const getFleetServerConfig = async (soClient: SavedObjectsClient): Promise<any> => {
   const fleetServerHosts = await listFleetServerHosts(soClient);
   const hosts = fleetServerHosts.items.map((item) => ({
     ...item,

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

@@ -27,16 +27,23 @@ export interface Usage {
   fleet_server: FleetServerUsage;
 }
 
-export const fetchFleetUsage = async (core: CoreSetup, config: FleetConfigType) => {
+export const fetchFleetUsage = async (
+  core: CoreSetup,
+  config: FleetConfigType,
+  abortController: AbortController
+) => {
   const [soClient, esClient] = await getInternalClients(core);
+  if (!soClient || !esClient) {
+    return;
+  }
   const usage = {
     agents_enabled: getIsAgentsEnabled(config),
     agents: await getAgentUsage(config, soClient, esClient),
     fleet_server: await getFleetServerUsage(soClient, esClient),
     packages: await getPackageUsage(soClient),
-    ...(await getAgentData(esClient)),
-    fleet_server_config: await getFleetServerConfig(soClient, esClient),
-    agent_policies: await getAgentPoliciesUsage(soClient, esClient),
+    ...(await getAgentData(esClient, abortController)),
+    fleet_server_config: await getFleetServerConfig(soClient),
+    agent_policies: await getAgentPoliciesUsage(esClient, abortController),
   };
   return usage;
 };

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

@@ -374,7 +374,8 @@ export class FleetPlugin
 
     // Register usage collection
     registerFleetUsageCollector(core, config, deps.usageCollection);
-    const fetch = async () => fetchFleetUsage(core, config);
+    const fetch = async (abortController: AbortController) =>
+      await fetchFleetUsage(core, config, abortController);
     this.fleetUsageSender = new FleetUsageSender(deps.taskManager, core, fetch);
     registerFleetUsageLogger(deps.taskManager, async () => fetchAgentsUsage(core, config));
 

x-pack/plugins/fleet/server/services/telemetry/fleet_usage_sender.ts

@@ -11,6 +11,7 @@ import type {
 } from '@kbn/task-manager-plugin/server';
 import { throwUnrecoverableError } from '@kbn/task-manager-plugin/server';
 import type { CoreSetup } from '@kbn/core/server';
+import { withSpan } from '@kbn/apm-utils';
 
 import type { Usage } from '../../collectors/register';
 
@@ -26,24 +27,30 @@ export class FleetUsageSender {
   private taskType = 'Fleet-Usage-Sender';
   private wasStarted: boolean = false;
   private interval = '1h';
+  private timeout = '1m';
+  private abortController = new AbortController();
 
   constructor(
     taskManager: TaskManagerSetupContract,
     core: CoreSetup,
-    fetchUsage: () => Promise<Usage>
+    fetchUsage: (abortController: AbortController) => Promise<Usage | undefined>
   ) {
     taskManager.registerTaskDefinitions({
       [this.taskType]: {
         title: 'Fleet Usage Sender',
-        timeout: '1m',
+        timeout: this.timeout,
         maxAttempts: 1,
         createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
           return {
             run: async () => {
-              return this.runTask(taskInstance, core, fetchUsage);
+              return withSpan({ name: this.taskType, type: 'telemetry' }, () =>
+                this.runTask(taskInstance, core, () => fetchUsage(this.abortController))
+              );
             },
 
-            cancel: async () => {},
+            cancel: async () => {
+              this.abortController.abort('task timed out');
+            },
           };
         },
       },
@@ -54,7 +61,7 @@ export class FleetUsageSender {
   private runTask = async (
     taskInstance: ConcreteTaskInstance,
     core: CoreSetup,
-    fetchUsage: () => Promise<Usage>
+    fetchUsage: () => Promise<Usage | undefined>
   ) => {
     if (!this.wasStarted) {
       appContextService.getLogger().debug('[runTask()] Aborted. Task not started yet');
@@ -69,6 +76,9 @@ export class FleetUsageSender {
 
     try {
       const usageData = await fetchUsage();
+      if (!usageData) {
+        return;
+      }
       appContextService.getLogger().debug(JSON.stringify(usageData));
       core.analytics.reportEvent(EVENT_TYPE, usageData);
     } catch (error) {