[8.x] Hook up discovery service to Task Manager health (elastic#194113)…

… (elastic#194685) # Backport This will backport the following commits from `main` to `8.x`: - [Hook up discovery service to Task Manager health (elastic#194113)](elastic#194113)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Mike Côté <[email protected]>
kibanamachine · Oct 2, 2024 · c24fee0 · c24fee0
1 parent 9f500cd
commit c24fee0
Show file tree

Hide file tree

Showing 12 changed files with 171 additions and 90 deletions.
diff --git a/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.test.ts b/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.test.ts
@@ -246,6 +246,7 @@ describe('KibanaDiscoveryService', () => {
     savedObjectsRepository.find.mockResolvedValueOnce(createFindResponse(mockActiveNodes));
 
     it('returns the active kibana nodes', async () => {
+      const onNodesCounted = jest.fn();
       const kibanaDiscoveryService = new KibanaDiscoveryService({
         savedObjectsRepository,
         logger,
@@ -254,6 +255,7 @@ describe('KibanaDiscoveryService', () => {
           active_nodes_lookback: DEFAULT_ACTIVE_NODES_LOOK_BACK_DURATION,
           interval: DEFAULT_DISCOVERY_INTERVAL_MS,
         },
+        onNodesCounted,
       });
 
       const activeNodes = await kibanaDiscoveryService.getActiveKibanaNodes();
@@ -265,6 +267,7 @@ describe('KibanaDiscoveryService', () => {
         type: BACKGROUND_TASK_NODE_SO_NAME,
       });
       expect(activeNodes).toEqual(mockActiveNodes);
+      expect(onNodesCounted).toHaveBeenCalledWith(mockActiveNodes.length);
     });
   });
 

diff --git a/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.ts b/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.ts
@@ -16,6 +16,7 @@ interface DiscoveryServiceParams {
   currentNode: string;
   savedObjectsRepository: ISavedObjectsRepository;
   logger: Logger;
+  onNodesCounted?: (numOfNodes: number) => void;
 }
 
 interface DiscoveryServiceUpsertParams {
@@ -34,13 +35,15 @@ export class KibanaDiscoveryService {
   private logger: Logger;
   private stopped = false;
   private timer: NodeJS.Timeout | undefined;
+  private onNodesCounted?: (numOfNodes: number) => void;
 
-  constructor({ config, currentNode, savedObjectsRepository, logger }: DiscoveryServiceParams) {
-    this.activeNodesLookBack = config.active_nodes_lookback;
-    this.discoveryInterval = config.interval;
-    this.savedObjectsRepository = savedObjectsRepository;
-    this.logger = logger;
-    this.currentNode = currentNode;
+  constructor(opts: DiscoveryServiceParams) {
+    this.activeNodesLookBack = opts.config.active_nodes_lookback;
+    this.discoveryInterval = opts.config.interval;
+    this.savedObjectsRepository = opts.savedObjectsRepository;
+    this.logger = opts.logger;
+    this.currentNode = opts.currentNode;
+    this.onNodesCounted = opts.onNodesCounted;
   }
 
   private async upsertCurrentNode({ id, lastSeen }: DiscoveryServiceUpsertParams) {
@@ -106,6 +109,10 @@ export class KibanaDiscoveryService {
         filter: `${BACKGROUND_TASK_NODE_SO_NAME}.attributes.last_seen > now-${this.activeNodesLookBack}`,
       });
 
+    if (this.onNodesCounted) {
+      this.onNodesCounted(activeNodes.length);
+    }
+
     return activeNodes;
   }
 

diff --git a/x-pack/plugins/task_manager/server/monitoring/capacity_estimation.test.ts b/x-pack/plugins/task_manager/server/monitoring/capacity_estimation.test.ts
@@ -63,7 +63,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,
@@ -119,7 +120,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,
@@ -158,7 +160,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,
@@ -214,7 +217,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,
@@ -271,7 +275,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,
@@ -327,7 +332,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        3
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 3,
@@ -396,7 +402,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        2
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: provisionedKibanaInstances,
@@ -477,7 +484,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        2
       ).value
     ).toMatchObject({
       observed: {
@@ -561,7 +569,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       )
     ).toMatchObject({
       status: 'OK',
@@ -626,7 +635,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       )
     ).toMatchObject({
       status: 'OK',
@@ -691,7 +701,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       )
     ).toMatchObject({
       status: 'OK',
@@ -755,7 +766,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       )
     ).toMatchObject({
       status: 'OK',
@@ -831,7 +843,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       )
     ).toMatchObject({
       status: 'OK',
@@ -905,7 +918,8 @@ describe('estimateCapacity', () => {
               result_frequency_percent_as_number: {},
             },
           }
-        )
+        ),
+        1
       ).value.observed
     ).toMatchObject({
       observed_kibana_instances: 1,

diff --git a/x-pack/plugins/task_manager/server/monitoring/capacity_estimation.ts b/x-pack/plugins/task_manager/server/monitoring/capacity_estimation.ts
@@ -46,11 +46,10 @@ function isCapacityEstimationParams(
 
 export function estimateCapacity(
   logger: Logger,
-  capacityStats: CapacityEstimationParams
+  capacityStats: CapacityEstimationParams,
+  assumedKibanaInstances: number
 ): RawMonitoredStat<CapacityEstimationStat> {
   const workload = capacityStats.workload.value;
-  // if there are no active owners right now, assume there's at least 1
-  const assumedKibanaInstances = Math.max(workload.owner_ids, 1);
 
   const {
     load: { p90: averageLoadPercentage },
@@ -262,12 +261,13 @@ function getHealthStatus(
 
 export function withCapacityEstimate(
   logger: Logger,
-  monitoredStats: RawMonitoringStats['stats']
+  monitoredStats: RawMonitoringStats['stats'],
+  assumedKibanaInstances: number
 ): RawMonitoringStats['stats'] {
   if (isCapacityEstimationParams(monitoredStats)) {
     return {
       ...monitoredStats,
-      capacity_estimation: estimateCapacity(logger, monitoredStats),
+      capacity_estimation: estimateCapacity(logger, monitoredStats, assumedKibanaInstances),
     };
   }
   return monitoredStats;

diff --git a/x-pack/plugins/task_manager/server/monitoring/monitoring_stats_stream.ts b/x-pack/plugins/task_manager/server/monitoring/monitoring_stats_stream.ts
@@ -158,42 +158,47 @@ export function summarizeMonitoringStats(
     last_update,
     stats: { runtime, workload, configuration, ephemeral, utilization },
   }: MonitoringStats,
-  config: TaskManagerConfig
+  config: TaskManagerConfig,
+  assumedKibanaInstances: number
 ): RawMonitoringStats {
-  const summarizedStats = withCapacityEstimate(logger, {
-    ...(configuration
-      ? {
-          configuration: {
-            ...configuration,
-            status: HealthStatus.OK,
-          },
-        }
-      : {}),
-    ...(runtime
-      ? {
-          runtime: {
-            timestamp: runtime.timestamp,
-            ...summarizeTaskRunStat(logger, runtime.value, config),
-          },
-        }
-      : {}),
-    ...(workload
-      ? {
-          workload: {
-            timestamp: workload.timestamp,
-            ...summarizeWorkloadStat(workload.value),
-          },
-        }
-      : {}),
-    ...(ephemeral
-      ? {
-          ephemeral: {
-            timestamp: ephemeral.timestamp,
-            ...summarizeEphemeralStat(ephemeral.value),
-          },
-        }
-      : {}),
-  });
+  const summarizedStats = withCapacityEstimate(
+    logger,
+    {
+      ...(configuration
+        ? {
+            configuration: {
+              ...configuration,
+              status: HealthStatus.OK,
+            },
+          }
+        : {}),
+      ...(runtime
+        ? {
+            runtime: {
+              timestamp: runtime.timestamp,
+              ...summarizeTaskRunStat(logger, runtime.value, config),
+            },
+          }
+        : {}),
+      ...(workload
+        ? {
+            workload: {
+              timestamp: workload.timestamp,
+              ...summarizeWorkloadStat(workload.value),
+            },
+          }
+        : {}),
+      ...(ephemeral
+        ? {
+            ephemeral: {
+              timestamp: ephemeral.timestamp,
+              ...summarizeEphemeralStat(ephemeral.value),
+            },
+          }
+        : {}),
+    },
+    assumedKibanaInstances
+  );
 
   return {
     last_update,

diff --git a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts
@@ -169,10 +169,6 @@ describe('Workload Statistics Aggregator', () => {
               missing: { field: 'task.schedule.interval' },
               aggs: { taskType: { terms: { size: 3, field: 'task.taskType' } } },
             },
-            ownerIds: {
-              filter: { range: { 'task.startedAt': { gte: 'now-1w/w' } } },
-              aggs: { ownerIds: { cardinality: { field: 'task.ownerId' } } },
-            },
             idleTasks: {
               filter: { term: { 'task.status': 'idle' } },
               aggs: {