[Metrics UI] Increase composite size to 10K for Metric Threshold Rule and optimize processing

x-pack/plugins/infra/server/lib/alerting/metric_threshold/lib/do_work.ts

@@ -0,0 +1,14 @@
+/*
+ * 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.
+ */
+
+export const doWork = async <T>(fn: () => T): Promise<T> => {
+  return new Promise((resolve) => {
+    setTimeout(() => {
+      resolve(fn());
+    });
+  });
+};

x-pack/plugins/infra/server/lib/alerting/metric_threshold/lib/evaluate_rule.ts

@@ -7,7 +7,7 @@
 
 import moment from 'moment';
 import { ElasticsearchClient } from 'kibana/server';
-import { mapValues, first, last, isNaN, isNumber, isObject, has } from 'lodash';
+import { difference, mapValues, first, last, isNaN, isNumber, isObject, has } from 'lodash';
 import {
   isTooManyBucketsPreviewException,
   TOO_MANY_BUCKETS_PREVIEW_EXCEPTION,
@@ -22,6 +22,7 @@ import { UNGROUPED_FACTORY_KEY } from '../../common/utils';
 import { MetricExpressionParams, Comparator, Aggregators } from '../types';
 import { getElasticsearchMetricQuery } from './metric_query';
 import { createTimerange } from './create_timerange';
+import { doWork } from './do_work';
 
 interface AggregationWithoutIntervals {
   aggregatedValue: { value: number; values?: Array<{ key: number; value: number }> };
@@ -66,6 +67,7 @@ export const evaluateRule = <Params extends EvaluatedRuleParams = EvaluatedRuleP
   params: Params,
   config: InfraSource['configuration'],
   prevGroups: string[],
+  compositeSize: number,
   timeframe?: { start?: number; end: number }
 ) => {
   const { criteria, groupBy, filterQuery, shouldDropPartialBuckets } = params;
@@ -78,6 +80,7 @@ export const evaluateRule = <Params extends EvaluatedRuleParams = EvaluatedRuleP
         config.metricAlias,
         groupBy,
         filterQuery,
+        compositeSize,
         timeframe,
         shouldDropPartialBuckets
       );
@@ -96,27 +99,22 @@ export const evaluateRule = <Params extends EvaluatedRuleParams = EvaluatedRuleP
       // If any previous groups are no longer being reported, backfill them with null values
       const currentGroups = Object.keys(currentValues);
 
-      const missingGroups = prevGroups.filter((g) => !currentGroups.includes(g));
+      const missingGroups = difference(prevGroups, currentGroups);
+
       if (currentGroups.length === 0 && missingGroups.length === 0) {
         missingGroups.push(UNGROUPED_FACTORY_KEY);
       }
       const backfillTimestamp =
         last(last(Object.values(currentValues)))?.key ?? new Date().toISOString();
-      const backfilledPrevGroups: Record<
-        string,
-        Array<{ key: string; value: number }>
-      > = missingGroups.reduce(
-        (result, group) => ({
-          ...result,
-          [group]: [
-            {
-              key: backfillTimestamp,
-              value: criterion.aggType === Aggregators.COUNT ? 0 : null,
-            },
-          ],
-        }),
-        {}
-      );
+      const backfilledPrevGroups: Record<string, Array<{ key: string; value: number | null }>> = {};
+      for (const group of missingGroups) {
+        backfilledPrevGroups[group] = [
+          {
+            key: backfillTimestamp,
+            value: criterion.aggType === Aggregators.COUNT ? 0 : null,
+          },
+        ];
+      }
       const currentValuesWithBackfilledPrevGroups = {
         ...currentValues,
         ...backfilledPrevGroups,
@@ -150,6 +148,7 @@ const getMetric: (
   index: string,
   groupBy: string | undefined | string[],
   filterQuery: string | undefined,
+  compositeSize: number,
   timeframe?: { start?: number; end: number },
   shouldDropPartialBuckets?: boolean
 ) => Promise<Record<string, Array<{ key: string; value: number }>>> = async function (
@@ -158,6 +157,7 @@ const getMetric: (
   index,
   groupBy,
   filterQuery,
+  compositeSize,
   timeframe,
   shouldDropPartialBuckets
 ) {
@@ -172,6 +172,7 @@ const getMetric: (
   const searchBody = getElasticsearchMetricQuery(
     params,
     calculatedTimerange,
+    compositeSize,
     hasGroupBy ? groupBy : undefined,
     filterQuery
   );
@@ -202,20 +203,23 @@ const getMetric: (
         bucketSelector,
         afterKeyHandler
       )) as Array<Aggregation & { key: Record<string, string>; doc_count: number }>;
-      const groupedResults = compositeBuckets.reduce(
-        (result, bucket) => ({
-          ...result,
-          [Object.values(bucket.key)
-            .map((value) => value)
-            .join(', ')]: getValuesFromAggregations(
-            bucket,
-            aggType,
-            dropPartialBucketsOptions,
-            calculatedTimerange,
-            bucket.doc_count
-          ),
-        }),
-        {}
+      const groupedResults: Record<string, any> = {};
+      await Promise.all(
+        compositeBuckets.map((bucket) => {
+          return doWork(() => {
+            const key = Object.values(bucket.key)
+              .map((value) => value)
+              .join(', ');
+            const value = getValuesFromAggregations(
+              bucket,
+              aggType,
+              dropPartialBucketsOptions,
+              calculatedTimerange,
+              bucket.doc_count
+            );
+            groupedResults[key] = value;
+          });
+        })
       );
       return groupedResults;
     }

x-pack/plugins/infra/server/lib/alerting/metric_threshold/lib/metric_query.test.ts

@@ -24,7 +24,7 @@ describe("The Metric Threshold Alert's getElasticsearchMetricQuery", () => {
   };
 
   describe('when passed no filterQuery', () => {
-    const searchBody = getElasticsearchMetricQuery(expressionParams, timeframe, groupBy);
+    const searchBody = getElasticsearchMetricQuery(expressionParams, timeframe, 100, groupBy);
     test('includes a range filter', () => {
       expect(
         searchBody.query.bool.filter.find((filter) => filter.hasOwnProperty('range'))
@@ -47,6 +47,7 @@ describe("The Metric Threshold Alert's getElasticsearchMetricQuery", () => {
     const searchBody = getElasticsearchMetricQuery(
       expressionParams,
       timeframe,
+      100,
       groupBy,
       filterQuery
     );

x-pack/plugins/infra/server/lib/alerting/metric_threshold/lib/metric_query.ts

@@ -11,8 +11,6 @@ import { MetricExpressionParams, Aggregators } from '../types';
 import { createPercentileAggregation } from './create_percentile_aggregation';
 import { calculateDateHistogramOffset } from '../../../metrics/lib/calculate_date_histogram_offset';
 
-const COMPOSITE_RESULTS_PER_PAGE = 100;
-
 const getParsedFilterQuery: (filterQuery: string | undefined) => Record<string, any> | null = (
   filterQuery
 ) => {
@@ -23,6 +21,7 @@ const getParsedFilterQuery: (filterQuery: string | undefined) => Record<string,
 export const getElasticsearchMetricQuery = (
   { metric, aggType, timeUnit, timeSize }: MetricExpressionParams,
   timeframe: { start: number; end: number },
+  compositeSize: number,
   groupBy?: string | string[],
   filterQuery?: string
 ) => {
@@ -73,7 +72,7 @@ export const getElasticsearchMetricQuery = (
     ? {
         groupings: {
           composite: {
-            size: COMPOSITE_RESULTS_PER_PAGE,
+            size: compositeSize,
             sources: Array.isArray(groupBy)
               ? groupBy.map((field, index) => ({
                   [`groupBy${index}`]: {

x-pack/plugins/infra/server/lib/alerting/metric_threshold/metric_threshold_executor.test.ts

@@ -747,6 +747,9 @@ describe('The metric threshold alert type', () => {
 });
 
 const createMockStaticConfiguration = (sources: any) => ({
+  alerting: {
+    group_by_page_size: 100,
+  },
   inventory: {
     compositeSize: 2000,
   },

x-pack/plugins/infra/server/lib/alerting/metric_threshold/metric_threshold_executor.ts

@@ -118,6 +118,7 @@ export const createMetricThresholdExecutor = (libs: InfraBackendLibs) =>
       sourceId || 'default'
     );
     const config = source.configuration;
+    const compositeSize = libs.configuration.alerting.group_by_page_size;
 
     const previousGroupBy = state.groupBy;
     const previousFilterQuery = state.filterQuery;
@@ -135,7 +136,8 @@ export const createMetricThresholdExecutor = (libs: InfraBackendLibs) =>
       services.scopedClusterClient.asCurrentUser,
       params as EvaluatedRuleParams,
       config,
-      prevGroups
+      prevGroups,
+      compositeSize
     );
 
     // Because each alert result has the same group definitions, just grab the groups from the first one.
@@ -248,7 +250,6 @@ export const createMetricThresholdExecutor = (libs: InfraBackendLibs) =>
         });
       }
     }
-
     return { groups, groupBy: params.groupBy, filterQuery: params.filterQuery };
   });
 

x-pack/plugins/infra/server/lib/infra_types.ts

@@ -6,7 +6,7 @@
  */
 
 import { handleEsError } from '../../../../../src/plugins/es_ui_shared/server';
-import { InfraConfig } from '../plugin';
+import { InfraConfig } from '../types';
 import { GetLogQueryFields } from '../services/log_queries/get_log_query_fields';
 import { RulesServiceSetup } from '../services/rules';
 import { KibanaFramework } from './adapters/framework/kibana_framework_adapter';

x-pack/plugins/infra/server/lib/sources/sources.test.ts

@@ -109,6 +109,9 @@ describe('the InfraSources lib', () => {
 });
 
 const createMockStaticConfiguration = (sources: any) => ({
+  alerting: {
+    group_by_page_size: 10000,
+  },
   enabled: true,
   inventory: {
     compositeSize: 2000,

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

@@ -6,7 +6,7 @@
  */
 
 import { Server } from '@hapi/hapi';
-import { schema, TypeOf } from '@kbn/config-schema';
+import { schema } from '@kbn/config-schema';
 import { i18n } from '@kbn/i18n';
 import { Logger } from '@kbn/logging';
 import {
@@ -35,15 +35,18 @@ import { InfraBackendLibs, InfraDomainLibs } from './lib/infra_types';
 import { infraSourceConfigurationSavedObjectType, InfraSources } from './lib/sources';
 import { InfraSourceStatus } from './lib/source_status';
 import { LogEntriesService } from './services/log_entries';
-import { InfraPluginRequestHandlerContext } from './types';
+import { InfraPluginRequestHandlerContext, InfraConfig } from './types';
 import { UsageCollector } from './usage/usage_collector';
 import { createGetLogQueryFields } from './services/log_queries/get_log_query_fields';
 import { handleEsError } from '../../../../src/plugins/es_ui_shared/server';
 import { RulesService } from './services/rules';
 import { configDeprecations, getInfraDeprecationsFactory } from './deprecations';
 
-export const config: PluginConfigDescriptor = {
+export const config: PluginConfigDescriptor<InfraConfig> = {
   schema: schema.object({
+    alerting: schema.object({
+      group_by_page_size: schema.number({ defaultValue: 10000 }),
+    }),
     inventory: schema.object({
       compositeSize: schema.number({ defaultValue: 2000 }),
     }),
@@ -64,7 +67,7 @@ export const config: PluginConfigDescriptor = {
   deprecations: configDeprecations,
 };
 
-export type InfraConfig = TypeOf<typeof config.schema>;
+export type { InfraConfig };
 
 export interface KbnServer extends Server {
   usage: any;

x-pack/plugins/infra/server/types.ts

@@ -31,3 +31,19 @@ export interface InfraPluginRequestHandlerContext extends RequestHandlerContext
   infra: InfraRequestHandlerContext;
   search: SearchRequestHandlerContext;
 }
+
+export interface InfraConfig {
+  alerting: {
+    group_by_page_size: number;
+  };
+  inventory: {
+    compositeSize: number;
+  };
+  sources?: {
+    default?: {
+      fields?: {
+        message?: string[];
+      };
+    };
+  };
+}