WIP: elasticsearch-js next

package.json

@@ -95,6 +95,7 @@
   },
   "dependencies": {
     "@elastic/datemath": "5.0.2",
+    "@elastic/elasticsearch": "elastic/elasticsearch-js#next",
     "@elastic/eui": "6.10.1",
     "@elastic/filesaver": "1.1.2",
     "@elastic/good": "8.1.1-kibana2",
@@ -137,7 +138,6 @@
     "d3-cloud": "1.2.1",
     "del": "^3.0.0",
     "dragula": "3.7.0",
-    "elasticsearch": "^15.4.1",
     "elasticsearch-browser": "^15.4.1",
     "encode-uri-query": "1.0.0",
     "execa": "^1.0.0",

packages/kbn-test/src/es/es_test_cluster.js

@@ -24,7 +24,7 @@ import toPath from 'lodash/internal/toPath';
 import { Cluster } from '@kbn/es';
 import { esTestConfig } from './es_test_config';
 import { KIBANA_ROOT } from '../';
-import elasticsearch from 'elasticsearch';
+import elasticsearch from '@elastic/elasticsearch';
 const path = require('path');
 const del = require('del');
 
@@ -105,7 +105,7 @@ export function createEsTestCluster(options = {}) {
      */
     getClient() {
       return new elasticsearch.Client({
-        host: this.getUrl(),
+        node: this.getUrl(),
       });
     }
 
@@ -130,13 +130,20 @@ export function createEsTestCluster(options = {}) {
  *  @return {Function}
  */
 function createCallCluster(esClient) {
-  return function callCluster(method, params) {
+  return async function callCluster(method, params) {
     const path = toPath(method);
     const contextPath = path.slice(0, -1);
 
     const action = get(esClient, path);
     const context = contextPath.length ? get(esClient, contextPath) : esClient;
 
-    return action.call(context, params);
+    const { ignore, headers, requestTimeout, maxRetries, ...esParams } = params;
+    const { body } = await action.call(context, esParams, {
+      ignore,
+      headers,
+      maxRetries,
+      requestTimeout,
+    });
+    return body;
   };
 }

src/dev/xml.test.ts

@@ -0,0 +1,28 @@
+/*
+ * 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 { escapeCdata } from './xml';
+
+it('returns original string when there is no invalid XML character', () => {
+  expect(escapeCdata('foo')).toBe('foo');
+});
+
+it('strips ansi escape sequences, including \\b', () => {
+  expect(escapeCdata('foo')).toBe('foo');
+});

src/dev/xml.ts

@@ -23,16 +23,27 @@ import stripAnsi from 'strip-ansi';
 
 // create a regular expression using regenerate() that selects any character that is explicitly allowed by https://www.w3.org/TR/xml/#NT-Char
 const validXmlCharsRE = new RegExp(
-  `(?:${regenerate()
+  `((?:${regenerate()
     .add(0x9, 0xa, 0xd)
     .addRange(0x20, 0xd7ff)
     .addRange(0xe000, 0xfffd)
     .addRange(0x10000, 0x10ffff)
-    .toString()})*`,
+    .toString()})+)`,
   'g'
 );
 
 export function escapeCdata(input: string) {
-  const match = stripAnsi(input).match(validXmlCharsRE) || [];
-  return match.join('');
+  let escaped = '';
+
+  const inputWithoutAnsi = stripAnsi(input);
+  while (true) {
+    const match = validXmlCharsRE.exec(inputWithoutAnsi);
+    if (!match) {
+      break;
+    }
+
+    escaped += match[0];
+  }
+
+  return escaped;
 }

src/es_archiver/cli.js

@@ -29,7 +29,7 @@ import { format as formatUrl } from 'url';
 import readline from 'readline';
 
 import { Command } from 'commander';
-import elasticsearch from 'elasticsearch';
+import elasticsearch from '@elastic/elasticsearch';
 
 import { EsArchiver } from './es_archiver';
 import { ToolingLog } from '@kbn/dev-utils';
@@ -132,7 +132,7 @@ async function execute(fn) {
 
     // run!
     const client = new elasticsearch.Client({
-      host: cmd.esUrl,
+      node: cmd.esUrl,
       log: cmd.verbose ? 'trace' : []
     });
 
@@ -148,6 +148,7 @@ async function execute(fn) {
       await client.close();
     }
   } catch (err) {
-    console.log('FATAL ERROR', err.stack);
+    console.log(err);
+    // console.log('FATAL ERROR', err.stack);
   }
 }

src/es_archiver/lib/docs/__tests__/stubs.js

@@ -48,7 +48,8 @@ export const createStubClient = (responses = []) => {
     }
 
     const response = responses.shift();
-    return await response(name, params);
+    const body = await response(name, params);
+    return { body };
   });
 
   return {

src/es_archiver/lib/docs/generate_doc_records_stream.js

@@ -33,19 +33,21 @@ export function createGenerateDocRecordsStream(client, stats) {
 
         while (!resp || remainingHits > 0) {
           if (!resp) {
-            resp = await client.search({
+            const result = await client.search({
               index: index,
               scroll: SCROLL_TIMEOUT,
               size: SCROLL_SIZE,
               _source: true,
               rest_total_hits_as_int: true
             });
+            resp = result.body;
             remainingHits = resp.hits.total;
           } else {
-            resp = await client.scroll({
+            const result = await client.scroll({
               scrollId: resp._scroll_id,
               scroll: SCROLL_TIMEOUT
             });
+            resp = result.body;
           }
 
           for (const hit of resp.hits.hits) {

src/es_archiver/lib/docs/index_doc_records_stream.js

@@ -38,7 +38,7 @@ export function createIndexDocRecordsStream(client, stats) {
       );
     });
 
-    const resp = await client.bulk({ body });
+    const { body: resp } = await client.bulk({ body });
     if (resp.errors) {
       throw new Error(`Failed to index all documents: ${JSON.stringify(resp, null, 2)}`);
     }

src/es_archiver/lib/indices/__tests__/create_index_stream.js

@@ -73,7 +73,8 @@ describe('esArchiver: createCreateIndexStream()', () => {
       ]);
 
       expect(client.indices.getAlias.calledOnce).to.be.ok();
-      expect(client.indices.getAlias.args[0][0]).to.eql({ name: 'existing-index', ignore: [404] });
+      expect(client.indices.getAlias.args[0][0]).to.eql({ name: 'existing-index' });
+      expect(client.indices.getAlias.args[0][1]).to.eql({ ignore: [404] });
       expect(client.indices.delete.calledOnce).to.be.ok();
       expect(client.indices.delete.args[0][0]).to.eql({ index: ['actual-index'] });
       sinon.assert.callCount(client.indices.create, 3); // one failed create because of existing

src/es_archiver/lib/indices/__tests__/stubs.js

@@ -66,9 +66,11 @@ export const createStubClient = (existingIndices = [], aliases = {}) => ({
       }
 
       return {
-        [index]: {
-          mappings: {},
-          settings: {},
+        body: {
+          [index]: {
+            mappings: {},
+            settings: {},
+          }
         }
       };
     }),
@@ -78,14 +80,14 @@ export const createStubClient = (existingIndices = [], aliases = {}) => ({
     getAlias: sinon.spy(async ({ index, name }) => {
       if (index && existingIndices.indexOf(index) >= 0) {
         const result = indexAlias(aliases, index);
-        return { [index]: { aliases: result ? { [result]: {} } : {} } };
+        return { body: { [index]: { aliases: result ? { [result]: {} } : {} } } };
       }
 
       if (name && aliases[name]) {
-        return { [aliases[name]]: { aliases: { [name]: {} } } };
+        return { body: { [aliases[name]]: { aliases: { [name]: {} } } } };
       }
 
-      return { status: 404 };
+      return { body: { status: 404 } };
     }),
     updateAliases: sinon.spy(async ({ body }) => {
       body.actions.forEach(({ add: { index, alias } }) => {
@@ -95,14 +97,14 @@ export const createStubClient = (existingIndices = [], aliases = {}) => ({
         existingIndices.push({ index, alias });
       });
 
-      return { ok: true };
+      return { body: { ok: true } };
     }),
     create: sinon.spy(async ({ index }) => {
       if (existingIndices.includes(index) || aliases.hasOwnProperty(index)) {
         throw createEsClientError('resource_already_exists_exception');
       } else {
         existingIndices.push(index);
-        return { ok: true };
+        return { body: { ok: true } };
       }
     }),
     delete: sinon.spy(async ({ index }) => {
@@ -116,7 +118,7 @@ export const createStubClient = (existingIndices = [], aliases = {}) => ({
           }
         });
         indices.forEach(ix => existingIndices.splice(existingIndices.indexOf(ix), 1));
-        return { ok: true };
+        return { body: { ok: true } };
       } else {
         throw createEsClientError('index_not_found_exception');
       }

src/es_archiver/lib/indices/delete_index.js

@@ -36,7 +36,7 @@ export async function deleteIndex(options) {
   } = options;
 
   const getIndicesToDelete = async () => {
-    const aliasInfo = await client.indices.getAlias({ name: index, ignore: [404] });
+    const { body: aliasInfo } = await client.indices.getAlias({ name: index }, { ignore: [404] });
     return aliasInfo.status === 404 ? index : Object.keys(aliasInfo);
   };
 
@@ -82,24 +82,26 @@ export function isDeleteWhileSnapshotInProgressError(error) {
  */
 export async function waitForSnapshotCompletion(client, index, log) {
   const isSnapshotPending = async (repository, snapshot) => {
-    const { snapshots: [status] } = await client.snapshot.status({
+    const result = await client.snapshot.status({
       repository,
       snapshot,
     });
-
+    const { snapshots: [status] } = result.body;
     log.debug(`Snapshot ${repository}/${snapshot} is ${status.state}`);
     return PENDING_SNAPSHOT_STATUSES.includes(status.state);
   };
 
   const getInProgressSnapshots = async (repository) => {
-    const { snapshots: inProgressSnapshots } = await client.snapshot.get({
+    const result = await client.snapshot.get({
       repository,
       snapshot: '_current'
     });
+    const { snapshots: inProgressSnapshots } = result.body;
     return inProgressSnapshots;
   };
 
-  for (const repository of Object.keys(await client.snapshot.getRepository())) {
+  const { body } = await client.snapshot.getRepository();
+  for (const repository of Object.keys(body)) {
     const allInProgress = await getInProgressSnapshots(repository);
     const found = allInProgress.find(s => s.indices.includes(index));
 

src/es_archiver/lib/indices/generate_index_records_stream.js

@@ -25,7 +25,7 @@ export function createGenerateIndexRecordsStream(client, stats) {
     readableObjectMode: true,
     async transform(indexOrAlias, enc, callback) {
       try {
-        const resp = await client.indices.get({
+        const { body: resp } = await client.indices.get({
           index: indexOrAlias,
           filterPath: [
             '*.settings',
@@ -39,7 +39,10 @@ export function createGenerateIndexRecordsStream(client, stats) {
         });
 
         for (const [index, { settings, mappings }] of Object.entries(resp)) {
-          const { [index]: { aliases } } = await client.indices.getAlias({ index });
+          const result = await client.indices.getAlias({ index });
+          const { [index]: { aliases } } = result.body;
+          const { settings, mappings } = resp[index];
+
 
           stats.archivedIndex(index, { settings, mappings });
           this.push({

src/es_archiver/lib/indices/kibana_index.js

@@ -85,7 +85,17 @@ export async function migrateKibanaIndex({ client, log }) {
   const ready = async () => undefined;
   const elasticsearch = {
     getCluster: () => ({
-      callWithInternalUser: (path, ...args) => _.get(client, path).call(client, ...args),
+      callWithInternalUser: async (path, clientParams) => {
+        const {
+          ignore,
+          headers,
+          requestTimeout,
+          maxRetries,
+          ...esParams
+        } = clientParams;
+        const result = await _.get(client, path).call(client, esParams, { ignore, headers, requestTimeout, maxRetries });
+        return result && result.body ? result.body : result;
+      },
     }),
     waitUntilReady: ready,
   };
@@ -160,7 +170,7 @@ export async function createDefaultSpace({ index, client }) {
  * @param {string} index
  */
 async function fetchKibanaIndices(client) {
-  const kibanaIndices = await client.cat.indices({ index: '.kibana*', format: 'json' });
+  const { body: kibanaIndices } = await client.cat.indices({ index: '.kibana*', format: 'json' });
   const isKibanaIndex = (index) => (/^\.kibana(:?_\d*)?$/).test(index);
   return kibanaIndices.map(x => x.index).filter(isKibanaIndex);
 }

src/legacy/core_plugins/elasticsearch/lib/__tests__/cluster.js

@@ -20,14 +20,14 @@
 import expect from 'expect.js';
 import { Cluster } from '../cluster';
 import sinon from 'sinon';
-import { errors as esErrors } from 'elasticsearch';
+import { errors as esErrors } from '@elastic/elasticsearch';
 import { set, partial } from 'lodash';
 
 describe('plugins/elasticsearch', function () {
   describe('cluster', function () {
     let cluster;
     const config = {
-      url: 'http://localhost:9200',
+      hosts: ['http://localhost:9200'],
       ssl: { verificationMode: 'full' },
       requestHeadersWhitelist: [ 'authorization' ]
     };
@@ -128,8 +128,11 @@ describe('plugins/elasticsearch', function () {
         cluster.callWithRequest(request, 'nodes.info');
 
         sinon.assert.calledOnce(client.nodes.info);
-        expect(client.nodes.info.getCall(0).args[0]).to.eql({
-          headers: headers
+        expect(client.nodes.info.getCall(0).args[1]).to.eql({
+          headers: headers,
+          ignore: undefined,
+          maxRetries: undefined,
+          requestTimeout: undefined
         });
       });
 

src/legacy/core_plugins/elasticsearch/lib/__tests__/create_clusters.js

@@ -47,7 +47,7 @@ describe('plugins/elasticsearch', function () {
     describe('createCluster', () => {
       let cluster;
       const config = {
-        url: 'http://localhost:9200',
+        hosts: ['http://localhost:9200'],
         ssl: {
           verificationMode: 'none'
         }

src/legacy/core_plugins/elasticsearch/lib/__tests__/health_check.js

@@ -21,7 +21,7 @@ import Promise from 'bluebird';
 import sinon from 'sinon';
 import expect from 'expect.js';
 
-const NoConnections = require('elasticsearch').errors.NoConnections;
+const NoConnections = require('@elastic/elasticsearch').errors.NoLivingConnectionsError;
 
 import mappings from './fixtures/mappings';
 import healthCheck from '../health_check';