[es/clusters] improve cleanup

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

@@ -34,6 +34,13 @@ describe('plugins/elasticsearch', function () {
       sinon.assert.calledOnce(cluster._noAuthClient.close);
     });
 
+    it('closes clients created with createClient', () => {
+      const client = cluster.createClient();
+      sinon.stub(client, 'close');
+      cluster.close();
+      sinon.assert.calledOnce(client.close);
+    });
+
     it('protects the config from changes', () => {
       const localRequestHeadersWhitelist = cluster.getRequestHeadersWhitelist();
       expect(localRequestHeadersWhitelist.length).to.not.equal(config.requestHeadersWhitelist);

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

@@ -52,15 +52,5 @@ describe('plugins/elasticsearch', function () {
       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

@@ -1,8 +1,10 @@
 import expect from 'expect.js';
 import { createClusters } from '../create_clusters';
-import { Cluster } from '../cluster';
 import sinon from 'sinon';
 import { partial } from 'lodash';
+import Hapi from 'hapi';
+
+import * as ClusterNS from '../cluster';
 
 describe('plugins/elasticsearch', function () {
   describe('createClusters', function () {
@@ -14,7 +16,8 @@ describe('plugins/elasticsearch', function () {
         plugins: {
           elasticsearch: {}
         },
-        expose: sinon.mock()
+        expose: sinon.mock(),
+        on: sinon.stub(),
       };
 
       clusters = createClusters(server);
@@ -34,11 +37,11 @@ describe('plugins/elasticsearch', function () {
       });
 
       it('returns a cluster', () => {
-        expect(cluster).to.be.a(Cluster);
+        expect(cluster).to.be.a(ClusterNS.Cluster);
       });
 
       it('persists the cluster', () => {
-        expect(clusters.get('admin')).to.be.a(Cluster);
+        expect(clusters.get('admin')).to.be.a(ClusterNS.Cluster);
       });
 
       it('throws if cluster already exists', () => {
@@ -47,4 +50,32 @@ describe('plugins/elasticsearch', function () {
       });
     });
   });
+
+  describe('server stop', () => {
+    const sandbox = sinon.sandbox.create();
+
+    beforeEach(() => {
+      sandbox.stub(ClusterNS, 'Cluster', function () {
+        this.stub = true;
+        this.close = sinon.stub();
+      });
+    });
+
+    after(() => {
+      sandbox.restore();
+    });
+
+    it('closes all clusters', async () => {
+      const server = new Hapi.Server();
+      server.connection({ port: 0 });
+      const clusters = createClusters(server);
+      const cluster = clusters.create('name', { config: true });
+      expect(cluster).to.have.property('stub', true);
+      sinon.assert.notCalled(cluster.close);
+      await server.start();
+      sinon.assert.notCalled(cluster.close);
+      await server.stop();
+      sinon.assert.calledOnce(cluster.close);
+    });
+  });
 });

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

@@ -69,17 +69,5 @@ describe('plugins/elasticsearch', function () {
       expect(logger.tags).to.eql(['data']);
       expect(logger.logQueries).to.eql(true);
     });
-
-    it('close cluster of server close', () => {
-      createDataCluster(server);
-
-      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/cluster.js

@@ -13,6 +13,7 @@ export class Cluster {
     };
     this.errors = elasticsearch.errors;
 
+    this._clients = new Set();
     this._client = this.createClient();
     this._noAuthClient = this.createClient({ auth: false });
 
@@ -45,21 +46,22 @@ export class Cluster {
   getClient = () => this._client;
 
   close() {
-    if (this._client) {
-      this._client.close();
+    for (const client of this._clients) {
+      client.close();
     }
 
-    if (this._noAuthClient) {
-      this._noAuthClient.close();
-    }
+    this._clients.clear();
   }
 
   createClient = configOverrides => {
     const config = {
       ...this._getClientConfig(),
       ...configOverrides
     };
-    return new elasticsearch.Client(parseConfig(config));
+
+    const client = new elasticsearch.Client(parseConfig(config));
+    this._clients.add(client);
+    return client;
   }
 
   _getClientConfig = () => {

src/core_plugins/elasticsearch/lib/create_admin_cluster.js

@@ -1,4 +1,3 @@
-import { bindKey } from 'lodash';
 import { clientLogger } from './client_logger';
 
 export function createAdminCluster(server) {
@@ -10,13 +9,11 @@ export function createAdminCluster(server) {
     logQueries = config.get('elasticsearch.logQueries');
   }
 
-  const adminCluster = server.plugins.elasticsearch.createCluster(
+  server.plugins.elasticsearch.createCluster(
     'admin',
     {
       log: AdminClientLogging,
       ...config.get('elasticsearch')
     }
   );
-
-  server.on('close', bindKey(adminCluster, 'close'));
 }

src/core_plugins/elasticsearch/lib/create_clusters.js

@@ -1,22 +1,28 @@
 import { Cluster } from './cluster';
 
 export function createClusters(server) {
-  const esPlugin = server.plugins.elasticsearch;
-  esPlugin._clusters = esPlugin._clusters || new Map();
+  const clusters = new Map();
+
+  server.on('stop', () => {
+    for (const [name, cluster] of clusters) {
+      cluster.close();
+      clusters.delete(name);
+    }
+  });
 
   return {
     get(name) {
-      return esPlugin._clusters.get(name);
+      return clusters.get(name);
     },
 
     create(name, config) {
       const cluster = new Cluster(config);
 
-      if (esPlugin._clusters.has(name)) {
+      if (clusters.has(name)) {
         throw new Error(`cluster '${name}' already exists`);
       }
 
-      esPlugin._clusters.set(name, cluster);
+      clusters.set(name, cluster);
 
       return cluster;
     }

src/core_plugins/elasticsearch/lib/create_data_cluster.js

@@ -1,4 +1,3 @@
-import { bindKey } from 'lodash';
 import { clientLogger } from './client_logger';
 
 export function createDataCluster(server) {
@@ -18,13 +17,11 @@ export function createDataCluster(server) {
     return config.get('elasticsearch');
   }
 
-  const dataCluster = server.plugins.elasticsearch.createCluster(
+  server.plugins.elasticsearch.createCluster(
     'data',
     {
       log: DataClientLogging,
       ...getConfig()
     }
   );
-
-  server.on('close', bindKey(dataCluster, 'close'));
 }