[Alerting v2] Flag rule ESQL query errors as user errors (#270643)

jcger · cnasikas · web-flow · commit 9365dc14c3df · 2026-05-27T08:17:11.000-07:00
## Summary Implements part of elastic/rna-program#430 This PR introduces `isEsqlUserError`: a small predicate that returns `true` for `ResponseError` with status 400 or 404, and applies it in two places: - **Main rule query** (`QueryService.executeQueryStream`): on a user error, wraps the thrown error with `createTaskRunError(..., TaskErrorSource.USER)` before rethrowing. Non-user errors (5xx, network, cancellation, Arrow parse errors) are rethrown unchanged. - **Recovery query** (`CreateRecoveryEventsStep`): same wrapping applied to the `recovery_policy.type === 'query'` execution path. --------- Co-authored-by: Christos Nasikas <xristosnasikas@gmail.com>
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/errors/esql_user_error.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/errors/esql_user_error.test.ts
@@ -0,0 +1,30 @@
+/*
+ * 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 { DiagnosticResult } from '@elastic/elasticsearch';
+import { errors } from '@elastic/elasticsearch';
+import { isEsqlUserError } from './esql_user_error';
+
+const makeResponseError = (statusCode: number) =>
+  new errors.ResponseError({ statusCode } as DiagnosticResult);
+
+describe('isEsqlUserError', () => {
+  it.each([400, 401, 403, 404])(
+    'returns true for ResponseError with statusCode %i',
+    (statusCode) => {
+      expect(isEsqlUserError(makeResponseError(statusCode))).toBe(true);
+    }
+  );
+
+  it('returns false for ResponseError with statusCode 503', () => {
+    expect(isEsqlUserError(makeResponseError(503))).toBe(false);
+  });
+
+  it('returns false for a plain Error', () => {
+    expect(isEsqlUserError(new Error('something went wrong'))).toBe(false);
+  });
+});
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/errors/esql_user_error.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/errors/esql_user_error.ts
@@ -0,0 +1,15 @@
+/*
+ * 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 { isResponseError } from '@kbn/es-errors';
+
+// 400: syntax/semantic ES|QL query errors (verification_exception, parsing_exception)
+// 404: unknown index referenced in the query
+const USER_ERROR_STATUS_CODES = new Set<number | undefined>([400, 401, 403, 404]);
+
+export const isEsqlUserError = (error: unknown): boolean =>
+  isResponseError(error) && USER_ERROR_STATUS_CODES.has(error.statusCode);
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/create_recovery_events_step.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/create_recovery_events_step.test.ts
@@ -5,7 +5,11 @@
  * 2.0.
  */
 
+import type { DiagnosticResult } from '@elastic/elasticsearch';
+import { errors } from '@elastic/elasticsearch';
 import { CreateRecoveryEventsStep } from './create_recovery_events_step';
+import { TaskErrorSource } from '@kbn/task-manager-plugin/server';
+import { getErrorSource } from '@kbn/task-manager-plugin/server/task_running';
 import {
   collectStreamResults,
   createPipelineStream,
@@ -14,6 +18,7 @@ import {
   createAlertEvent,
   createRuleResponse,
   createEsqlResponse,
+  getStepError,
 } from '../test_utils';
 import { createLoggerService } from '../../services/logger_service/logger_service.mock';
 import { createQueryService } from '../../services/query_service/query_service.mock';
@@ -331,6 +336,80 @@ describe('CreateRecoveryEventsStep', () => {
       expect(result.state.alertEventsBatch![0].status).toBe('breached');
       expect(result.state.alertEventsBatch![0].group_hash).toBe('hash-new');
     });
+
+    it('marks ResponseError(400) recovery query errors as TaskErrorSource.USER', async () => {
+      const { step, internalEsClient, scopedEsClient } = createStep();
+
+      internalEsClient.esql.query.mockResolvedValue(createActiveGroupHashesResponse(['hash-1']));
+      scopedEsClient.esql.query.mockRejectedValue(
+        // @ts-expect-error: Not all params are needed for the test.
+        new errors.ResponseError({ statusCode: 400 })
+      );
+
+      const state = createRulePipelineState({
+        rule: createRuleResponse({
+          kind: 'alert',
+          recovery_policy: {
+            type: 'query',
+            query: { base: 'FROM logs-* | WHERE invalid syntax' },
+          },
+        }),
+        alertEventsBatch: [],
+      });
+
+      const error = await getStepError(step, state);
+
+      expect(error).toBeInstanceOf(Error);
+      expect(getErrorSource(error!)).toBe(TaskErrorSource.USER);
+    });
+
+    it('does not mark ResponseError(503) recovery query errors as TaskErrorSource.USER', async () => {
+      const { step, internalEsClient, scopedEsClient } = createStep();
+
+      internalEsClient.esql.query.mockResolvedValue(createActiveGroupHashesResponse(['hash-1']));
+      scopedEsClient.esql.query.mockRejectedValue(
+        new errors.ResponseError({ statusCode: 503 } as DiagnosticResult)
+      );
+
+      const state = createRulePipelineState({
+        rule: createRuleResponse({
+          kind: 'alert',
+          recovery_policy: {
+            type: 'query',
+            query: { base: 'FROM logs-*' },
+          },
+        }),
+        alertEventsBatch: [],
+      });
+
+      const error = await getStepError(step, state);
+
+      expect(error).toBeInstanceOf(Error);
+      expect(getErrorSource(error!)).toBeUndefined();
+    });
+
+    it('does not mark plain recovery query errors as TaskErrorSource.USER', async () => {
+      const { step, internalEsClient, scopedEsClient } = createStep();
+
+      internalEsClient.esql.query.mockResolvedValue(createActiveGroupHashesResponse(['hash-1']));
+      scopedEsClient.esql.query.mockRejectedValue(new Error('connection reset'));
+
+      const state = createRulePipelineState({
+        rule: createRuleResponse({
+          kind: 'alert',
+          recovery_policy: {
+            type: 'query',
+            query: { base: 'FROM logs-*' },
+          },
+        }),
+        alertEventsBatch: [],
+      });
+
+      const error = await getStepError(step, state);
+
+      expect(error).toBeInstanceOf(Error);
+      expect(getErrorSource(error!)).toBeUndefined();
+    });
   });
 
   describe('abort signal', () => {
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/create_recovery_events_step.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/create_recovery_events_step.ts
@@ -6,8 +6,10 @@
  */
 
 import { inject, injectable } from 'inversify';
+import { createTaskRunError, TaskErrorSource } from '@kbn/task-manager-plugin/server';
 import { stableStringify } from '@kbn/std';
 import { recoveryPolicyType } from '@kbn/alerting-v2-schemas';
+import { isEsqlUserError } from '../../errors/esql_user_error';
 import type { PipelineStateStream, RuleExecutionStep, RulePipelineState } from '../types';
 import { buildRecoveryAlertEvents, buildQueryRecoveryAlertEvents } from '../build_alert_events';
 import { getQueryPayload } from '../get_query_payload';
@@ -126,22 +128,29 @@ export class CreateRecoveryEventsStep implements RuleExecutionStep {
         })}`,
     });
 
-    const esqlResponse = await this.scopedQueryService.executeQuery({
-      query: effectiveQuery,
-      filter: queryPayload.filter,
-      params: queryPayload.params,
-      abortSignal: input.executionContext.signal,
-    });
+    try {
+      const esqlResponse = await this.scopedQueryService.executeQuery({
+        query: effectiveQuery,
+        filter: queryPayload.filter,
+        params: queryPayload.params,
+        abortSignal: input.executionContext.signal,
+      });
 
-    return buildQueryRecoveryAlertEvents({
-      ruleId: rule.id,
-      ruleVersion: 1,
-      spaceId: input.spaceId,
-      ruleAttributes: rule,
-      activeGroupHashes,
-      esqlResponse,
-      scheduledTimestamp: input.scheduledAt,
-    });
+      return buildQueryRecoveryAlertEvents({
+        ruleId: rule.id,
+        ruleVersion: 1,
+        spaceId: input.spaceId,
+        ruleAttributes: rule,
+        activeGroupHashes,
+        esqlResponse,
+        scheduledTimestamp: input.scheduledAt,
+      });
+    } catch (error) {
+      if (isEsqlUserError(error)) {
+        throw createTaskRunError(error as Error, TaskErrorSource.USER);
+      }
+      throw error;
+    }
   }
 
   private async fetchActiveAlertGroupHashes(
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/execute_rule_query_step.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/execute_rule_query_step.test.ts
@@ -5,13 +5,18 @@
  * 2.0.
  */
 
+import type { DiagnosticResult } from '@elastic/elasticsearch';
+import { errors } from '@elastic/elasticsearch';
+import { TaskErrorSource } from '@kbn/task-manager-plugin/server';
+import { getErrorSource } from '@kbn/task-manager-plugin/server/task_running';
 import { ExecuteRuleQueryStep } from './execute_rule_query_step';
 import {
   collectStreamResults,
   createPipelineStream,
   createRuleExecutionInput,
   createRuleResponse,
   createRulePipelineState,
+  getStepError,
   mockHelpersEsqlArrowBatches,
   mockHelpersEsqlToArrowReader,
 } from '../test_utils';
@@ -91,6 +96,34 @@ describe('ExecuteRuleQueryStep', () => {
     ).rejects.toThrow('Query execution failed');
   });
 
+  it('marks ResponseError(400) ES|QL errors as TaskErrorSource.USER', async () => {
+    mockHelpersEsqlToArrowReader(
+      mockEsClient,
+      jest.fn().mockRejectedValue(new errors.ResponseError({ statusCode: 400 } as DiagnosticResult))
+    );
+
+    const state = createRulePipelineState({ rule: createRuleResponse() });
+
+    const error = await getStepError(step, state);
+
+    expect(error).toBeInstanceOf(Error);
+    expect(getErrorSource(error!)).toBe(TaskErrorSource.USER);
+  });
+
+  it('does not mark plain ES|QL errors as TaskErrorSource.USER', async () => {
+    mockHelpersEsqlToArrowReader(
+      mockEsClient,
+      jest.fn().mockRejectedValue(new Error('ES query failed'))
+    );
+
+    const state = createRulePipelineState({ rule: createRuleResponse() });
+
+    const error = await getStepError(step, state);
+
+    expect(error).toBeInstanceOf(Error);
+    expect(getErrorSource(error!)).toBeUndefined();
+  });
+
   it('yields rows from query results', async () => {
     mockHelpersEsqlArrowBatches(mockEsClient, [
       {
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/execute_rule_query_step.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/steps/execute_rule_query_step.ts
@@ -6,6 +6,8 @@
  */
 
 import { inject, injectable } from 'inversify';
+import { createTaskRunError, TaskErrorSource } from '@kbn/task-manager-plugin/server';
+import { isEsqlUserError } from '../../errors/esql_user_error';
 import type { PipelineStateStream, RuleExecutionStep } from '../types';
 import { getQueryPayload } from '../get_query_payload';
 import {
@@ -60,28 +62,35 @@ export class ExecuteRuleQueryStep implements RuleExecutionStep {
           })}`,
       });
 
-      const esqlRowBatchStream = step.queryService.executeQueryStream({
-        query: effectiveQuery,
-        filter: queryPayload.filter,
-        params: queryPayload.params,
-        abortSignal: input.executionContext.signal,
-      });
+      try {
+        const esqlRowBatchStream = step.queryService.executeQueryStream({
+          query: effectiveQuery,
+          filter: queryPayload.filter,
+          params: queryPayload.params,
+          abortSignal: input.executionContext.signal,
+        });
 
-      let yielded = false;
+        let yielded = false;
 
-      for await (const batch of esqlRowBatchStream) {
-        yielded = true;
-        yield {
-          type: 'continue',
-          state: { ...state, queryPayload, esqlRowBatch: batch },
-        };
-      }
+        for await (const batch of esqlRowBatchStream) {
+          yielded = true;
+          yield {
+            type: 'continue',
+            state: { ...state, queryPayload, esqlRowBatch: batch },
+          };
+        }
 
-      if (!yielded) {
-        yield {
-          type: 'continue',
-          state: { ...state, queryPayload, esqlRowBatch: [] },
-        };
+        if (!yielded) {
+          yield {
+            type: 'continue',
+            state: { ...state, queryPayload, esqlRowBatch: [] },
+          };
+        }
+      } catch (error) {
+        if (isEsqlUserError(error)) {
+          throw createTaskRunError(error as Error, TaskErrorSource.USER);
+        }
+        throw error;
       }
     });
   }
diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/test_utils.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/rule_executor/test_utils.ts
@@ -5,4 +5,19 @@
  * 2.0.
  */
 
+import { collectStreamResults, createPipelineStream } from '../test_utils';
+import type { RuleExecutionStep, RulePipelineState } from './types';
+
+export async function getStepError(
+  step: RuleExecutionStep,
+  state: RulePipelineState
+): Promise<Error | undefined> {
+  try {
+    await collectStreamResults(step.executeStream(createPipelineStream([state])));
+    return undefined;
+  } catch (error) {
+    return error as Error;
+  }
+}
+
 export * from '../test_utils';
