Reuse the destination clients across replication entries

A queue processor built a new ClientManager for every entry it
replicated, and another one each time a retry moved to the next host.
Each of them was left behind with its credentials cache, its keepAlive
agents and the sweep timer holding it, so a role was assumed again for
every single object.

Hold them on the queue processor, keyed by destination host and role, the
way the copy location tasks already share theirs, and release them on
shutdown. They assume their role on the destination STS, so they cannot
share the cache the copy location tasks read their sources with.

Issue: BB-877

Author: francoisferrand

extensions/replication/queueProcessor/QueueProcessor.js

@@ -229,10 +229,16 @@ class QueueProcessor extends EventEmitter {
         this.logger = new Logger(
             `Backbeat:Replication:QueueProcessor:${this.site}`);
 
-        // clients to read data straight from the sites we replicate to,
-        // keyed by endpoint and role, shared by all copy location tasks
+        // clients to read data straight from the sites we replicate to, keyed
+        // by endpoint and role, shared by all copy location tasks: they assume
+        // their role on the remote site's STS
         this.sourceClientManagers = {};
 
+        // clients to write to the destination site, keyed by host and role,
+        // shared by all replication tasks: they assume their role on the
+        // destination STS, so they cannot be shared with the ones above
+        this.destClientManagers = {};
+
         // global variables
         if (sourceConfig.transport === 'https') {
             this.sourceHTTPAgent = new HttpsAgent.Agent({
@@ -703,6 +709,7 @@ class QueueProcessor extends EventEmitter {
             vaultclientCache: this.vaultclientCache,
             accountCredsCache: this.accountCredsCache,
             sourceClientManagers: this.sourceClientManagers,
+            destClientManagers: this.destClientManagers,
             replicationStatusProducer: this.replicationStatusProducer,
             mProducer: this._mProducer,
             logger: this.logger,
@@ -866,11 +873,15 @@ class QueueProcessor extends EventEmitter {
                 return next();
             },
         ], err => {
-            // the tasks hold a reference to this map, so empty it in place
+            // the tasks hold a reference to these maps, so empty them in place
             Object.keys(this.sourceClientManagers).forEach(key => {
                 this.sourceClientManagers[key].close();
                 delete this.sourceClientManagers[key];
             });
+            Object.keys(this.destClientManagers).forEach(key => {
+                this.destClientManagers[key].close();
+                delete this.destClientManagers[key];
+            });
             return done(err);
         });
     }

extensions/replication/tasks/ReplicateObject.js

@@ -837,21 +837,26 @@ class ReplicateObject extends BackbeatTask {
         if (this.destConfig.auth.type === authTypeAssumeRole) {
             const accountId = _extractAccountIdFromRole(targetRole);
             const roleName = _extractRoleNameFromRole(targetRole);
-            this.clientManager = new ClientManager({
-                id: accountId,
-                authConfig: {
-                    type: authTypeAssumeRole,
-                    roleName,
-                    sts: this.destConfig.auth.sts,
-                },
-                s3Config: {
-                    host: this.destBackbeatHost.host,
-                    port: this.destBackbeatHost.port,
-                },
-                transport: this.destConfig.transport,
-            }, this.logger);
-            this.clientManager.initSTSConfig();
-            this.clientManager.initCredentialsManager();
+            const { host, port } = this.destBackbeatHost;
+            // one manager per destination host and role, shared by every entry:
+            // it holds the assumed-role credentials and the clients they opened
+            const cacheKey = `${host}:${port}::${targetRole}`;
+            this.clientManager = this.destClientManagers[cacheKey];
+            if (!this.clientManager) {
+                this.clientManager = new ClientManager({
+                    id: accountId,
+                    authConfig: {
+                        type: authTypeAssumeRole,
+                        roleName,
+                        sts: this.destConfig.auth.sts,
+                    },
+                    s3Config: { host, port },
+                    transport: this.destConfig.transport,
+                }, this.logger);
+                this.clientManager.initSTSConfig();
+                this.clientManager.initCredentialsManager();
+                this.destClientManagers[cacheKey] = this.clientManager;
+            }
             this.backbeatDest = this.clientManager.getBackbeatClient(accountId);
             return;
         }

tests/unit/replication/CopyLocationTask.spec.js

@@ -627,6 +627,24 @@ describe('CopyLocationTask', () => {
             assert(getBackbeatClient.alwaysCalledWith('123456789012'));
         });
 
+        it('should reuse the client manager across tasks', () => {
+            const sourceClientManagers = task.sourceClientManagers;
+            const otherEntry = new CopyLocationTask({
+                getStateVars: () => ({
+                    mProducer: { getProducer: () => {} },
+                    sourceConfig: { transport: 'http' },
+                    logger: fakeLogger,
+                    sourceClientManagers,
+                }),
+            });
+
+            const client1 = task._getAssumedRoleS3Client(siteConfig, roleArn, fakeLogger);
+            const client2 = otherEntry._getAssumedRoleS3Client(siteConfig, roleArn, fakeLogger);
+
+            assert.strictEqual(client1, client2);
+            assert.strictEqual(Object.keys(sourceClientManagers).length, 1);
+        });
+
         it('should default the port when the location carries none', () => {
             task._getAssumedRoleS3Client(
                 { ...siteConfig, endpoint: 'production.example.com' }, roleArn, fakeLogger);

tests/unit/replication/QueueProcessor.spec.js

@@ -335,6 +335,15 @@ describe('Queue Processor', () => {
         });
     });
 
+    describe('getStateVars', () => {
+        it('should share the client manager caches with the tasks', () => {
+            const stateVars = qp.getStateVars();
+
+            assert.strictEqual(stateVars.sourceClientManagers, qp.sourceClientManagers);
+            assert.strictEqual(stateVars.destClientManagers, qp.destClientManagers);
+        });
+    });
+
     describe('constructor', () => {
         it('should use s3c site\'s host as a destination host', () => {
             const config = getQueueProcessorConfig();
@@ -378,13 +387,17 @@ describe('Queue Processor', () => {
     });
 
     describe('stop', () => {
-        it('should close the cached source client managers', done => {
-            const close = sinon.stub();
-            qp.sourceClientManagers['http://site:8000::read-role'] = { close };
+        it('should close the cached source and destination client managers', done => {
+            const closeSource = sinon.stub();
+            const closeDest = sinon.stub();
+            qp.sourceClientManagers['http://site:8000::read-role'] = { close: closeSource };
+            qp.destClientManagers['site:8000::write-role'] = { close: closeDest };
 
             qp.stop(() => {
-                assert(close.calledOnce);
+                assert(closeSource.calledOnce);
+                assert(closeDest.calledOnce);
                 assert.deepStrictEqual(qp.sourceClientManagers, {});
+                assert.deepStrictEqual(qp.destClientManagers, {});
                 done();
             });
         });
@@ -393,7 +406,7 @@ describe('Queue Processor', () => {
             const consumerClosed = sinon.stub();
             const close = sinon.stub();
             qp._consumer = { close: cb => { consumerClosed(); cb(); } };
-            qp.sourceClientManagers['http://site:8000::read-role'] = { close };
+            qp.destClientManagers['site:8000::write-role'] = { close };
 
             qp.stop(() => {
                 assert(close.calledOnce);
@@ -402,14 +415,18 @@ describe('Queue Processor', () => {
             });
         });
 
-        it('should empty the client manager cache in place', done => {
-            // the tasks hold a reference to that same object
-            const cache = qp.sourceClientManagers;
-            cache['http://site:8000::read-role'] = { close: () => {} };
+        it('should empty the client manager caches in place', done => {
+            // the tasks hold a reference to those same objects
+            const sourceCache = qp.sourceClientManagers;
+            const destCache = qp.destClientManagers;
+            sourceCache['http://site:8000::read-role'] = { close: () => {} };
+            destCache['site:8000::write-role'] = { close: () => {} };
 
             qp.stop(() => {
-                assert.strictEqual(qp.sourceClientManagers, cache);
-                assert.deepStrictEqual(cache, {});
+                assert.strictEqual(qp.sourceClientManagers, sourceCache);
+                assert.strictEqual(qp.destClientManagers, destCache);
+                assert.deepStrictEqual(sourceCache, {});
+                assert.deepStrictEqual(destCache, {});
                 done();
             });
         });

tests/unit/replication/ReplicateObject.spec.js

@@ -73,6 +73,7 @@ describe('ReplicateObject', () => {
                         port: 80,
                     }),
                 },
+                destClientManagers: {},
                 logger: fakeLogger,
             }),
         });
@@ -1031,6 +1032,51 @@ describe('ReplicateObject', () => {
             });
         });
 
+        it('should reuse the client manager across entries for the same host and role', () => {
+            sinon.stub(ClientManager.prototype, 'initCredentialsManager').returns(null);
+            sinon.stub(ClientManager.prototype, 'getBackbeatClient').returns(null);
+            const destClientManagers = task.destClientManagers;
+            const otherEntry = new ReplicateObject({
+                getStateVars: () => ({ ...task, destClientManagers }),
+            });
+
+            task._setupDestClients('arn:aws:iam::123456789012:role/crr-role', fakeLogger);
+            otherEntry._setupDestClients('arn:aws:iam::123456789012:role/crr-role', fakeLogger);
+
+            assert.strictEqual(otherEntry.clientManager, task.clientManager);
+            assert.strictEqual(Object.keys(destClientManagers).length, 1);
+        });
+
+        it('should use a separate client manager per role', () => {
+            sinon.stub(ClientManager.prototype, 'initCredentialsManager').returns(null);
+            sinon.stub(ClientManager.prototype, 'getBackbeatClient').returns(null);
+
+            task._setupDestClients('arn:aws:iam::123456789012:role/crr-role', fakeLogger);
+            const first = task.clientManager;
+            task._setupDestClients('arn:aws:iam::210987654321:role/other-role', fakeLogger);
+
+            assert.notStrictEqual(task.clientManager, first);
+            assert.strictEqual(task.clientManager._id, '210987654321');
+            assert.strictEqual(Object.keys(task.destClientManagers).length, 2);
+        });
+
+        it('should not hand back the manager of the failed host when retrying', () => {
+            sinon.stub(ClientManager.prototype, 'initCredentialsManager').returns(null);
+            sinon.stub(ClientManager.prototype, 'getBackbeatClient').returns(null);
+            const role = 'arn:aws:iam::123456789012:role/crr-role';
+
+            task._setupDestClients(role, fakeLogger);
+            const failedHostManager = task.clientManager;
+
+            // what the retry hooks do: rotate the host, then set the clients up again
+            task.destHosts.pickHost = () => ({ host: 's3-2.zenko.local', port: 80 });
+            task._setupDestClients(role, fakeLogger);
+
+            assert.notStrictEqual(task.clientManager, failedHostManager);
+            assert.deepStrictEqual(task.clientManager._s3Config,
+                { host: 's3-2.zenko.local', port: 80 });
+        });
+
         it('should setup destination BackbeatClient with proper creds when not in assumeRole', async () => {
             task.destConfig.auth = {
                 type: 'service',