Perform successful Elasticsearch version check before migrations (#51311

) (#56600)

* Convert parts of Elasticsearch version check to ts
* Move ES version check to NP
* Improve types
* Wait till for compatible ES nodes before SO migrations
* Don't wait for ES compatibility if skipMigrations=true
* Legacy Elasticsearch plugin integration test
* Make ES compatibility check and migrations logging more visible
* Test for isCompatible=false when ES version check throws
* Start pollEsNodesVersion immediately
* Refactor pollEsNodesVersion

src/core/server/elasticsearch/elasticsearch_config.ts

@@ -103,7 +103,19 @@ const configSchema = schema.object({
   ),
   apiVersion: schema.string({ defaultValue: DEFAULT_API_VERSION }),
   healthCheck: schema.object({ delay: schema.duration({ defaultValue: 2500 }) }),
-  ignoreVersionMismatch: schema.boolean({ defaultValue: false }),
+  ignoreVersionMismatch: schema.conditional(
+    schema.contextRef('dev'),
+    false,
+    schema.boolean({
+      validate: rawValue => {
+        if (rawValue === true) {
+          return '"ignoreVersionMismatch" can only be set to true in development mode';
+        }
+      },
+      defaultValue: false,
+    }),
+    schema.boolean({ defaultValue: false })
+  ),
 });
 
 const deprecations: ConfigDeprecationProvider = () => [

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

@@ -23,6 +23,7 @@ import { IScopedClusterClient } from './scoped_cluster_client';
 import { ElasticsearchConfig } from './elasticsearch_config';
 import { ElasticsearchService } from './elasticsearch_service';
 import { InternalElasticsearchServiceSetup, ElasticsearchServiceSetup } from './types';
+import { NodesVersionCompatibility } from './version_check/ensure_es_version';
 
 const createScopedClusterClientMock = (): jest.Mocked<IScopedClusterClient> => ({
   callAsInternalUser: jest.fn(),
@@ -71,6 +72,12 @@ type MockedInternalElasticSearchServiceSetup = jest.Mocked<
 const createInternalSetupContractMock = () => {
   const setupContract: MockedInternalElasticSearchServiceSetup = {
     ...createSetupContractMock(),
+    esNodesCompatibility$: new BehaviorSubject<NodesVersionCompatibility>({
+      isCompatible: true,
+      incompatibleNodes: [],
+      warningNodes: [],
+      kibanaVersion: '8.0.0',
+    }),
     legacy: {
       config$: new BehaviorSubject({} as ElasticsearchConfig),
     },

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

@@ -31,6 +31,7 @@ import { httpServiceMock } from '../http/http_service.mock';
 import { ElasticsearchConfig } from './elasticsearch_config';
 import { ElasticsearchService } from './elasticsearch_service';
 import { elasticsearchServiceMock } from './elasticsearch_service.mock';
+import { duration } from 'moment';
 
 let elasticsearchService: ElasticsearchService;
 const configService = configServiceMock.create();
@@ -41,7 +42,7 @@ configService.atPath.mockReturnValue(
   new BehaviorSubject({
     hosts: ['http://1.2.3.4'],
     healthCheck: {
-      delay: 2000,
+      delay: duration(2000),
     },
     ssl: {
       verificationMode: 'none',
@@ -125,7 +126,7 @@ describe('#setup', () => {
       const config = MockClusterClient.mock.calls[0][0];
       expect(config).toMatchInlineSnapshot(`
 Object {
-  "healthCheckDelay": 2000,
+  "healthCheckDelay": "PT2S",
   "hosts": Array [
     "http://8.8.8.8",
   ],
@@ -150,7 +151,7 @@ Object {
       const config = MockClusterClient.mock.calls[0][0];
       expect(config).toMatchInlineSnapshot(`
 Object {
-  "healthCheckDelay": 2000,
+  "healthCheckDelay": "PT2S",
   "hosts": Array [
     "http://1.2.3.4",
   ],
@@ -174,7 +175,7 @@ Object {
         new BehaviorSubject({
           hosts: ['http://1.2.3.4', 'http://9.8.7.6'],
           healthCheck: {
-            delay: 2000,
+            delay: duration(2000),
           },
           ssl: {
             verificationMode: 'none',
@@ -196,7 +197,7 @@ Object {
       const config = MockClusterClient.mock.calls[0][0];
       expect(config).toMatchInlineSnapshot(`
         Object {
-          "healthCheckDelay": 2000,
+          "healthCheckDelay": "PT2S",
           "hosts": Array [
             "http://8.8.8.8",
           ],

src/core/server/elasticsearch/elasticsearch_service.ts

@@ -30,6 +30,7 @@ import { ElasticsearchConfig, ElasticsearchConfigType } from './elasticsearch_co
 import { InternalHttpServiceSetup, GetAuthHeaders } from '../http/';
 import { InternalElasticsearchServiceSetup } from './types';
 import { CallAPIOptions } from './api_types';
+import { pollEsNodesVersion } from './version_check/ensure_es_version';
 
 /** @internal */
 interface CoreClusterClients {
@@ -46,9 +47,17 @@ interface SetupDeps {
 export class ElasticsearchService implements CoreService<InternalElasticsearchServiceSetup> {
   private readonly log: Logger;
   private readonly config$: Observable<ElasticsearchConfig>;
-  private subscription?: Subscription;
+  private subscriptions: {
+    client?: Subscription;
+    esNodesCompatibility?: Subscription;
+  } = {
+    client: undefined,
+    esNodesCompatibility: undefined,
+  };
+  private kibanaVersion: string;
 
   constructor(private readonly coreContext: CoreContext) {
+    this.kibanaVersion = coreContext.env.packageInfo.version;
     this.log = coreContext.logger.get('elasticsearch-service');
     this.config$ = coreContext.configService
       .atPath<ElasticsearchConfigType>('elasticsearch')
@@ -60,7 +69,7 @@ export class ElasticsearchService implements CoreService<InternalElasticsearchSe
 
     const clients$ = this.config$.pipe(
       filter(() => {
-        if (this.subscription !== undefined) {
+        if (this.subscriptions.client !== undefined) {
           this.log.error('Clients cannot be changed after they are created');
           return false;
         }
@@ -91,7 +100,7 @@ export class ElasticsearchService implements CoreService<InternalElasticsearchSe
       publishReplay(1)
     ) as ConnectableObservable<CoreClusterClients>;
 
-    this.subscription = clients$.connect();
+    this.subscriptions.client = clients$.connect();
 
     const config = await this.config$.pipe(first()).toPromise();
 
@@ -149,11 +158,31 @@ export class ElasticsearchService implements CoreService<InternalElasticsearchSe
       },
     };
 
+    const esNodesCompatibility$ = pollEsNodesVersion({
+      callWithInternalUser: adminClient.callAsInternalUser,
+      log: this.log,
+      ignoreVersionMismatch: config.ignoreVersionMismatch,
+      esVersionCheckInterval: config.healthCheckDelay.asMilliseconds(),
+      kibanaVersion: this.kibanaVersion,
+    }).pipe(publishReplay(1));
+
+    this.subscriptions.esNodesCompatibility = (esNodesCompatibility$ as ConnectableObservable<
+      unknown
+    >).connect();
+
+    // TODO: Move to Status Service https://github.com/elastic/kibana/issues/41983
+    esNodesCompatibility$.subscribe(({ isCompatible, message }) => {
+      if (!isCompatible && message) {
+        this.log.error(message);
+      }
+    });
+
     return {
       legacy: { config$: clients$.pipe(map(clients => clients.config)) },
 
       adminClient,
       dataClient,
+      esNodesCompatibility$,
 
       createClient: (type: string, clientConfig: Partial<ElasticsearchClientConfig> = {}) => {
         const finalConfig = merge({}, config, clientConfig);
@@ -166,11 +195,12 @@ export class ElasticsearchService implements CoreService<InternalElasticsearchSe
 
   public async stop() {
     this.log.debug('Stopping elasticsearch service');
-
-    if (this.subscription !== undefined) {
-      this.subscription.unsubscribe();
-      this.subscription = undefined;
-    }
+    // TODO(TS-3.7-ESLINT)
+    // eslint-disable-next-line no-unused-expressions
+    this.subscriptions.client?.unsubscribe();
+    // eslint-disable-next-line no-unused-expressions
+    this.subscriptions.esNodesCompatibility?.unsubscribe();
+    this.subscriptions = { client: undefined, esNodesCompatibility: undefined };
   }
 
   private createClusterClient(

src/core/server/elasticsearch/types.ts

@@ -21,6 +21,7 @@ import { Observable } from 'rxjs';
 import { ElasticsearchConfig } from './elasticsearch_config';
 import { ElasticsearchClientConfig } from './elasticsearch_client_config';
 import { IClusterClient, ICustomClusterClient } from './cluster_client';
+import { NodesVersionCompatibility } from './version_check/ensure_es_version';
 
 /**
  * @public
@@ -77,4 +78,5 @@ export interface InternalElasticsearchServiceSetup extends ElasticsearchServiceS
   readonly legacy: {
     readonly config$: Observable<ElasticsearchConfig>;
   };
+  esNodesCompatibility$: Observable<NodesVersionCompatibility>;
 }