Perform successful Elasticsearch version check before migrations

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.any({
+      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,24 @@ 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();
+
     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 +188,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>;
 }

src/core/server/elasticsearch/version_check/ensure_es_version.test.ts

@@ -0,0 +1,177 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { mapNodesVersionCompatibility, pollEsNodesVersion, NodesInfo } from './ensure_es_version';
+import { loggingServiceMock } from '../../logging/logging_service.mock';
+import { take } from 'rxjs/operators';
+
+const mockLoggerFactory = loggingServiceMock.create();
+const mockLogger = mockLoggerFactory.get('mock logger');
+
+const KIBANA_VERSION = '5.1.0';
+
+function createNodes(...versions: string[]): NodesInfo {
+  const nodes = {} as any;
+  versions
+    .map(version => {
+      return {
+        version,
+        http: {
+          publish_address: 'http_address',
+        },
+        ip: 'ip',
+      };
+    })
+    .forEach((node, i) => {
+      nodes[`node-${i}`] = node;
+    });
+
+  return { nodes };
+}
+
+describe('mapNodesVersionCompatibility', () => {
+  function createNodesInfoWithoutHTTP(version: string): NodesInfo {
+    return { nodes: { 'node-without-http': { version, ip: 'ip' } } } as any;
+  }
+
+  it('returns isCompatible=true with a single node that matches', async () => {
+    const nodesInfo = createNodes('5.1.0');
+    const result = await mapNodesVersionCompatibility(nodesInfo, KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(true);
+  });
+
+  it('returns isCompatible=true with multiple nodes that satisfy', async () => {
+    const nodesInfo = createNodes('5.1.0', '5.2.0', '5.1.1-Beta1');
+    const result = await mapNodesVersionCompatibility(nodesInfo, KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(true);
+  });
+
+  it('returns isCompatible=false for a single node that is out of date', () => {
+    // 5.0.0 ES is too old to work with a 5.1.0 version of Kibana.
+    const nodesInfo = createNodes('5.1.0', '5.2.0', '5.0.0');
+    const result = mapNodesVersionCompatibility(nodesInfo, KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(false);
+    expect(result.message).toMatchInlineSnapshot(
+      `"This version of Kibana (v5.1.0) is incompatible with the following Elasticsearch nodes in your cluster: v5.0.0 @ http_address (ip)"`
+    );
+  });
+
+  it('returns isCompatible=false for an incompatible node without http publish address', async () => {
+    const nodesInfo = createNodesInfoWithoutHTTP('6.1.1');
+    const result = mapNodesVersionCompatibility(nodesInfo, KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(false);
+    expect(result.message).toMatchInlineSnapshot(
+      `"This version of Kibana (v5.1.0) is incompatible with the following Elasticsearch nodes in your cluster: v6.1.1 @ undefined (ip)"`
+    );
+  });
+
+  it('returns isCompatible=true for outdated nodes when ignoreVersionMismatch=true', async () => {
+    // 5.0.0 ES is too old to work with a 5.1.0 version of Kibana.
+    const nodesInfo = createNodes('5.1.0', '5.2.0', '5.0.0');
+    const ignoreVersionMismatch = true;
+    const result = mapNodesVersionCompatibility(nodesInfo, KIBANA_VERSION, ignoreVersionMismatch);
+    expect(result.isCompatible).toBe(true);
+    expect(result.message).toMatchInlineSnapshot(
+      `"Ignoring version incompatibility between Kibana v5.1.0 and the following Elasticsearch nodes: v5.0.0 @ http_address (ip)"`
+    );
+  });
+
+  it('returns isCompatible=true with a message if a node is only off by a patch version', () => {
+    const result = mapNodesVersionCompatibility(createNodes('5.1.1'), KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(true);
+    expect(result.message).toMatchInlineSnapshot(
+      `"You're running Kibana 5.1.0 with some different versions of Elasticsearch. Update Kibana or Elasticsearch to the same version to prevent compatibility issues: v5.1.1 @ http_address (ip)"`
+    );
+  });
+
+  it('returns isCompatible=true with a message if a node is only off by a patch version and without http publish address', async () => {
+    const result = mapNodesVersionCompatibility(createNodes('5.1.1'), KIBANA_VERSION, false);
+    expect(result.isCompatible).toBe(true);
+    expect(result.message).toMatchInlineSnapshot(
+      `"You're running Kibana 5.1.0 with some different versions of Elasticsearch. Update Kibana or Elasticsearch to the same version to prevent compatibility issues: v5.1.1 @ http_address (ip)"`
+    );
+  });
+});
+
+describe('pollEsNodesVersion', () => {
+  const callWithInternalUser = jest.fn();
+  it('keeps polling when a poll request throws', done => {
+    expect.assertions(2);
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.0', '5.2.0', '5.0.0'));
+    callWithInternalUser.mockRejectedValueOnce(new Error('mock request error'));
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.0', '5.2.0', '5.1.1-Beta1'));
+    pollEsNodesVersion({
+      callWithInternalUser,
+      esVersionCheckInterval: 1,
+      ignoreVersionMismatch: false,
+      kibanaVersion: KIBANA_VERSION,
+      log: mockLogger,
+    })
+      .pipe(take(2))
+      .subscribe({
+        next: result => expect(result.isCompatible).toBeDefined(),
+        complete: done,
+        error: done,
+      });
+  });
+
+  it('returns compatibility results', done => {
+    expect.assertions(1);
+    const nodes = createNodes('5.1.0', '5.2.0', '5.0.0');
+    callWithInternalUser.mockResolvedValueOnce(nodes);
+    pollEsNodesVersion({
+      callWithInternalUser,
+      esVersionCheckInterval: 1,
+      ignoreVersionMismatch: false,
+      kibanaVersion: KIBANA_VERSION,
+      log: mockLogger,
+    })
+      .pipe(take(1))
+      .subscribe({
+        next: result => {
+          expect(result).toEqual(mapNodesVersionCompatibility(nodes, KIBANA_VERSION, false));
+        },
+        complete: done,
+        error: done,
+      });
+  });
+
+  it('only emits if the node versions changed since the previous poll', done => {
+    expect.assertions(4);
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.0', '5.2.0', '5.0.0')); // emit
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.0.0', '5.1.0', '5.2.0')); // ignore, same versions, different ordering
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.1', '5.2.0', '5.0.0')); // emit
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.1', '5.1.2', '5.1.3')); // emit
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.1.1', '5.1.2', '5.1.3')); // ignore
+    callWithInternalUser.mockResolvedValueOnce(createNodes('5.0.0', '5.1.0', '5.2.0')); // emit, different from previous version
+
+    pollEsNodesVersion({
+      callWithInternalUser,
+      esVersionCheckInterval: 1,
+      ignoreVersionMismatch: false,
+      kibanaVersion: KIBANA_VERSION,
+      log: mockLogger,
+    })
+      .pipe(take(4))
+      .subscribe({
+        next: result => expect(result).toBeDefined(),
+        complete: done,
+        error: done,
+      });
+  });
+});