Enable Product check from @elastic/elasticsearch-js (#107663)

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

packages/kbn-es/src/integration_tests/__fixtures__/es_bin.js

@@ -31,11 +31,36 @@ const { ES_KEY_PATH, ES_CERT_PATH } = require('@kbn/dev-utils');
     },
     (req, res) => {
       const url = new URL(req.url, serverUrl);
-      const send = (code, body) => {
-        res.writeHead(code, { 'content-type': 'application/json' });
+      const send = (code, body, headers = {}) => {
+        res.writeHead(code, { 'content-type': 'application/json', ...headers });
         res.end(JSON.stringify(body));
       };
 
+      // ES client's Product check request: it checks some fields in the body and the header
+      if (url.pathname === '/') {
+        return send(
+          200,
+          {
+            name: 'es-bin',
+            cluster_name: 'elasticsearch',
+            cluster_uuid: 'k0sr2gr9S4OBtygmu9ndzA',
+            version: {
+              number: '8.0.0-SNAPSHOT',
+              build_flavor: 'default',
+              build_type: 'tar',
+              build_hash: 'b11c15b7e0af64f90c3eb9c52c2534b4f143a070',
+              build_date: '2021-08-03T19:32:39.781056185Z',
+              build_snapshot: true,
+              lucene_version: '8.9.0',
+              minimum_wire_compatibility_version: '7.15.0',
+              minimum_index_compatibility_version: '7.0.0',
+            },
+            tagline: 'You Know, for Search',
+          },
+          { 'x-elastic-product': 'Elasticsearch' }
+        );
+      }
+
       if (url.pathname === '/_xpack') {
         return send(400, {
           error: {

packages/kbn-es/src/utils/native_realm.js

@@ -11,23 +11,6 @@ const chalk = require('chalk');
 
 const { log: defaultLog } = require('./log');
 
-/**
- * Hack to skip the Product Check performed by the Elasticsearch-js client.
- * We noticed a couple of bugs that may need to be fixed before taking full
- * advantage of this feature.
- *
- * The bugs are detailed in this issue: https://github.com/elastic/kibana/issues/105557
- *
- * The hack is copied from the test/utils in the elasticsearch-js repo
- * (https://github.com/elastic/elasticsearch-js/blob/master/test/utils/index.js#L45-L56)
- */
-function skipProductCheck(client) {
-  const tSymbol = Object.getOwnPropertySymbols(client.transport || client).filter(
-    (symbol) => symbol.description === 'product check'
-  )[0];
-  (client.transport || client)[tSymbol] = 2;
-}
-
 exports.NativeRealm = class NativeRealm {
   constructor({ elasticPassword, port, log = defaultLog, ssl = false, caCert }) {
     this._client = new Client({
@@ -39,8 +22,6 @@ exports.NativeRealm = class NativeRealm {
           }
         : undefined,
     });
-    // TODO: @elastic/es-clients I had to disable the product check here because the client is getting 404 while ES is initializing, but the requests here auto retry them.
-    skipProductCheck(this._client);
     this._elasticPassword = elasticPassword;
     this._log = log;
   }

src/cli/serve/integration_tests/__fixtures__/reload_logging_config/kibana.test.yml

@@ -9,3 +9,5 @@ plugins:
   initialize: false
 migrations:
   skip: true
+elasticsearch:
+  skipStartupConnectionCheck: true

src/cli/serve/integration_tests/__fixtures__/reload_logging_config/kibana_log_console.test.yml

@@ -20,3 +20,5 @@ plugins:
   initialize: false
 migrations:
   skip: true
+elasticsearch:
+  skipStartupConnectionCheck: true

src/cli/serve/integration_tests/__fixtures__/reload_logging_config/kibana_log_file.test.yml

@@ -20,3 +20,5 @@ plugins:
   initialize: false
 migrations:
   skip: true
+elasticsearch:
+  skipStartupConnectionCheck: true

src/core/server/core_app/integration_tests/core_app_routes.test.ts

@@ -15,6 +15,7 @@ describe('Core app routes', () => {
   beforeAll(async function () {
     root = kbnTestServer.createRoot({
       plugins: { initialize: false },
+      elasticsearch: { skipStartupConnectionCheck: true },
       server: {
         basePath: '/base-path',
       },

src/core/server/core_app/integration_tests/static_assets.test.ts

@@ -13,7 +13,10 @@ describe('Platform assets', function () {
   let root: Root;
 
   beforeAll(async function () {
-    root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+    root = kbnTestServer.createRoot({
+      plugins: { initialize: false },
+      elasticsearch: { skipStartupConnectionCheck: true },
+    });
 
     await root.preboot();
     await root.setup();

src/core/server/elasticsearch/client/configure_client.ts

@@ -49,12 +49,6 @@ export const configureClient = (
   const client = new Client({ ...clientOptions, Transport: KibanaTransport });
   addLogging(client, logger.get('query', type));
 
-  // ------------------------------------------------------------------------ //
-  // Hack to disable the "Product check" while the bugs in                    //
-  // https://github.com/elastic/kibana/issues/105557 are handled.             //
-  skipProductCheck(client);
-  // ------------------------------------------------------------------------ //
-
   return client;
 };
 
@@ -137,21 +131,3 @@ const addLogging = (client: Client, logger: Logger) => {
     }
   });
 };
-
-/**
- * Hack to skip the Product Check performed by the Elasticsearch-js client.
- * We noticed a couple of bugs that may need to be fixed before taking full
- * advantage of this feature.
- *
- * The bugs are detailed in this issue: https://github.com/elastic/kibana/issues/105557
- *
- * The hack is copied from the test/utils in the elasticsearch-js repo
- * (https://github.com/elastic/elasticsearch-js/blob/master/test/utils/index.js#L45-L56)
- */
-function skipProductCheck(client: Client) {
-  const tSymbol = Object.getOwnPropertySymbols(client.transport || client).filter(
-    (symbol) => symbol.description === 'product check'
-  )[0];
-  // @ts-expect-error `tSymbol` is missing in the index signature of Transport
-  (client.transport || client)[tSymbol] = 2;
-}

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

@@ -43,6 +43,7 @@ test('set correct defaults', () => {
       "requestTimeout": "PT30S",
       "serviceAccountToken": undefined,
       "shardTimeout": "PT30S",
+      "skipStartupConnectionCheck": false,
       "sniffInterval": false,
       "sniffOnConnectionFault": false,
       "sniffOnStart": false,
@@ -397,3 +398,33 @@ test('serviceAccountToken does not throw if username is not set', () => {
 
   expect(() => config.schema.validate(obj)).not.toThrow();
 });
+
+describe('skipStartupConnectionCheck', () => {
+  test('defaults to `false`', () => {
+    const obj = {};
+    expect(() => config.schema.validate(obj)).not.toThrow();
+    expect(config.schema.validate(obj)).toEqual(
+      expect.objectContaining({
+        skipStartupConnectionCheck: false,
+      })
+    );
+  });
+
+  test('accepts `false` on both prod and dev mode', () => {
+    const obj = {
+      skipStartupConnectionCheck: false,
+    };
+    expect(() => config.schema.validate(obj, { dist: false })).not.toThrow();
+    expect(() => config.schema.validate(obj, { dist: true })).not.toThrow();
+  });
+
+  test('accepts `true` only when running from source to allow integration tests to run without an ES server', () => {
+    const obj = {
+      skipStartupConnectionCheck: true,
+    };
+    expect(() => config.schema.validate(obj, { dist: false })).not.toThrow();
+    expect(() => config.schema.validate(obj, { dist: true })).toThrowErrorMatchingInlineSnapshot(
+      `"[skipStartupConnectionCheck]: \\"skipStartupConnectionCheck\\" can only be set to true when running from source to allow integration tests to run without an ES server"`
+    );
+  });
+});

src/core/server/elasticsearch/elasticsearch_config.ts

@@ -153,6 +153,21 @@ export const configSchema = schema.object({
     }),
     schema.boolean({ defaultValue: false })
   ),
+  skipStartupConnectionCheck: schema.conditional(
+    // Using dist over dev because integration_tests run with dev: false,
+    // and this config is solely introduced to allow some of the integration tests to run without an ES server.
+    schema.contextRef('dist'),
+    true,
+    schema.boolean({
+      validate: (rawValue) => {
+        if (rawValue === true) {
+          return '"skipStartupConnectionCheck" can only be set to true when running from source to allow integration tests to run without an ES server';
+        }
+      },
+      defaultValue: false,
+    }),
+    schema.boolean({ defaultValue: false })
+  ),
 });
 
 const deprecations: ConfigDeprecationProvider = () => [
@@ -220,6 +235,17 @@ export const config: ServiceConfigDescriptor<ElasticsearchConfigType> = {
  * @public
  */
 export class ElasticsearchConfig {
+  /**
+   * @internal
+   * Only valid in dev mode. Skip the valid connection check during startup. The connection check allows
+   * Kibana to ensure that the Elasticsearch connection is valid before allowing
+   * any other services to be set up.
+   *
+   * @remarks
+   * You should disable this check at your own risk: Other services in Kibana
+   * may fail if this step is not completed.
+   */
+  public readonly skipStartupConnectionCheck: boolean;
   /**
    * The interval between health check requests Kibana sends to the Elasticsearch.
    */
@@ -337,6 +363,7 @@ export class ElasticsearchConfig {
     this.password = rawConfig.password;
     this.serviceAccountToken = rawConfig.serviceAccountToken;
     this.customHeaders = rawConfig.customHeaders;
+    this.skipStartupConnectionCheck = rawConfig.skipStartupConnectionCheck;
 
     const { alwaysPresentCertificate, verificationMode } = rawConfig.ssl;
     const { key, keyPassphrase, certificate, certificateAuthorities } = readKeyAndCerts(rawConfig);

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

@@ -6,6 +6,17 @@
  * Side Public License, v 1.
  */
 
+// Mocking the module to avoid waiting for a valid ES connection during these unit tests
+jest.mock('./is_valid_connection', () => ({
+  isValidConnection: jest.fn(),
+}));
+
+// Mocking this module to force different statuses to help with the unit tests
+jest.mock('./version_check/ensure_es_version', () => ({
+  pollEsNodesVersion: jest.fn(),
+}));
+
+import type { NodesVersionCompatibility } from './version_check/ensure_es_version';
 import { MockClusterClient } from './elasticsearch_service.test.mocks';
 import { BehaviorSubject } from 'rxjs';
 import { first } from 'rxjs/operators';
@@ -20,6 +31,11 @@ import { configSchema, ElasticsearchConfig } from './elasticsearch_config';
 import { ElasticsearchService } from './elasticsearch_service';
 import { elasticsearchClientMock } from './client/mocks';
 import { duration } from 'moment';
+import { isValidConnection as isValidConnectionMock } from './is_valid_connection';
+import { pollEsNodesVersion as pollEsNodesVersionMocked } from './version_check/ensure_es_version';
+const { pollEsNodesVersion: pollEsNodesVersionActual } = jest.requireActual(
+  './version_check/ensure_es_version'
+);
 
 const delay = async (durationMs: number) =>
   await new Promise((resolve) => setTimeout(resolve, durationMs));
@@ -33,7 +49,6 @@ const setupDeps = {
 
 let env: Env;
 let coreContext: CoreContext;
-const logger = loggingSystemMock.create();
 
 let mockClusterClientInstance: ReturnType<typeof elasticsearchClientMock.createCustomClusterClient>;
 
@@ -52,12 +67,16 @@ beforeEach(() => {
   });
   configService.atPath.mockReturnValue(mockConfig$);
 
+  const logger = loggingSystemMock.create();
   coreContext = { coreId: Symbol(), env, logger, configService: configService as any };
   elasticsearchService = new ElasticsearchService(coreContext);
 
   MockClusterClient.mockClear();
   mockClusterClientInstance = elasticsearchClientMock.createCustomClusterClient();
   MockClusterClient.mockImplementation(() => mockClusterClientInstance);
+
+  // @ts-expect-error TS does not get that `pollEsNodesVersion` is mocked
+  pollEsNodesVersionMocked.mockImplementation(pollEsNodesVersionActual);
 });
 
 afterEach(() => jest.clearAllMocks());
@@ -204,6 +223,62 @@ describe('#start', () => {
     expect(client.asInternalUser).toBe(mockClusterClientInstance.asInternalUser);
   });
 
+  it('should log.error non-compatible nodes error', async () => {
+    const defaultMessage = {
+      isCompatible: true,
+      kibanaVersion: '8.0.0',
+      incompatibleNodes: [],
+      warningNodes: [],
+    };
+    const observable$ = new BehaviorSubject<NodesVersionCompatibility>(defaultMessage);
+
+    // @ts-expect-error this module is mocked, so `mockImplementation` is an allowed property
+    pollEsNodesVersionMocked.mockImplementation(() => observable$);
+
+    await elasticsearchService.setup(setupDeps);
+    await elasticsearchService.start();
+    expect(loggingSystemMock.collect(coreContext.logger).error).toEqual([]);
+    observable$.next({
+      ...defaultMessage,
+      isCompatible: false,
+      message: 'Something went terribly wrong!',
+    });
+    expect(loggingSystemMock.collect(coreContext.logger).error).toEqual([
+      ['Something went terribly wrong!'],
+    ]);
+  });
+
+  describe('skipStartupConnectionCheck', () => {
+    it('should validate the connection by default', async () => {
+      await elasticsearchService.setup(setupDeps);
+      expect(isValidConnectionMock).not.toHaveBeenCalled();
+      await elasticsearchService.start();
+      expect(isValidConnectionMock).toHaveBeenCalledTimes(1);
+    });
+
+    it('should validate the connection when `false`', async () => {
+      mockConfig$.next({
+        ...(await mockConfig$.pipe(first()).toPromise()),
+        skipStartupConnectionCheck: false,
+      });
+      await elasticsearchService.setup(setupDeps);
+      expect(isValidConnectionMock).not.toHaveBeenCalled();
+      await elasticsearchService.start();
+      expect(isValidConnectionMock).toHaveBeenCalledTimes(1);
+    });
+
+    it('should not validate the connection when `true`', async () => {
+      mockConfig$.next({
+        ...(await mockConfig$.pipe(first()).toPromise()),
+        skipStartupConnectionCheck: true,
+      });
+      await elasticsearchService.setup(setupDeps);
+      expect(isValidConnectionMock).not.toHaveBeenCalled();
+      await elasticsearchService.start();
+      expect(isValidConnectionMock).not.toHaveBeenCalled();
+    });
+  });
+
   describe('#createClient', () => {
     it('allows to specify config properties', async () => {
       await elasticsearchService.setup(setupDeps);
@@ -281,7 +356,7 @@ describe('#stop', () => {
   });
 
   it('stops pollEsNodeVersions even if there are active subscriptions', async (done) => {
-    expect.assertions(2);
+    expect.assertions(3);
 
     const mockedClient = mockClusterClientInstance.asInternalUser;
     mockedClient.nodes.info.mockImplementation(() =>
@@ -292,10 +367,12 @@ describe('#stop', () => {
 
     setupContract.esNodesCompatibility$.subscribe(async () => {
       expect(mockedClient.nodes.info).toHaveBeenCalledTimes(1);
+      await delay(10);
+      expect(mockedClient.nodes.info).toHaveBeenCalledTimes(2);
 
       await elasticsearchService.stop();
       await delay(100);
-      expect(mockedClient.nodes.info).toHaveBeenCalledTimes(1);
+      expect(mockedClient.nodes.info).toHaveBeenCalledTimes(2);
       done();
     });
   });

src/core/server/elasticsearch/elasticsearch_service.ts

@@ -23,8 +23,10 @@ import {
   InternalElasticsearchServiceSetup,
   InternalElasticsearchServiceStart,
 } from './types';
+import type { NodesVersionCompatibility } from './version_check/ensure_es_version';
 import { pollEsNodesVersion } from './version_check/ensure_es_version';
 import { calculateStatus$ } from './status';
+import { isValidConnection } from './is_valid_connection';
 
 interface SetupDeps {
   http: InternalHttpServiceSetup;
@@ -40,7 +42,7 @@ export class ElasticsearchService
   private kibanaVersion: string;
   private getAuthHeaders?: GetAuthHeaders;
   private executionContextClient?: IExecutionContext;
-
+  private esNodesCompatibility$?: Observable<NodesVersionCompatibility>;
   private client?: ClusterClient;
 
   constructor(private readonly coreContext: CoreContext) {
@@ -84,6 +86,8 @@ export class ElasticsearchService
       kibanaVersion: this.kibanaVersion,
     }).pipe(takeUntil(this.stop$), shareReplay({ refCount: true, bufferSize: 1 }));
 
+    this.esNodesCompatibility$ = esNodesCompatibility$;
+
     return {
       legacy: {
         config$: this.config$,
@@ -93,11 +97,24 @@ export class ElasticsearchService
     };
   }
   public async start(): Promise<InternalElasticsearchServiceStart> {
-    if (!this.client) {
+    if (!this.client || !this.esNodesCompatibility$) {
       throw new Error('ElasticsearchService needs to be setup before calling start');
     }
 
     const config = await this.config$.pipe(first()).toPromise();
+
+    // Log every error we may encounter in the connection to Elasticsearch
+    this.esNodesCompatibility$.subscribe(({ isCompatible, message }) => {
+      if (!isCompatible && message) {
+        this.log.error(message);
+      }
+    });
+
+    if (!config.skipStartupConnectionCheck) {
+      // Ensure that the connection is established and the product is valid before moving on
+      await isValidConnection(this.esNodesCompatibility$);
+    }
+
     return {
       client: this.client!,
       createClient: (type, clientConfig) => this.createClusterClient(type, config, clientConfig),

src/core/server/elasticsearch/integration_tests/client.test.ts

@@ -6,11 +6,17 @@
  * Side Public License, v 1.
  */
 
+import { esTestConfig } from '@kbn/test';
+import * as http from 'http';
+import supertest from 'supertest';
+
 import {
+  createRootWithCorePlugins,
   createTestServers,
   TestElasticsearchUtils,
   TestKibanaUtils,
 } from '../../../test_helpers/kbn_server';
+import { Root } from '../../root';
 
 describe('elasticsearch clients', () => {
   let esServer: TestElasticsearchUtils;
@@ -55,3 +61,50 @@ describe('elasticsearch clients', () => {
     expect(resp.headers!.warning).toMatch('system indices');
   });
 });
+
+function createFakeElasticsearchServer() {
+  const server = http.createServer((req, res) => {
+    // Reply with a 200 and empty response by default (intentionally malformed response)
+    res.writeHead(200);
+    res.end();
+  });
+  server.listen(esTestConfig.getPort());
+
+  return server;
+}
+
+describe('fake elasticsearch', () => {
+  let esServer: http.Server;
+  let kibanaServer: Root;
+  let kibanaHttpServer: http.Server;
+
+  beforeAll(async () => {
+    kibanaServer = createRootWithCorePlugins({ status: { allowAnonymous: true } });
+    esServer = createFakeElasticsearchServer();
+
+    const kibanaPreboot = await kibanaServer.preboot();
+    kibanaHttpServer = kibanaPreboot.http.server.listener; // Mind that we are using the prebootServer at this point because the migration gets hanging, while waiting for ES to be correct
+    await kibanaServer.setup();
+  });
+
+  afterAll(async () => {
+    await kibanaServer.shutdown();
+    await new Promise<void>((resolve, reject) =>
+      esServer.close((err) => (err ? reject(err) : resolve()))
+    );
+  });
+
+  test('should return unknown product when it cannot perform the Product check (503 response)', async () => {
+    const resp = await supertest(kibanaHttpServer).get('/api/status').expect(503);
+    expect(resp.body.status.overall.state).toBe('red');
+    expect(resp.body.status.statuses[0].message).toBe(
+      'Unable to retrieve version information from Elasticsearch nodes. The client noticed that the server is not Elasticsearch and we do not support this unknown product.'
+    );
+  });
+
+  test('should fail to start Kibana because of the Product Check Error', async () => {
+    await expect(kibanaServer.start()).rejects.toThrowError(
+      'The client noticed that the server is not Elasticsearch and we do not support this unknown product.'
+    );
+  });
+});

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

@@ -0,0 +1,71 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Subject } from 'rxjs';
+import { errors } from '@elastic/elasticsearch';
+import { isValidConnection } from './is_valid_connection';
+import { NodesVersionCompatibility } from './version_check/ensure_es_version';
+
+describe('isValidConnection', () => {
+  const esNodesCompatibilityRequired: NodesVersionCompatibility = {
+    isCompatible: true,
+    incompatibleNodes: [],
+    warningNodes: [],
+    kibanaVersion: '8.0.0',
+  };
+  const incompatible = {
+    ...esNodesCompatibilityRequired,
+    isCompatible: false,
+    message: 'Something is wrong!',
+  };
+  const compatible = {
+    ...esNodesCompatibilityRequired,
+    isCompatible: true,
+    message: 'All OK!',
+  };
+  const errored = {
+    ...incompatible,
+    nodesInfoRequestError: new errors.ConnectionError('Something went terribly wrong', {} as any),
+  };
+
+  test('should resolve only on compatible nodes', async () => {
+    const esNodesCompatibility$ = new Subject<NodesVersionCompatibility>();
+    const promise = isValidConnection(esNodesCompatibility$);
+
+    esNodesCompatibility$.next(incompatible);
+    esNodesCompatibility$.next(errored);
+    esNodesCompatibility$.next(compatible);
+
+    await expect(promise).resolves.toStrictEqual(compatible);
+  });
+
+  test('should throw an error only on ProductCheckError', async () => {
+    const esNodesCompatibility$ = new Subject<NodesVersionCompatibility>();
+    const promise = isValidConnection(esNodesCompatibility$);
+
+    const { ProductNotSupportedError, ConnectionError, ConfigurationError } = errors;
+
+    // Emit some other errors declared by the ES client
+    esNodesCompatibility$.next({
+      ...errored,
+      nodesInfoRequestError: new ConnectionError('Something went terribly wrong', {} as any),
+    });
+    esNodesCompatibility$.next({
+      ...errored,
+      nodesInfoRequestError: new ConfigurationError('Something went terribly wrong'),
+    });
+
+    const productCheckErrored = {
+      ...incompatible,
+      nodesInfoRequestError: new ProductNotSupportedError({} as any),
+    };
+    esNodesCompatibility$.next(productCheckErrored);
+
+    await expect(promise).rejects.toThrow(productCheckErrored.nodesInfoRequestError);
+  });
+});

src/core/server/elasticsearch/is_valid_connection.ts

@@ -0,0 +1,42 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { filter, first } from 'rxjs/operators';
+import { errors } from '@elastic/elasticsearch';
+import { Observable } from 'rxjs';
+import { NodesVersionCompatibility } from './version_check/ensure_es_version';
+
+/**
+ * Validates the output of the ES Compatibility Check and waits for a valid connection.
+ * It may also throw on specific config/connection errors to make Kibana halt.
+ *
+ * @param esNodesCompatibility$ ES Compatibility Check's observable
+ *
+ * @remarks: Ideally, this will be called during the start lifecycle to figure
+ * out any configuration issue as soon as possible.
+ */
+export async function isValidConnection(
+  esNodesCompatibility$: Observable<NodesVersionCompatibility>
+) {
+  return await esNodesCompatibility$
+    .pipe(
+      filter(({ nodesInfoRequestError, isCompatible }) => {
+        if (
+          nodesInfoRequestError &&
+          nodesInfoRequestError instanceof errors.ProductNotSupportedError
+        ) {
+          // Throw on the specific error of ProductNotSupported.
+          // We explicitly want Kibana to halt in this case.
+          throw nodesInfoRequestError;
+        }
+        return isCompatible;
+      }),
+      first()
+    )
+    .toPromise();
+}

src/core/server/http/integration_tests/core_services.test.ts

@@ -34,7 +34,10 @@ describe('http service', () => {
   describe('auth', () => {
     let root: ReturnType<typeof kbnTestServer.createRoot>;
     beforeEach(async () => {
-      root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+      root = kbnTestServer.createRoot({
+        plugins: { initialize: false },
+        elasticsearch: { skipStartupConnectionCheck: true },
+      });
       await root.preboot();
     }, 30000);
 
@@ -182,7 +185,10 @@ describe('http service', () => {
     let root: ReturnType<typeof kbnTestServer.createRoot>;
 
     beforeEach(async () => {
-      root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+      root = kbnTestServer.createRoot({
+        plugins: { initialize: false },
+        elasticsearch: { skipStartupConnectionCheck: true },
+      });
       await root.preboot();
     }, 30000);
 

src/core/server/http/integration_tests/http_auth.test.ts

@@ -14,7 +14,10 @@ describe('http auth', () => {
   let root: ReturnType<typeof kbnTestServer.createRoot>;
 
   beforeEach(async () => {
-    root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+    root = kbnTestServer.createRoot({
+      plugins: { initialize: false },
+      elasticsearch: { skipStartupConnectionCheck: true },
+    });
     await root.preboot();
   }, 30000);
 

src/core/server/http/integration_tests/logging.test.ts

@@ -27,7 +27,10 @@ describe('request logging', () => {
   describe('http server response logging', () => {
     describe('configuration', () => {
       it('does not log with a default config', async () => {
-        const root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+        const root = kbnTestServer.createRoot({
+          plugins: { initialize: false },
+          elasticsearch: { skipStartupConnectionCheck: true },
+        });
         await root.preboot();
         const { http } = await root.setup();
 
@@ -69,6 +72,7 @@ describe('request logging', () => {
           plugins: {
             initialize: false,
           },
+          elasticsearch: { skipStartupConnectionCheck: true },
         });
         await root.preboot();
         const { http } = await root.setup();
@@ -116,6 +120,7 @@ describe('request logging', () => {
         plugins: {
           initialize: false,
         },
+        elasticsearch: { skipStartupConnectionCheck: true },
       };
 
       beforeEach(() => {
@@ -327,6 +332,7 @@ describe('request logging', () => {
             plugins: {
               initialize: false,
             },
+            elasticsearch: { skipStartupConnectionCheck: true },
           });
           await root.preboot();
           const { http } = await root.setup();
@@ -426,6 +432,7 @@ describe('request logging', () => {
             plugins: {
               initialize: false,
             },
+            elasticsearch: { skipStartupConnectionCheck: true },
           });
           await root.preboot();
           const { http } = await root.setup();

src/core/server/http_resources/integration_tests/http_resources_service.test.ts

@@ -19,6 +19,7 @@ describe('http resources service', () => {
           rules: [defaultCspRules],
         },
         plugins: { initialize: false },
+        elasticsearch: { skipStartupConnectionCheck: true },
       });
       await root.preboot();
     }, 30000);

src/core/server/legacy/integration_tests/logging.test.ts

@@ -18,6 +18,7 @@ function createRoot(legacyLoggingConfig: LegacyLoggingConfig = {}) {
   return kbnTestServer.createRoot({
     migrations: { skip: true }, // otherwise stuck in polling ES
     plugins: { initialize: false },
+    elasticsearch: { skipStartupConnectionCheck: true },
     logging: {
       // legacy platform config
       silent: false,

src/core/server/saved_objects/routes/integration_tests/migrate.test.ts

@@ -13,7 +13,11 @@ describe('SavedObjects /_migrate endpoint', () => {
   let root: ReturnType<typeof kbnTestServer.createRoot>;
 
   beforeEach(async () => {
-    root = kbnTestServer.createRoot({ migrations: { skip: true }, plugins: { initialize: false } });
+    root = kbnTestServer.createRoot({
+      migrations: { skip: true },
+      plugins: { initialize: false },
+      elasticsearch: { skipStartupConnectionCheck: true },
+    });
     await root.preboot();
     await root.setup();
     await root.start();

src/core/server/saved_objects/saved_objects_service.ts

@@ -399,20 +399,20 @@ export class SavedObjectsService
         'Waiting until all Elasticsearch nodes are compatible with Kibana before starting saved objects migrations...'
       );
 
-      // TODO: Move to Status Service https://github.com/elastic/kibana/issues/41983
-      this.setupDeps!.elasticsearch.esNodesCompatibility$.subscribe(({ isCompatible, message }) => {
-        if (!isCompatible && message) {
-          this.logger.error(message);
-        }
-      });
-
-      await this.setupDeps!.elasticsearch.esNodesCompatibility$.pipe(
+      // The Elasticsearch service should already ensure that, but let's double check just in case.
+      // Should it be replaced with elasticsearch.status$ API instead?
+      const compatibleNodes = await this.setupDeps!.elasticsearch.esNodesCompatibility$.pipe(
         filter((nodes) => nodes.isCompatible),
         take(1)
       ).toPromise();
 
-      this.logger.info('Starting saved objects migrations');
-      await migrator.runMigrations();
+      // Running migrations only if we got compatible nodes.
+      // It may happen that the observable completes due to Kibana shutting down
+      // and the promise above fulfils as undefined. We shouldn't trigger migrations at that point.
+      if (compatibleNodes) {
+        this.logger.info('Starting saved objects migrations');
+        await migrator.runMigrations();
+      }
     }
 
     const createRepository = (

src/core/server/server.api.md

@@ -237,6 +237,7 @@ export const config: {
                 delay: Type<import("moment").Duration>;
             }>;
             ignoreVersionMismatch: import("@kbn/config-schema/target_types/types").ConditionalType<false, boolean, boolean>;
+            skipStartupConnectionCheck: import("@kbn/config-schema/target_types/types").ConditionalType<true, boolean, boolean>;
         }>;
     };
     logging: {
@@ -825,6 +826,8 @@ export class ElasticsearchConfig {
     readonly requestTimeout: Duration;
     readonly serviceAccountToken?: string;
     readonly shardTimeout: Duration;
+    // @internal
+    readonly skipStartupConnectionCheck: boolean;
     readonly sniffInterval: false | Duration;
     readonly sniffOnConnectionFault: boolean;
     readonly sniffOnStart: boolean;

src/core/server/ui_settings/integration_tests/routes.test.ts

@@ -13,7 +13,10 @@ describe('ui settings service', () => {
   describe('routes', () => {
     let root: ReturnType<typeof kbnTestServer.createRoot>;
     beforeAll(async () => {
-      root = kbnTestServer.createRoot({ plugins: { initialize: false } });
+      root = kbnTestServer.createRoot({
+        plugins: { initialize: false },
+        elasticsearch: { skipStartupConnectionCheck: true },
+      });
 
       await root.preboot();
       const { uiSettings } = await root.setup();

x-pack/plugins/monitoring/server/config.test.ts

@@ -87,6 +87,7 @@ describe('config schema', () => {
             ],
             "requestTimeout": "PT30S",
             "shardTimeout": "PT30S",
+            "skipStartupConnectionCheck": false,
             "sniffInterval": false,
             "sniffOnConnectionFault": false,
             "sniffOnStart": false,