Retry authentication and other connection failures in migrations

src/core/server/elasticsearch/retry_call_cluster.test.ts

@@ -18,17 +18,16 @@
  */
 import * as legacyElasticsearch from 'elasticsearch';
 
-import { retryCallCluster } from './retry_call_cluster';
+import { retryCallCluster, migrationsRetryCallCluster } from './retry_call_cluster';
 
 describe('retryCallCluster', () => {
-  it('retries ES API calls that rejects with NoConnection errors', () => {
+  it('retries ES API calls that rejects with NoConnections', () => {
     expect.assertions(1);
     const callEsApi = jest.fn();
     let i = 0;
+    const ErrorConstructor = legacyElasticsearch.errors.NoConnections;
     callEsApi.mockImplementation(() => {
-      return i++ <= 2
-        ? Promise.reject(new legacyElasticsearch.errors.NoConnections())
-        : Promise.resolve('success');
+      return i++ <= 2 ? Promise.reject(new ErrorConstructor()) : Promise.resolve('success');
     });
     const retried = retryCallCluster(callEsApi);
     return expect(retried('endpoint')).resolves.toMatchInlineSnapshot(`"success"`);
@@ -57,3 +56,49 @@ describe('retryCallCluster', () => {
     return expect(retried('endpoint')).rejects.toMatchInlineSnapshot(`[Error: unknown error]`);
   });
 });
+
+describe('migrationsRetryCallCluster', () => {
+  const errors = [
+    'NoConnections',
+    'ConnectionFault',
+    'ServiceUnavailable',
+    'RequestTimeout',
+    'AuthenticationException',
+  ];
+  errors.forEach(errorName => {
+    it('retries ES API calls that rejects with ' + errorName, () => {
+      expect.assertions(1);
+      const callEsApi = jest.fn();
+      let i = 0;
+      const ErrorConstructor = (legacyElasticsearch.errors as any)[errorName];
+      callEsApi.mockImplementation(() => {
+        return i++ <= 2 ? Promise.reject(new ErrorConstructor()) : Promise.resolve('success');
+      });
+      const retried = migrationsRetryCallCluster(callEsApi);
+      return expect(retried('endpoint')).resolves.toMatchInlineSnapshot(`"success"`);
+    });
+  });
+
+  it('rejects when ES API calls reject with other errors', async () => {
+    expect.assertions(3);
+    const callEsApi = jest.fn();
+    let i = 0;
+    callEsApi.mockImplementation(() => {
+      i++;
+
+      return i === 1
+        ? Promise.reject(new Error('unknown error'))
+        : i === 2
+        ? Promise.resolve('success')
+        : i === 3 || i === 4
+        ? Promise.reject(new legacyElasticsearch.errors.NoConnections())
+        : i === 5
+        ? Promise.reject(new Error('unknown error'))
+        : null;
+    });
+    const retried = migrationsRetryCallCluster(callEsApi);
+    await expect(retried('endpoint')).rejects.toMatchInlineSnapshot(`[Error: unknown error]`);
+    await expect(retried('endpoint')).resolves.toMatchInlineSnapshot(`"success"`);
+    return expect(retried('endpoint')).rejects.toMatchInlineSnapshot(`[Error: unknown error]`);
+  });
+});

src/core/server/elasticsearch/retry_call_cluster.ts

@@ -23,6 +23,53 @@ import * as legacyElasticsearch from 'elasticsearch';
 
 import { CallAPIOptions } from '.';
 
+const esErrors = legacyElasticsearch.errors;
+
+/**
+ * Retries the provided Elasticsearch API call when an error such as
+ * `AuthenticationException` `NoConnections`, `ConnectionFault`,
+ * `ServiceUnavailable` or `RequestTimeout` are encountered. The API call will
+ * be retried once a second, indefinitely, until a successful response or a
+ * different error is received.
+ *
+ * @param apiCaller
+ */
+
+// TODO: Replace with APICaller from './scoped_cluster_client' once #46668 is merged
+export function migrationsRetryCallCluster(
+  apiCaller: (
+    endpoint: string,
+    clientParams: Record<string, any>,
+    options?: CallAPIOptions
+  ) => Promise<any>
+) {
+  return (endpoint: string, clientParams: Record<string, any> = {}, options?: CallAPIOptions) => {
+    return defer(() => apiCaller(endpoint, clientParams, options))
+      .pipe(
+        retryWhen(errors =>
+          errors.pipe(
+            concatMap((error, i) =>
+              iif(
+                () => {
+                  return (
+                    error instanceof esErrors.NoConnections ||
+                    error instanceof esErrors.ConnectionFault ||
+                    error instanceof esErrors.ServiceUnavailable ||
+                    error instanceof esErrors.RequestTimeout ||
+                    error instanceof esErrors.AuthenticationException
+                  );
+                },
+                timer(1000),
+                throwError(error)
+              )
+            )
+          )
+        )
+      )
+      .toPromise();
+  };
+}
+
 /**
  * Retries the provided Elasticsearch API call when a `NoConnections` error is
  * encountered. The API call will be retried once a second, indefinitely, until

src/core/server/saved_objects/saved_objects_service.ts

@@ -33,7 +33,7 @@ import { CoreContext } from '../core_context';
 import { LegacyServiceSetup } from '../legacy/legacy_service';
 import { ElasticsearchServiceSetup } from '../elasticsearch';
 import { KibanaConfigType } from '../kibana_config';
-import { retryCallCluster } from '../elasticsearch/retry_call_cluster';
+import { retryCallCluster, migrationsRetryCallCluster } from '../elasticsearch/retry_call_cluster';
 import { SavedObjectsConfigType } from './saved_objects_config';
 import { KibanaRequest } from '../http';
 import { Logger } from '..';
@@ -105,7 +105,7 @@ export class SavedObjectsService
       config: coreSetup.legacy.pluginExtendedConfig,
       savedObjectsConfig,
       kibanaConfig,
-      callCluster: retryCallCluster(adminClient.callAsInternalUser),
+      callCluster: migrationsRetryCallCluster(adminClient.callAsInternalUser),
     }));
 
     const mappings = this.migrator.getActiveMappings();