diff --git a/extensions/replication/queueProcessor/QueueProcessor.js b/extensions/replication/queueProcessor/QueueProcessor.js index 24edf7d05..5b2fddd58 100644 --- a/extensions/replication/queueProcessor/QueueProcessor.js +++ b/extensions/replication/queueProcessor/QueueProcessor.js @@ -229,6 +229,10 @@ 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 + this.sourceClientManagers = {}; + // global variables if (sourceConfig.transport === 'https') { this.sourceHTTPAgent = new HttpsAgent.Agent({ @@ -690,6 +694,7 @@ class QueueProcessor extends EventEmitter { destHTTPAgent: this.destHTTPAgent, vaultclientCache: this.vaultclientCache, accountCredsCache: this.accountCredsCache, + sourceClientManagers: this.sourceClientManagers, replicationStatusProducer: this.replicationStatusProducer, mProducer: this._mProducer, logger: this.logger, diff --git a/extensions/replication/tasks/CopyLocationTask.js b/extensions/replication/tasks/CopyLocationTask.js index b7d455e90..37af87519 100644 --- a/extensions/replication/tasks/CopyLocationTask.js +++ b/extensions/replication/tasks/CopyLocationTask.js @@ -6,7 +6,7 @@ const { ObjectMD } = models; const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); -const { +const { BackbeatRoutesClient, GetObjectCommand, MultipleBackendPutObjectCommand, @@ -24,8 +24,11 @@ const { getAccountCredentials } = require('../../../lib/credentials/AccountCredentials'); const RoleCredentials = require('../../../lib/credentials/RoleCredentials'); +const ClientManager = require('../../../lib/clients/ClientManager'); +const { authTypeAssumeRole } = require('../../../lib/constants'); const { metricsExtension, metricsTypeQueued, metricsTypeCompleted } = require('../constants'); +const locations = require('../../../conf/locationConfig.json') || {}; const MPU_GCP_MAX_PARTS = 1024; @@ -101,6 +104,154 @@ class CopyLocationTask extends BackbeatTask { .setSourceClient(log); } + /** + * Get a cached client on a remote site, authenticated with the + * assumed-role credentials for the role carried on a location part. + * @param {Object} siteConfig - transport, endpoint and STS of the source site + * @param {String} roleArn - the role ARN carried by the location part + * @param {Werelogs} log - the logger instance + * @return {BackbeatRoutesClient} the client + * @throws {ArsenalError} AccessDenied (retryable) if credentials + * could not be obtained for the role + */ + _getAssumedRoleS3Client(siteConfig, roleArn, log) { + const { transport, endpoint, sts } = siteConfig; + // a location may carry its servers without an explicit port + const [host, port = transport === 'https' ? 443 : 80] = endpoint.split(':'); + const s3Endpoint = `${transport}://${host}:${port}`; + const accountId = roleArn.split(':')[4]; + const roleName = roleArn.split(':role/')[1]; + // one manager per endpoint and role: it holds the credentials and + // clients of every account we assume that role on, and expires them + const cacheKey = `${s3Endpoint}::${roleName}`; + let clientManager = this.sourceClientManagers[cacheKey]; + if (!clientManager) { + clientManager = new ClientManager({ + id: 'replication-copy-location', + authConfig: { + type: authTypeAssumeRole, + roleName, + sts, + }, + s3Config: { host, port }, + transport, + }, this.logger); + clientManager.initSTSConfig(); + clientManager.initCredentialsManager(); + this.sourceClientManagers[cacheKey] = clientManager; + } + const client = clientManager.getBackbeatClient(accountId); + if (!client) { + log.error('unable to obtain assumed-role credentials for source location', { + method: 'CopyLocationTask._getAssumedRoleS3Client', + roleArn, + endpoint: s3Endpoint, + }); + const err = errors.AccessDenied.customizeDescription( + `unable to assume role ${roleArn} on source location`); + err.retryable = true; + throw err; + } + return client; + } + + /** + * Get a client to read the object data straight from the site holding + * it, for the locations Cloudserver has no data client for: the remote + * sites we replicate to over CRR, which are exactly the locations a + * clean room reads its source data from. + * @param {ObjectMD} objMD - metadata object + * @param {Werelogs} log - the logger instance + * @return {Object|null} the client and the location part to read, or + * null if the data is reachable through Cloudserver + * @throws {ArsenalError} if the site cannot be read from: no endpoint to + * reach it, no single part to read, no role to assume, or no credentials + * for that role + */ + _getSourceLocationClient(objMD, log) { + const site = objMD.getDataStoreName(); + if (!locations[site]?.isCRR) { + return null; + } + // The location itself carries how to reach the site holding the data: + // the servers to read from, and the STS to assume roles on. + const { servers, transport, sts } = locations[site].details || {}; + if (!servers?.length || !sts) { + log.error('no endpoint to reach source location', { + method: 'CopyLocationTask._getSourceLocationClient', + site, + }); + // retryable: a config fix should be picked up without losing the action + const err = errors.InternalError.customizeDescription( + `no endpoint to reach source location ${site}`); + err.retryable = true; + throw err; + } + // A CRR location holds the whole object in a single part, written by + // the source site's own replication: it carries where the data landed + // there, and the role to assume to read it back. Ranged reads below + // assume that, and would read the wrong bytes if it did not hold. + const parts = objMD.getLocation(); + if (parts?.length !== 1) { + log.error('unexpected number of location parts on source location', { + method: 'CopyLocationTask._getSourceLocationClient', + site, + parts: parts?.length ?? 0, + }); + throw errors.InternalError.customizeDescription( + `expected a single location part for source location ${site}`); + } + const part = parts[0]; + if (!part.role) { + log.error('missing role on location part for source location', { + method: 'CopyLocationTask._getSourceLocationClient', + site, + }); + const err = errors.AccessDenied.customizeDescription( + `missing role on location part for source location ${site}`); + err.retryable = true; + throw err; + } + const client = this._getAssumedRoleS3Client( + { transport, endpoint: servers[0], sts }, part.role, log); + const { bucket, key, dataStoreVersionId: version } = part; + return { client, target: { bucket, key, version }, dataStoreName: undefined }; + } + + /** + * Send a GetObject request for the object's data, either through the + * local Cloudserver, or straight to the remote site holding the data + * via an assumed role. A CRR location is Scality on both ends, so the + * same command is used either way: only the target differs, and the + * location constraint, which is meaningless on the site that owns the + * data. + * @param {ActionQueueEntry} actionEntry - the action entry + * @param {ObjectMD} objMD - metadata object + * @param {Object} [range] - byte range to request, or undefined for the whole object + * @param {Werelogs} log - the logger instance + * @param {AbortController} abortController - abort controller for the GET request + * @return {Promise} resolves to the GetObject response + */ + async _sendGetObject(actionEntry, objMD, range, log, abortController) { + // read from the site holding the data, where it is native and there + // is no location to redirect to, or locally through Cloudserver + const { client, target: { bucket, key, version }, dataStoreName } = + this._getSourceLocationClient(objMD, log) ?? { + client: this.backbeatClient, + target: actionEntry.getAttribute('target'), + dataStoreName: objMD.getDataStoreName(), + }; + const command = new GetObjectCommand({ + Bucket: bucket, + Key: key, + VersionId: version, + LocationConstraint: dataStoreName, + Range: range && `bytes=${range.start}-${range.end}`, + RequestUids: log.getSerializedUids(), + }); + return await client.send(command, { abortSignal: abortController.signal }); + } + processQueueEntry(actionEntry, kafkaEntry, done) { const startTime = Date.now(); const log = this.logger.newRequestLogger(); @@ -237,16 +388,8 @@ class CopyLocationTask extends BackbeatTask { let sourceStreamAborted = false; let abortedByPut = false; const abortController = new AbortController(); - const { bucket, key, version } = actionEntry.getAttribute('target'); - const getObjectCommand = new GetObjectCommand({ - Bucket: bucket, - Key: key, - VersionId: version, - LocationConstraint: objMD.getDataStoreName(), - RequestUids: log.getSerializedUids(), - }); - return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal }) + return this._sendGetObject(actionEntry, objMD, undefined, log, abortController) .then(response => { const incomingMsg = response.Body; incomingMsg.on('error', err => { @@ -307,6 +450,7 @@ class CopyLocationTask extends BackbeatTask { method: 'CopyLocationTask._getAndPutObjectOnce', peer: this.sourceConfig.s3, error: err.message, + errorName: err.name, httpStatus: err.$metadata?.httpStatusCode, }, actionEntry.getLogInfo())); return doneOnce(err); @@ -418,17 +562,8 @@ class CopyLocationTask extends BackbeatTask { } const abortController = new AbortController(); - const { bucket, key, version } = actionEntry.getAttribute('target'); - const getObjectCommand = new GetObjectCommand({ - Bucket: bucket, - Key: key, - VersionId: version, - Range: range && `bytes=${range.start}-${range.end}`, - LocationConstraint: objMD.getDataStoreName(), - RequestUids: log.getSerializedUids(), - }); - return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal }) + return this._sendGetObject(actionEntry, objMD, range, log, abortController) .then(response => this._putMPUPart(actionEntry, objMD, response.Body, size, uploadId, partNumber, log, abortController, done)) .catch(err => { @@ -439,6 +574,7 @@ class CopyLocationTask extends BackbeatTask { Object.assign({ method: 'CopyLocationTask._getRangeAndPutMPUPartOnce', error: err.message, + errorName: err.name, httpStatus: err.$metadata?.httpStatusCode, }, actionEntry.getLogInfo())); return done(err); diff --git a/tests/unit/replication/CopyLocationTask.spec.js b/tests/unit/replication/CopyLocationTask.spec.js index dcf9d8481..904488918 100644 --- a/tests/unit/replication/CopyLocationTask.spec.js +++ b/tests/unit/replication/CopyLocationTask.spec.js @@ -2,9 +2,11 @@ const assert = require('assert'); const sinon = require('sinon'); const CopyLocationTask = require('../../../extensions/replication/tasks/CopyLocationTask'); +const ClientManager = require('../../../lib/clients/ClientManager'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const { errors } = require('arsenal'); const { ObjectMD } = require('arsenal').models; +const locationConfig = require('../../../conf/locationConfig.json'); const fakeLogger = require('../../utils/fakeLogger'); @@ -299,4 +301,337 @@ describe('CopyLocationTask', () => { assert.strictEqual(task.retryParams.maxRetries, 13); }); }); + + describe('_sendGetObject', () => { + let task; + + // what the operator writes for a location we replicate to over CRR: + // the servers to reach it, and an STS to assume roles on it + const remoteSiteDetails = { + transport: 'https', + servers: ['production.example.com:443'], + sts: { + host: 'sts.production.example.com', + port: '443', + accessKey: 'AK', + secretKey: 'SK', + }, + }; + + const sourcePart = { + key: 'backups/vm001.vbk', + size: 1048576, + start: 0, + dataStoreName: 'source-site', + dataStoreType: 'aws_s3', + dataStoreETag: '1:9b2cf535f27731c974343645a3985328', + dataStoreVersionId: 'aJdO95zrzY5BKLXf9GHFItC0d1CkQ0Ei', + bucket: 'backup-repo-01', + role: 'arn:aws:iam::123456789012:role/clean-room-read', + }; + + function sourceObjectMD(location = [sourcePart]) { + const objMd = new ObjectMD(); + objMd.setDataStoreName('source-site'); + objMd.setKey('vm001.vbk'); + objMd.setLocation(location); + return objMd; + } + + beforeEach(() => { + locationConfig['source-site'] = + { type: 'aws_s3', isCRR: true, details: remoteSiteDetails }; + sinon.stub(fakeLogger, 'getSerializedUids').returns('req-uid-1'); + task = new CopyLocationTask({ + getStateVars: () => ({ + mProducer: { getProducer: () => {} }, + sourceConfig: { transport: 'http' }, + }), + }); + }); + + afterEach(() => { + delete locationConfig['source-site']; + sinon.restore(); + }); + + it('should read through Cloudserver when the location is not a CRR location', () => { + task.backbeatClient = { send: sinon.stub().resolves({ Body: 'stream' }) }; + + const entry = new ActionQueueEntry({ + target: { bucket: 'bucket', key: 'key', version: 'v1' }, + }); + const objMd = new ObjectMD(); + objMd.setDataStoreName('some-location'); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(response => { + assert.deepStrictEqual(response, { Body: 'stream' }); + assert(task.backbeatClient.send.calledOnce); + const command = task.backbeatClient.send.firstCall.args[0]; + assert.strictEqual(command.input.Bucket, 'bucket'); + assert.strictEqual(command.input.Key, 'key'); + assert.strictEqual(command.input.VersionId, 'v1'); + assert.strictEqual(command.input.LocationConstraint, 'some-location'); + assert.strictEqual(command.input.RequestUids, 'req-uid-1'); + }); + }); + + it('should read directly from the source location when it is a CRR location', () => { + const remoteClient = { send: sinon.stub().resolves({ Body: 'remote-stream' }) }; + sinon.stub(task, '_getAssumedRoleS3Client').returns(remoteClient); + + const entry = new ActionQueueEntry({ + target: { bucket: 'local-bucket', key: 'key', version: 'v1' }, + }); + + return task._sendGetObject(entry, sourceObjectMD(), undefined, fakeLogger, + new AbortController()) + .then(response => { + assert.deepStrictEqual(response, { Body: 'remote-stream' }); + assert(task._getAssumedRoleS3Client.calledOnce); + const [siteConfig, roleArn] = task._getAssumedRoleS3Client.firstCall.args; + assert.deepStrictEqual(siteConfig, { + transport: 'https', + endpoint: 'production.example.com:443', + sts: remoteSiteDetails.sts, + }); + assert.strictEqual(roleArn, 'arn:aws:iam::123456789012:role/clean-room-read'); + assert(remoteClient.send.calledOnce); + const command = remoteClient.send.firstCall.args[0]; + // bucket, key and version all describe the data on the + // source site, not the object we are copying + assert.strictEqual(command.input.Bucket, 'backup-repo-01'); + assert.strictEqual(command.input.Key, 'backups/vm001.vbk'); + assert.strictEqual(command.input.VersionId, 'aJdO95zrzY5BKLXf9GHFItC0d1CkQ0Ei'); + // the source site is Scality too: same command, and the + // request uids let us follow the read across both sites + assert.strictEqual(command.input.RequestUids, 'req-uid-1'); + // the data is native there, it must not be redirected + assert.strictEqual(command.input.LocationConstraint, undefined); + }); + }); + + it('should pass the range on to the source location', () => { + const remoteClient = { send: sinon.stub().resolves({}) }; + sinon.stub(task, '_getAssumedRoleS3Client').returns(remoteClient); + + const entry = new ActionQueueEntry({ target: {} }); + + return task._sendGetObject(entry, sourceObjectMD(), { start: 0, end: 99 }, + fakeLogger, new AbortController()) + .then(() => { + const command = remoteClient.send.firstCall.args[0]; + assert.strictEqual(command.input.Range, 'bytes=0-99'); + }); + }); + + it('should reject when the CRR location has no endpoint to reach it', () => { + // flagged isCRR, but carrying none of the details telling us + // where to read the data from + locationConfig['source-site'] = { type: 'aws_s3', isCRR: true, details: {} }; + task.backbeatClient = { send: sinon.stub() }; + + const entry = new ActionQueueEntry({ target: {} }); + + return task._sendGetObject(entry, sourceObjectMD(), undefined, fakeLogger, + new AbortController()) + .then(() => assert.fail('expected rejection')) + .catch(err => { + assert(err.InternalError); + assert.strictEqual(err.retryable, true); + assert(task.backbeatClient.send.notCalled); + }); + }); + + it('should reject when the source location holds more than one part', () => { + task.backbeatClient = { send: sinon.stub() }; + sinon.stub(task, '_getAssumedRoleS3Client'); + + const entry = new ActionQueueEntry({ target: {} }); + const objMd = sourceObjectMD([sourcePart, { ...sourcePart, start: 1048576 }]); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(() => assert.fail('expected rejection')) + .catch(err => { + assert(err.InternalError); + // the metadata will not change: retrying cannot help + assert.notStrictEqual(err.retryable, true); + assert(task.backbeatClient.send.notCalled); + assert(task._getAssumedRoleS3Client.notCalled); + }); + }); + + it('should reject without calling Cloudserver or the remote site when the role is missing', () => { + task.backbeatClient = { send: sinon.stub() }; + sinon.stub(task, '_getAssumedRoleS3Client'); + + const entry = new ActionQueueEntry({ target: {} }); + // no role: owner absent from the ownerId->role map + const objMd = sourceObjectMD([{ key: 'k', bucket: 'b', dataStoreName: 'source-site' }]); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(() => assert.fail('expected rejection')) + .catch(err => { + assert(err.AccessDenied); + assert.strictEqual(err.retryable, true); + assert(task.backbeatClient.send.notCalled); + assert(task._getAssumedRoleS3Client.notCalled); + }); + }); + }); + + describe('source version disappearing mid-copy', () => { + let task; + + function noSuchVersion() { + const err = new Error('The version does not exist.'); + err.name = 'NoSuchVersion'; + err.$metadata = { httpStatusCode: 404 }; + return err; + } + + beforeEach(() => { + task = new CopyLocationTask({ + getStateVars: () => ({ + site: 'test-site', + mProducer: { getProducer: () => {} }, + sourceConfig: { transport: 'http', s3: {} }, + }), + }); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should fail the copy, not skip it', done => { + sinon.stub(task, '_sendGetObject').rejects(noSuchVersion()); + const put = sinon.stub(task, '_sendMultipleBackendPutObject'); + + const entry = new ActionQueueEntry({ target: {} }); + const objMd = new ObjectMD(); + objMd.setContentLength(200); + + task._getAndPutObjectOnce(entry, objMd, fakeLogger, err => { + assert(err); + // an InvalidObjectState would be committed as a skip, leaving + // the object flagged in transition for good + assert(!err.InvalidObjectState); + assert(put.notCalled); + done(); + }); + }); + + it('should fail the copy, not skip it, while streaming an MPU part', done => { + sinon.stub(task, '_sendGetObject').rejects(noSuchVersion()); + const putPart = sinon.stub(task, '_putMPUPart'); + + const entry = new ActionQueueEntry({ target: {} }); + const objMd = new ObjectMD(); + + task._getRangeAndPutMPUPartOnce(entry, objMd, { start: 0, end: 99 }, 1, + 'upload-id', fakeLogger, err => { + assert(err); + assert(!err.InvalidObjectState); + assert(putPart.notCalled); + done(); + }); + }); + + it('should report the failure so the transition can be reset', () => { + const entry = new ActionQueueEntry({ target: {} }); + entry.setResultsTopic('backbeat-lifecycle-transition-tasks'); + task.replicationStatusProducer = { sendToTopic: sinon.stub() }; + task.dataMoverConsumer = { onEntryCommittable: sinon.stub() }; + + const res = task._publishCopyLocationStatus( + noSuchVersion(), entry, null, fakeLogger); + + // committing here would drop the entry before the lifecycle side + // resets transitionInProgress and counts the attempt + assert.strictEqual(res.committable, false); + assert(task.replicationStatusProducer.sendToTopic.calledOnce); + assert.strictEqual(task.replicationStatusProducer.sendToTopic.firstCall.args[0], + 'backbeat-lifecycle-transition-tasks'); + }); + }); + + describe('_getAssumedRoleS3Client', () => { + let task; + const siteConfig = { + transport: 'https', + endpoint: 'production.example.com:443', + sts: { + host: 'sts.production.example.com', + port: '443', + accessKey: 'AK', + secretKey: 'SK', + }, + }; + const roleArn = 'arn:aws:iam::123456789012:role/clean-room-read'; + + let getBackbeatClient; + + beforeEach(() => { + sinon.stub(ClientManager.prototype, 'initSTSConfig'); + sinon.stub(ClientManager.prototype, 'initCredentialsManager'); + getBackbeatClient = sinon.stub(ClientManager.prototype, 'getBackbeatClient') + .returns({ send: () => {} }); + task = new CopyLocationTask({ + getStateVars: () => ({ + mProducer: { getProducer: () => {} }, + sourceConfig: { transport: 'http' }, + logger: fakeLogger, + sourceClientManagers: {}, + }), + }); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should reuse the client manager for the same endpoint and role', () => { + const client1 = task._getAssumedRoleS3Client(siteConfig, roleArn, fakeLogger); + const client2 = task._getAssumedRoleS3Client(siteConfig, roleArn, fakeLogger); + + assert.strictEqual(client1, client2); + assert.strictEqual(Object.keys(task.sourceClientManagers).length, 1); + const clientManager = Object.values(task.sourceClientManagers)[0]; + assert.strictEqual(clientManager._transport, 'https'); + assert.deepStrictEqual(clientManager._s3Config, + { host: 'production.example.com', port: '443' }); + assert.deepStrictEqual(clientManager._authConfig.sts, siteConfig.sts); + assert(getBackbeatClient.alwaysCalledWith('123456789012')); + }); + + it('should default the port when the location carries none', () => { + task._getAssumedRoleS3Client( + { ...siteConfig, endpoint: 'production.example.com' }, roleArn, fakeLogger); + + const clientManager = Object.values(task.sourceClientManagers)[0]; + assert.deepStrictEqual(clientManager._s3Config, + { host: 'production.example.com', port: 443 }); + }); + + it('should log and throw a retryable AccessDenied when credentials cannot be obtained', () => { + getBackbeatClient.returns(null); + const logSpy = sinon.spy(fakeLogger, 'error'); + + assert.throws( + () => task._getAssumedRoleS3Client(siteConfig, roleArn, fakeLogger), + err => err.AccessDenied && err.retryable === true); + assert(logSpy.calledOnce); + }); + + it('should keep the full role name, including any path, when the role ARN has one', () => { + const pathedRoleArn = 'arn:aws:iam::123456789012:role/service-role/clean-room-read'; + + task._getAssumedRoleS3Client(siteConfig, pathedRoleArn, fakeLogger); + + const clientManager = Object.values(task.sourceClientManagers)[0]; + assert.strictEqual(clientManager._authConfig.roleName, 'service-role/clean-room-read'); + }); + }); });