[8.10] [Security Solution] [Detections] Display shard failure messages for threshold rules (#164231) (#164843)

# Backport

This will backport the following commits from `main` to `8.10`:
- [[Security Solution] [Detections] Display shard failure messages for
threshold rules
(#164231)](https://github.com/elastic/kibana/pull/164231)

<!--- Backport version: 8.9.7 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Devin W.
Hurley","email":"devin.hurley@elastic.co"},"sourceCommit":{"committedDate":"2023-08-25T13:52:35Z","message":"[Security
Solution] [Detections] Display shard failure messages for threshold
rules (#164231)\n\n## Summary\r\n\r\nref:
https://github.com/elastic/kibana/issues/163369\r\n\r\nprevents
threshold rules from throwing error message that covers up\r\nshard
failures messages\r\n\r\n\r\n<img width=\"1245\"
alt=\"threshold_search_errors\"\r\nsrc=\"9ed9050b-dcc8-456a-957b-a96407da6fe0\">","sha":"19fdc91af9fe81ba360829ab9b0c2250868ce3c1","branchLabelMapping":{"^v8.11.0$":"main","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["review","release_note:fix","fixed","Feature:Threshold
Rule","Team:Detection
Engine","v8.10.0","v8.11.0"],"number":164231,"url":"https://github.com/elastic/kibana/pull/164231","mergeCommit":{"message":"[Security
Solution] [Detections] Display shard failure messages for threshold
rules (#164231)\n\n## Summary\r\n\r\nref:
https://github.com/elastic/kibana/issues/163369\r\n\r\nprevents
threshold rules from throwing error message that covers up\r\nshard
failures messages\r\n\r\n\r\n<img width=\"1245\"
alt=\"threshold_search_errors\"\r\nsrc=\"9ed9050b-dcc8-456a-957b-a96407da6fe0\">","sha":"19fdc91af9fe81ba360829ab9b0c2250868ce3c1"}},"sourceBranch":"main","suggestedTargetBranches":["8.10"],"targetPullRequestStates":[{"branch":"8.10","label":"v8.10.0","labelRegex":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"},{"branch":"main","label":"v8.11.0","labelRegex":"^v8.11.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/164231","number":164231,"mergeCommit":{"message":"[Security
Solution] [Detections] Display shard failure messages for threshold
rules (#164231)\n\n## Summary\r\n\r\nref:
https://github.com/elastic/kibana/issues/163369\r\n\r\nprevents
threshold rules from throwing error message that covers up\r\nshard
failures messages\r\n\r\n\r\n<img width=\"1245\"
alt=\"threshold_search_errors\"\r\nsrc=\"9ed9050b-dcc8-456a-957b-a96407da6fe0\">","sha":"19fdc91af9fe81ba360829ab9b0c2250868ce3c1"}}]}]
BACKPORT-->

Co-authored-by: Devin W. Hurley <devin.hurley@elastic.co>

x-pack/plugins/security_solution/server/lib/detection_engine/rule_types/threshold/find_threshold_signals.ts

@@ -6,6 +6,7 @@
  */
 
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
+import { isEmpty } from 'lodash';
 
 import type {
   AlertInstanceContext,
@@ -28,7 +29,7 @@ import type {
   ThresholdBucket,
   ThresholdSingleBucketAggregationResult,
 } from './types';
-import { shouldFilterByCardinality } from './utils';
+import { shouldFilterByCardinality, searchResultHasAggs } from './utils';
 import type { IRuleExecutionLogForExecutors } from '../../rule_monitoring';
 import { getMaxSignalsWarning } from '../utils/utils';
 
@@ -74,7 +75,6 @@ export const findThresholdSignals = async ({
   warnings: string[];
 }> => {
   // Leaf aggregations used below
-  let sortKeys;
   const buckets: ThresholdBucket[] = [];
   const searchAfterResults: SearchAfterResults = {
     searchDurations: [],
@@ -85,6 +85,7 @@ export const findThresholdSignals = async ({
   const includeCardinalityFilter = shouldFilterByCardinality(threshold);
 
   if (hasThresholdFields(threshold)) {
+    let sortKeys: Record<string, string | number | null> | undefined;
     do {
       const { searchResult, searchDuration, searchErrors } = await singleSearchAfter({
         aggregations: buildThresholdMultiBucketAggregation({
@@ -106,20 +107,22 @@ export const findThresholdSignals = async ({
         secondaryTimestamp,
       });
 
-      const searchResultWithAggs = searchResult as ThresholdMultiBucketAggregationResult;
-      if (!searchResultWithAggs.aggregations) {
-        throw new Error('Aggregations were missing on threshold rule search result');
-      }
-
       searchAfterResults.searchDurations.push(searchDuration);
+      if (!isEmpty(searchErrors)) {
         searchAfterResults.searchErrors.push(...searchErrors);
-
-      const thresholdTerms = searchResultWithAggs.aggregations?.thresholdTerms;
-      sortKeys = thresholdTerms.after_key;
+        sortKeys = undefined; // this will eject us out of the loop
+        // if a search failure occurs on a secondary iteration,
+        // we will return early.
+      } else if (searchResultHasAggs<ThresholdMultiBucketAggregationResult>(searchResult)) {
+        const thresholdTerms = searchResult.aggregations?.thresholdTerms;
+        sortKeys = thresholdTerms?.after_key;
 
         buckets.push(
-        ...(searchResultWithAggs.aggregations.thresholdTerms.buckets as ThresholdBucket[])
+          ...((searchResult.aggregations?.thresholdTerms.buckets as ThresholdBucket[]) ?? [])
         );
+      } else {
+        throw new Error('Aggregations were missing on threshold rule search result');
+      }
     } while (sortKeys && buckets.length <= maxSignals);
   } else {
     const { searchResult, searchDuration, searchErrors } = await singleSearchAfter({
@@ -142,32 +145,34 @@ export const findThresholdSignals = async ({
       secondaryTimestamp,
     });
 
-    const searchResultWithAggs = searchResult as ThresholdSingleBucketAggregationResult;
-    if (!searchResultWithAggs.aggregations) {
-      throw new Error('Aggregations were missing on threshold rule search result');
-    }
-
     searchAfterResults.searchDurations.push(searchDuration);
     searchAfterResults.searchErrors.push(...searchErrors);
 
-    const docCount = searchResultWithAggs.hits.total.value;
+    if (
+      !searchResultHasAggs<ThresholdSingleBucketAggregationResult>(searchResult) &&
+      isEmpty(searchErrors)
+    ) {
+      throw new Error('Aggregations were missing on threshold rule search result');
+    } else if (searchResultHasAggs<ThresholdSingleBucketAggregationResult>(searchResult)) {
+      const docCount = searchResult.hits.total.value;
       if (
         docCount >= threshold.value &&
         (!includeCardinalityFilter ||
-        (searchResultWithAggs.aggregations.cardinality_count?.value ?? 0) >=
+          (searchResult?.aggregations?.cardinality_count?.value ?? 0) >=
             threshold.cardinality[0].value)
       ) {
         buckets.push({
           doc_count: docCount,
           key: {},
-        max_timestamp: searchResultWithAggs.aggregations.max_timestamp,
-        min_timestamp: searchResultWithAggs.aggregations.min_timestamp,
+          max_timestamp: searchResult.aggregations?.max_timestamp ?? { value: null },
+          min_timestamp: searchResult.aggregations?.min_timestamp ?? { value: null },
           ...(includeCardinalityFilter
-          ? { cardinality_count: searchResultWithAggs.aggregations.cardinality_count }
+            ? { cardinality_count: searchResult.aggregations?.cardinality_count }
             : {}),
         });
       }
     }
+  }
 
   if (buckets.length > maxSignals) {
     warnings.push(getMaxSignalsWarning());

x-pack/plugins/security_solution/server/lib/detection_engine/rule_types/threshold/threshold.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { isEmpty } from 'lodash';
 import type { SearchHit } from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 import type { ExceptionListItemSchema } from '@kbn/securitysolution-io-ts-list-types';
@@ -135,8 +136,6 @@ export const thresholdExecutor = async ({
       aggregatableTimestampField,
     });
 
-    // Build and index new alerts
-
     const createResult = await bulkCreateThresholdSignals({
       buckets,
       completeRule,
@@ -152,7 +151,10 @@ export const thresholdExecutor = async ({
       ruleExecutionLogger,
     });
 
-    addToSearchAfterReturn({ current: result, next: createResult });
+    addToSearchAfterReturn({
+      current: result,
+      next: { ...createResult, success: createResult.success && isEmpty(searchErrors) },
+    });
 
     result.errors.push(...previousSearchErrors);
     result.errors.push(...searchErrors);

x-pack/plugins/security_solution/server/lib/detection_engine/rule_types/threshold/utils.ts

@@ -5,14 +5,20 @@
  * 2.0.
  */
 
+import type { estypes } from '@elastic/elasticsearch';
 import { createHash } from 'crypto';
 import { v5 as uuidv5 } from 'uuid';
 import type {
   ThresholdNormalized,
   ThresholdWithCardinality,
 } from '../../../../../common/api/detection_engine/model/rule_schema';
-import type { RuleRangeTuple } from '../types';
-import type { ThresholdSignalHistory, ThresholdAlertState } from './types';
+import type { RuleRangeTuple, SignalSearchResponse } from '../types';
+import type {
+  ThresholdSignalHistory,
+  ThresholdAlertState,
+  ThresholdSingleBucketAggregationResult,
+  ThresholdMultiBucketAggregationResult,
+} from './types';
 
 /**
  * Returns a new signal history based on what the previous
@@ -82,3 +88,9 @@ export const getThresholdTermsHash = (
     )
     .digest('hex');
 };
+
+export const searchResultHasAggs = <
+  T extends ThresholdSingleBucketAggregationResult | ThresholdMultiBucketAggregationResult
+>(
+  obj: SignalSearchResponse<Record<estypes.AggregateName, estypes.AggregationsAggregate>>
+): obj is T => obj?.aggregations != null;