Handle transient Elasticsearch errors during package installation

x-pack/plugins/fleet/server/services/epm/elasticsearch/datastream_ilm/install.ts

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import type { ElasticsearchClient, SavedObjectsClientContract } from 'kibana/server';
+import type { ElasticsearchClient, Logger, SavedObjectsClientContract } from 'kibana/server';
 
 import { ElasticsearchAssetType } from '../../../../../common/types/models';
 import type {
@@ -18,6 +18,7 @@ import { saveInstalledEsRefs } from '../../packages/install';
 import { getAsset } from '../transform/common';
 
 import { getESAssetMetadata } from '../meta';
+import { retryTransientEsErrors } from '../retry';
 
 import { deleteIlmRefs, deleteIlms } from './remove';
 
@@ -35,7 +36,8 @@ export const installIlmForDataStream = async (
   registryPackage: InstallablePackage,
   paths: string[],
   esClient: ElasticsearchClient,
-  savedObjectsClient: SavedObjectsClientContract
+  savedObjectsClient: SavedObjectsClientContract,
+  logger: Logger
 ) => {
   const installation = await getInstallation({ savedObjectsClient, pkgName: registryPackage.name });
   let previousInstalledIlmEsAssets: EsAssetReference[] = [];
@@ -90,7 +92,7 @@ export const installIlmForDataStream = async (
     );
 
     const installationPromises = ilmInstallations.map(async (ilmInstallation) => {
-      return handleIlmInstall({ esClient, ilmInstallation });
+      return handleIlmInstall({ esClient, ilmInstallation, logger });
     });
 
     installedIlms = await Promise.all(installationPromises).then((results) => results.flat());
@@ -117,15 +119,21 @@ export const installIlmForDataStream = async (
 async function handleIlmInstall({
   esClient,
   ilmInstallation,
+  logger,
 }: {
   esClient: ElasticsearchClient;
   ilmInstallation: IlmInstallation;
+  logger: Logger;
 }): Promise<EsAssetReference> {
-  await esClient.transport.request({
-    method: 'PUT',
-    path: `/_ilm/policy/${ilmInstallation.installationName}`,
-    body: ilmInstallation.content,
-  });
+  await retryTransientEsErrors(
+    () =>
+      esClient.transport.request({
+        method: 'PUT',
+        path: `/_ilm/policy/${ilmInstallation.installationName}`,
+        body: ilmInstallation.content,
+      }),
+    { logger }
+  );
 
   return { id: ilmInstallation.installationName, type: ElasticsearchAssetType.dataStreamIlmPolicy };
 }

x-pack/plugins/fleet/server/services/epm/elasticsearch/ilm/install.ts

@@ -5,18 +5,20 @@
  * 2.0.
  */
 
-import type { ElasticsearchClient } from 'kibana/server';
+import type { ElasticsearchClient, Logger } from 'kibana/server';
 
 import type { InstallablePackage } from '../../../../types';
 
 import { ElasticsearchAssetType } from '../../../../types';
 import { getAsset, getPathParts } from '../../archive';
 import { getESAssetMetadata } from '../meta';
+import { retryTransientEsErrors } from '../retry';
 
 export async function installILMPolicy(
   packageInfo: InstallablePackage,
   paths: string[],
-  esClient: ElasticsearchClient
+  esClient: ElasticsearchClient,
+  logger: Logger
 ) {
   const ilmPaths = paths.filter((path) => isILMPolicy(path));
   if (!ilmPaths.length) return;
@@ -29,11 +31,15 @@ export async function installILMPolicy(
       const { file } = getPathParts(path);
       const name = file.substr(0, file.lastIndexOf('.'));
       try {
-        await esClient.transport.request({
-          method: 'PUT',
-          path: '/_ilm/policy/' + name,
-          body,
-        });
+        await retryTransientEsErrors(
+          () =>
+            esClient.transport.request({
+              method: 'PUT',
+              path: '/_ilm/policy/' + name,
+              body,
+            }),
+          { logger }
+        );
       } catch (err) {
         throw new Error(err.message);
       }

x-pack/plugins/fleet/server/services/epm/elasticsearch/ingest_pipeline/install.ts

@@ -6,7 +6,7 @@
  */
 
 import type { TransportRequestOptions } from '@elastic/elasticsearch';
-import type { ElasticsearchClient, SavedObjectsClientContract } from 'src/core/server';
+import type { ElasticsearchClient, Logger, SavedObjectsClientContract } from 'src/core/server';
 
 import { ElasticsearchAssetType } from '../../../../types';
 import type { EsAssetReference, RegistryDataStream, InstallablePackage } from '../../../../types';
@@ -22,6 +22,8 @@ import {
 
 import { appendMetadataToIngestPipeline } from '../meta';
 
+import { retryTransientEsErrors } from '../retry';
+
 import { deletePipelineRefs } from './remove';
 
 interface RewriteSubstitution {
@@ -41,7 +43,8 @@ export const installPipelines = async (
   installablePackage: InstallablePackage,
   paths: string[],
   esClient: ElasticsearchClient,
-  savedObjectsClient: SavedObjectsClientContract
+  savedObjectsClient: SavedObjectsClientContract,
+  logger: Logger
 ) => {
   // unlike other ES assets, pipeline names are versioned so after a template is updated
   // it can be created pointing to the new template, without removing the old one and effecting data
@@ -105,6 +108,7 @@ export const installPipelines = async (
             installAllPipelines({
               dataStream,
               esClient,
+              logger,
               paths: pipelinePaths,
               installablePackage,
             })
@@ -119,6 +123,7 @@ export const installPipelines = async (
       installAllPipelines({
         dataStream: undefined,
         esClient,
+        logger,
         paths: topLevelPipelinePaths,
         installablePackage,
       })
@@ -151,11 +156,13 @@ export function rewriteIngestPipeline(
 
 export async function installAllPipelines({
   esClient,
+  logger,
   paths,
   dataStream,
   installablePackage,
 }: {
   esClient: ElasticsearchClient;
+  logger: Logger;
   paths: string[];
   dataStream?: RegistryDataStream;
   installablePackage: InstallablePackage;
@@ -195,18 +202,20 @@ export async function installAllPipelines({
   });
 
   const installationPromises = pipelines.map(async (pipeline) => {
-    return installPipeline({ esClient, pipeline, installablePackage });
+    return installPipeline({ esClient, pipeline, installablePackage, logger });
   });
 
   return Promise.all(installationPromises);
 }
 
 async function installPipeline({
   esClient,
+  logger,
   pipeline,
   installablePackage,
 }: {
   esClient: ElasticsearchClient;
+  logger: Logger;
   pipeline: any;
   installablePackage?: InstallablePackage;
 }): Promise<EsAssetReference> {
@@ -233,15 +242,21 @@ async function installPipeline({
     };
   }
 
-  await esClient.ingest.putPipeline(esClientParams, esClientRequestOptions);
+  await retryTransientEsErrors(
+    () => esClient.ingest.putPipeline(esClientParams, esClientRequestOptions),
+    { logger }
+  );
 
   return {
     id: pipelineWithMetadata.nameForInstallation,
     type: ElasticsearchAssetType.ingestPipeline,
   };
 }
 
-export async function ensureFleetFinalPipelineIsInstalled(esClient: ElasticsearchClient) {
+export async function ensureFleetFinalPipelineIsInstalled(
+  esClient: ElasticsearchClient,
+  logger: Logger
+) {
   const esClientRequestOptions: TransportRequestOptions = {
     ignore: [404],
   };
@@ -258,6 +273,7 @@ export async function ensureFleetFinalPipelineIsInstalled(esClient: Elasticsearc
   ) {
     await installPipeline({
       esClient,
+      logger,
       pipeline: {
         nameForInstallation: FLEET_FINAL_PIPELINE_ID,
         contentForInstallation: FLEET_FINAL_PIPELINE_CONTENT,

x-pack/plugins/fleet/server/services/epm/elasticsearch/ml_model/install.ts

@@ -5,14 +5,16 @@
  * 2.0.
  */
 
-import type { ElasticsearchClient, SavedObjectsClientContract } from 'kibana/server';
+import type { ElasticsearchClient, Logger, SavedObjectsClientContract } from 'kibana/server';
 import { errors } from '@elastic/elasticsearch';
 
 import { saveInstalledEsRefs } from '../../packages/install';
 import { getPathParts } from '../../archive';
 import { ElasticsearchAssetType } from '../../../../../common/types/models';
 import type { EsAssetReference, InstallablePackage } from '../../../../../common/types/models';
 
+import { retryTransientEsErrors } from '../retry';
+
 import { getAsset } from './common';
 
 interface MlModelInstallation {
@@ -24,7 +26,8 @@ export const installMlModel = async (
   installablePackage: InstallablePackage,
   paths: string[],
   esClient: ElasticsearchClient,
-  savedObjectsClient: SavedObjectsClientContract
+  savedObjectsClient: SavedObjectsClientContract,
+  logger: Logger
 ) => {
   const mlModelPath = paths.find((path) => isMlModel(path));
 
@@ -47,7 +50,7 @@ export const installMlModel = async (
       content,
     };
 
-    const result = await handleMlModelInstall({ esClient, mlModel });
+    const result = await handleMlModelInstall({ esClient, logger, mlModel });
     installedMlModels.push(result);
   }
   return installedMlModels;
@@ -61,19 +64,25 @@ const isMlModel = (path: string) => {
 
 async function handleMlModelInstall({
   esClient,
+  logger,
   mlModel,
 }: {
   esClient: ElasticsearchClient;
+  logger: Logger;
   mlModel: MlModelInstallation;
 }): Promise<EsAssetReference> {
   try {
-    await esClient.ml.putTrainedModel({
-      model_id: mlModel.installationName,
-      defer_definition_decompression: true,
-      timeout: '45s',
-      // @ts-expect-error expects an object not a string
-      body: mlModel.content,
-    });
+    await retryTransientEsErrors(
+      () =>
+        esClient.ml.putTrainedModel({
+          model_id: mlModel.installationName,
+          defer_definition_decompression: true,
+          timeout: '45s',
+          // @ts-expect-error expects an object not a string
+          body: mlModel.content,
+        }),
+      { logger }
+    );
   } catch (err) {
     // swallow the error if the ml model already exists.
     const isAlreadyExistError =

x-pack/plugins/fleet/server/services/epm/elasticsearch/retry.test.ts

@@ -0,0 +1,88 @@
+/*
+ * 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.
+ */
+
+jest.mock('timers/promises');
+import { setTimeout } from 'timers/promises';
+
+import { loggerMock } from '@kbn/logging/mocks';
+import { errors as EsErrors } from '@elastic/elasticsearch';
+
+import { retryTransientEsErrors } from './retry';
+
+const setTimeoutMock = setTimeout as jest.Mock<
+  ReturnType<typeof setTimeout>,
+  Parameters<typeof setTimeout>
+>;
+
+describe('retryTransientErrors', () => {
+  beforeEach(() => {
+    setTimeoutMock.mockClear();
+  });
+
+  it("doesn't retry if operation is successful", async () => {
+    const esCallMock = jest.fn().mockResolvedValue('success');
+    expect(await retryTransientEsErrors(esCallMock)).toEqual('success');
+    expect(esCallMock).toHaveBeenCalledTimes(1);
+  });
+
+  it('logs an warning message on retry', async () => {
+    const logger = loggerMock.create();
+    const esCallMock = jest
+      .fn()
+      .mockRejectedValueOnce(new EsErrors.ConnectionError('foo'))
+      .mockResolvedValue('success');
+
+    await retryTransientEsErrors(esCallMock, { logger });
+    expect(logger.warn).toHaveBeenCalledTimes(1);
+    expect(logger.warn.mock.calls[0][0]).toMatch(
+      `Retrying Elasticsearch operation after [2s] due to error: ConnectionError: foo ConnectionError: foo`
+    );
+  });
+
+  it('retries with an exponential backoff', async () => {
+    let attempt = 0;
+    const esCallMock = jest.fn(async () => {
+      attempt++;
+      if (attempt < 5) {
+        throw new EsErrors.ConnectionError('foo');
+      } else {
+        return 'success';
+      }
+    });
+
+    expect(await retryTransientEsErrors(esCallMock)).toEqual('success');
+    expect(setTimeoutMock.mock.calls).toEqual([[2000], [4000], [8000], [16000]]);
+    expect(esCallMock).toHaveBeenCalledTimes(5);
+  });
+
+  it('retries each supported error type', async () => {
+    const errors = [
+      new EsErrors.NoLivingConnectionsError('no living connection', {
+        warnings: [],
+        meta: {} as any,
+      }),
+      new EsErrors.ConnectionError('no connection'),
+      new EsErrors.TimeoutError('timeout'),
+      new EsErrors.ResponseError({ statusCode: 503, meta: {} as any, warnings: [] }),
+      new EsErrors.ResponseError({ statusCode: 408, meta: {} as any, warnings: [] }),
+      new EsErrors.ResponseError({ statusCode: 410, meta: {} as any, warnings: [] }),
+    ];
+
+    for (const error of errors) {
+      const esCallMock = jest.fn().mockRejectedValueOnce(error).mockResolvedValue('success');
+      expect(await retryTransientEsErrors(esCallMock)).toEqual('success');
+      expect(esCallMock).toHaveBeenCalledTimes(2);
+    }
+  });
+
+  it('does not retry unsupported errors', async () => {
+    const error = new Error('foo!');
+    const esCallMock = jest.fn().mockRejectedValueOnce(error).mockResolvedValue('success');
+    await expect(retryTransientEsErrors(esCallMock)).rejects.toThrow(error);
+    expect(esCallMock).toHaveBeenCalledTimes(1);
+  });
+});