[Entity Manager] Exposing stop and start transforms actions in the En…

…tityClient (elastic#192186)

This PR exposes two new methods (`startTransforms` and `stopTransforms`)
in the EntityClient as proposed
[here](elastic/elastic-entity-model#160)
This work is also required by the Entity Analytics team:
elastic/security-team#10230

In addition, it splits up `stop_and_delete_transforms` into more
granular actions, one for stopping and another for deleting. The
uninstall function is now responsible for appropriately calling those
actions in sequence.

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

@@ -0,0 +1,58 @@
+/*
+ * 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 { ElasticsearchClient, Logger } from '@kbn/core/server';
+import { EntityDefinition } from '@kbn/entities-schema';
+
+import {
+  generateHistoryTransformId,
+  generateHistoryBackfillTransformId,
+  generateLatestTransformId,
+} from './helpers/generate_component_id';
+import { retryTransientEsErrors } from './helpers/retry';
+import { isBackfillEnabled } from './helpers/is_backfill_enabled';
+
+export async function deleteTransforms(
+  esClient: ElasticsearchClient,
+  definition: EntityDefinition,
+  logger: Logger
+) {
+  try {
+    const historyTransformId = generateHistoryTransformId(definition);
+    const latestTransformId = generateLatestTransformId(definition);
+    await retryTransientEsErrors(
+      () =>
+        esClient.transform.deleteTransform(
+          { transform_id: historyTransformId, force: true },
+          { ignore: [404] }
+        ),
+      { logger }
+    );
+    if (isBackfillEnabled(definition)) {
+      const historyBackfillTransformId = generateHistoryBackfillTransformId(definition);
+      await retryTransientEsErrors(
+        () =>
+          esClient.transform.deleteTransform(
+            { transform_id: historyBackfillTransformId, force: true },
+            { ignore: [404] }
+          ),
+        { logger }
+      );
+    }
+    await retryTransientEsErrors(
+      () =>
+        esClient.transform.deleteTransform(
+          { transform_id: latestTransformId, force: true },
+          { ignore: [404] }
+        ),
+      { logger }
+    );
+  } catch (e) {
+    logger.error(`Cannot delete history transform [${definition.id}]: ${e}`);
+    throw e;
+  }
+}

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

@@ -32,11 +32,7 @@ import {
   saveEntityDefinition,
   updateEntityDefinition,
 } from './save_entity_definition';
-import {
-  stopAndDeleteHistoryBackfillTransform,
-  stopAndDeleteHistoryTransform,
-  stopAndDeleteLatestTransform,
-} from './stop_and_delete_transform';
+
 import { isBackfillEnabled } from './helpers/is_backfill_enabled';
 import { deleteTemplate, upsertTemplate } from '../manage_index_templates';
 import { generateEntitiesLatestIndexTemplateConfig } from './templates/entities_latest_template';
@@ -45,6 +41,8 @@ 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 { stopTransforms } from './stop_transforms';
+import { deleteTransforms } from './delete_transforms';
 
 export interface InstallDefinitionParams {
   esClient: ElasticsearchClient;
@@ -91,14 +89,7 @@ export async function installEntityDefinition({
     return await install({ esClient, soClient, logger, definition: entityDefinition });
   } catch (e) {
     logger.error(`Failed to install entity definition ${definition.id}: ${e}`);
-
-    await Promise.all([
-      stopAndDeleteHistoryTransform(esClient, definition, logger),
-      isBackfillEnabled(definition)
-        ? stopAndDeleteHistoryBackfillTransform(esClient, definition, logger)
-        : Promise.resolve(),
-      stopAndDeleteLatestTransform(esClient, definition, logger),
-    ]);
+    await stopAndDeleteTransforms(esClient, definition, logger);
 
     await Promise.all([
       deleteHistoryIngestPipeline(esClient, definition, logger),
@@ -253,13 +244,7 @@ export async function reinstallEntityDefinition({
     });
 
     logger.debug(`Deleting transforms for definition ${definition.id} v${definition.version}`);
-    await Promise.all([
-      stopAndDeleteHistoryTransform(esClient, definition, logger),
-      isBackfillEnabled(definition)
-        ? stopAndDeleteHistoryBackfillTransform(esClient, definition, logger)
-        : Promise.resolve(),
-      stopAndDeleteLatestTransform(esClient, definition, logger),
-    ]);
+    await stopAndDeleteTransforms(esClient, definition, logger);
 
     return await install({
       esClient,
@@ -308,3 +293,12 @@ const shouldReinstallBuiltinDefinition = (
 
   return timedOut || outdated || failed || partial;
 };
+
+const stopAndDeleteTransforms = async (
+  esClient: ElasticsearchClient,
+  definition: EntityDefinition,
+  logger: Logger
+) => {
+  await stopTransforms(esClient, definition, logger);
+  await deleteTransforms(esClient, definition, logger);
+};

x-pack/plugins/observability_solution/entity_manager/server/lib/entities/start_transform.ts → x-pack/plugins/observability_solution/entity_manager/server/lib/entities/start_transforms.ts

@@ -15,7 +15,7 @@ import {
 import { retryTransientEsErrors } from './helpers/retry';
 import { isBackfillEnabled } from './helpers/is_backfill_enabled';
 
-export async function startTransform(
+export async function startTransforms(
   esClient: ElasticsearchClient,
   definition: EntityDefinition,
   logger: Logger

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

@@ -0,0 +1,65 @@
+/*
+ * 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 { ElasticsearchClient, Logger } from '@kbn/core/server';
+import { EntityDefinition } from '@kbn/entities-schema';
+
+import {
+  generateHistoryTransformId,
+  generateHistoryBackfillTransformId,
+  generateLatestTransformId,
+} from './helpers/generate_component_id';
+import { retryTransientEsErrors } from './helpers/retry';
+
+import { isBackfillEnabled } from './helpers/is_backfill_enabled';
+
+export async function stopTransforms(
+  esClient: ElasticsearchClient,
+  definition: EntityDefinition,
+  logger: Logger
+) {
+  try {
+    const historyTransformId = generateHistoryTransformId(definition);
+    const latestTransformId = generateLatestTransformId(definition);
+
+    await retryTransientEsErrors(
+      () =>
+        esClient.transform.stopTransform(
+          { transform_id: historyTransformId, wait_for_completion: true, force: true },
+          { ignore: [409, 404] }
+        ),
+      { logger }
+    );
+
+    if (isBackfillEnabled(definition)) {
+      const historyBackfillTransformId = generateHistoryBackfillTransformId(definition);
+      await retryTransientEsErrors(
+        () =>
+          esClient.transform.stopTransform(
+            {
+              transform_id: historyBackfillTransformId,
+              wait_for_completion: true,
+              force: true,
+            },
+            { ignore: [409, 404] }
+          ),
+        { logger }
+      );
+    }
+    await retryTransientEsErrors(
+      () =>
+        esClient.transform.stopTransform(
+          { transform_id: latestTransformId, wait_for_completion: true, force: true },
+          { ignore: [409, 404] }
+        ),
+      { logger }
+    );
+  } catch (e) {
+    logger.error(`Cannot stop entity transforms [${definition.id}]: ${e}`);
+    throw e;
+  }
+}

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

@@ -13,18 +13,17 @@ import { deleteEntityDefinition } from './delete_entity_definition';
 import { deleteIndices } from './delete_index';
 import { deleteHistoryIngestPipeline, deleteLatestIngestPipeline } from './delete_ingest_pipeline';
 import { findEntityDefinitions } from './find_entity_definition';
-import {
-  stopAndDeleteHistoryBackfillTransform,
-  stopAndDeleteHistoryTransform,
-  stopAndDeleteLatestTransform,
-} from './stop_and_delete_transform';
-import { isBackfillEnabled } from './helpers/is_backfill_enabled';
+
 import {
   generateHistoryIndexTemplateId,
   generateLatestIndexTemplateId,
 } from './helpers/generate_component_id';
 import { deleteTemplate } from '../manage_index_templates';
 
+import { stopTransforms } from './stop_transforms';
+
+import { deleteTransforms } from './delete_transforms';
+
 export async function uninstallEntityDefinition({
   definition,
   esClient,
@@ -38,13 +37,8 @@ export async function uninstallEntityDefinition({
   logger: Logger;
   deleteData?: boolean;
 }) {
-  await Promise.all([
-    stopAndDeleteHistoryTransform(esClient, definition, logger),
-    stopAndDeleteLatestTransform(esClient, definition, logger),
-    isBackfillEnabled(definition)
-      ? stopAndDeleteHistoryBackfillTransform(esClient, definition, logger)
-      : Promise.resolve(),
-  ]);
+  await stopTransforms(esClient, definition, logger);
+  await deleteTransforms(esClient, definition, logger);
 
   await Promise.all([
     deleteHistoryIngestPipeline(esClient, definition, logger),

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

@@ -7,7 +7,7 @@
 
 import { EntityDefinition } from '@kbn/entities-schema';
 import { installBuiltInEntityDefinitions } from './install_entity_definition';
-import { startTransform } from './start_transform';
+import { startTransforms } from './start_transforms';
 import { EntityManagerServerSetup } from '../../types';
 import { checkIfEntityDiscoveryAPIKeyIsValid, readEntityDiscoveryAPIKey } from '../auth';
 import { getClientsFromAPIKey } from '../utils';
@@ -46,7 +46,7 @@ export async function upgradeBuiltInEntityDefinitions({
   });
 
   await Promise.all(
-    upgradedDefinitions.map((definition) => startTransform(esClient, definition, logger))
+    upgradedDefinitions.map((definition) => startTransforms(esClient, definition, logger))
   );
 
   return { success: true, definitions: upgradedDefinitions };

x-pack/plugins/observability_solution/entity_manager/server/lib/entity_client.ts

@@ -10,11 +10,13 @@ import { SavedObjectsClientContract } from '@kbn/core-saved-objects-api-server';
 import { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
 import { Logger } from '@kbn/logging';
 import { installEntityDefinition } from './entities/install_entity_definition';
-import { startTransform } from './entities/start_transform';
+import { startTransforms } from './entities/start_transforms';
 import { findEntityDefinitions } from './entities/find_entity_definition';
 import { uninstallEntityDefinition } from './entities/uninstall_entity_definition';
 import { EntityDefinitionNotFound } from './entities/errors/entity_not_found';
 
+import { stopTransforms } from './entities/stop_transforms';
+
 export class EntityClient {
   constructor(
     private options: {
@@ -39,7 +41,7 @@ export class EntityClient {
     });
 
     if (!installOnly) {
-      await startTransform(this.options.esClient, definition, this.options.logger);
+      await startTransforms(this.options.esClient, definition, this.options.logger);
     }
 
     return installedDefinition;
@@ -78,4 +80,12 @@ export class EntityClient {
 
     return { definitions };
   }
+
+  async startEntityDefinition(definition: EntityDefinition) {
+    return startTransforms(this.options.esClient, definition, this.options.logger);
+  }
+
+  async stopEntityDefinition(definition: EntityDefinition) {
+    return stopTransforms(this.options.esClient, definition, this.options.logger);
+  }
 }