Tribe node support

src/core_plugins/elasticsearch/index.js

@@ -1,8 +1,13 @@
-import { trim, trimRight } from 'lodash';
+import { trim, trimRight, bindKey, get } from 'lodash';
 import { methodNotAllowed } from 'boom';
 
 import healthCheck from './lib/health_check';
-import exposeClient from './lib/expose_client';
+import { createDataCluster } from './lib/create_data_cluster';
+import { createAdminCluster } from './lib/create_admin_cluster';
+import { clientLogger } from './lib/client_logger';
+import { createClusters } from './lib/create_clusters';
+import filterHeaders from './lib/filter_headers';
+
 import createProxy, { createPath } from './lib/create_proxy';
 
 const DEFAULT_REQUEST_HEADERS = [ 'authorization' ];
@@ -26,13 +31,37 @@ module.exports = function ({ Plugin }) {
         customHeaders: object().default({}),
         pingTimeout: number().default(ref('requestTimeout')),
         startupTimeout: number().default(5000),
+        logQueries: boolean().default(false),
         ssl: object({
           verify: boolean().default(true),
           ca: array().single().items(string()),
           cert: string(),
           key: string()
         }).default(),
         apiVersion: Joi.string().default('master'),
+        healthCheck: object({
+          delay: number().default(2500)
+        }).default(),
+        tribe: object({
+          url: string().uri({ scheme: ['http', 'https'] }),
+          preserveHost: boolean().default(true),
+          username: string(),
+          password: string(),
+          shardTimeout: number().default(0),
+          requestTimeout: number().default(30000),
+          requestHeadersWhitelist: array().items().single().default(DEFAULT_REQUEST_HEADERS),
+          customHeaders: object().default({}),
+          pingTimeout: number().default(ref('requestTimeout')),
+          startupTimeout: number().default(5000),
+          logQueries: boolean().default(false),
+          ssl: object({
+            verify: boolean().default(true),
+            ca: array().single().items(string()),
+            cert: string(),
+            key: string()
+          }).default(),
+          apiVersion: Joi.string().default('master'),
+        }).default()
       }).default();
     },
 
@@ -42,15 +71,24 @@ module.exports = function ({ Plugin }) {
           esRequestTimeout: options.requestTimeout,
           esShardTimeout: options.shardTimeout,
           esApiVersion: options.apiVersion,
+          esDataIsTribe: get(options, 'tribe.url') ? true : false,
         };
       }
     },
 
     init(server, options) {
       const kibanaIndex = server.config().get('kibana.index');
+      const clusters = createClusters(server);
+
+      server.expose('getCluster', clusters.get);
+      server.expose('createCluster', clusters.create);
+
+      server.expose('filterHeaders', filterHeaders);
+      server.expose('ElasticsearchClientLogging', clientLogger(server));
+
+      createDataCluster(server);
+      createAdminCluster(server);
 
-      // Expose the client to the server
-      exposeClient(server);
       createProxy(server, 'GET', '/{paths*}');
       createProxy(server, 'POST', '/_mget');
       createProxy(server, 'POST', '/{index}/_search');
@@ -69,7 +107,7 @@ module.exports = function ({ Plugin }) {
 
       function noDirectIndex({ path }, reply) {
         const requestPath = trimRight(trim(path), '/');
-        const matchPath = createPath(kibanaIndex);
+        const matchPath = createPath('/elasticsearch', kibanaIndex);
 
         if (requestPath === matchPath) {
           return reply(methodNotAllowed('You cannot modify the primary kibana index through this interface.'));

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

@@ -0,0 +1,134 @@
+import expect from 'expect.js';
+import { Cluster } from '../cluster';
+import sinon from 'sinon';
+import { errors as esErrors } from 'elasticsearch';
+import { set, partial, cloneDeep } from 'lodash';
+import Boom from 'boom';
+
+describe('plugins/elasticsearch', function () {
+  describe('cluster', function () {
+    let cluster;
+    const config = {
+      url: 'http://localhost:9200',
+      ssl: { verify: false },
+      requestHeadersWhitelist: [ 'authorization' ]
+    };
+
+    beforeEach(() => {
+      cluster = new Cluster(config);
+    });
+
+    it('persists the config', () => {
+      expect(cluster._config).to.eql(config);
+    });
+
+    it('exposes error definitions', () => {
+      expect(cluster.errors).to.be(esErrors);
+    });
+
+    it('closes the clients', () => {
+      cluster._client.close = sinon.spy();
+      cluster._noAuthClient.close = sinon.spy();
+      cluster.close();
+
+      sinon.assert.calledOnce(cluster._client.close);
+      sinon.assert.calledOnce(cluster._noAuthClient.close);
+    });
+
+    it('protects the config from changes', () => {
+      const localRequestHeadersWhitelist = cluster.getRequestHeadersWhitelist();
+      expect(localRequestHeadersWhitelist.length).to.not.equal(config.requestHeadersWhitelist);
+    });
+
+    describe('callWithInternalUser', () => {
+      let client;
+
+      beforeEach(() => {
+        client = cluster._client = sinon.stub();
+        set(client, 'nodes.info', sinon.stub().returns(Promise.resolve()));
+      });
+
+      it('should return a function', () => {
+        expect(cluster.callWithInternalUser).to.be.a('function');
+      });
+
+      it('throws an error for an invalid endpoint', () => {
+        const fn = partial(cluster.callWithInternalUser, 'foo');
+        expect(fn).to.throwException(/called with an invalid endpoint: foo/);
+      });
+
+      it('calls the client with params', () => {
+        const params = { foo: 'Foo' };
+        cluster.callWithInternalUser('nodes.info', params);
+
+        sinon.assert.calledOnce(client.nodes.info);
+        expect(client.nodes.info.getCall(0).args[0]).to.eql(params);
+      });
+    });
+
+    describe('callWithRequest', () => {
+      let client;
+
+      beforeEach(() => {
+        client = cluster._noAuthClient = sinon.stub();
+        set(client, 'nodes.info', sinon.stub().returns(Promise.resolve()));
+      });
+
+      it('should return a function', () => {
+        expect(cluster.callWithRequest).to.be.a('function');
+      });
+
+      it('throws an error for an invalid endpoint', () => {
+        const fn = partial(cluster.callWithRequest, {}, 'foo');
+        expect(fn).to.throwException(/called with an invalid endpoint: foo/);
+      });
+
+      it('calls the client with params', () => {
+        const params = { foo: 'Foo' };
+        cluster.callWithRequest({}, 'nodes.info', params);
+
+        sinon.assert.calledOnce(client.nodes.info);
+        expect(client.nodes.info.getCall(0).args[0]).to.eql(params);
+      });
+
+      it('passes only whitelisted headers', () => {
+        const headers = { authorization: 'Basic TEST' };
+        const request = { headers: Object.assign({}, headers, { foo: 'Foo' }) };
+
+        cluster.callWithRequest(request, 'nodes.info');
+
+        sinon.assert.calledOnce(client.nodes.info);
+        expect(client.nodes.info.getCall(0).args[0]).to.eql({
+          headers: headers
+        });
+      });
+
+      describe('wrap401Errors', () => {
+        let handler;
+        const error = new Error('Authentication required');
+        error.statusCode = 401;
+
+        beforeEach(() => {
+          handler = sinon.stub();
+        });
+
+        it('ensures WWW-Authenticate header', async () => {
+          set(client, 'mock.401', sinon.stub().returns(Promise.reject(error)));
+          await cluster.callWithRequest({}, 'mock.401', {}, { wrap401Errors: true }).catch(handler);
+
+          sinon.assert.calledOnce(handler);
+          expect(handler.getCall(0).args[0].output.headers['WWW-Authenticate']).to.eql('Basic realm="Authorization Required"');
+        });
+
+        it('persists WWW-Authenticate header', async () => {
+          set(error, 'body.error.header[WWW-Authenticate]', 'Basic realm="Test"');
+          set(client, 'mock.401', sinon.stub().returns(Promise.reject(error)));
+          await cluster.callWithRequest({}, 'mock.401', {}, { wrap401Errors: true }).catch(handler);
+
+          sinon.assert.calledOnce(handler);
+          expect(handler.getCall(0).args[0].output.headers['WWW-Authenticate']).to.eql('Basic realm="Test"');
+        });
+      });
+    });
+  });
+});

src/core_plugins/elasticsearch/lib/__tests__/create_admin_cluster.js

@@ -0,0 +1,66 @@
+import expect from 'expect.js';
+import sinon from 'sinon';
+import { bindKey, set, get, partial } from 'lodash';
+import { createAdminCluster } from '../create_admin_cluster';
+
+describe('plugins/elasticsearch', function () {
+  describe('create_admin_cluster', function () {
+    let cluster;
+    let server;
+
+    beforeEach(() => {
+      const config = {
+        elasticsearch: {
+          url: 'http://localhost:9200',
+          logQueries: true
+        }
+      };
+
+      server = sinon.spy();
+
+      cluster = {
+        close: sinon.spy()
+      };
+
+      set(server, 'plugins.elasticsearch.createCluster', sinon.mock().returns(cluster));
+      set(server, 'on', sinon.spy());
+
+      server.config = () => {
+        return { get: partial(get, config) };
+      };
+
+      createAdminCluster(server);
+    });
+
+    it('creates the cluster', () => {
+      const { createCluster } = server.plugins.elasticsearch;
+
+      sinon.assert.calledOnce(createCluster);
+      expect(createCluster.getCall(0).args[0]).to.eql('admin');
+      expect(createCluster.getCall(0).args[1].url).to.eql('http://localhost:9200');
+    });
+
+    it('sets client logger for cluster options', () => {
+      const { createCluster } = server.plugins.elasticsearch;
+      const firstCall = createCluster.getCall(0);
+      const Log = firstCall.args[1].log;
+      const logger = new Log;
+
+      sinon.assert.calledOnce(createCluster);
+      expect(firstCall.args[0]).to.eql('admin');
+      expect(firstCall.args[1].url).to.eql('http://localhost:9200');
+      expect(logger.tags).to.eql(['admin']);
+      expect(logger.logQueries).to.eql(true);
+    });
+
+    it('close cluster of server close', () => {
+      const clusterClose = server.on.getCall(0).args[1];
+
+      clusterClose();
+
+      sinon.assert.calledOnce(cluster.close);
+      sinon.assert.calledOnce(server.on);
+      expect(server.on.getCall(0).args[0]).to.eql('close');
+    });
+  });
+});

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

@@ -0,0 +1,50 @@
+import expect from 'expect.js';
+import { createClusters } from '../create_clusters';
+import { Cluster } from '../cluster';
+import sinon from 'sinon';
+import { partial } from 'lodash';
+
+describe('plugins/elasticsearch', function () {
+  describe('createClusters', function () {
+    let clusters;
+    let server;
+
+    beforeEach(() => {
+      server = {
+        plugins: {
+          elasticsearch: {}
+        },
+        expose: sinon.mock()
+      };
+
+      clusters = createClusters(server);
+    });
+
+    describe('createCluster', () => {
+      let cluster;
+      const config = {
+        url: 'http://localhost:9200',
+        ssl: {
+          verify: false
+        }
+      };
+
+      beforeEach(() => {
+        cluster = clusters.create('admin', config);
+      });
+
+      it('returns a cluster', () => {
+        expect(cluster).to.be.a(Cluster);
+      });
+
+      it('persists the cluster', () => {
+        expect(clusters.get('admin')).to.be.a(Cluster);
+      });
+
+      it('throws if cluster already exists', () => {
+        const fn = partial(clusters.create, 'admin', config);
+        expect(fn).to.throwException(/cluster \'admin\' already exists/);
+      });
+    });
+  });
+});