Incorporate feedback from alerting framework team

Author: sdesalas

x-pack/platform/plugins/shared/alerting/server/application/rule/methods/bulk_create/bulk_create_rules.test.ts

@@ -5,66 +5,138 @@
  * 2.0.
  */
 
+import Boom from '@hapi/boom';
 import pMap from 'p-map';
 import { withSpan } from '@kbn/apm-utils';
-import type { SavedObject, SavedObjectsBulkCreateObject } from '@kbn/core/server';
+import type { SavedObjectsBulkCreateObject } from '@kbn/core/server';
 import { SavedObjectsUtils } from '@kbn/core/server';
 import { RULE_SAVED_OBJECT_TYPE } from '../../../../saved_objects';
 import { getRuleCircuitBreakerErrorMessage } from '../../../../../common';
 import { updateMeta } from '../../../../rules_client/lib';
-import { API_KEY_GENERATE_CONCURRENCY } from '../../../../rules_client/common/constants';
+import {
+  API_KEY_GENERATE_CONCURRENCY,
+  DEFAULT_BULK_CREATE_BATCH_SIZE,
+  MAX_BULK_CREATE_BATCH_SIZE,
+  MAX_RULES_NUMBER_FOR_BULK_OPERATION,
+} from '../../../../rules_client/common/constants';
 import { ruleAuditEvent, RuleAuditAction } from '../../../../rules_client/common/audit_events';
-import { bulkEnableTasks } from '../bulk_enable_tasks';
 import { bulkCreateRulesSo } from '../../../../data/rule';
-import type { RawRule, SanitizedRule } from '../../../../types';
-import type { BulkOperationError, RulesClientContext } from '../../../../rules_client/types';
+import type { RawRule } from '../../../../types';
+import type { RulesClientContext } from '../../../../rules_client/types';
 import type { RuleParams } from '../../types';
 import { validateScheduleLimit } from '../get_schedule_frequency';
 import type {
   ApiKeyEntry,
-  PreparedRule,
+  BatchResult,
+  BulkCreateOperationError,
+  BulkCreateRulesItem,
   BulkCreateRulesParams,
   BulkCreateRulesResult,
+  PreparedRule,
 } from './types';
 import {
   buildTaskInstance,
   collectNewKeysToInvalidate,
   demotePreparedRules,
   flushKeysToInvalidate,
   prepareRule,
-  toSanitizedRule,
 } from './utils';
 
 export async function bulkCreateRules<Params extends RuleParams = never>(
   context: RulesClientContext,
   params: BulkCreateRulesParams<Params>
-): Promise<BulkCreateRulesResult<Params>> {
-  const { rules } = params;
+): Promise<BulkCreateRulesResult> {
+  const { rules, exitEarlyOnError = false } = params;
   const { logger } = context;
   const total = rules.length;
 
   if (total === 0) {
-    return { rules: [], errors: [], total: 0, taskIdsFailedToBeEnabled: [] };
+    return { successfulIds: [], errors: [], total: 0 };
+  }
+
+  if (total > MAX_RULES_NUMBER_FOR_BULK_OPERATION) {
+    throw Boom.badRequest(
+      `bulkCreateRules: ${total} rules exceeds the hard limit of ${MAX_RULES_NUMBER_FOR_BULK_OPERATION}. ` +
+        `Callers should enforce request-level limits before invoking this method.`
+    );
+  }
+
+  const requestedBatchSize = params.batchSize ?? DEFAULT_BULK_CREATE_BATCH_SIZE;
+  const batchSize = Math.max(1, Math.min(MAX_BULK_CREATE_BATCH_SIZE, requestedBatchSize));
+
+  if (requestedBatchSize !== batchSize) {
+    logger.warn(
+      `bulkCreateRules: batchSize ${requestedBatchSize} clamped to ${batchSize} (hard cap ${MAX_BULK_CREATE_BATCH_SIZE}).`
+    );
   }
 
   const username = await context.getUserName();
   const actionsClient = await context.getActionsClient();
 
-  const inputsWithIds = rules.map((rule) => ({
+  const successfulIds: string[] = [];
+  const errors: BulkCreateOperationError[] = [];
+
+  const totalBatches = Math.ceil(total / batchSize);
+  logger.debug(`bulkCreateRules: ${total} rules, ${totalBatches}x batches of ${batchSize} rules.`);
+
+  for (let batchIndex = 0; batchIndex < totalBatches; batchIndex++) {
+    const start = batchIndex * batchSize;
+    const batch = rules.slice(start, start + batchSize);
+
+    const result = await runBatch<Params>({
+      context,
+      username,
+      actionsClient,
+      batch,
+    });
+
+    successfulIds.push(...result.successfulIds);
+    errors.push(...result.errors);
+
+    if (exitEarlyOnError && result.soFailureOccurred) {
+      logger.warn(
+        `bulkCreateRules: exiting early on SO failure at batch ${
+          batchIndex + 1
+        }/${totalBatches}. ` +
+          `${successfulIds.length} rule(s) created, ${
+            total - start - batch.length
+          } rule(s) skipped.`
+      );
+      break;
+    }
+  }
+
+  return { successfulIds, errors, total };
+}
+
+interface RunBatchArgs<Params extends RuleParams> {
+  context: RulesClientContext;
+  username: string | null;
+  actionsClient: Awaited<ReturnType<RulesClientContext['getActionsClient']>>;
+  batch: Array<BulkCreateRulesItem<Params>>;
+}
+
+async function runBatch<Params extends RuleParams>({
+  context,
+  username,
+  actionsClient,
+  batch,
+}: RunBatchArgs<Params>): Promise<BatchResult> {
+  const { logger } = context;
+
+  const inputsWithIds = batch.map((rule) => ({
     id: rule.options?.id ?? SavedObjectsUtils.generateId(),
     rule,
   }));
 
-  logger.debug(`Bulk creating batch of ${total} rules`);
-
   // Phase 1: per-rule prepare (validation + api key generation).
   // NOTE: in order to minimise external calls, the values below get mutated
   // at different stages in the process (ie if we fail schedule creation),
   // so that we invalidate keys and create rule SOs in a single batch at the end.
   const preparedRules = new Map<string, PreparedRule>();
   const keysToInvalidate = new Set<string>();
   const apiKeysMap = new Map<string, ApiKeyEntry>();
-  const errors: BulkOperationError[] = [];
+  const errors: BulkCreateOperationError[] = [];
   const authzCache = new Map<string, Promise<void>>();
 
   await pMap(
@@ -86,10 +158,10 @@ export async function bulkCreateRules<Params extends RuleParams = never>(
     { concurrency: API_KEY_GENERATE_CONCURRENCY }
   );
 
-  // No survivors? Flush out any keys created and return
+  // No survivors? Flush any keys created and return.
   if (preparedRules.size === 0) {
     await flushKeysToInvalidate(keysToInvalidate, context);
-    return { rules: [], errors, total, taskIdsFailedToBeEnabled: [] };
+    return { successfulIds: [], errors, soFailureOccurred: false };
   }
 
   // Phase 2: validate schedule-limits, enabled subset only.
@@ -211,10 +283,9 @@ export async function bulkCreateRules<Params extends RuleParams = never>(
         })
     );
   } catch (error) {
-    // Whole-call SO failure (auth, timeout, etc): invalidate every newly-minted
-    // key, best-effort orphan-task cleanup, then rethrow.
+    // Whole-call SO failure: invalidate keys, best-effort task cleanup.
+    // Surface as batch-wide SO failure so exitEarlyOnError can honour it.
     for (const k of collectNewKeysToInvalidate(apiKeysMap.values())) keysToInvalidate.add(k);
-    await flushKeysToInvalidate(keysToInvalidate, context);
     if (newlyScheduledTaskIds.size > 0) {
       try {
         await context.taskManager.bulkRemove([...newlyScheduledTaskIds]);
@@ -226,16 +297,23 @@ export async function bulkCreateRules<Params extends RuleParams = never>(
         );
       }
     }
-    throw error;
+    await flushKeysToInvalidate(keysToInvalidate, context);
+    errors.push({
+      message: `Failed to bulk create rule saved objects: ${error.message}`,
+      status: error.output?.statusCode,
+      rule: { id: 'n/a', name: 'n/a' },
+    });
+    return { successfulIds: [], errors, soFailureOccurred: true };
   }
 
   // Phase 4 per-row outcomes.
-  const successfulSos: Array<SavedObject<RawRule>> = [];
-  const taskIdsToEnable: string[] = [];
+  const batchSuccessfulIds: string[] = [];
   const taskIdsToCleanUp: string[] = [];
+  let perRowFailureOccurred = false;
 
   for (const so of bulkResponse.saved_objects) {
     if (so.error) {
+      perRowFailureOccurred = true;
       errors.push({
         message: so.error.message ?? 'n/a',
         status: so.error.statusCode,
@@ -253,9 +331,8 @@ export async function bulkCreateRules<Params extends RuleParams = never>(
         taskIdsToCleanUp.push(so.id);
       }
     } else {
-      successfulSos.push(so as SavedObject<RawRule>);
+      batchSuccessfulIds.push(so.id);
       if (newlyScheduledTaskIds.has(so.id)) {
-        taskIdsToEnable.push(so.id);
         // Audit per-rule ENABLE for the enabled subset (mirrors single-rule semantics).
         context.auditLogger?.log(
           ruleAuditEvent({
@@ -286,23 +363,12 @@ export async function bulkCreateRules<Params extends RuleParams = never>(
     }
   }
 
-  // Phase 5: enable scheduled tasks. If skipTaskEnabling, caller enables them later.
-  const taskIdsFailedToBeEnabled = params.skipTaskEnabling
-    ? [...taskIdsToEnable]
-    : (await bulkEnableTasks(context, { taskIds: taskIdsToEnable })).taskIdsFailedToBeEnabled;
-
-  // Single end-of-function flush for all collected key invalidations.
+  // Single per-batch flush for all collected key invalidations.
   await flushKeysToInvalidate(keysToInvalidate, context);
 
-  // Phase 6: domain transform + return.
-  const sanitizedRules: Array<SanitizedRule<Params>> = successfulSos.map((so) =>
-    toSanitizedRule<Params>(context, so, context.ruleTypeRegistry)
-  );
-
   return {
-    rules: sanitizedRules,
+    successfulIds: batchSuccessfulIds,
     errors,
-    total,
-    taskIdsFailedToBeEnabled,
+    soFailureOccurred: perRowFailureOccurred,
   };
 }

x-pack/platform/plugins/shared/alerting/server/application/rule/methods/bulk_create/bulk_create_rules.ts

@@ -11,7 +11,7 @@ import type { RuleParams } from '../../types';
 import type { CreateRuleData } from '../create/types';
 import type { CreateRuleOptions } from '../create/create_rule';
 import type { BulkOperationError, RulesClientContext } from '../../../../rules_client/types';
-import type { RawRule, SanitizedRule } from '../../../../types';
+import type { RawRule } from '../../../../types';
 
 export interface PreparedRule {
   id: string;
@@ -49,8 +49,10 @@ export interface BulkCreateRulesItem<Params extends RuleParams = never> {
 
 export interface BulkCreateRulesParams<Params extends RuleParams = never> {
   rules: Array<BulkCreateRulesItem<Params>>;
-  // If true, skip Phase 5 (taskManager.bulkEnable);
-  skipTaskEnabling?: boolean;
+  /** Per-batch size; clamped to [1, MAX_BULK_CREATE_BATCH_SIZE], defaults to DEFAULT_BULK_CREATE_BATCH_SIZE. Total is bounded by MAX_RULES_NUMBER_FOR_BULK_OPERATION (callers should enforce request-level limits). */
+  batchSize?: number;
+  /** If true, stop further batches on SO-level failure (whole-call throw or any per-row SO error). Phase 1/2/3 demotions never halt the loop. Defaults to false. */
+  exitEarlyOnError?: boolean;
 }
 
 export type BulkCreateDisabledReason =
@@ -63,9 +65,16 @@ export interface BulkCreateOperationError extends BulkOperationError {
   disabledReason?: BulkCreateDisabledReason;
 }
 
-export interface BulkCreateRulesResult<Params extends RuleParams = never> {
-  rules: Array<SanitizedRule<Params>>;
+export interface BulkCreateRulesResult {
+  /** IDs of rules whose SO was successfully persisted. */
+  successfulIds: string[];
   errors: BulkCreateOperationError[];
   total: number;
-  taskIdsFailedToBeEnabled: string[];
+}
+
+export interface BatchResult {
+  successfulIds: string[];
+  errors: BulkCreateOperationError[];
+  /** True if SO bulkCreate threw or any per-row SO error was returned; drives the exitEarlyOnError short-circuit. */
+  soFailureOccurred: boolean;
 }

x-pack/platform/plugins/shared/alerting/server/application/rule/methods/bulk_create/types.ts

@@ -8,7 +8,6 @@
 import Boom from '@hapi/boom';
 import Semver from 'semver';
 import { i18n } from '@kbn/i18n';
-import type { SavedObject } from '@kbn/core/server';
 import type { TaskInstanceWithDeprecatedFields } from '@kbn/task-manager-plugin/server/task';
 
 import { validateAndAuthorizeSystemActions } from '../../../../lib/validate_authorize_system_actions';
@@ -32,17 +31,12 @@ import {
   apiKeyAsAlertAttributes,
   apiKeyAsRuleDomainProperties,
 } from '../../../../rules_client/common';
+// import { BULK_TM_SCHEDULE_DELAY } from '../../../../rules_client/common/constants';
 import { ruleAuditEvent, RuleAuditAction } from '../../../../rules_client/common/audit_events';
 import { bulkMarkApiKeysForInvalidation } from '../../../../invalidate_pending_api_keys/bulk_mark_api_keys_for_invalidation';
-import type { RawRule, RuleTypeRegistry, SanitizedRule } from '../../../../types';
 import type { BulkOperationError, RulesClientContext } from '../../../../rules_client/types';
 import type { RuleDomain, RuleParams } from '../../types';
-import {
-  transformRuleAttributesToRuleDomain,
-  transformRuleDomainToRule,
-  transformRuleDomainToRuleAttributes,
-} from '../../transforms';
-import { ruleDomainSchema } from '../../schemas';
+import { transformRuleDomainToRuleAttributes } from '../../transforms';
 import { createRuleDataSchema } from '../create/schemas';
 import type {
   PreparedRule,
@@ -67,55 +61,37 @@ export const collectNewKeysToInvalidate = (entries: Iterable<ApiKeyEntry>): stri
   return keys;
 };
 
+// Task is scheduled `enabled: true` in the future.
 export const buildTaskInstance = (
   context: RulesClientContext,
   prepared: PreparedRule
-): TaskInstanceWithDeprecatedFields => ({
-  id: prepared.id,
-  taskType: `alerting:${prepared.ruleTypeId}`,
-  schedule: prepared.schedule,
-  params: {
-    alertId: prepared.id,
-    spaceId: context.spaceId,
-    consumer: prepared.consumer,
-  },
-  state: {
-    previousStartedAt: null,
-    alertTypeState: {},
-    alertInstances: {},
-  },
-  scope: ['alerting'],
-  // Tasks are scheduled disabled. Phase 5 enables them via taskManager.bulkEnable
-  // so per-task validation drops never produce a running task without a rule SO,
-  // and the activation can randomise schedule datetimes across the batch.
-  enabled: false,
-});
+): TaskInstanceWithDeprecatedFields => {
+  // const FIVE_MINUTES = 5 * 60 * 1000;
+  // const intervalMs = parseDuration(prepared.schedule.interval);
+  // const jitterCeilingMs = Math.min(intervalMs, FIVE_MINUTES);
+  // const jitterMs = Math.floor(Math.random() * jitterCeilingMs);
+  // // future `runAt` delay so SO bulkCreate lands before TM claims the task
+  // const runAt = new Date(Date.now() + BULK_TM_SCHEDULE_DELAY + jitterMs);
 
-export const toSanitizedRule = <Params extends RuleParams = never>(
-  context: RulesClientContext,
-  so: SavedObject<RawRule>,
-  ruleTypeRegistry: RuleTypeRegistry
-): SanitizedRule<Params> => {
-  const ruleType = ruleTypeRegistry.get(so.attributes.alertTypeId);
-  const ruleDomain: RuleDomain<Params> = transformRuleAttributesToRuleDomain<Params>(
-    so.attributes,
-    {
-      id: so.id,
-      logger: context.logger,
-      ruleType,
-      references: so.references,
-      omitGeneratedValues: false,
+  return {
+    id: prepared.id,
+    taskType: `alerting:${prepared.ruleTypeId}`,
+    schedule: prepared.schedule,
+    params: {
+      alertId: prepared.id,
+      spaceId: context.spaceId,
+      consumer: prepared.consumer,
     },
-    context.isSystemAction
-  );
-
-  try {
-    ruleDomainSchema.validate(ruleDomain);
-  } catch (e) {
-    context.logger.warn(`Error validating bulk-created rule domain object for id: ${so.id}, ${e}`);
-  }
-
-  return transformRuleDomainToRule<Params>(ruleDomain, { isPublic: true }) as SanitizedRule<Params>;
+    state: {
+      previousStartedAt: null,
+      alertTypeState: {},
+      alertInstances: {},
+    },
+    scope: ['alerting'],
+    enabled: true,
+    runAt: new Date(),
+    scheduledAt: new Date(),
+  };
 };
 
 export const prepareRule = async <Params extends RuleParams>({