[EEM] Add basic integration test (#190097)

## Summary

This PR adds some basic integration tests. I also had to modify the
Entity schemas to allow for parsing the dynamic metadata fields. I don't
love the type solution and would ideally like to have something like:
```typescript
const entitySchema = createEntitySchemaFromDefintion(entityDefinintion);
```
which would dynamically generate an entity schema from the definition
based on the identity and metadata fields. I have a prototype that sort
of works but still needs a lot of work and started to become a huge
"time suck".

Closes [139](https://github.com/elastic/elastic-entity-model/issues/139)

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

@@ -0,0 +1,76 @@
+/*
+ * 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 { entityLatestSchema, entityMetadataSchema } from './entity';
+
+const entity = {
+  entity: {
+    lastSeenTimestamp: '2024-08-06T17:03:50.722Z',
+    schemaVersion: 'v1',
+    definitionVersion: '999.999.999',
+    displayName: 'message_processor',
+    identityFields: ['log.logger', 'event.category'],
+    id: '6UHVPiduEC2qk6rMjs1Jzg==',
+    metrics: {
+      logRate: 100,
+      errorRate: 0,
+    },
+    type: 'service',
+    firstSeenTimestamp: '2024-08-06T16:50:00.000Z',
+    definitionId: 'admin-console-services',
+  },
+};
+
+const metadata = {
+  host: {
+    os: {
+      name: [],
+    },
+    name: [
+      'message_processor.prod.002',
+      'message_processor.prod.001',
+      'message_processor.prod.010',
+      'message_processor.prod.006',
+      'message_processor.prod.005',
+      'message_processor.prod.004',
+      'message_processor.prod.003',
+      'message_processor.prod.009',
+      'message_processor.prod.008',
+      'message_processor.prod.007',
+    ],
+  },
+  event: {
+    ingested: '2024-08-06T17:06:24.444700Z',
+    category: '',
+  },
+  sourceIndex: ['kbn-data-forge-fake_stack.message_processor-2024-08-01'],
+  log: {
+    logger: 'message_processor',
+  },
+  tags: ['infra:message_processor'],
+};
+
+describe('Entity Schemas', () => {
+  describe('entityMetadataSchema', () => {
+    it('should parse metadata object', () => {
+      const results = entityMetadataSchema.safeParse(metadata);
+      expect(results).toHaveProperty('success', true);
+    });
+  });
+
+  describe('entitySummarySchema', () => {
+    it('should parse an entity with metadata', () => {
+      const doc = {
+        ...entity,
+        ...metadata,
+      };
+
+      const result = entityLatestSchema.safeParse(doc);
+      expect(result).toHaveProperty('success', true);
+    });
+  });
+});

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

@@ -8,24 +8,42 @@
 import { z } from 'zod';
 import { arrayOfStringsSchema } from './common';
 
-const entitySchema = z.object({
-  entity: z.object({
-    id: z.string(),
-    identityFields: arrayOfStringsSchema,
-    displayName: z.string(),
-    metrics: z.record(z.string(), z.number()),
-  }),
+export const entityBaseSchema = z.object({
+  id: z.string(),
+  type: z.string(),
+  identityFields: arrayOfStringsSchema,
+  displayName: z.string(),
+  metrics: z.record(z.string(), z.number()),
+  definitionVersion: z.string(),
+  schemaVersion: z.string(),
+  definitionId: z.string(),
 });
 
-export const entitySummarySchema = z.intersection(
-  entitySchema.extend({
-    lastSeenTimestamp: z.string(),
-    firstSeenTimestamp: z.string(),
-  }),
-  z.record(z.string(), z.string().or(z.number()))
+export interface MetadataRecord {
+  [key: string]: string[] | MetadataRecord | string;
+}
+
+const literalSchema = z.union([z.string(), z.number(), z.boolean(), z.null()]);
+type Literal = z.infer<typeof literalSchema>;
+type Metadata = Literal | { [key: string]: Metadata } | Metadata[];
+export const entityMetadataSchema: z.ZodType<Metadata> = z.lazy(() =>
+  z.union([literalSchema, z.array(entityMetadataSchema), z.record(entityMetadataSchema)])
 );
 
-export const entityHistorySchema = z.intersection(
-  entitySchema.extend({ ['@timestamp']: z.string() }),
-  z.record(z.string(), z.string().or(z.number()))
-);
+export const entityLatestSchema = z
+  .object({
+    entity: entityBaseSchema.merge(
+      z.object({
+        lastSeenTimestamp: z.string(),
+        firstSeenTimestamp: z.string(),
+      })
+    ),
+  })
+  .and(entityMetadataSchema);
+
+export const entityHistorySchema = z
+  .object({
+    '@timestamp': z.string(),
+    entity: entityBaseSchema,
+  })
+  .and(entityMetadataSchema);

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

@@ -15,8 +15,12 @@ export async function deleteIndices(
   logger: Logger
 ) {
   try {
+    const { indices: historyIndices } = await esClient.indices.resolveIndex({
+      name: `${generateHistoryIndexName(definition)}.*`,
+      expand_wildcards: 'all',
+    });
     const indices = [
-      `${generateHistoryIndexName(definition)}.*`,
+      ...historyIndices.map(({ name }) => name),
       generateLatestIndexName(definition),
     ];
     await esClient.indices.delete({ index: indices, ignore_unavailable: true });

x-pack/test/alerting_api_integration/observability/helpers/alerting_wait_for_helpers.ts

@@ -60,12 +60,18 @@ export async function waitForDocumentInIndex<T>({
   docCountTarget = 1,
   retryService,
   logger,
+  timeout = TIMEOUT,
+  retries = RETRIES,
+  retryDelay = RETRY_DELAY,
 }: {
   esClient: Client;
   indexName: string;
   docCountTarget?: number;
   retryService: RetryService;
   logger: ToolingLog;
+  timeout?: number;
+  retries?: number;
+  retryDelay?: number;
 }): Promise<SearchResponse<T, Record<string, AggregationsAggregate>>> {
   return await retry<SearchResponse<T, Record<string, AggregationsAggregate>>>({
     test: async () => {
@@ -75,6 +81,7 @@ export async function waitForDocumentInIndex<T>({
         ignore_unavailable: true,
       });
       if (!response.hits.total || (response.hits.total as number) < docCountTarget) {
+        logger.debug(`Document count is ${response.hits.total}, should be ${docCountTarget}`);
         throw new Error(
           `Number of hits does not match expectation (total: ${response.hits.total}, target: ${docCountTarget})`
         );
@@ -85,9 +92,9 @@ export async function waitForDocumentInIndex<T>({
     utilityName: `waiting for documents in ${indexName} index`,
     logger,
     retryService,
-    timeout: TIMEOUT,
-    retries: RETRIES,
-    retryDelay: RETRY_DELAY,
+    timeout,
+    retries,
+    retryDelay,
   });
 }
 

x-pack/test/api_integration/apis/entity_manager/definitions.ts

@@ -6,15 +6,23 @@
  */
 
 import expect from '@kbn/expect';
+import { entityLatestSchema } from '@kbn/entities-schema';
 import {
   entityDefinition as mockDefinition,
   entityDefinitionWithBackfill as mockBackfillDefinition,
 } from '@kbn/entityManager-plugin/server/lib/entities/helpers/fixtures';
+import { PartialConfig, cleanup, generate } from '@kbn/data-forge';
+import { generateLatestIndexName } from '@kbn/entityManager-plugin/server/lib/entities/helpers/generate_component_id';
 import { FtrProviderContext } from '../../ftr_provider_context';
 import { installDefinition, uninstallDefinition, getInstalledDefinitions } from './helpers/request';
+import { waitForDocumentInIndex } from '../../../alerting_api_integration/observability/helpers/alerting_wait_for_helpers';
 
 export default function ({ getService }: FtrProviderContext) {
   const supertest = getService('supertest');
+  const logger = getService('log');
+  const esClient = getService('es');
+  const retryService = getService('retry');
+  const esDeleteAllIndices = getService('esDeleteAllIndices');
 
   describe('Entity definitions', () => {
     describe('definitions installations', () => {
@@ -56,5 +64,53 @@ export default function ({ getService }: FtrProviderContext) {
         await uninstallDefinition(supertest, mockDefinition.id);
       });
     });
+    describe('entity data', () => {
+      let dataForgeConfig: PartialConfig;
+      let dataForgeIndices: string[];
+
+      before(async () => {
+        dataForgeConfig = {
+          indexing: {
+            dataset: 'fake_stack',
+            eventsPerCycle: 100,
+            interval: 60_000,
+          },
+          schedule: [
+            {
+              template: 'good',
+              start: 'now-15m',
+              end: 'now+5m',
+            },
+          ],
+        };
+        dataForgeIndices = await generate({ client: esClient, config: dataForgeConfig, logger });
+        await waitForDocumentInIndex({
+          esClient,
+          indexName: 'kbn-data-forge-fake_stack.admin-console-*',
+          docCountTarget: 2020,
+          retryService,
+          logger,
+        });
+      });
+
+      after(async () => {
+        await esDeleteAllIndices(dataForgeIndices);
+        await uninstallDefinition(supertest, mockDefinition.id, true);
+        await cleanup({ client: esClient, config: dataForgeConfig, logger });
+      });
+
+      it('should create the proper entities in the latest index', async () => {
+        await installDefinition(supertest, mockDefinition);
+        const sample = await waitForDocumentInIndex({
+          esClient,
+          indexName: generateLatestIndexName(mockDefinition),
+          docCountTarget: 5,
+          retryService,
+          logger,
+        });
+        const parsedSample = entityLatestSchema.safeParse(sample.hits.hits[0]._source);
+        expect(parsedSample.success).to.be(true);
+      });
+    });
   });
 }

x-pack/test/api_integration/apis/entity_manager/helpers/request.ts

@@ -39,9 +39,10 @@ export const installDefinition = async (
     .expect(200);
 };
 
-export const uninstallDefinition = (supertest: Agent, id: string) => {
+export const uninstallDefinition = (supertest: Agent, id: string, deleteData = false) => {
   return supertest
     .delete(`/internal/entities/definition/${id}`)
+    .query({ deleteData })
     .set('kbn-xsrf', 'xxx')
     .send()
     .expect(200);