Disconnect the redis client the metrics consumer opens

MetricsConsumer built its RedisClient as a local and handed it straight to
the StatsModel, so nothing could reach it afterwards: close() only tore down
the kafka consumer, and the socket stayed open for the life of the process.
Keep the client as a member and disconnect it alongside the consumer. Both
populators already close their metrics consumer, so there is nothing to wire
up on their side.

close() also dereferenced the consumer unconditionally, while start() only
assigns it on the ready event: closing during startup threw a TypeError and
left the caller's series waiting on a callback that never came. Guard it, and
still call back.

Issue: BB-882

Author: francoisferrand

lib/MetricsConsumer.js

@@ -43,8 +43,8 @@ class MetricsConsumer {
         this._consumer = null;
 
         this.logger = new Logger('Backbeat:MetricsConsumer');
-        const redisClient = new RedisClient(rConfig, this.logger);
-        this._statsClient = new StatsModel(redisClient, INTERVAL, EXPIRY);
+        this._redisClient = new RedisClient(rConfig, this.logger);
+        this._statsClient = new StatsModel(this._redisClient, INTERVAL, EXPIRY);
     }
 
     /**
@@ -230,7 +230,18 @@ class MetricsConsumer {
     }
 
     close(cb) {
-        this._consumer.close(cb);
+        return (next => {
+            // start() only sets the consumer once it reports ready, so a
+            // close racing the startup has nothing to tear down on the
+            // kafka side
+            if (!this._consumer) {
+                return process.nextTick(next);
+            }
+            return this._consumer.close(next);
+        })(err => {
+            this._redisClient.disconnect();
+            cb(err);
+        });
     }
 }
 

tests/unit/lib/MetricsConsumer.spec.js

@@ -0,0 +1,82 @@
+const assert = require('assert');
+const sinon = require('sinon');
+
+const MetricsConsumer = require('../../../lib/MetricsConsumer');
+
+// lazyConnect keeps ioredis from opening a socket for the tests that only
+// check the wiring
+const redisConfig = { host: 'localhost', port: 6379, lazyConnect: true };
+const mConfig = { topic: 'backbeat-metrics', groupIdPrefix: 'backbeat-metrics-group' };
+const kafkaConfig = { hosts: 'localhost:9092' };
+
+describe('MetricsConsumer', () => {
+    let mConsumer;
+
+    beforeEach(() => {
+        mConsumer = new MetricsConsumer(redisConfig, mConfig, kafkaConfig, 'crr');
+    });
+
+    afterEach(() => {
+        sinon.restore();
+    });
+
+    it('should keep the redis client it hands to the stats model', () => {
+        assert(mConsumer._redisClient);
+        assert.strictEqual(mConsumer._statsClient._redis, mConsumer._redisClient);
+    });
+
+    describe('close', () => {
+        let disconnect;
+
+        beforeEach(() => {
+            disconnect = sinon.stub(mConsumer._redisClient, 'disconnect');
+        });
+
+        it('should disconnect redis along with the kafka consumer', done => {
+            mConsumer._consumer = { close: sinon.stub().yields() };
+
+            mConsumer.close(err => {
+                assert.ifError(err);
+                sinon.assert.calledOnce(mConsumer._consumer.close);
+                sinon.assert.calledOnce(disconnect);
+                return done();
+            });
+        });
+
+        it('should disconnect redis and forward a kafka consumer error', done => {
+            const closeError = new Error('close failed');
+            mConsumer._consumer = { close: sinon.stub().yields(closeError) };
+
+            mConsumer.close(err => {
+                assert.strictEqual(err, closeError);
+                sinon.assert.calledOnce(disconnect);
+                return done();
+            });
+        });
+
+        it('should still call back when the consumer is not ready yet', done => {
+            assert.strictEqual(mConsumer._consumer, null);
+
+            assert.doesNotThrow(() => mConsumer.close(err => {
+                assert.ifError(err);
+                sinon.assert.calledOnce(disconnect);
+                return done();
+            }));
+        });
+    });
+
+    describe('close, on a live redis client', () => {
+        it('should leave the connection unusable', done => {
+            const consumer = new MetricsConsumer(
+                { host: 'localhost', port: 6379 }, mConfig, kafkaConfig, 'crr');
+
+            consumer.close(err => {
+                assert.ifError(err);
+                return consumer._redisClient.incrby('test-key', 1, err => {
+                    assert(err, 'expected the redis connection to be closed');
+                    return done();
+                });
+            });
+        });
+    });
+});