Release the credentials sweep timer held by ClientManager

The interval sweeping inactive credentials was started and never
stopped: it kept the event loop alive, so a process could not exit on
its own, and it kept the whole ClientManager reachable along with its
keepAlive agents and client caches.

The lifecycle processors meant to clear it, but did so on themselves
rather than on the ClientManager, which never had anything to clear it
with. Give the class a close() and call it from the services owning a
manager, once their consumers and producers are done with the clients.

Issue: BB-877

Author: francoisferrand

extensions/gc/GarbageCollector.js

@@ -184,8 +184,12 @@ class GarbageCollector extends EventEmitter {
     close(cb) {
         this._logger.debug('closing garbage collector consumer');
         if (this._consumer) {
-            this._consumer.close(cb);
+            this._consumer.close(err => {
+                this.clientManager.close();
+                cb(err);
+            });
         } else {
+            this.clientManager.close();
             cb();
         }
     }

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -529,10 +529,6 @@ class LifecycleBucketProcessor {
      * @return {undefined}
      */
     close(cb) {
-        if (this._deleteInactiveCredentialsInterval) {
-            clearInterval(this._deleteInactiveCredentialsInterval);
-        }
-
         async.parallel([
             done => {
                 this._log.debug('closing bucket tasks consumer');
@@ -542,7 +538,10 @@ class LifecycleBucketProcessor {
                 this._log.debug('closing producer');
                 this._producer.close(done);
             },
-        ], () => cb());
+        ], () => {
+            this.clientManager.close();
+            cb();
+        });
     }
 
     isReady() {

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -1205,6 +1205,7 @@ class LifecycleConductor {
                 });
             },
         ], err => {
+            this.clientManager.close();
             this._circuitBreaker.stop();
             this._started = false;
             return done(err);

extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js

@@ -142,13 +142,13 @@ class LifecycleObjectProcessor extends EventEmitter {
     close(cb) {
         this._log.debug('closing object tasks consumer');
 
-        if (this._deleteInactiveCredentialsInterval) {
-            clearInterval(this._deleteInactiveCredentialsInterval);
-        }
-
         if (this._consumers) {
-            this._consumers.close(cb);
+            this._consumers.close(err => {
+                this.clientManager.close();
+                cb(err);
+            });
         } else {
+            this.clientManager.close();
             cb();
         }
     }

extensions/replication/queueProcessor/QueueProcessor.js

@@ -865,7 +865,14 @@ class QueueProcessor extends EventEmitter {
                 });
                 return next();
             },
-        ], done);
+        ], err => {
+            // the tasks hold a reference to this map, so empty it in place
+            Object.keys(this.sourceClientManagers).forEach(key => {
+                this.sourceClientManagers[key].close();
+                delete this.sourceClientManagers[key];
+            });
+            return done(err);
+        });
     }
 
     /**

lib/clients/ClientManager.js

@@ -38,6 +38,7 @@ class ClientManager {
         }
 
         this._stsConfig = null;
+        this._deleteInactiveCredentialsInterval = null;
         this.s3Clients = {};
         this.backbeatClients = {};
         this.credentialsManager = new CredentialsManager(this._id, this._log);
@@ -71,6 +72,21 @@ class ClientManager {
         this._deleteInactiveCredentialsInterval = setInterval(() => {
             this.credentialsManager.removeInactiveCredentials(MAX_INACTIVE_DURATION);
         }, DELETE_INACTIVE_CREDENTIALS_INTERVAL);
+        // housekeeping only: it must not keep the process alive on its own
+        this._deleteInactiveCredentialsInterval.unref();
+    }
+
+    /**
+     * Release the resources held by this client manager. Safe to call
+     * whether or not initCredentialsManager() was called.
+     * @return {undefined}
+     */
+    close() {
+        if (this._deleteInactiveCredentialsInterval) {
+            clearInterval(this._deleteInactiveCredentialsInterval);
+            this._deleteInactiveCredentialsInterval = null;
+        }
+        this.credentialsManager.removeAllListeners('deleteCredentials');
     }
 
     /**

tests/unit/clients/ClientManager.spec.js

@@ -0,0 +1,77 @@
+const assert = require('assert');
+
+const ClientManager = require('../../../lib/clients/ClientManager');
+
+const fakeLogger = require('../../utils/fakeLogger');
+
+function createClientManager() {
+    return new ClientManager({
+        id: 'test-extension',
+        authConfig: { type: 'account', account: 'bart' },
+        s3Config: { host: 's3.zenko.local', port: 80 },
+        transport: 'http',
+    }, fakeLogger);
+}
+
+describe('ClientManager', () => {
+    let clientManager;
+
+    beforeEach(() => {
+        clientManager = createClientManager();
+    });
+
+    afterEach(() => {
+        clientManager.close();
+    });
+
+    describe('initCredentialsManager', () => {
+        it('should not let the credentials sweep hold the event loop open', () => {
+            clientManager.initCredentialsManager();
+
+            assert.strictEqual(
+                clientManager._deleteInactiveCredentialsInterval.hasRef(), false);
+        });
+
+        it('should drop the clients of credentials that got removed', () => {
+            clientManager.initCredentialsManager();
+            clientManager.s3Clients['123456789012'] = {};
+            clientManager.backbeatClients['123456789012'] = {};
+
+            clientManager.credentialsManager.emit('deleteCredentials', '123456789012');
+
+            assert.deepStrictEqual(clientManager.s3Clients, {});
+            assert.deepStrictEqual(clientManager.backbeatClients, {});
+        });
+    });
+
+    describe('close', () => {
+        it('should clear the credentials sweep', () => {
+            clientManager.initCredentialsManager();
+
+            clientManager.close();
+
+            assert.strictEqual(clientManager._deleteInactiveCredentialsInterval, null);
+        });
+
+        it('should stop listening for credentials removal', () => {
+            clientManager.initCredentialsManager();
+
+            clientManager.close();
+
+            assert.strictEqual(
+                clientManager.credentialsManager.listenerCount('deleteCredentials'), 0);
+        });
+
+        it('should do nothing when initCredentialsManager was never called', () => {
+            assert.doesNotThrow(() => clientManager.close());
+            assert.strictEqual(clientManager._deleteInactiveCredentialsInterval, null);
+        });
+
+        it('should be safe to call twice', () => {
+            clientManager.initCredentialsManager();
+
+            clientManager.close();
+            assert.doesNotThrow(() => clientManager.close());
+        });
+    });
+});

tests/unit/gc/GarbageCollector.spec.js

@@ -144,5 +144,35 @@ describe('garbage collector', function garbageCollector() {
                 done();
             });
         });
+
+        it('should close the client manager', done => {
+            const gcToClose = new GarbageCollector({
+                kafkaConfig: {},
+                s3Config: { host: 'localhost', port: 7777 },
+                gcConfig: {
+                    topic: 'backbeat-gc',
+                    auth: { type: 'account', account: 'bart' },
+                    consumer: { groupId: 'backbeat-gc-consumer-group' },
+                },
+            });
+            gcToClose.clientManager.initCredentialsManager();
+            let consumerClosed = false;
+            gcToClose._consumer = {
+                close: cb => {
+                    consumerClosed = true;
+                    cb();
+                },
+            };
+
+            gcToClose.close(err => {
+                assert.ifError(err);
+                assert.strictEqual(
+                    gcToClose.clientManager._deleteInactiveCredentialsInterval, null);
+                done();
+            });
+
+            // the consumer must stop using the clients before they are released
+            assert.strictEqual(consumerClosed, true);
+        });
     });
 });

tests/unit/lifecycle/LifecycleBucketProcessor.spec.js

@@ -340,4 +340,21 @@ describe('Lifecycle Bucket Processor', () => {
             });
         });
     });
+
+    describe('close', () => {
+        it('should close the client manager once the consumer and producer are closed', done => {
+            const closeStub = sinon.stub(lbp.clientManager, 'close');
+            const consumerClosed = sinon.stub();
+            const producerClosed = sinon.stub();
+            lbp._consumer = { close: cb => { consumerClosed(); cb(); } };
+            lbp._producer = { close: cb => { producerClosed(); cb(); } };
+
+            lbp.close(() => {
+                assert(closeStub.calledOnce);
+                assert(closeStub.calledAfter(consumerClosed));
+                assert(closeStub.calledAfter(producerClosed));
+                done();
+            });
+        });
+    });
 });

tests/unit/lifecycle/LifecycleConductor.spec.js

@@ -1277,4 +1277,19 @@ describe('Lifecycle Conductor', () => {
             });
         });
     });
+
+    describe('stop', () => {
+        it('should close the client manager once the producer is closed', done => {
+            const closeStub = sinon.stub(conductor.clientManager, 'close');
+            const producerClosed = sinon.stub();
+            conductor._producer = { close: cb => { producerClosed(); cb(); } };
+
+            conductor.stop(err => {
+                assert.ifError(err);
+                assert(closeStub.calledOnce);
+                assert(closeStub.calledAfter(producerClosed));
+                done();
+            });
+        });
+    });
 });