[eem] add entity definition state (elastic#191933)

~blocked by elastic#192004 This change adds an `includeState: boolean` option to methods querying entity definitions. When true this adds an `EntityDefinitionState` object containing all the definition components and their state (installed or not) and stats. Since this may only be used internally (eg builtin definition installation process) and for troubleshooting, `includeState` is false by default #### Testing - install a definition - call `GET kbn:/internal/entities/definition/<definition-id>?includeState=true` - check and validate the definition `state` block - manually remove transform/pipeline/template components - check and validate the definition `state` block
pgayvallet · Sep 13, 2024 · 2f1d0cd · 2f1d0cd
1 parent 3226eb6
commit 2f1d0cd
Show file tree

Hide file tree

Showing 12 changed files with 216 additions and 57 deletions.
diff --git a/x-pack/packages/kbn-entities-schema/src/rest_spec/get.ts b/x-pack/packages/kbn-entities-schema/src/rest_spec/get.ts
@@ -6,8 +6,12 @@
  */
 
 import { z } from '@kbn/zod';
+import { BooleanFromString } from '@kbn/zod-helpers';
 
 export const getEntityDefinitionQuerySchema = z.object({
   page: z.optional(z.coerce.number()),
   perPage: z.optional(z.coerce.number()),
+  includeState: z.optional(BooleanFromString).default(false),
 });
+
+export type GetEntityDefinitionQuerySchema = z.infer<typeof getEntityDefinitionQuerySchema>;
diff --git a/x-pack/plugins/observability_solution/entity_manager/server/lib/client/index.ts b/x-pack/plugins/observability_solution/entity_manager/server/lib/client/index.ts
@@ -6,6 +6,7 @@
  */
 
 import type { IScopedClusterClient, SavedObjectsClientContract } from '@kbn/core/server';
+import { EntityDefinition } from '@kbn/entities-schema';
 import { findEntityDefinitions } from '../entities/find_entity_definition';
 import type { EntityDefinitionWithState } from '../entities/types';
 
@@ -16,7 +17,7 @@ export class EntityManagerClient {
   ) {}
 
   findEntityDefinitions({ page, perPage }: { page?: number; perPage?: number } = {}): Promise<
-    EntityDefinitionWithState[]
+    EntityDefinition[] | EntityDefinitionWithState[]
   > {
     return findEntityDefinitions({
       esClient: this.esClient.asCurrentUser,

diff --git a/...ugins/observability_solution/entity_manager/server/lib/entities/find_entity_definition.ts b/...ugins/observability_solution/entity_manager/server/lib/entities/find_entity_definition.ts
@@ -5,9 +5,10 @@
  * 2.0.
  */
 
-import { compact } from 'lodash';
+import { compact, forEach, reduce } from 'lodash';
 import { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';
 import { EntityDefinition } from '@kbn/entities-schema';
+import { NodesIngestTotal } from '@elastic/elasticsearch/lib/api/types';
 import { SO_ENTITY_DEFINITION_TYPE } from '../../saved_objects';
 import {
   generateHistoryTransformId,
@@ -19,7 +20,7 @@ import {
   generateLatestIndexTemplateId,
 } from './helpers/generate_component_id';
 import { BUILT_IN_ID_PREFIX } from './built_in';
-import { EntityDefinitionWithState } from './types';
+import { EntityDefinitionState, EntityDefinitionWithState } from './types';
 import { isBackfillEnabled } from './helpers/is_backfill_enabled';
 
 export async function findEntityDefinitions({
@@ -29,14 +30,16 @@ export async function findEntityDefinitions({
   id,
   page = 1,
   perPage = 10,
+  includeState = false,
 }: {
   soClient: SavedObjectsClientContract;
   esClient: ElasticsearchClient;
   builtIn?: boolean;
   id?: string;
   page?: number;
   perPage?: number;
-}): Promise<EntityDefinitionWithState[]> {
+  includeState?: boolean;
+}): Promise<EntityDefinition[] | EntityDefinitionWithState[]> {
   const filter = compact([
     typeof builtIn === 'boolean'
       ? `${SO_ENTITY_DEFINITION_TYPE}.attributes.id:(${BUILT_IN_ID_PREFIX}*)`
@@ -50,6 +53,10 @@ export async function findEntityDefinitions({
     perPage,
   });
 
+  if (!includeState) {
+    return response.saved_objects.map(({ attributes }) => attributes);
+  }
+
   return Promise.all(
     response.saved_objects.map(async ({ attributes }) => {
       const state = await getEntityDefinitionState(esClient, attributes);
@@ -62,15 +69,18 @@ export async function findEntityDefinitionById({
   id,
   esClient,
   soClient,
+  includeState = false,
 }: {
   id: string;
   esClient: ElasticsearchClient;
   soClient: SavedObjectsClientContract;
+  includeState?: boolean;
 }) {
   const [definition] = await findEntityDefinitions({
     esClient,
     soClient,
     id,
+    includeState,
     perPage: 1,
   });
 
@@ -80,43 +90,126 @@ export async function findEntityDefinitionById({
 async function getEntityDefinitionState(
   esClient: ElasticsearchClient,
   definition: EntityDefinition
-) {
-  const historyIngestPipelineId = generateHistoryIngestPipelineId(definition);
-  const latestIngestPipelineId = generateLatestIngestPipelineId(definition);
+): Promise<EntityDefinitionState> {
+  const [ingestPipelines, transforms, indexTemplates] = await Promise.all([
+    getIngestPipelineState({ definition, esClient }),
+    getTransformState({ definition, esClient }),
+    getIndexTemplatesState({ definition, esClient }),
+  ]);
+
+  const installed =
+    ingestPipelines.every((pipeline) => pipeline.installed) &&
+    transforms.every((transform) => transform.installed) &&
+    indexTemplates.every((template) => template.installed);
+  const running = transforms.every((transform) => transform.running);
+
+  return {
+    installed,
+    running,
+    components: { transforms, ingestPipelines, indexTemplates },
+  };
+}
+
+async function getTransformState({
+  definition,
+  esClient,
+}: {
+  definition: EntityDefinition;
+  esClient: ElasticsearchClient;
+}) {
   const transformIds = [
     generateHistoryTransformId(definition),
     generateLatestTransformId(definition),
     ...(isBackfillEnabled(definition) ? [generateHistoryBackfillTransformId(definition)] : []),
   ];
-  const [ingestPipelines, indexTemplatesInstalled, transforms] = await Promise.all([
-    esClient.ingest.getPipeline(
-      {
-        id: `${historyIngestPipelineId},${latestIngestPipelineId}`,
-      },
-      { ignore: [404] }
-    ),
-    esClient.indices.existsIndexTemplate({
-      name: `${
-        (generateLatestIndexTemplateId(definition), generateHistoryIndexTemplateId(definition))
-      }`,
-    }),
-    esClient.transform.getTransformStats({
-      transform_id: transformIds,
+
+  const transformStats = await Promise.all(
+    transformIds.map((id) => esClient.transform.getTransformStats({ transform_id: id }))
+  ).then((results) => results.map(({ transforms }) => transforms).flat());
+
+  return transformIds.map((id) => {
+    const stats = transformStats.find((transform) => transform.id === id);
+    if (!stats) {
+      return { id, installed: false, running: false };
+    }
+
+    return {
+      id,
+      installed: true,
+      running: stats.state === 'started' || stats.state === 'indexing',
+      stats,
+    };
+  });
+}
+
+async function getIngestPipelineState({
+  definition,
+  esClient,
+}: {
+  definition: EntityDefinition;
+  esClient: ElasticsearchClient;
+}) {
+  const ingestPipelineIds = [
+    generateHistoryIngestPipelineId(definition),
+    generateLatestIngestPipelineId(definition),
+  ];
+  const [ingestPipelines, ingestPipelinesStats] = await Promise.all([
+    esClient.ingest.getPipeline({ id: ingestPipelineIds.join(',') }, { ignore: [404] }),
+    esClient.nodes.stats({
+      metric: 'ingest',
+      filter_path: ingestPipelineIds.map((id) => `nodes.*.ingest.pipelines.${id}`),
     }),
   ]);
 
-  const ingestPipelinesInstalled = !!(
-    ingestPipelines[historyIngestPipelineId] && ingestPipelines[latestIngestPipelineId]
+  const ingestStatsByPipeline = reduce(
+    ingestPipelinesStats.nodes,
+    (pipelines, { ingest }) => {
+      forEach(ingest?.pipelines, (value: NodesIngestTotal, key: string) => {
+        if (!pipelines[key]) {
+          pipelines[key] = { count: 0, failed: 0 };
+        }
+        pipelines[key].count += value.count ?? 0;
+        pipelines[key].failed += value.failed ?? 0;
+      });
+      return pipelines;
+    },
+    {} as Record<string, { count: number; failed: number }>
   );
-  const transformsInstalled = transforms.count === transformIds.length;
-  const transformsRunning =
-    transformsInstalled &&
-    transforms.transforms.every(
-      (transform) => transform.state === 'started' || transform.state === 'indexing'
-    );
 
-  return {
-    installed: ingestPipelinesInstalled && transformsInstalled && indexTemplatesInstalled,
-    running: transformsRunning,
-  };
+  return ingestPipelineIds.map((id) => ({
+    id,
+    installed: !!ingestPipelines[id],
+    stats: ingestStatsByPipeline[id],
+  }));
+}
+
+async function getIndexTemplatesState({
+  definition,
+  esClient,
+}: {
+  definition: EntityDefinition;
+  esClient: ElasticsearchClient;
+}) {
+  const indexTemplatesIds = [
+    generateLatestIndexTemplateId(definition),
+    generateHistoryIndexTemplateId(definition),
+  ];
+  const templates = await Promise.all(
+    indexTemplatesIds.map((id) =>
+      esClient.indices
+        .getIndexTemplate({ name: id }, { ignore: [404] })
+        .then(({ index_templates: indexTemplates }) => indexTemplates?.[0])
+    )
+  ).then(compact);
+  return indexTemplatesIds.map((id) => {
+    const template = templates.find(({ name }) => name === id);
+    if (!template) {
+      return { id, installed: false };
+    }
+    return {
+      id,
+      installed: true,
+      stats: template.index_template,
+    };
+  });
 }
diff --git a/...servability_solution/entity_manager/server/lib/entities/install_entity_definition.test.ts b/...servability_solution/entity_manager/server/lib/entities/install_entity_definition.test.ts
@@ -354,6 +354,7 @@ describe('install_entity_definition', () => {
         version: semver.inc(mockEntityDefinition.version, 'major') ?? '0.0.0',
       };
       const esClient = elasticsearchClientMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValue({ transforms: [], count: 0 });
       const soClient = savedObjectsClientMock.create();
 
       soClient.find.mockResolvedValueOnce({
@@ -391,6 +392,7 @@ describe('install_entity_definition', () => {
         version: semver.inc(mockEntityDefinition.version, 'major') ?? '0.0.0',
       };
       const esClient = elasticsearchClientMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValue({ transforms: [], count: 0 });
       const soClient = savedObjectsClientMock.create();
 
       soClient.find.mockResolvedValueOnce({
@@ -426,6 +428,7 @@ describe('install_entity_definition', () => {
 
     it('should reinstall when failed installation', async () => {
       const esClient = elasticsearchClientMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValue({ transforms: [], count: 0 });
       const soClient = savedObjectsClientMock.create();
 
       soClient.find.mockResolvedValueOnce({

diff --git a/...ns/observability_solution/entity_manager/server/lib/entities/install_entity_definition.ts b/...ns/observability_solution/entity_manager/server/lib/entities/install_entity_definition.ts
@@ -39,8 +39,8 @@ import { generateEntitiesLatestIndexTemplateConfig } from './templates/entities_
 import { generateEntitiesHistoryIndexTemplateConfig } from './templates/entities_history_template';
 import { EntityIdConflict } from './errors/entity_id_conflict_error';
 import { EntityDefinitionNotFound } from './errors/entity_not_found';
-import { EntityDefinitionWithState } from './types';
 import { mergeEntityDefinitionUpdate } from './helpers/merge_definition_update';
+import { EntityDefinitionWithState } from './types';
 import { stopTransforms } from './stop_transforms';
 import { deleteTransforms } from './delete_transforms';
 
@@ -136,6 +136,7 @@ export async function installBuiltInEntityDefinitions({
       esClient,
       soClient,
       id: builtInDefinition.id,
+      includeState: true,
     });
 
     if (!installedDefinition) {
@@ -148,7 +149,12 @@ export async function installBuiltInEntityDefinitions({
     }
 
     // verify existing installation
-    if (!shouldReinstallBuiltinDefinition(installedDefinition, builtInDefinition)) {
+    if (
+      !shouldReinstallBuiltinDefinition(
+        installedDefinition as EntityDefinitionWithState,
+        builtInDefinition
+      )
+    ) {
       return installedDefinition;
     }
 

diff --git a/x-pack/plugins/observability_solution/entity_manager/server/lib/entities/types.ts b/x-pack/plugins/observability_solution/entity_manager/server/lib/entities/types.ts
@@ -5,12 +5,43 @@
  * 2.0.
  */
 
+import {
+  IndicesIndexTemplate,
+  TransformGetTransformStatsTransformStats,
+} from '@elastic/elasticsearch/lib/api/types';
 import { EntityDefinition } from '@kbn/entities-schema';
 
+interface TransformState {
+  id: string;
+  installed: boolean;
+  running: boolean;
+  stats?: TransformGetTransformStatsTransformStats;
+}
+
+interface IngestPipelineState {
+  id: string;
+  installed: boolean;
+  stats?: { count: number; failed: number };
+}
+
+interface IndexTemplateState {
+  id: string;
+  installed: boolean;
+  stats?: IndicesIndexTemplate;
+}
+
 // state is the *live* state of the definition. since a definition
 // is composed of several elasticsearch components that can be
 // modified or deleted outside of the entity manager apis, this can
 // be used to verify the actual installation is complete and running
-export type EntityDefinitionWithState = EntityDefinition & {
-  state: { installed: boolean; running: boolean };
-};
+export interface EntityDefinitionState {
+  installed: boolean;
+  running: boolean;
+  components: {
+    transforms: TransformState[];
+    ingestPipelines: IngestPipelineState[];
+    indexTemplates: IndexTemplateState[];
+  };
+}
+
+export type EntityDefinitionWithState = EntityDefinition & { state: EntityDefinitionState };
diff --git a/x-pack/plugins/observability_solution/entity_manager/server/lib/entity_client.ts b/x-pack/plugins/observability_solution/entity_manager/server/lib/entity_client.ts
@@ -70,12 +70,24 @@ export class EntityClient {
     });
   }
 
-  async getEntityDefinitions({ page = 1, perPage = 10 }: { page?: number; perPage?: number }) {
+  async getEntityDefinitions({
+    id,
+    page = 1,
+    perPage = 10,
+    includeState = false,
+  }: {
+    id?: string;
+    page?: number;
+    perPage?: number;
+    includeState?: boolean;
+  }) {
     const definitions = await findEntityDefinitions({
       esClient: this.options.esClient,
       soClient: this.options.soClient,
       page,
       perPage,
+      id,
+      includeState,
     });
 
     return { definitions };

diff --git a/x-pack/plugins/observability_solution/entity_manager/server/lib/utils.ts b/x-pack/plugins/observability_solution/entity_manager/server/lib/utils.ts
@@ -19,7 +19,7 @@ export const getClientsFromAPIKey = ({
   server: EntityManagerServerSetup;
 }): { esClient: ElasticsearchClient; soClient: SavedObjectsClientContract } => {
   const fakeRequest = getFakeKibanaRequest({ id: apiKey.id, api_key: apiKey.apiKey });
-  const esClient = server.core.elasticsearch.client.asScoped(fakeRequest).asCurrentUser;
+  const esClient = server.core.elasticsearch.client.asScoped(fakeRequest).asSecondaryAuthUser;
   const soClient = server.core.savedObjects.getScopedClient(fakeRequest);
   return { esClient, soClient };
 };