[ML] Add rule registry for ML alerting rules

packages/kbn-rule-data-utils/src/alerts_as_data_rbac.ts

@@ -25,6 +25,7 @@ export const AlertConsumers = {
   OBSERVABILITY: 'observability',
   SIEM: 'siem',
   SYNTHETICS: 'synthetics',
+  ML: 'ml',
 } as const;
 export type AlertConsumers = typeof AlertConsumers[keyof typeof AlertConsumers];
 export type STATUS_VALUES = 'open' | 'acknowledged' | 'closed';
@@ -36,6 +37,7 @@ export const mapConsumerToIndexName: Record<AlertConsumers, string | string[]> =
   observability: '.alerts-observability',
   siem: ['.alerts-security.alerts', '.siem-signals'],
   synthetics: '.alerts-observability-synthetics',
+  ml: '.alerts-ml',
 };
 export type ValidFeatureId = keyof typeof mapConsumerToIndexName;
 

x-pack/plugins/ml/common/constants/alerts.ts

@@ -21,8 +21,12 @@ export const TOP_N_BUCKETS_COUNT = 1;
 
 export const ALL_JOBS_SELECTION = '*';
 
-export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; description: string }> = {
+export const HEALTH_CHECK_NAMES: Record<
+  JobsHealthTests,
+  { id: string; name: string; description: string }
+> = {
   datafeed: {
+    id: 'datafeed_not_started',
     name: i18n.translate('xpack.ml.alertTypes.jobsHealthAlertingRule.datafeedCheckName', {
       defaultMessage: 'Datafeed is not started',
     }),
@@ -34,6 +38,7 @@ export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; descrip
     ),
   },
   mml: {
+    id: 'mml',
     name: i18n.translate('xpack.ml.alertTypes.jobsHealthAlertingRule.mmlCheckName', {
       defaultMessage: 'Model memory limit reached',
     }),
@@ -42,6 +47,7 @@ export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; descrip
     }),
   },
   delayedData: {
+    id: 'delayed_data',
     name: i18n.translate('xpack.ml.alertTypes.jobsHealthAlertingRule.delayedDataCheckName', {
       defaultMessage: 'Data delay has occurred',
     }),
@@ -53,6 +59,7 @@ export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; descrip
     ),
   },
   errorMessages: {
+    id: 'error_messages',
     name: i18n.translate('xpack.ml.alertTypes.jobsHealthAlertingRule.errorMessagesCheckName', {
       defaultMessage: 'There are errors in the job messages',
     }),
@@ -64,6 +71,7 @@ export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; descrip
     ),
   },
   behindRealtime: {
+    id: 'behind_realtime',
     name: i18n.translate('xpack.ml.alertTypes.jobsHealthAlertingRule.behindRealtimeCheckName', {
       defaultMessage: 'Job is running behind real-time',
     }),
@@ -75,3 +83,19 @@ export const HEALTH_CHECK_NAMES: Record<JobsHealthTests, { name: string; descrip
     ),
   },
 };
+
+const ML_NAMESPACE = 'ml' as const;
+
+export const JOB_ID = `${ML_NAMESPACE}.job_id` as const;
+export const JOB_STATE = `${ML_NAMESPACE}.job_state` as const;
+export const DATAFEED_ID = `${ML_NAMESPACE}.datafeed_id` as const;
+export const DATAFEED_STATE = `${ML_NAMESPACE}.datafeed_state` as const;
+export const MEMORY_STATUS = `${ML_NAMESPACE}.memory_status` as const;
+export const MEMORY_LOG_TIME = `${ML_NAMESPACE}.memory_log_time` as const;
+export const MODEL_BYTES = `${ML_NAMESPACE}.model_bytes` as const;
+export const MODEL_BYTES_MEMORY_LIMIT = `${ML_NAMESPACE}.model_bytes_memory_limit` as const;
+export const PEAK_MODEL_BYTES = `${ML_NAMESPACE}.peak_model_bytes` as const;
+export const MODEL_BYTES_EXCEEDED = `${ML_NAMESPACE}.model_bytes_exceeded` as const;
+export const ANNOTATION = `${ML_NAMESPACE}.annotation` as const;
+export const MISSED_DOC_COUNT = `${ML_NAMESPACE}.missed_docs_count` as const;
+export const END_TIMESTAMP = `${ML_NAMESPACE}.end_timestamp` as const;

x-pack/plugins/ml/kibana.json

@@ -27,7 +27,8 @@
     "management",
     "licenseManagement",
     "maps",
-    "usageCollection"
+    "usageCollection",
+    "ruleRegistry"
   ],
   "server": true,
   "ui": true,

x-pack/plugins/ml/server/lib/alerts/get_rule_data_client.ts

@@ -0,0 +1,127 @@
+/*
+ * 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 { Logger } from 'kibana/server';
+import { once } from 'lodash';
+import { IRuleDataClient, RuleRegistryPluginSetupContract } from '../../../../rule_registry/server';
+import { mappingFromFieldMap } from '../../../../rule_registry/common/mapping_from_field_map';
+import { TECHNICAL_COMPONENT_TEMPLATE_NAME } from '../../../../rule_registry/common/assets';
+import {
+  DATAFEED_ID,
+  DATAFEED_STATE,
+  JOB_ID,
+  JOB_STATE,
+  MEMORY_STATUS,
+  MEMORY_LOG_TIME,
+  MODEL_BYTES,
+  MODEL_BYTES_MEMORY_LIMIT,
+  PEAK_MODEL_BYTES,
+  MODEL_BYTES_EXCEEDED,
+  ANNOTATION,
+  MISSED_DOC_COUNT,
+  END_TIMESTAMP,
+} from '../../../common/constants/alerts';
+
+export function getRuleDataClient(
+  ruleRegistry: RuleRegistryPluginSetupContract,
+  logger: Logger
+): IRuleDataClient {
+  const { ruleDataService } = ruleRegistry;
+
+  const alertsIndexPattern = ruleDataService.getFullAssetName('ml*');
+
+  const initializeRuleDataTemplates = once(async () => {
+    const componentTemplateName = ruleDataService.getFullAssetName('ml-mappings');
+
+    if (!ruleDataService.isWriteEnabled()) {
+      return;
+    }
+
+    await ruleDataService.createOrUpdateComponentTemplate({
+      name: componentTemplateName,
+      body: {
+        template: {
+          settings: {
+            number_of_shards: 1,
+          },
+          // Mappings based on {@link AnomalyDetectionJobHealthResult}
+          mappings: mappingFromFieldMap(
+            {
+              [JOB_ID]: {
+                type: 'keyword',
+              },
+              [JOB_STATE]: {
+                type: 'keyword',
+              },
+              // datafeed
+              [DATAFEED_ID]: {
+                type: 'keyword',
+              },
+              [DATAFEED_STATE]: {
+                type: 'keyword',
+              },
+              // mml
+              [MEMORY_STATUS]: {
+                type: 'keyword',
+              },
+              [MEMORY_LOG_TIME]: {
+                type: 'date',
+              },
+              [MODEL_BYTES]: {
+                type: 'long',
+              },
+              [MODEL_BYTES_MEMORY_LIMIT]: {
+                type: 'long',
+              },
+              [PEAK_MODEL_BYTES]: {
+                type: 'long',
+              },
+              [MODEL_BYTES_EXCEEDED]: {
+                type: 'long',
+              },
+              // {@link DelayedDataResponse)
+              [ANNOTATION]: {
+                type: 'text',
+              },
+              [MISSED_DOC_COUNT]: {
+                type: 'long',
+              },
+              [END_TIMESTAMP]: {
+                type: 'date',
+              },
+            },
+            'strict'
+          ),
+        },
+      },
+    });
+
+    await ruleDataService.createOrUpdateIndexTemplate({
+      name: ruleDataService.getFullAssetName('ml-index-template'),
+      body: {
+        index_patterns: [alertsIndexPattern],
+        composed_of: [
+          ruleDataService.getFullAssetName(TECHNICAL_COMPONENT_TEMPLATE_NAME),
+          componentTemplateName,
+        ],
+      },
+    });
+    await ruleDataService.updateIndexMappingsMatchingPattern(alertsIndexPattern);
+  });
+
+  const initializeRuleDataTemplatesPromise = initializeRuleDataTemplates().catch((err) => {
+    logger!.error(err);
+  });
+
+  const ruleDataClient = ruleDataService.getRuleDataClient(
+    'ml',
+    ruleDataService.getFullAssetName('ml'),
+    () => initializeRuleDataTemplatesPromise
+  );
+
+  return ruleDataClient;
+}

x-pack/plugins/ml/server/lib/alerts/jobs_health_service.test.ts

@@ -11,9 +11,34 @@ import type { Logger } from 'kibana/server';
 import { MlClient } from '../ml_client';
 import { MlJob, MlJobStats } from '@elastic/elasticsearch/api/types';
 import { AnnotationService } from '../../models/annotation_service/annotation';
+import { JobsHealthExecutorOptions } from './register_jobs_monitoring_rule_type';
 
 const MOCK_DATE_NOW = 1487076708000;
 
+function getDefaultExecutorOptions(): JobsHealthExecutorOptions {
+  return ({
+    state: {},
+    startedAt: new Date('2021-08-12T13:13:39.396Z'),
+    previousStartedAt: new Date('2021-08-12T13:13:27.396Z'),
+    spaceId: 'default',
+    namespace: undefined,
+    name: 'ml-health-check',
+    tags: [],
+    createdBy: 'elastic',
+    updatedBy: 'elastic',
+    rule: {
+      name: 'ml-health-check',
+      tags: [],
+      consumer: 'alerts',
+      producer: 'ml',
+      ruleTypeId: 'xpack.ml.anomaly_detection_jobs_health',
+      ruleTypeName: 'Anomaly detection jobs health',
+      enabled: true,
+      schedule: { interval: '10s' },
+    },
+  } as unknown) as JobsHealthExecutorOptions;
+}
+
 describe('JobsHealthService', () => {
   const mlClient = ({
     getJobs: jest.fn().mockImplementation(({ job_id: jobIds = [] }) => {
@@ -61,7 +86,7 @@ describe('JobsHealthService', () => {
               state: j === 'test_job_02' || 'test_job_01' ? 'opened' : 'closed',
               model_size_stats: {
                 memory_status: j === 'test_job_01' ? 'hard_limit' : 'ok',
-                log_time: 1626935914540,
+                memory_log_time: 1626935914540,
               },
             };
           }) as MlJobStats,
@@ -123,11 +148,14 @@ describe('JobsHealthService', () => {
     debug: jest.fn(),
   } as unknown) as jest.Mocked<Logger>;
 
+  const ruleDataClient = null;
+
   const jobHealthService: JobsHealthService = jobsHealthServiceProvider(
     mlClient,
     datafeedsService,
     annotationService,
-    logger
+    logger,
+    ruleDataClient
   );
 
   let dateNowSpy: jest.SpyInstance;
@@ -143,21 +171,23 @@ describe('JobsHealthService', () => {
 
   test('returns empty results when no jobs provided', async () => {
     // act
-    const executionResult = await jobHealthService.getTestsResults('testRule', {
+    const executionResult = await jobHealthService.getTestsResults(getDefaultExecutorOptions(), {
       testsConfig: null,
       includeJobs: {
         jobIds: ['*'],
         groupIds: [],
       },
       excludeJobs: null,
     });
-    expect(logger.warn).toHaveBeenCalledWith('Rule "testRule" does not have associated jobs.');
+    expect(logger.warn).toHaveBeenCalledWith(
+      'Rule "ml-health-check" does not have associated jobs.'
+    );
     expect(datafeedsService.getDatafeedByJobId).not.toHaveBeenCalled();
     expect(executionResult).toEqual([]);
   });
 
   test('returns empty results and does not perform datafeed check when test is disabled', async () => {
-    const executionResult = await jobHealthService.getTestsResults('testRule', {
+    const executionResult = await jobHealthService.getTestsResults(getDefaultExecutorOptions(), {
       testsConfig: {
         datafeed: {
           enabled: false,
@@ -186,7 +216,7 @@ describe('JobsHealthService', () => {
   });
 
   test('takes into account delayed data params', async () => {
-    const executionResult = await jobHealthService.getTestsResults('testRule_04', {
+    const executionResult = await jobHealthService.getTestsResults(getDefaultExecutorOptions(), {
       testsConfig: {
         delayedData: {
           enabled: true,
@@ -216,6 +246,7 @@ describe('JobsHealthService', () => {
 
     expect(executionResult).toEqual([
       {
+        id: 'delayed_data',
         name: 'Data delay has occurred',
         context: {
           results: [
@@ -234,7 +265,7 @@ describe('JobsHealthService', () => {
   });
 
   test('returns results based on provided selection', async () => {
-    const executionResult = await jobHealthService.getTestsResults('testRule_03', {
+    const executionResult = await jobHealthService.getTestsResults(getDefaultExecutorOptions(), {
       testsConfig: null,
       includeJobs: {
         jobIds: [],
@@ -266,6 +297,7 @@ describe('JobsHealthService', () => {
 
     expect(executionResult).toEqual([
       {
+        id: 'datafeed_not_started',
         name: 'Datafeed is not started',
         context: {
           results: [
@@ -280,12 +312,12 @@ describe('JobsHealthService', () => {
         },
       },
       {
+        id: 'mml',
         name: 'Model memory limit reached',
         context: {
           results: [
             {
               job_id: 'test_job_01',
-              log_time: 1626935914540,
               memory_status: 'hard_limit',
             },
           ],
@@ -294,6 +326,7 @@ describe('JobsHealthService', () => {
         },
       },
       {
+        id: 'delayed_data',
         name: 'Data delay has occurred',
         context: {
           results: [