diff --git a/conf/locationConfig.json b/conf/locationConfig.json index 8ba3dc334..dceb31dda 100644 --- a/conf/locationConfig.json +++ b/conf/locationConfig.json @@ -42,5 +42,12 @@ "legacyAwsBehavior": false, "isCold": true, "details": {} + }, + "location-crr-source": { + "type": "scality", + "objectId": "location-crr-source", + "legacyAwsBehavior": false, + "isCRR": true, + "details": {} } } diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 883146ff6..42aea6bde 100755 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -154,6 +154,16 @@ if [[ "$EXTENSIONS_REPLICATION_DEST_BOOTSTRAPLIST" ]]; then fi fi +# Clean room: localize objects whose data still lives on the source (isCRR) +# location. Setting the target location enables the trigger. +if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION" ]]; then + JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.toLocation=\"$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION\"" +fi + +if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC" ]]; then + JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.resultsTopic=\"$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC\"" +fi + # START Retry config # AWS_S3 diff --git a/extensions/lifecycle/LifecycleConfigValidator.js b/extensions/lifecycle/LifecycleConfigValidator.js index b6487862e..0d5a1c1a0 100644 --- a/extensions/lifecycle/LifecycleConfigValidator.js +++ b/extensions/lifecycle/LifecycleConfigValidator.js @@ -71,6 +71,7 @@ const joiSchema = joi.object({ concurrency: joi.number().greater(0).default(10), maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT), probeServer: probeServerJoi.default(), + vaultAdmin: hostPortJoi, circuitBreaker: joi.object().optional(), }, coldStorageArchiveTopicPrefix: joi.string().default('cold-archive-req-'), diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 794449bb5..5352e58b6 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -2,11 +2,15 @@ const { EventEmitter } = require('events'); const Logger = require('werelogs').Logger; +const { errors } = require('arsenal'); const BackbeatConsumerManager = require('../../../lib/BackbeatConsumerManager'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const ClientManager = require('../../../lib/clients/ClientManager'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); +const VaultClientWrapper = require('../../utils/VaultClientWrapper'); +const { AccountIdCache } = require('../../utils/AccountIdCache'); +const { authTypeAssumeRole } = require('../../../lib/constants'); const logIdFromType = { 'object-processor': 'Backbeat:Lifecycle:ObjectProcessor', @@ -61,9 +65,81 @@ class LifecycleObjectProcessor extends EventEmitter { transport, }, this._log); + this.vaultClientWrapper = new VaultClientWrapper( + `lifecycle:${this.getProcessorType()}`, + this._processConfig.vaultAdmin, + this.getAuthConfig(this._lcConfig), + this._log, + ); + this._accountIdCache = new AccountIdCache( + this._processConfig.concurrency); + this.retryWrapper = new BackbeatTask(this._processConfig.retry); } + /** + * Whether this processor can resolve canonical ids through Vault. Only the + * transition processor receives actions published without an account id + * (clean room localization), and only it is configured with a Vault admin + * endpoint - so leave the Vault client alone everywhere else. + * @return {Boolean} true if account id lookups are available + */ + _accountIdLookupEnabled() { + const authConfig = this.getAuthConfig(this._lcConfig); + return authConfig.type === authTypeAssumeRole && + !!(this._processConfig.vaultAdmin || authConfig.vault); + } + + /** + * Resolve the account id of a canonical id. Actions published by the + * lifecycle conductor already carry the account id; those published by the + * queue populator (clean room localization) only know the canonical id. + * @param {String} ownerId - canonical id of the object owner + * @param {Logger} log - logger instance + * @param {Function} cb - callback: cb(err, accountId) + * @return {undefined} + */ + getAccountId(ownerId, log, cb) { + if (this.getAuthConfig(this._lcConfig).type !== authTypeAssumeRole) { + log.debug('skipping: not assume role auth type'); + return process.nextTick(cb); + } + + if (!this._accountIdLookupEnabled()) { + log.error('cannot resolve canonical id: no vault endpoint configured'); + return process.nextTick(cb, errors.InternalError.customizeDescription( + 'account id resolution requires a vault endpoint')); + } + + // A cached miss must fail like a fresh lookup would: `isKnown()` is also + // true for misses, and `get()` would then hand back `undefined`. + if (this._accountIdCache.isMiss(ownerId)) { + log.error('canonical id does not exist (cached)', { ownerId }); + return process.nextTick(cb, errors.NoSuchEntity); + } + + if (this._accountIdCache.has(ownerId)) { + return process.nextTick(cb, null, this._accountIdCache.get(ownerId)); + } + + return this.vaultClientWrapper.getAccountId(ownerId, (err, accountId) => { + if (err) { + if (err.NoSuchEntity) { + log.error('canonical id does not exist', { error: err, ownerId }); + this._accountIdCache.miss(ownerId); + } else { + log.error('could not get account id', { error: err, ownerId }); + } + return cb(err); + } + + this._accountIdCache.set(ownerId, accountId); + this._accountIdCache.expireOldest(); + + return cb(null, accountId); + }); + } + getProcessorType() { return 'object-processor'; } @@ -130,6 +206,9 @@ class LifecycleObjectProcessor extends EventEmitter { start(done) { this.clientManager.initSTSConfig(); this.clientManager.initCredentialsManager(); + if (this._accountIdLookupEnabled()) { + this.vaultClientWrapper.init(); + } this._setupConsumers(done); } @@ -225,12 +304,15 @@ class LifecycleObjectProcessor extends EventEmitter { this.clientManager.getBackbeatClient.bind(this.clientManager), getBackbeatMetadataProxy: this.clientManager.getBackbeatMetadataProxy.bind(this.clientManager), + getAccountId: this.getAccountId.bind(this), logger: this._log, }; } isReady() { - return this._consumers && this._consumers.isReady(); + return this._consumers && this._consumers.isReady() && + (!this._accountIdLookupEnabled() || + this.vaultClientWrapper.tempCredentialsReady()); } } diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a8311840..6bf7556f8 100644 --- a/extensions/lifecycle/tasks/LifecycleTask.js +++ b/extensions/lifecycle/tasks/LifecycleTask.js @@ -31,6 +31,8 @@ const errorTransitionInProgress = errors.InternalError. customizeDescription('transition is currently in progress'); const errorTransitionColdObject = errors.InternalError. customizeDescription('transitioning a cold object is forbidden'); +const errorTransitionNonLocalizedObject = errors.InternalError. + customizeDescription('transitioning a non-localized object is forbidden'); const errorObjectTemporarilyRestored = errors.InternalError. customizeDescription('object temporarily restored'); const errorReplicationInProgress = errors.InternalError. @@ -1270,12 +1272,21 @@ class LifecycleTask extends BackbeatTask { return next(errorReplicationInProgress); } const dataStoreName = objectMD.getDataStoreName(); - const isObjectCold = dataStoreName && locationsConfig[dataStoreName] - && locationsConfig[dataStoreName].isCold; + const locationConfig = (dataStoreName + && locationsConfig[dataStoreName]) || {}; // We do not transition cold objects - if (isObjectCold) { + if (locationConfig.isCold) { return next(errorTransitionColdObject); } + // Clean room: the object data still lives on the source + // (isCRR) location. Localization is the only valid transition + // out of such a location, and it is triggered by its own path + // (the queue populator), not by lifecycle rules. The version + // is simply re-evaluated on a later scan, once localization + // has completed. + if (locationConfig.isCRR) { + return next(errorTransitionNonLocalizedObject); + } // If transition is in progress, do not re-publish entry // to data-mover or cold-archive topic. if (objectMD.getTransitionInProgress()) { diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 4f57c33c7..2fe958e04 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -252,6 +252,36 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { ], done); } + /** + * Actions published by the lifecycle conductor carry the account id; + * those published by the queue populator (clean room localization) only + * know the object owner's canonical id. Resolve it once, up-front, so the + * rest of the task - and the garbage collection entry it emits - can use + * `target.accountId` as usual. + * @param {ActionQueueEntry} entry - action entry to execute + * @param {Logger} log - logger instance + * @param {Function} cb - callback function + * @return {undefined} + */ + _resolveAccountId(entry, log, cb) { + const { accountId, owner } = this.getTargetAttribute(entry); + if (accountId || !owner) { + return process.nextTick(cb); + } + + log.debug('no account id in entry, resolving from canonical id', + { owner }); + return this.getAccountId(owner, log, (err, resolvedAccountId) => { + if (err) { + return cb(err); + } + if (resolvedAccountId) { + entry.setAttribute('target.accountId', resolvedAccountId); + } + return cb(); + }); + } + /** * * @param {ActionQueueEntry} entry - action entry to execute @@ -268,11 +298,17 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { lastModified: 'target.lastModified', }); log.addDefaultFields(entry.getLogInfo()); - if (entry.getStatus() === 'success') { - return this.handleSuccessfullTransition(entry, log, done); - } - return this.handleFailedTransition(entry, log, done); + return this._resolveAccountId(entry, log, err => { + if (err) { + return done(err); + } + if (entry.getStatus() === 'success') { + return this.handleSuccessfullTransition(entry, log, done); + } + + return this.handleFailedTransition(entry, log, done); + }); } } diff --git a/extensions/replication/ReplicationConfigValidator.js b/extensions/replication/ReplicationConfigValidator.js index 8f748c807..47d1c2663 100644 --- a/extensions/replication/ReplicationConfigValidator.js +++ b/extensions/replication/ReplicationConfigValidator.js @@ -139,6 +139,13 @@ const joiSchema = joi.object({ probeServer: probeServerPerSite, }).optional(), objectSizeMetrics: joi.array().items(joi.number()).default(OBJECT_SIZE_METRICS), + // Clean room: localization of objects whose data still lives on the source + // (isCRR) location. Enabled by setting `toLocation`. + localization: joi.object({ + toLocation: joi.string().required(), + resultsTopic: joi.string() + .default('backbeat-lifecycle-transition-tasks'), + }).optional(), }); /** diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index 22c31ad0b..9d9d89907 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -1,18 +1,25 @@ const { isMasterKey } = require('arsenal').versioning; +const { encode } = require('arsenal').versioning.VersionID; const { usersBucket, mpuBucketPrefix } = require('arsenal').constants; const QueuePopulatorExtension = require('../../lib/queuePopulator/QueuePopulatorExtension'); const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry'); +const ReplicationAPI = require('./ReplicationAPI'); const locationsConfig = require('../../conf/locationConfig.json') || {}; const safeJsonParse = require('../../lib/util/safeJsonParse'); const { traceHeadersFromEntry } = require('arsenal/build/lib/tracing').kafka; +const TRANSITION_ATTEMPT_MD = 'x-amz-meta-scal-s3-transition-attempt'; + class ReplicationQueuePopulator extends QueuePopulatorExtension { constructor(params) { super(params); this.repConfig = params.config; this.metricsHandler = params.metricsHandler; + // Clean room: when set, objects whose data still lives on the source + // (isCRR) location are queued for localization instead of replication. + this.localizationConfig = params.config.localization; } filter(entry) { @@ -73,6 +80,19 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (sanityCheckRes) { return; } + const dataStoreName = queueEntry.getDataStoreName(); + const locationConfig = (dataStoreName && locationsConfig[dataStoreName]) + || {}; + // Clean room: the object data still lives on the source (isCRR) + // location and first needs to be localized. This is unrelated to + // replicationInfo, which tracks replication of a *local* object to + // remote sites, hence the check before any replication condition. + if (locationConfig.isCRR) { + if (this.localizationConfig) { + this._publishLocalizationAction(entry, queueEntry, value); + } + return; + } // Allow a non-versioned object if being replicated from an NFS bucket. // Or if the master key is of a non versioned object if (!this._entryCanBeReplicated(queueEntry)) { @@ -81,11 +101,8 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (queueEntry.getReplicationStatus() !== 'PENDING') { return; } - const dataStoreName = queueEntry.getDataStoreName(); - const isObjectCold = dataStoreName && locationsConfig[dataStoreName] - && locationsConfig[dataStoreName].isCold; // We do not replicate cold objects. - if (isObjectCold) { + if (locationConfig.isCold) { return; } @@ -124,6 +141,119 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { traceHeaders); } + /** + * Queue a copyLocation action for an object whose data still lives on the + * source (isCRR) location, so the data mover copies it to the local + * location and the transition processor merges the new location back into + * the object metadata. + * + * Duplicates are expected (and harmless): the same object may show up + * several times in the oplog, and the copy is idempotent. + * + * @param {Object} entry - raw metadata log entry + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @param {Object} value - parsed entry metadata + * @return {undefined} + */ + _publishLocalizationAction(entry, queueEntry, value) { + // Clean room buckets are versioned: the master key is repaired by the + // metadata layer once the version has been localized. + if (isMasterKey(queueEntry.getObjectVersionedKey())) { + return; + } + if (queueEntry.getIsDeleteMarker()) { + return; + } + const locations = queueEntry.getLocation(); + if (!locations || locations.length === 0) { + // Empty objects hold no data, there is nothing to localize. Any + // other object without location information is inconsistent. + if (queueEntry.getContentLength() > 0) { + this.log.error( + 'non-empty object without location, skipping localization', + { + method: 'ReplicationQueuePopulator.' + + '_publishLocalizationAction', + ...queueEntry.getLogInfo(), + dataStoreName: queueEntry.getDataStoreName(), + contentLength: queueEntry.getContentLength(), + }); + } + return; + } + + const bucket = queueEntry.getBucket(); + const objectKey = queueEntry.getObjectKey(); + const contentLength = queueEntry.getContentLength(); + const action = ReplicationAPI.createCopyLocationAction({ + bucketName: bucket, + objectKey, + owner: queueEntry.getOwnerId(), + versionId: value.versionId ? encode(value.versionId) : undefined, + eTag: `"${queueEntry.getContentMd5()}"`, + lastModified: queueEntry.getLastModified(), + toLocation: this.localizationConfig.toLocation, + originLabel: 'localization', + fromLocation: queueEntry.getDataStoreName(), + contentLength, + resultsTopic: this.localizationConfig.resultsTopic, + transitionTime: new Date( + entry.overheadFields?.commitTimestamp ?? Date.now() + ).toISOString(), + attempt: this._getTransitionAttempt(queueEntry), + }); + // 'transition' is what the lifecycle transition processor dispatches + // on to pick up the copyLocation result. + action.addContext({ + origin: 'localization', + ruleType: 'transition', + bucketName: bucket, + objectKey, + versionId: value.versionId, + }); + action.setAttribute('source', { + bucket, + objectKey, + storageClass: queueEntry.getDataStoreName(), + }); + + this.metricsHandler.localizationBytes( + entry.logReader.getMetricLabels(), + contentLength + ); + this.metricsHandler.localizationObjects( + entry.logReader.getMetricLabels() + ); + + this.log.trace('publishing object localization entry', + { entry: queueEntry.getLogInfo() }); + this.publish(ReplicationAPI.getDataMoverTopic(), + `${bucket}/${objectKey}`, + action.toKafkaMessage(), + undefined, + traceHeadersFromEntry(value)); + } + + /** + * Number of times the data mover already tried to copy this object. The + * transition processor bumps the counter on failure, which produces a new + * oplog entry and re-triggers the copy. + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @return {Number|undefined} attempt count, if any + */ + _getTransitionAttempt(queueEntry) { + const umd = queueEntry.getUserMetadata(); + if (!umd) { + return undefined; + } + const { error, result } = safeJsonParse(umd); + if (error) { + return undefined; + } + const attempt = Number.parseInt(result[TRANSITION_ATTEMPT_MD], 10); + return Number.isInteger(attempt) ? attempt : undefined; + } + /** * Filter if the entry is considered a valid master key entry. * There is a case where a single null entry looks like a master key and diff --git a/lib/Config.js b/lib/Config.js index 57a9aaa88..4a1d18646 100644 --- a/lib/Config.js +++ b/lib/Config.js @@ -89,6 +89,9 @@ class Config extends EventEmitter { if (backbeatSupportsTransition && !replicationConfig.dataMoverTopic) { throw new Error('dataMoverTopic is required when lifecycle transitions is supported'); } + if (replicationConfig?.localization && !replicationConfig.dataMoverTopic) { + throw new Error('dataMoverTopic is required when localization is enabled'); + } const destination = parsedConfig.extensions?.replication?.destination; this.bootstrapList = destination?.bootstrapList?.map(endpoint => { diff --git a/lib/queuePopulator/QueuePopulator.js b/lib/queuePopulator/QueuePopulator.js index 8b63e3d05..d2a716bf1 100644 --- a/lib/queuePopulator/QueuePopulator.js +++ b/lib/queuePopulator/QueuePopulator.js @@ -79,12 +79,26 @@ const notificationEvent = ZenkoMetrics.createCounter({ help: 'Total number of oplog events processed by notification extension', }); +const localizationObjectMetrics = ZenkoMetrics.createCounter({ + name: 's3_backbeat_populator_localization_objects_total', + help: 'Total objects queued for clean room localization', + labelNames: metricLabels, +}); + +const localizationByteMetrics = ZenkoMetrics.createCounter({ + name: 's3_backbeat_populator_localization_bytes_total', + help: 'Total number of bytes queued for clean room localization', + labelNames: metricLabels, +}); + /** * Contains methods to incrememt different metrics * @typedef {Object} MetricsHandler * @property {CounterInc} messages - Increments the message metric * @property {CounterInc} objects - Increments the objects metric * @property {CounterInc} bytes - Increments the bytes metric + * @property {CounterInc} localizationObjects - Increments the localized objects metric + * @property {CounterInc} localizationBytes - Increments the localized bytes metric * @property {GaugeSet} logReadOffset - Set the log read offset metric * @property {GaugeSet} logSize - Set the log size metric */ @@ -92,6 +106,8 @@ const metricsHandler = { messages: wrapCounterInc(messageMetrics, {}), objects: wrapCounterInc(objectMetrics, {}), bytes: wrapCounterInc(byteMetrics, {}), + localizationObjects: wrapCounterInc(localizationObjectMetrics, {}), + localizationBytes: wrapCounterInc(localizationByteMetrics, {}), logReadOffset: wrapGaugeSet(logReadOffsetMetric, {}), logSize: wrapGaugeSet(logSizeMetric, {}), logTimestamp: wrapGaugeSet(logTimestamp, {}), diff --git a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js index 56e825d45..c8c8737b6 100644 --- a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js +++ b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js @@ -436,6 +436,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], @@ -493,6 +498,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa5138851..faa903283 100644 --- a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js +++ b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js @@ -1,5 +1,6 @@ const assert = require('assert'); const sinon = require('sinon'); +const { errors } = require('arsenal'); const config = require('../../config.json'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const LifecycleObjectTransitionProcessor = @@ -125,4 +126,126 @@ describe('LifecycleObjectTransitionProcessor', () => { }); }); }); + + describe('getAccountId', () => { + const ownerId = 'canonical-id-1'; + const accountId = '834789881858'; + let processor; + let log; + + beforeEach(() => { + processor = new LifecycleObjectTransitionProcessor( + config.zookeeper, + config.kafka, + { + ...config.extensions.lifecycle, + transitionProcessor: { + ...config.extensions.lifecycle.transitionProcessor, + auth: { type: 'assumeRole', roleName: 'role' }, + vaultAdmin: { host: 'localhost', port: 8600 }, + }, + }, + config.s3, + ); + log = { debug: () => {}, error: () => {} }; + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should skip the lookup when auth type is not assume role', done => { + const spy = sinon.spy(objectProcessor.vaultClientWrapper, 'getAccountId'); + objectProcessor.getAccountId(ownerId, log, (err, id) => { + assert.ifError(err); + assert.strictEqual(id, undefined); + assert.strictEqual(spy.callCount, 0); + done(); + }); + }); + + it('should resolve through vault and cache the result', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(null, accountId); + + processor.getAccountId(ownerId, log, (err, id) => { + assert.ifError(err); + assert.strictEqual(id, accountId); + assert.strictEqual(stub.callCount, 1); + + processor.getAccountId(ownerId, log, (err2, id2) => { + assert.ifError(err2); + assert.strictEqual(id2, accountId); + assert.strictEqual(stub.callCount, 1); + done(); + }); + }); + }); + + it('should fail on a cached miss instead of returning no account id', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(errors.NoSuchEntity); + + processor.getAccountId(ownerId, log, err => { + assert(err.NoSuchEntity); + assert.strictEqual(stub.callCount, 1); + + // the miss is cached, but must still surface as an error + processor.getAccountId(ownerId, log, (err2, id2) => { + assert(err2.NoSuchEntity); + assert.strictEqual(id2, undefined); + assert.strictEqual(stub.callCount, 1); + done(); + }); + }); + }); + + it('should propagate other vault errors without caching them', done => { + const stub = sinon.stub(processor.vaultClientWrapper, 'getAccountId') + .yields(errors.InternalError); + + processor.getAccountId(ownerId, log, err => { + assert(err.InternalError); + + processor.getAccountId(ownerId, log, err2 => { + assert(err2.InternalError); + assert.strictEqual(stub.callCount, 2); + done(); + }); + }); + }); + + it('should not touch vault when no vault endpoint is configured', done => { + // assume role auth, but no vaultAdmin and no auth.vault: the + // expiration processor is deployed this way, and must neither + // start a vault client nor be held back by its readiness. + const noVault = new LifecycleObjectTransitionProcessor( + config.zookeeper, + config.kafka, + { + ...config.extensions.lifecycle, + auth: { type: 'assumeRole', roleName: 'role', sts: {} }, + transitionProcessor: { + ...config.extensions.lifecycle.transitionProcessor, + vaultAdmin: undefined, + }, + }, + config.s3, + ); + const spy = sinon.spy(noVault.vaultClientWrapper, 'getAccountId'); + + assert.strictEqual(noVault._accountIdLookupEnabled(), false); + // readiness must not wait on credentials that are never fetched + assert.strictEqual( + noVault.vaultClientWrapper.tempCredentialsReady(), false); + noVault._consumers = { isReady: () => true }; + assert.strictEqual(noVault.isReady(), true); + + noVault.getAccountId(ownerId, log, err => { + assert(err.InternalError); + assert.strictEqual(spy.callCount, 0); + done(); + }); + }); + }); }); diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe14..ae46563b2 100644 --- a/tests/unit/lifecycle/LifecycleTask.spec.js +++ b/tests/unit/lifecycle/LifecycleTask.spec.js @@ -11,6 +11,7 @@ const LifecycleTask = require( const LifecycleTaskV2 = require( '../../../extensions/lifecycle/tasks/LifecycleTaskV2'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics'); const fakeLogger = require('../../utils/fakeLogger'); const { withActiveSpan } = require('../../utils/withActiveSpan'); @@ -2473,6 +2474,89 @@ describe('lifecycle task helper methods', () => { }); }); + describe('_applyTransitionRule', () => { + const CRR_LOCATION = 'location-crr-source'; + const testParams = { + bucket: 'test-bucket', + owner: 'test-owner', + objectKey: 'test-key', + versionId: 'test-version-id', + eTag: '"test-etag"', + lastModified: '2023-01-01T00:00:00.000Z', + site: 'test-site', + accountId: 'test-account-id', + transitionTime: Date.now(), + bucketData: { + target: { + bucket: 'test-bucket', + owner: 'test-owner', + accountId: 'test-account-id', + }, + }, + }; + + let lifecycleTask; + let objectMD; + let sendDataMoverAction; + let putObjectMD; + + function setupObjectMD(dataStoreName) { + objectMD = { + getReplicationStatus: () => 'COMPLETED', + getDataStoreName: () => dataStoreName, + getDataStoreVersionId: () => 'version-123', + getTransitionInProgress: () => false, + getArchive: () => undefined, + getContentLength: () => 1024, + getUserMetadata: () => null, + setTransitionInProgress: sinon.spy(), + setOriginOp: sinon.spy(), + getSerialized: () => '{}', + }; + sinon.stub(lifecycleTask, '_getObjectMD') + .callsFake((params, log, cb) => cb(null, objectMD)); + } + + beforeEach(() => { + lifecycleTask = new LifecycleTask(lp); + lifecycleTask.pausedLocations = new Set(); + lifecycleTask.circuitBreakers = { tripped: () => false }; + lifecycleTask.producer = {}; + lifecycleTask.transitionTasksTopic = 'test-transition-topic'; + sendDataMoverAction = sinon.stub( + ReplicationAPI, 'sendDataMoverAction') + .callsFake((producer, entry, log, cb) => cb()); + putObjectMD = sinon.stub(lifecycleTask, '_putObjectMD') + .callsFake((params, log, cb) => cb()); + }); + + it('should not transition an object still on an isCRR location', + done => { + setupObjectMD(CRR_LOCATION); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.strictEqual(err.description, + 'transitioning a non-localized object is forbidden'); + sinon.assert.notCalled(sendDataMoverAction); + sinon.assert.notCalled(objectMD.setTransitionInProgress); + sinon.assert.notCalled(putObjectMD); + done(); + }); + }); + + it('should transition an object on a regular location', done => { + setupObjectMD('us-east-1'); + + lifecycleTask._applyTransitionRule(testParams, fakeLogger, err => { + assert.ifError(err); + sinon.assert.calledOnce(sendDataMoverAction); + sinon.assert.calledOnce(objectMD.setTransitionInProgress); + sinon.assert.calledOnce(putObjectMD); + done(); + }); + }); + }); + describe('_sendObjectAction', () => { it('should emit trigger metrics with the entry location', done => { const lifecycleTask = new LifecycleTask(lp); diff --git a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js index 61748e96e..bebdc9c57 100644 --- a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js +++ b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js @@ -169,4 +169,47 @@ describe('LifecycleUpdateTransitionTask', () => { done(); }); }); + + // clean room localization actions are published by the queue populator, + // which only knows the object owner's canonical id + describe('account id resolution', () => { + it('should not look up the account id when the entry has one', done => { + actionEntry.setAttribute('target.accountId', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 0); + done(); + }); + }); + + it('should resolve the account id from the owner canonical id', done => { + objectProcessor.setAccountId('some-canonical-id', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 1); + assert.strictEqual( + actionEntry.getAttribute('target.accountId'), + '000000000042'); + // the garbage collection entry must not resolve it again + const receivedGcEntry = gcProducer.getReceivedEntry(); + assert.strictEqual( + receivedGcEntry.getAttribute('target.accountId'), + '000000000042'); + done(); + }); + }); + + it('should fail the entry when the account id cannot be resolved', + done => { + actionEntry.setAttribute('target.owner', 'unknown-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert(err); + assert.strictEqual( + backbeatMetadataProxyClient.getReceivedMd(), null); + done(); + }); + }); + }); }); diff --git a/tests/unit/mocks.js b/tests/unit/mocks.js index 2b9096f88..99ebb0871 100644 --- a/tests/unit/mocks.js +++ b/tests/unit/mocks.js @@ -1,4 +1,5 @@ const assert = require('assert'); +const { errors } = require('arsenal'); const { ObjectMD } = require('arsenal').models; class GarbageCollectorProducerMock { @@ -157,6 +158,21 @@ class ProcessorMock { this.coldProducer = coldProducer; this._gcConfig = gcConfig; this.logger = logger; + this.accountIds = {}; + this.accountIdLookups = 0; + } + + setAccountId(ownerId, accountId) { + this.accountIds[ownerId] = accountId; + } + + getAccountId(ownerId, log, cb) { + this.accountIdLookups += 1; + const accountId = this.accountIds[ownerId]; + if (!accountId) { + return process.nextTick(cb, errors.NoSuchEntity); + } + return process.nextTick(cb, null, accountId); } getStateVars() { @@ -170,6 +186,7 @@ class ProcessorMock { getBackbeatClient: () => this.backbeatClient, getBackbeatMetadataProxy: () => this.backbeatMetadataProxy, getS3Client: () => this.s3Client, + getAccountId: this.getAccountId.bind(this), }; } } diff --git a/tests/unit/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index cc384b695..baa31e1dc 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -1,8 +1,11 @@ const assert = require('assert'); const sinon = require('sinon'); +const { encode } = require('arsenal').versioning.VersionID; + const ReplicationQueuePopulator = require('../../../extensions/replication/ReplicationQueuePopulator'); +const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); const fakeLogger = require('../../utils/fakeLogger'); @@ -382,3 +385,245 @@ describe('replication queue populator', () => { assert.deepStrictEqual(rqp.getState(), {}); }); }); + +/** + * Records every published message, whatever the topic, so localization + * entries (data mover topic) can be inspected. + * @class + */ +class RecordingQueuePopulatorMock extends ReplicationQueuePopulator { + constructor(params) { + super(params); + + this.published = []; + } + + publish(topic, key, message) { + this.published.push({ topic, key, message }); + } +} + +describe('replication queue populator: clean room localization', () => { + const CRR_LOCATION = 'location-crr-source'; + const LOCAL_LOCATION = 'us-east-1'; + const RESULTS_TOPIC = 'test-transition-results'; + const VERSION_ID = '98477724999464999999RG001 1.30.12'; + const VERSIONED_KEY = `a-test-key\u0000${VERSION_ID}`; + + let params; + let rqp; + + function makeValue(overrides = {}) { + return JSON.stringify({ + ...kafkaValue, + dataStoreName: CRR_LOCATION, + location: [{ + key: 'some-data-key', + size: 128, + start: 0, + dataStoreName: CRR_LOCATION, + dataStoreETag: '1:d41d8cd98f00b204e9800118ecf8427e', + }], + ...overrides, + }); + } + + function makeEntry(value, key = VERSIONED_KEY) { + return { + type: 'put', + bucket: 'test-bucket-source', + key, + value, + overheadFields: { commitTimestamp: '2024-05-06T10:11:12.000Z' }, + logReader: { getMetricLabels: stubMetricLabels() }, + }; + } + + beforeEach(() => { + params = { + config: { + topic: TOPIC, + localization: { + toLocation: LOCAL_LOCATION, + resultsTopic: RESULTS_TOPIC, + }, + }, + logger: fakeLogger, + metricsHandler: { + bytes: sinon.spy(), + objects: sinon.spy(), + localizationBytes: sinon.spy(), + localizationObjects: sinon.spy(), + }, + }; + rqp = new RecordingQueuePopulatorMock(params); + }); + + it('should publish a copyLocation action for a non-localized object', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 1); + const [{ topic, key, message }] = rqp.published; + assert.strictEqual(topic, ReplicationAPI.getDataMoverTopic()); + assert.strictEqual(key, 'test-bucket-source/a-test-key'); + + const action = JSON.parse(message); + assert.strictEqual(action.action, 'copyLocation'); + assert.strictEqual(action.toLocation, LOCAL_LOCATION); + assert.strictEqual(action.resultsTopic, RESULTS_TOPIC); + assert.strictEqual(action.contextInfo.ruleType, 'transition'); + assert.strictEqual(action.contextInfo.origin, 'localization'); + assert.deepStrictEqual(action.target, { + owner: kafkaValue['owner-id'], + bucket: 'test-bucket-source', + key: 'a-test-key', + version: encode(VERSION_ID), + eTag: `"${kafkaValue['content-md5']}"`, + lastModified: kafkaValue['last-modified'], + }); + // resolved by the transition processor, not by the populator + assert.strictEqual(action.target.accountId, undefined); + assert.deepStrictEqual(action.source, { + bucket: 'test-bucket-source', + objectKey: 'a-test-key', + storageClass: CRR_LOCATION, + }); + assert.strictEqual(action.metrics.fromLocation, CRR_LOCATION); + assert.strictEqual(action.metrics.contentLength, 128); + assert.strictEqual(action.metrics.transitionTime, + '2024-05-06T10:11:12.000Z'); + }); + + it('should account localized objects and bytes', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + sinon.assert.calledOnceWithExactly( + params.metricsHandler.localizationBytes, labels, 128); + sinon.assert.calledOnceWithExactly( + params.metricsHandler.localizationObjects, labels); + sinon.assert.notCalled(params.metricsHandler.objects); + }); + + // localization is about where the data lives, forward replication is + // about where it has been copied to: the two are independent. + ['PENDING', 'COMPLETED', 'FAILED'].forEach(status => { + it(`should publish regardless of replication status ${status}`, () => { + const value = makeValue({ + replicationInfo: { ...repInfo, status }, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + }); + }); + + it('should publish when there is no replication configured', () => { + const value = makeValue({ replicationInfo: null }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + }); + + it('should propagate the transition attempt count', () => { + const value = makeValue({ + 'x-amz-meta-scal-s3-transition-attempt': '3', + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + const action = JSON.parse(rqp.published[0].message); + assert.strictEqual(action.target.attempt, 3); + }); + + it('should not set an attempt count for a first copy', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + const action = JSON.parse(rqp.published[0].message); + assert.strictEqual(action.target.attempt, undefined); + }); + + it('should skip master keys', () => { + rqp._filterKeyOp(makeEntry(makeValue(), 'a-test-key')); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip delete markers', () => { + const value = makeValue({ isDeleteMarker: true }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip empty objects', () => { + const value = makeValue({ + 'location': null, + 'content-length': 0, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + }); + + it('should skip and report non-empty objects without location', () => { + const errorSpy = sinon.spy(rqp.log, 'error'); + const value = makeValue({ location: null }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 0); + sinon.assert.calledOnce(errorSpy); + errorSpy.restore(); + }); + + // partial oplog projections (change stream `update` events) may not carry + // the location: they cannot be localized, and behave as before. + it('should not localize entries with no dataStoreName', () => { + const value = makeValue({ dataStoreName: undefined }); + rqp._filterKeyOp(makeEntry(value)); + + sinon.assert.notCalled(params.metricsHandler.localizationObjects); + assert.strictEqual( + rqp.published.filter( + p => p.topic === ReplicationAPI.getDataMoverTopic()).length, + 0); + }); + + it('should not localize objects on a regular location', () => { + const value = makeValue({ dataStoreName: LOCAL_LOCATION }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + assert.strictEqual(rqp.published[0].topic, TOPIC); + sinon.assert.notCalled(params.metricsHandler.localizationObjects); + }); + + it('should not replicate a non-localized object when localization is ' + + 'disabled', () => { + delete params.config.localization; + rqp = new RecordingQueuePopulatorMock(params); + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 0); + }); + + // the data still lives on the source location: there is nothing local to + // replicate, whatever replicationInfo says. + [true, false].forEach(localizationEnabled => { + it('should never replicate a pending non-localized object ' + + `(localization ${localizationEnabled ? 'enabled' : 'disabled'})`, + () => { + if (!localizationEnabled) { + delete params.config.localization; + } + rqp = new RecordingQueuePopulatorMock(params); + const value = makeValue({ + replicationInfo: { ...repInfo, status: 'PENDING' }, + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual( + rqp.published.filter(p => p.topic === TOPIC).length, 0); + sinon.assert.notCalled(params.metricsHandler.objects); + }); + }); +});