[eem] top_value metadata aggregation

x-pack/packages/kbn-entities-schema/src/schema/common.test.ts

@@ -28,7 +28,7 @@ describe('schemas', () => {
       const result = metadataSchema.safeParse({
         source: 'host.name',
         destination: 'host.name',
-        limit: 0,
+        aggregation: { type: 'terms', limit: 0 },
       });
       expect(result.success).toBeFalsy();
       expect(result).toMatchSnapshot();
@@ -52,11 +52,41 @@ describe('schemas', () => {
       const result = metadataSchema.safeParse({
         source: 'host.name',
         destination: 'hostName',
-        size: 1,
       });
       expect(result.success).toBeTruthy();
       expect(result).toMatchSnapshot();
     });
+
+    it('should default to terms aggregation when none provided', () => {
+      const result = metadataSchema.safeParse({
+        source: 'host.name',
+        destination: 'hostName',
+      });
+      expect(result.success).toBeTruthy();
+      expect(result.data).toEqual({
+        source: 'host.name',
+        destination: 'hostName',
+        aggregation: { type: 'terms', limit: 1000 },
+      });
+    });
+
+    it('should parse supported aggregations', () => {
+      const result = metadataSchema.safeParse({
+        source: 'host.name',
+        destination: 'hostName',
+        aggregation: { type: 'top_value', sort: { '@timestamp': 'desc' } },
+      });
+      expect(result.success).toBeTruthy();
+    });
+
+    it('should reject unsupported aggregation', () => {
+      const result = metadataSchema.safeParse({
+        source: 'host.name',
+        destination: 'hostName',
+        aggregation: { type: 'unknown_agg', limit: 10 },
+      });
+      expect(result.success).toBeFalsy();
+    });
   });
 
   describe('durationSchema', () => {

x-pack/packages/kbn-entities-schema/src/schema/common.ts

@@ -84,24 +84,40 @@ export const keyMetricSchema = z.object({
 
 export type KeyMetric = z.infer<typeof keyMetricSchema>;
 
+export const metadataAggregation = z.union([
+  z.object({ type: z.literal('terms'), limit: z.number().default(1000) }),
+  z.object({
+    type: z.literal('top_value'),
+    sort: z.record(z.string(), z.union([z.literal('asc'), z.literal('desc')])),
+    lookbackPeriod: z.optional(durationSchema),
+  }),
+]);
+
 export const metadataSchema = z
   .object({
     source: z.string(),
     destination: z.optional(z.string()),
-    limit: z.optional(z.number().default(1000)),
+    aggregation: z
+      .optional(metadataAggregation)
+      .default({ type: z.literal('terms').value, limit: 1000 }),
   })
+  .or(
+    z.string().transform((value) => ({
+      source: value,
+      destination: value,
+      aggregation: { type: z.literal('terms').value, limit: 1000 },
+    }))
+  )
   .transform((metadata) => ({
     ...metadata,
     destination: metadata.destination ?? metadata.source,
-    limit: metadata.limit ?? 1000,
   }))
-  .or(z.string().transform((value) => ({ source: value, destination: value, limit: 1000 })))
   .superRefine((value, ctx) => {
-    if (value.limit < 1) {
+    if (value.aggregation.type === 'terms' && value.aggregation.limit < 1) {
       ctx.addIssue({
         path: ['limit'],
         code: z.ZodIssueCode.custom,
-        message: 'limit should be greater than 1',
+        message: 'limit for terms aggregation should be greater than 1',
       });
     }
     if (value.source.length === 0) {
@@ -120,6 +136,8 @@ export const metadataSchema = z
     }
   });
 
+export type MetadataField = z.infer<typeof metadataSchema>;
+
 export const identityFieldsSchema = z
   .object({
     field: z.string(),

x-pack/plugins/observability_solution/entity_manager/server/lib/entities/built_in/services.ts

@@ -46,7 +46,7 @@ export const builtInServicesFromLogsEntityDefinition: EntityDefinition =
     displayNameTemplate: '{{service.name}}{{#service.environment}}:{{.}}{{/service.environment}}',
     metadata: [
       { source: '_index', destination: 'sourceIndex' },
-      { source: 'agent.name', limit: 100 },
+      { source: 'agent.name', aggregation: { type: 'terms', limit: 100 } },
       'data_stream.type',
       'service.environment',
       'service.name',

x-pack/plugins/observability_solution/entity_manager/server/lib/entities/ingest_pipeline/generate_history_processors.ts

@@ -5,18 +5,27 @@
  * 2.0.
  */
 
-import { EntityDefinition, ENTITY_SCHEMA_VERSION_V1 } from '@kbn/entities-schema';
+import { EntityDefinition, ENTITY_SCHEMA_VERSION_V1, MetadataField } from '@kbn/entities-schema';
 import {
   initializePathScript,
   cleanScript,
 } from '../helpers/ingest_pipeline_script_processor_helpers';
 import { generateHistoryIndexName } from '../helpers/generate_component_id';
 import { isBuiltinDefinition } from '../helpers/is_builtin_definition';
 
-function mapDestinationToPainless(field: string) {
+function getMetadataSourceField({ aggregation, destination, source }: MetadataField) {
+  if (aggregation.type === 'terms') {
+    return `ctx.entity.metadata.${destination}.keySet()`;
+  } else if (aggregation.type === 'top_value') {
+    return `ctx.entity.metadata.${destination}.top_value["${source}"]`;
+  }
+}
+
+function mapDestinationToPainless(metadata: MetadataField) {
+  const field = metadata.destination;
   return `
     ${initializePathScript(field)}
-    ctx.${field} = ctx.entity.metadata.${field}.keySet();
+    ctx.${field} = ${getMetadataSourceField(metadata)};
   `;
 }
 
@@ -25,15 +34,27 @@ function createMetadataPainlessScript(definition: EntityDefinition) {
     return '';
   }
 
-  return definition.metadata.reduce((acc, def) => {
-    const destination = def.destination;
+  return definition.metadata.reduce((acc, metadata) => {
+    const { destination, source } = metadata;
     const optionalFieldPath = destination.replaceAll('.', '?.');
-    const next = `
-      if (ctx.entity?.metadata?.${optionalFieldPath} != null) {
-        ${mapDestinationToPainless(destination)}
-      }
-    `;
-    return `${acc}\n${next}`;
+
+    if (metadata.aggregation.type === 'terms') {
+      const next = `
+        if (ctx.entity?.metadata?.${optionalFieldPath} != null) {
+          ${mapDestinationToPainless(metadata)}
+        }
+      `;
+      return `${acc}\n${next}`;
+    } else if (metadata.aggregation.type === 'top_value') {
+      const next = `
+        if (ctx.entity?.metadata?.${optionalFieldPath}?.top_value["${source}"] != null) {
+          ${mapDestinationToPainless(metadata)}
+        }
+      `;
+      return `${acc}\n${next}`;
+    }
+
+    return acc;
   }, '');
 }
 

x-pack/plugins/observability_solution/entity_manager/server/lib/entities/ingest_pipeline/generate_latest_processors.ts

@@ -5,18 +5,27 @@
  * 2.0.
  */
 
-import { EntityDefinition, ENTITY_SCHEMA_VERSION_V1 } from '@kbn/entities-schema';
+import { EntityDefinition, ENTITY_SCHEMA_VERSION_V1, MetadataField } from '@kbn/entities-schema';
 import {
   initializePathScript,
   cleanScript,
 } from '../helpers/ingest_pipeline_script_processor_helpers';
 import { generateLatestIndexName } from '../helpers/generate_component_id';
 import { isBuiltinDefinition } from '../helpers/is_builtin_definition';
 
-function mapDestinationToPainless(field: string) {
+function getMetadataSourceField({ aggregation, destination, source }: MetadataField) {
+  if (aggregation.type === 'terms') {
+    return `ctx.entity.metadata.${destination}.data.keySet()`;
+  } else if (aggregation.type === 'top_value') {
+    return `ctx.entity.metadata.${destination}.top_value["${destination}"]`;
+  }
+}
+
+function mapDestinationToPainless(metadata: MetadataField) {
+  const field = metadata.destination;
   return `
     ${initializePathScript(field)}
-    ctx.${field} = ctx.entity.metadata.${field}.data.keySet();
+    ctx.${field} = ${getMetadataSourceField(metadata)};
   `;
 }
 
@@ -25,15 +34,27 @@ function createMetadataPainlessScript(definition: EntityDefinition) {
     return '';
   }
 
-  return definition.metadata.reduce((acc, def) => {
-    const destination = def.destination || def.source;
+  return definition.metadata.reduce((acc, metadata) => {
+    const destination = metadata.destination;
     const optionalFieldPath = destination.replaceAll('.', '?.');
-    const next = `
-      if (ctx.entity?.metadata?.${optionalFieldPath}.data != null) {
-        ${mapDestinationToPainless(destination)}
-      }
-    `;
-    return `${acc}\n${next}`;
+
+    if (metadata.aggregation.type === 'terms') {
+      const next = `
+        if (ctx.entity?.metadata?.${optionalFieldPath}.data != null) {
+          ${mapDestinationToPainless(metadata)}
+        }
+      `;
+      return `${acc}\n${next}`;
+    } else if (metadata.aggregation.type === 'top_value') {
+      const next = `
+        if (ctx.entity?.metadata?.${optionalFieldPath}?.top_value["${destination}"] != null) {
+          ${mapDestinationToPainless(metadata)}
+        }
+      `;
+      return `${acc}\n${next}`;
+    }
+
+    return acc;
   }, '');
 }