[ML] Adding runtime_mappings to job wizards (#85817) (#86118)

* [ML] Adding runtime_mappings to job wizards

* fixing test

* adding runtime fields to fields list in datafeed preview

* fixing cardinality count for runtime field

* fixing cardinality check in data visualizer

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/ml/common/types/anomaly_detection_jobs/datafeed.ts

@@ -20,6 +20,7 @@ export interface Datafeed {
   query: object;
   query_delay?: string;
   script_fields?: Record<string, any>;
+  runtime_mappings?: Record<string, any>;
   scroll_size?: number;
   delayed_data_check_config?: object;
   indices_options?: IndicesOptions;

x-pack/plugins/ml/public/application/jobs/new_job/common/job_creator/job_creator.ts

@@ -56,6 +56,7 @@ export class JobCreator {
   protected _aggs: Aggregation[] = [];
   protected _fields: Field[] = [];
   protected _scriptFields: Field[] = [];
+  protected _runtimeMappings: Field[] = [];
   protected _aggregationFields: Field[] = [];
   protected _sparseData: boolean = false;
   private _stopAllRefreshPolls: {
@@ -487,12 +488,16 @@ export class JobCreator {
     return this._scriptFields;
   }
 
+  public get runtimeMappings(): Field[] {
+    return this._runtimeMappings;
+  }
+
   public get aggregationFields(): Field[] {
     return this._aggregationFields;
   }
 
   public get additionalFields(): Field[] {
-    return [...this._scriptFields, ...this._aggregationFields];
+    return [...this._scriptFields, ...this._runtimeMappings, ...this._aggregationFields];
   }
 
   public get subscribers(): ProgressSubscriber[] {
@@ -688,6 +693,16 @@ export class JobCreator {
       }));
     }
 
+    this._runtimeMappings = [];
+    if (this._datafeed_config.runtime_mappings !== undefined) {
+      this._runtimeMappings = Object.keys(this._datafeed_config.runtime_mappings).map((f) => ({
+        id: f,
+        name: f,
+        type: ES_FIELD_TYPES.KEYWORD,
+        aggregatable: true,
+      }));
+    }
+
     this._aggregationFields = [];
     const aggs = getDatafeedAggregations(this._datafeed_config);
     if (aggs !== undefined) {

x-pack/plugins/ml/public/application/services/job_service.js

@@ -457,6 +457,12 @@ class JobService {
               if (scriptFields && Object.keys(scriptFields).length) {
                 body.script_fields = scriptFields;
               }
+
+              // add runtime_mappings if present
+              const runtimeMappings = job.datafeed_config.runtime_mappings;
+              if (runtimeMappings && Object.keys(runtimeMappings).length) {
+                body.runtime_mappings = runtimeMappings;
+              }
             } else {
               // if aggregations is not set and retrieveWholeSource is not set, add all of the fields from the job
               body.size = ML_DATA_PREVIEW_COUNT;
@@ -467,6 +473,12 @@ class JobService {
                 body.script_fields = scriptFields;
               }
 
+              // add runtime_mappings if present
+              const runtimeMappings = job.datafeed_config.runtime_mappings;
+              if (runtimeMappings && Object.keys(runtimeMappings).length) {
+                body.runtime_mappings = runtimeMappings;
+              }
+
               const fields = {};
 
               // get fields from detectors
@@ -509,6 +521,13 @@ class JobService {
                 fields[job.data_description.time_field] = {};
               }
 
+              // add runtime fields
+              if (runtimeMappings) {
+                Object.keys(runtimeMappings).forEach((fieldName) => {
+                  fields[fieldName] = {};
+                });
+              }
+
               const fieldsList = Object.keys(fields);
               if (fieldsList.length) {
                 body.fields = fieldsList;

x-pack/plugins/ml/server/models/data_visualizer/data_visualizer.ts

@@ -605,6 +605,7 @@ export class DataVisualizer {
     // Value count aggregation faster way of checking if field exists than using
     // filter aggregation with exists query.
     const aggs: Aggs = datafeedAggregations !== undefined ? { ...datafeedAggregations } : {};
+    const runtimeMappings: any = {};
 
     aggregatableFields.forEach((field, i) => {
       const safeFieldName = getSafeAggregationName(field, i);
@@ -617,6 +618,11 @@ export class DataVisualizer {
         cardinalityField = aggs[`${safeFieldName}_cardinality`] = {
           cardinality: { script: datafeedConfig?.script_fields[field].script },
         };
+      } else if (datafeedConfig?.runtime_mappings?.hasOwnProperty(field)) {
+        cardinalityField = {
+          cardinality: { field },
+        };
+        runtimeMappings.runtime_mappings = datafeedConfig.runtime_mappings;
       } else {
         cardinalityField = {
           cardinality: { field },
@@ -632,6 +638,7 @@ export class DataVisualizer {
         },
       },
       aggs: buildSamplerAggregation(aggs, samplerShardSize),
+      ...runtimeMappings,
     };
 
     const { body } = await this._asCurrentUser.search({
@@ -670,7 +677,10 @@ export class DataVisualizer {
           },
         });
       } else {
-        if (datafeedConfig?.script_fields?.hasOwnProperty(field)) {
+        if (
+          datafeedConfig?.script_fields?.hasOwnProperty(field) ||
+          datafeedConfig?.runtime_mappings?.hasOwnProperty(field)
+        ) {
           const cardinality = get(
             aggregations,
             [...aggsPath, `${safeFieldName}_cardinality`, 'value'],

x-pack/plugins/ml/server/models/fields_service/fields_service.ts

@@ -56,6 +56,12 @@ export function fieldsServiceProvider({ asCurrentUser }: IScopedClusterClient) {
       ) {
         aggregatableFields.push(fieldName);
       }
+      if (
+        typeof datafeedConfig?.runtime_mappings === 'object' &&
+        datafeedConfig.runtime_mappings.hasOwnProperty(fieldName)
+      ) {
+        aggregatableFields.push(fieldName);
+      }
       if (
         datafeedAggregations !== undefined &&
         isValidAggregationField(datafeedAggregations, fieldName)
@@ -133,6 +139,7 @@ export function fieldsServiceProvider({ asCurrentUser }: IScopedClusterClient) {
       mustCriteria.push(query);
     }
 
+    const runtimeMappings: any = {};
     const aggs = fieldsToAgg.reduce(
       (obj, field) => {
         if (
@@ -140,6 +147,12 @@ export function fieldsServiceProvider({ asCurrentUser }: IScopedClusterClient) {
           datafeedConfig.script_fields.hasOwnProperty(field)
         ) {
           obj[field] = { cardinality: { script: datafeedConfig.script_fields[field].script } };
+        } else if (
+          typeof datafeedConfig?.runtime_mappings === 'object' &&
+          datafeedConfig.runtime_mappings.hasOwnProperty(field)
+        ) {
+          obj[field] = { cardinality: { field } };
+          runtimeMappings.runtime_mappings = datafeedConfig.runtime_mappings;
         } else {
           obj[field] = { cardinality: { field } };
         }
@@ -161,6 +174,7 @@ export function fieldsServiceProvider({ asCurrentUser }: IScopedClusterClient) {
         excludes: [],
       },
       aggs,
+      ...runtimeMappings,
     };
 
     const {

x-pack/plugins/ml/server/models/job_validation/validate_cardinality.ts

@@ -26,6 +26,11 @@ function isScriptField(job: CombinedJob, fieldName: string): boolean {
   return scriptFields.includes(fieldName);
 }
 
+function isRuntimeMapping(job: CombinedJob, fieldName: string): boolean {
+  const runtimeMappings = Object.keys(job.datafeed_config.runtime_mappings ?? {});
+  return runtimeMappings.includes(fieldName);
+}
+
 // Thresholds to determine whether cardinality is
 // too high or low for certain fields analysis
 const OVER_FIELD_CARDINALITY_THRESHOLD_LOW = 10;
@@ -93,6 +98,13 @@ const validateFactory = (client: IScopedClusterClient, job: CombinedJob): Valida
             ) {
               return true;
             }
+            if (
+              typeof datafeedConfig?.runtime_mappings === 'object' &&
+              datafeedConfig?.runtime_mappings.hasOwnProperty(field)
+            ) {
+              return true;
+            }
+
             // if datafeed has aggregation fields, check recursively if field exist
             if (
               datafeedAggregations !== undefined &&
@@ -136,10 +148,11 @@ const validateFactory = (client: IScopedClusterClient, job: CombinedJob): Valida
             }
           } else {
             // only report uniqueFieldName as not aggregatable if it's not part
-            // of a valid categorization configuration and if it's not a scripted field.
+            // of a valid categorization configuration and if it's not a scripted field or runtime mapping.
             if (
               !isValidCategorizationConfig(job, uniqueFieldName) &&
-              !isScriptField(job, uniqueFieldName)
+              !isScriptField(job, uniqueFieldName) &&
+              !isRuntimeMapping(job, uniqueFieldName)
             ) {
               messages.push({
                 id: 'field_not_aggregatable',

x-pack/plugins/ml/server/routes/apidoc_scripts/schema_extractor.test.ts

@@ -109,6 +109,11 @@ describe('schema_extractor', () => {
         documentation: '',
         type: 'any',
       },
+      {
+        name: 'runtime_mappings',
+        documentation: '',
+        type: 'any',
+      },
       {
         name: 'scroll_size',
         documentation: '',

x-pack/plugins/ml/server/routes/schemas/datafeeds_schema.ts

@@ -31,6 +31,7 @@ export const datafeedConfigSchema = schema.object({
   max_empty_searches: schema.maybe(schema.number()),
   query_delay: schema.maybe(schema.string()),
   script_fields: schema.maybe(schema.any()),
+  runtime_mappings: schema.maybe(schema.any()),
   scroll_size: schema.maybe(schema.number()),
   delayed_data_check_config: schema.maybe(schema.any()),
   indices_options: schema.maybe(