From 6e783fdaccd25b7c4a795f430c71341947f6fc38 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 09:31:45 +0200 Subject: [PATCH 1/2] Trigger localization of clean room objects In a clean room, objects are created locally but their metadata still points at the source cluster's location: the data itself has not been copied over yet. Something has to notice those objects and ask for the data to be pulled in. The queue populator is the natural place for it, since bootstrap, re-bootstrap and streamed updates all go through the same oplog. When an object lands on a location flagged isCRR, publish a copyLocation action on the data mover topic and let the existing data mover + transition merge pipeline do the actual copy. This is unrelated to replicationInfo, which describes replication of a *local* object to remote sites, so the check sits before any replication condition. Activation is simply the presence of extensions.replication.localization.toLocation in the config; without it nothing changes. The populator only knows the object owner's canonical id, and resolving an account id per entry would throttle the whole populator, so the lookup is deferred to the transition processor, which now resolves it once up front and stamps it back on the action - the same way the garbage collector already does. Issue: BB-814 --- conf/locationConfig.json | 7 + docker-entrypoint.sh | 10 + .../lifecycle/LifecycleConfigValidator.js | 1 + .../LifecycleObjectProcessor.js | 56 ++++- .../tasks/LifecycleUpdateTransitionTask.js | 42 +++- .../replication/ReplicationConfigValidator.js | 7 + .../replication/ReplicationQueuePopulator.js | 136 ++++++++++- lib/Config.js | 3 + lib/queuePopulator/QueuePopulator.js | 16 ++ .../LifecycleUpdateTransitionTask.spec.js | 43 ++++ tests/unit/mocks.js | 17 ++ .../ReplicationQueuePopulator.spec.js | 224 ++++++++++++++++++ 12 files changed, 553 insertions(+), 9 deletions(-) diff --git a/conf/locationConfig.json b/conf/locationConfig.json index 8ba3dc334c..dceb31ddac 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 883146ff62..42aea6bdec 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 b6487862e5..0d5a1c1a0b 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 794449bb52..39a8b8ae03 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -7,6 +7,9 @@ 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 +64,55 @@ 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); } + /** + * 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._accountIdCache.isKnown(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 +179,9 @@ class LifecycleObjectProcessor extends EventEmitter { start(done) { this.clientManager.initSTSConfig(); this.clientManager.initCredentialsManager(); + if (this.getAuthConfig(this._lcConfig).type === authTypeAssumeRole) { + this.vaultClientWrapper.init(); + } this._setupConsumers(done); } @@ -225,12 +277,14 @@ 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.vaultClientWrapper.tempCredentialsReady(); } } diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 4f57c33c7f..a3450f48ad 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -252,6 +252,34 @@ 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); + } + entry.setAttribute('target.accountId', resolvedAccountId); + return cb(); + }); + } + /** * * @param {ActionQueueEntry} entry - action entry to execute @@ -268,11 +296,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 8f748c8077..47d1c26636 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 22c31ad0bb..e8d9b3234e 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,17 @@ 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 && 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 +99,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 +139,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 57a9aaa880..4a1d18646f 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 8b63e3d055..d2a716bf15 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/LifecycleUpdateTransitionTask.spec.js b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js index 61748e96e3..bebdc9c572 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 2b9096f88a..99ebb0871d 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 cc384b6950..bb9a95d16f 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,224 @@ 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 fall back to replication when localization is disabled', () => { + delete params.config.localization; + rqp = new RecordingQueuePopulatorMock(params); + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 1); + assert.strictEqual(rqp.published[0].topic, TOPIC); + }); +}); From 7139bf2287c49a5f29dd36313a8b1762158da4e7 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 09:48:34 +0200 Subject: [PATCH 2/2] Fail account id resolution on a cached vault miss AccountIdCache.isKnown() is also true for cached misses, where get() returns undefined. The localization flow would then carry on with target.accountId unset and surface a confusing "failed to get backbeat client" InternalError instead of the NoSuchEntity that actually caused it, making a bad canonical id hard to diagnose in production. Treat a cached miss as the error it is, and only stamp the resolved account id onto the entry when there is one. Issue: BB-814 --- .../LifecycleObjectProcessor.js | 10 ++- .../tasks/LifecycleUpdateTransitionTask.js | 4 +- ...LifecycleObjectTransitionProcessor.spec.js | 89 +++++++++++++++++++ 3 files changed, 101 insertions(+), 2 deletions(-) diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 39a8b8ae03..14e1ee5181 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js @@ -2,6 +2,7 @@ const { EventEmitter } = require('events'); const Logger = require('werelogs').Logger; +const { errors } = require('arsenal'); const BackbeatConsumerManager = require('../../../lib/BackbeatConsumerManager'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); @@ -91,7 +92,14 @@ class LifecycleObjectProcessor extends EventEmitter { return process.nextTick(cb); } - if (this._accountIdCache.isKnown(ownerId)) { + // 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)); } diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index a3450f48ad..2fe958e045 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -275,7 +275,9 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { if (err) { return cb(err); } - entry.setAttribute('target.accountId', resolvedAccountId); + if (resolvedAccountId) { + entry.setAttribute('target.accountId', resolvedAccountId); + } return cb(); }); } diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa51388517..94b8c5b417 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,92 @@ 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' }, + }, + }, + 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(); + }); + }); + }); + }); });