[kbn/es] add support for --ready-timeout

packages/kbn-es/src/cli_commands/archive.js

@@ -10,6 +10,7 @@ const dedent = require('dedent');
 const getopts = require('getopts');
 const { Cluster } = require('../cluster');
 const { createCliError } = require('../errors');
+const { parseTimeoutToMs } = require('../utils');
 
 exports.description = 'Install and run from an Elasticsearch tar';
 
@@ -27,6 +28,8 @@ exports.help = (defaults = {}) => {
       --password.[user] Sets password for native realm user [default: ${password}]
       --ssl             Sets up SSL on Elasticsearch
       -E                Additional key=value settings to pass to Elasticsearch
+      --no-ready-check  Disable the ready check,
+      --ready-timeout   Customize the ready check timeout, in seconds or "Xm" format, defaults to 1m
 
     Example:
 
@@ -41,8 +44,13 @@ exports.run = async (defaults = {}) => {
       basePath: 'base-path',
       installPath: 'install-path',
       esArgs: 'E',
+      skipReadyCheck: 'no-ready-check',
+      readyTimeout: 'ready-timeout',
     },
 
+    string: ['ready-timeout'],
+    boolean: ['no-ready-check'],
+
     default: defaults,
   });
 
@@ -54,5 +62,8 @@ exports.run = async (defaults = {}) => {
   }
 
   const { installPath } = await cluster.installArchive(path, options);
-  await cluster.run(installPath, options);
+  await cluster.run(installPath, {
+    ...options,
+    readyTimeout: parseTimeoutToMs(options.readyTimeout),
+  });
 };

packages/kbn-es/src/cli_commands/snapshot.js

@@ -10,6 +10,7 @@ const dedent = require('dedent');
 const getopts = require('getopts');
 import { ToolingLog, getTimeReporter } from '@kbn/dev-utils';
 const { Cluster } = require('../cluster');
+const { parseTimeoutToMs } = require('../utils');
 
 exports.description = 'Downloads and run from a nightly snapshot';
 
@@ -30,6 +31,8 @@ exports.help = (defaults = {}) => {
       --download-only   Download the snapshot but don't actually start it
       --ssl             Sets up SSL on Elasticsearch
       --use-cached      Skips cache verification and use cached ES snapshot.
+      --no-ready-check  Disable the ready check,
+      --ready-timeout   Customize the ready check timeout, in seconds or "Xm" format, defaults to 1m
 
     Example:
 
@@ -53,11 +56,12 @@ exports.run = async (defaults = {}) => {
       dataArchive: 'data-archive',
       esArgs: 'E',
       useCached: 'use-cached',
+      skipReadyCheck: 'no-ready-check',
+      readyTimeout: 'ready-timeout',
     },
 
-    string: ['version'],
-
-    boolean: ['download-only', 'use-cached'],
+    string: ['version', 'ready-timeout'],
+    boolean: ['download-only', 'use-cached', 'no-ready-check'],
 
     default: defaults,
   });
@@ -82,6 +86,7 @@ exports.run = async (defaults = {}) => {
       reportTime,
       startTime: runStartTime,
       ...options,
+      readyTimeout: parseTimeoutToMs(options.readyTimeout),
     });
   }
 };

packages/kbn-es/src/cli_commands/source.js

@@ -9,6 +9,7 @@
 const dedent = require('dedent');
 const getopts = require('getopts');
 const { Cluster } = require('../cluster');
+const { parseTimeoutToMs } = require('../utils');
 
 exports.description = 'Build and run from source';
 
@@ -27,6 +28,8 @@ exports.help = (defaults = {}) => {
       --password.[user] Sets password for native realm user [default: ${password}]
       --ssl             Sets up SSL on Elasticsearch
       -E                Additional key=value settings to pass to Elasticsearch
+      --no-ready-check  Disable the ready check,
+      --ready-timeout   Customize the ready check timeout, in seconds or "Xm" format, defaults to 1m
 
     Example:
 
@@ -42,9 +45,14 @@ exports.run = async (defaults = {}) => {
       installPath: 'install-path',
       sourcePath: 'source-path',
       dataArchive: 'data-archive',
+      skipReadyCheck: 'no-ready-check',
+      readyTimeout: 'ready-timeout',
       esArgs: 'E',
     },
 
+    string: ['ready-timeout'],
+    boolean: ['no-ready-check'],
+
     default: defaults,
   });
 
@@ -55,5 +63,8 @@ exports.run = async (defaults = {}) => {
     await cluster.extractDataDirectory(installPath, options.dataArchive);
   }
 
-  await cluster.run(installPath, options);
+  await cluster.run(installPath, {
+    ...options,
+    readyTimeout: parseTimeoutToMs(options.readyTimeout),
+  });
 };

packages/kbn-es/src/cluster.js

@@ -6,20 +6,29 @@
  * Side Public License, v 1.
  */
 
-const fs = require('fs');
-const util = require('util');
+const fsp = require('fs/promises');
 const execa = require('execa');
 const chalk = require('chalk');
 const path = require('path');
+const { Client } = require('@elastic/elasticsearch');
 const { downloadSnapshot, installSnapshot, installSource, installArchive } = require('./install');
 const { ES_BIN } = require('./paths');
-const { log: defaultLog, parseEsLog, extractConfigFiles, NativeRealm } = require('./utils');
+const {
+  log: defaultLog,
+  parseEsLog,
+  extractConfigFiles,
+  NativeRealm,
+  parseTimeoutToMs,
+} = require('./utils');
 const { createCliError } = require('./errors');
 const { promisify } = require('util');
 const treeKillAsync = promisify(require('tree-kill'));
 const { parseSettings, SettingsFilter } = require('./settings');
 const { CA_CERT_PATH, ES_NOPASSWORD_P12_PATH, extract } = require('@kbn/dev-utils');
-const readFile = util.promisify(fs.readFile);
+
+const DEFAULT_READY_TIMEOUT = parseTimeoutToMs('1m');
+
+/** @typedef {import('./cluster_exec_options').EsClusterExecOptions} ExecOptions */
 
 // listen to data on stream until map returns anything but undefined
 const first = (stream, map) =>
@@ -38,7 +47,6 @@ exports.Cluster = class Cluster {
   constructor({ log = defaultLog, ssl = false } = {}) {
     this._log = log.withType('@kbn/es Cluster');
     this._ssl = ssl;
-    this._caCertPromise = ssl ? readFile(CA_CERT_PATH) : undefined;
   }
 
   /**
@@ -157,10 +165,8 @@ exports.Cluster = class Cluster {
    * Starts ES and returns resolved promise once started
    *
    * @param {String} installPath
-   * @param {Object} options
-   * @property {Array} options.esArgs
-   * @property {String} options.password - super user password used to bootstrap
-   * @returns {Promise}
+   * @param {ExecOptions} options
+   * @returns {Promise<void>}
    */
   async start(installPath, options = {}) {
     this._exec(installPath, options);
@@ -173,7 +179,7 @@ exports.Cluster = class Cluster {
             return true;
           }
         }),
-        this._nativeRealmSetup,
+        this._setupPromise,
       ]),
 
       // await the outcome of the process in case it exits before starting
@@ -187,15 +193,14 @@ exports.Cluster = class Cluster {
    * Starts Elasticsearch and waits for Elasticsearch to exit
    *
    * @param {String} installPath
-   * @param {Object} options
-   * @property {Array} options.esArgs
-   * @returns {Promise<undefined>}
+   * @param {ExecOptions} options
+   * @returns {Promise<void>}
    */
   async run(installPath, options = {}) {
     this._exec(installPath, options);
 
     // log native realm setup errors so they aren't uncaught
-    this._nativeRealmSetup.catch((error) => {
+    this._setupPromise.catch((error) => {
       this._log.error(error);
       this.stop();
     });
@@ -233,14 +238,17 @@ exports.Cluster = class Cluster {
    *
    * @private
    * @param {String} installPath
-   * @param {Object} options
-   * @property {string|Array} options.esArgs
-   * @property {string} options.esJavaOpts
-   * @property {Boolean} options.skipNativeRealmSetup
-   * @return {undefined}
+   * @param {ExecOptions} opts
    */
   _exec(installPath, opts = {}) {
-    const { skipNativeRealmSetup = false, reportTime = () => {}, startTime, ...options } = opts;
+    const {
+      skipNativeRealmSetup = false,
+      reportTime = () => {},
+      startTime,
+      skipReadyCheck,
+      readyTimeout,
+      ...options
+    } = opts;
 
     if (this._process || this._outcome) {
       throw new Error('ES has already been started');
@@ -300,30 +308,49 @@ exports.Cluster = class Cluster {
       stdio: ['ignore', 'pipe', 'pipe'],
     });
 
-    // parse log output to find http port
-    const httpPort = first(this._process.stdout, (data) => {
-      const match = data.toString('utf8').match(/HttpServer.+publish_address {[0-9.]+:([0-9]+)/);
+    this._setupPromise = Promise.all([
+      // parse log output to find http port
+      first(this._process.stdout, (data) => {
+        const match = data.toString('utf8').match(/HttpServer.+publish_address {[0-9.]+:([0-9]+)/);
 
-      if (match) {
-        return match[1];
+        if (match) {
+          return match[1];
+        }
+      }),
+
+      // load the CA cert from disk if necessary
+      this._ssl ? fsp.readFile(CA_CERT_PATH) : null,
+    ]).then(async ([port, caCert]) => {
+      const client = new Client({
+        node: `${caCert ? 'https:' : 'http:'}//localhost:${port}`,
+        auth: {
+          username: 'elastic',
+          password: options.password,
+        },
+        tls: caCert
+          ? {
+              ca: caCert,
+              rejectUnauthorized: true,
+            }
+          : undefined,
+      });
+
+      if (!skipReadyCheck) {
+        await this._waitForClusterReady(client, readyTimeout);
       }
-    });
 
-    // once the http port is available setup the native realm
-    this._nativeRealmSetup = httpPort.then(async (port) => {
-      if (skipNativeRealmSetup) {
-        return;
+      // once the cluster is ready setup the native realm
+      if (!skipNativeRealmSetup) {
+        const nativeRealm = new NativeRealm({
+          log: this._log,
+          elasticPassword: options.password,
+          client,
+        });
+
+        await nativeRealm.setPasswords(options);
       }
 
-      const caCert = await this._caCertPromise;
-      const nativeRealm = new NativeRealm({
-        port,
-        caCert,
-        log: this._log,
-        elasticPassword: options.password,
-        ssl: this._ssl,
-      });
-      await nativeRealm.setPasswords(options);
+      this._log.success('kbn/es setup complete');
     });
 
     let reportSent = false;
@@ -366,4 +393,43 @@ exports.Cluster = class Cluster {
       }
     });
   }
+
+  async _waitForClusterReady(client, readyTimeout = DEFAULT_READY_TIMEOUT) {
+    let attempt = 0;
+    const start = Date.now();
+
+    this._log.info('waiting for ES cluster to report a yellow or green status');
+
+    while (true) {
+      attempt += 1;
+
+      try {
+        const resp = await client.cluster.health();
+        if (resp.status !== 'red') {
+          return;
+        }
+
+        throw new Error(`not ready, cluster health is ${resp.status}`);
+      } catch (error) {
+        const timeSinceStart = Date.now() - start;
+        if (timeSinceStart > readyTimeout) {
+          const sec = readyTimeout / 1000;
+          throw new Error(`ES cluster failed to come online with the ${sec} second timeout`);
+        }
+
+        if (error.message.startsWith('not ready,')) {
+          if (timeSinceStart > 10_000) {
+            this._log.warning(error.message);
+          }
+        } else {
+          this._log.warning(
+            `waiting for ES cluster to come online, attempt ${attempt} failed with: ${error.message}`
+          );
+        }
+
+        const waitSec = attempt * 1.5;
+        await new Promise((resolve) => setTimeout(resolve, waitSec * 1000));
+      }
+    }
+  }
 };

packages/kbn-es/src/cluster_exec_options.ts

@@ -0,0 +1,18 @@
+/*
+ * 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.
+ */
+
+export interface EsClusterExecOptions {
+  skipNativeRealmSetup?: boolean;
+  reportTime?: (...args: any[]) => void;
+  startTime?: number;
+  esArgs?: string[];
+  esJavaOpts?: string;
+  password?: string;
+  skipReadyCheck?: boolean;
+  readyTimeout?: number;
+}

packages/kbn-es/src/utils/index.ts

@@ -17,3 +17,4 @@ export { extractConfigFiles } from './extract_config_files';
 export { NativeRealm, SYSTEM_INDICES_SUPERUSER } from './native_realm';
 export { buildSnapshot } from './build_snapshot';
 export { archiveForPlatform } from './build_snapshot';
+export * from './parse_timeout_to_ms';