From 8d16e58edf3d5ee81e91bb520523807457232219 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:27:36 +0200 Subject: [PATCH 1/4] Share the transition attempt metadata helper The x-amz-meta-scal-s3-transition-attempt key was open-coded in five places across lifecycle and gc, each with a slightly different way of reading or clearing it - one of which would throw on user metadata that does not parse. Move it behind a small helper, which the localization work needs to read as well. Issue: BB-814 --- extensions/gc/tasks/GarbageCollectorTask.js | 7 ++- .../tasks/LifecycleColdStatusArchiveTask.js | 7 ++- .../LifecycleResetTransitionInProgressTask.js | 5 +- extensions/lifecycle/tasks/LifecycleTask.js | 11 +--- .../tasks/LifecycleUpdateTransitionTask.js | 25 ++++----- lib/util/transitionAttempt.js | 52 +++++++++++++++++++ 6 files changed, 71 insertions(+), 36 deletions(-) create mode 100644 lib/util/transitionAttempt.js diff --git a/extensions/gc/tasks/GarbageCollectorTask.js b/extensions/gc/tasks/GarbageCollectorTask.js index dae27c5807..e745652a02 100644 --- a/extensions/gc/tasks/GarbageCollectorTask.js +++ b/extensions/gc/tasks/GarbageCollectorTask.js @@ -5,6 +5,7 @@ const { ObjectMD } = require('arsenal').models; const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const { BatchDeleteCommand } = require('@scality/cloudserverclient'); const { GarbageCollectorMetrics } = require('../GarbageCollectorMetrics'); +const { clearTransitionAttempt } = require('../../../lib/util/transitionAttempt'); /** @typedef { import('../GarbageCollector.js') } GarbageCollector */ class GarbageCollectorTask extends BackbeatTask { @@ -290,10 +291,8 @@ class GarbageCollectorTask extends BackbeatTask { .setDataStoreName(newLocation) .setAmzStorageClass(newLocation) .setOriginOp('s3:LifecycleTransition') - .setTransitionInProgress(false) - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }); + .setTransitionInProgress(false); + clearTransitionAttempt(objMD); this._putMetadata(entry, objMD, log, err => { GarbageCollectorMetrics.onS3Request(log, 'putMetadata', 'archive', err); if (!err) { diff --git a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js index 8cea3c62bc..f81eece662 100644 --- a/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js +++ b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js @@ -4,6 +4,7 @@ const ObjectMDArchive = require('arsenal').models.ObjectMDArchive; const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const LifecycleUpdateTransitionTask = require('./LifecycleUpdateTransitionTask'); const { LifecycleMetrics } = require('../LifecycleMetrics'); +const { clearTransitionAttempt } = require('../../../lib/util/transitionAttempt'); class SkipMdUpdateError extends Error {} @@ -115,10 +116,8 @@ class LifecycleColdStatusArchiveTask extends LifecycleUpdateTransitionTask { objectMD.setDataStoreName(coldLocation) .setAmzStorageClass(coldLocation) .setTransitionInProgress(false) - .setOriginOp('s3:LifecycleTransition') - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }); + .setOriginOp('s3:LifecycleTransition'); + clearTransitionAttempt(objectMD); } this._putMetadata(entry, objectMD, log, err => { diff --git a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js index 8768d577fe..95b27bf6aa 100644 --- a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js +++ b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js @@ -1,6 +1,7 @@ 'use strict'; const { LifecycleRequeueTask } = require('./LifecycleRequeueTask'); +const { setTransitionAttempt } = require('../../../lib/util/transitionAttempt'); class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { /** @@ -19,9 +20,7 @@ class LifecycleResetTransitionInProgressTask extends LifecycleRequeueTask { } md.setOriginOp('s3:LifecycleTransition:Retry'); md.setTransitionInProgress(false); - md.setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': try_, - }); + setTransitionAttempt(md, try_); return true; } diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a83118404..ff3bcf9df2 100644 --- a/extensions/lifecycle/tasks/LifecycleTask.js +++ b/extensions/lifecycle/tasks/LifecycleTask.js @@ -24,6 +24,7 @@ const ReplicationAPI = require('../../replication/ReplicationAPI'); const { LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION } = require('../LifecycleMetrics'); const locationsConfig = require('../../../conf/locationConfig.json') || {}; const { rulesSupportTransition } = require('../util/rules'); +const { getTransitionAttempt } = require('../../../lib/util/transitionAttempt'); const { stampTraceHeaders } = require('arsenal/build/lib/tracing').kafka; const { decode } = versioning.VersionID; @@ -1176,15 +1177,7 @@ class LifecycleTask extends BackbeatTask { } _getTransitionActionEntry(params, objectMD, log, cb) { - let attempt; - const umd = objectMD.getUserMetadata(); - if (umd) { - const parsed = JSON.parse(umd); - const rawAttempt = parsed['x-amz-meta-scal-s3-transition-attempt']; - if (rawAttempt) { - attempt = Number.parseInt(rawAttempt, 10); - } - } + const attempt = getTransitionAttempt(objectMD.getUserMetadata()); const entry = ReplicationAPI.createCopyLocationAction({ bucketName: params.bucket, diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 4f57c33c7f..a5f9cae3c4 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -6,6 +6,11 @@ const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const ObjectMD = require('arsenal').models.ObjectMD; const { LifecycleMetrics } = require('../LifecycleMetrics'); +const { + getTransitionAttempt, + setTransitionAttempt, + clearTransitionAttempt, +} = require('../../../lib/util/transitionAttempt'); /** @typedef { import('../objectProcessor/LifecycleObjectProcessor.js') } LifecycleObjectProcessor */ class LifecycleUpdateTransitionTask extends BackbeatTask { @@ -71,10 +76,8 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { .setDataStoreName(newLocationName) .setAmzStorageClass(newLocationName) .setOriginOp('s3:LifecycleTransition') - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }) .setTransitionInProgress(false); + clearTransitionAttempt(objMD); } _putMetadata(entry, objMD, log, done) { @@ -232,20 +235,10 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { next(err, objMD); }), (objMD, next) => { - const userMDStr = objMD.getUserMetadata() || '{}'; - const userMD = JSON.parse(userMDStr); + const tryCount = (getTransitionAttempt(objMD.getUserMetadata()) || 0) + 1; - let tryCount = userMD['x-amz-meta-scal-s3-transition-attempt']; - if (tryCount === undefined) { - tryCount = 1; - } else { - tryCount = parseInt(tryCount, 10) + 1; - } - - objMD.setTransitionInProgress(false) - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': tryCount, - }); + objMD.setTransitionInProgress(false); + setTransitionAttempt(objMD, tryCount); return this._putMetadata(entry, objMD, log, next); }, diff --git a/lib/util/transitionAttempt.js b/lib/util/transitionAttempt.js new file mode 100644 index 0000000000..51d8260e6a --- /dev/null +++ b/lib/util/transitionAttempt.js @@ -0,0 +1,52 @@ +const safeJsonParse = require('./safeJsonParse'); + +// How many times the transition of an object was attempted. Kept in the +// object user metadata so that it survives across processes: the transition +// processor bumps it on failure, and it is cleared once the object made it to +// its new location. +const TRANSITION_ATTEMPT_MD = 'x-amz-meta-scal-s3-transition-attempt'; + +/** + * Read the transition attempt count from raw object user metadata. + * @param {String|undefined} userMetadata - serialized user metadata, as + * returned by ObjectMD.getUserMetadata() + * @return {Number|undefined} attempt count, or undefined if the object was + * never transitioned, or the metadata cannot be read + */ +function getTransitionAttempt(userMetadata) { + if (!userMetadata) { + return undefined; + } + const { error, result } = safeJsonParse(userMetadata); + if (error) { + return undefined; + } + const attempt = Number.parseInt(result[TRANSITION_ATTEMPT_MD], 10); + return Number.isInteger(attempt) ? attempt : undefined; +} + +/** + * Set the transition attempt count on an object. + * @param {ObjectMD} objMD - object metadata to update in place + * @param {Number} attempt - attempt count + * @return {ObjectMD} the updated object metadata + */ +function setTransitionAttempt(objMD, attempt) { + return objMD.setUserMetadata({ [TRANSITION_ATTEMPT_MD]: attempt }); +} + +/** + * Forget the transition attempt count, once the transition succeeded. + * @param {ObjectMD} objMD - object metadata to update in place + * @return {ObjectMD} the updated object metadata + */ +function clearTransitionAttempt(objMD) { + return setTransitionAttempt(objMD, undefined); +} + +module.exports = { + TRANSITION_ATTEMPT_MD, + getTransitionAttempt, + setTransitionAttempt, + clearTransitionAttempt, +}; From 3d98764ebd89680dd26481f8f2e97aaff002a320 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:27:54 +0200 Subject: [PATCH 2/4] Report copyLocation results under the metrics type of their flow copyLocation actions were all reported as transitions, which is about to stop being true: localization uses the same pipeline but has a trigger latency of milliseconds where a transition is minutes to hours, and the two would share histogram buckets that start at a minute. Derive the metrics type from the origin of the action instead, so trigger, start and completion keep pairing up. Issue: BB-814 --- extensions/lifecycle/LifecycleMetrics.js | 19 +++++++++++++++++++ .../tasks/LifecycleUpdateTransitionTask.js | 4 ++-- .../replication/tasks/CopyLocationTask.js | 4 ++-- tests/unit/lifecycle/LifecycleMetrics.spec.js | 19 +++++++++++++++++++ 4 files changed, 42 insertions(+), 4 deletions(-) diff --git a/extensions/lifecycle/LifecycleMetrics.js b/extensions/lifecycle/LifecycleMetrics.js index bde4315530..020a3c87a0 100644 --- a/extensions/lifecycle/LifecycleMetrics.js +++ b/extensions/lifecycle/LifecycleMetrics.js @@ -12,6 +12,9 @@ const LIFECYCLE_LABEL_CONDUCTOR_SCAN_ID = 'conductor_scan_id'; const LIFECYCLE_MARKER_METRICS_LOCATION = '-delete-marker-'; +const TRANSITION_TYPE = 'transition'; +const LOCALIZATION_TYPE = 'localization'; + // Keep per-scan series long enough for scraping and debugging recent overlap, // but remove them from prom-client after a configurable retention interval. // We intentionally do not cap the number of tracked scan IDs: if overlapping @@ -447,9 +450,25 @@ class LifecycleMetrics { } } +/** + * Metrics type of a copyLocation action: clean room localization goes through + * the same pipeline as a lifecycle transition, but is triggered from the oplog + * rather than from a bucket scan, so its latencies are orders of magnitude + * apart and would skew the transition ones if reported under the same type. + * + * @param {ActionQueueEntry} actionEntry - copyLocation action + * @return {string} metrics type + */ +function getCopyLocationMetricsType(actionEntry) { + return actionEntry.getAttribute('metrics.origin') === LOCALIZATION_TYPE ? + LOCALIZATION_TYPE : TRANSITION_TYPE; +} + module.exports = { DEFAULT_SCAN_METRIC_RETENTION_S, LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION, + LOCALIZATION_TYPE, + getCopyLocationMetricsType, resetLifecycleScanMetricCleanupTimers, }; diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index a5f9cae3c4..aa2d1af478 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -5,7 +5,7 @@ const errors = require('arsenal').errors; const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const ObjectMD = require('arsenal').models.ObjectMD; -const { LifecycleMetrics } = require('../LifecycleMetrics'); +const { LifecycleMetrics, getCopyLocationMetricsType } = require('../LifecycleMetrics'); const { getTransitionAttempt, setTransitionAttempt, @@ -203,7 +203,7 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { const transitionTime = entry.getAttribute('metrics.transitionTime') || objMD.getTransitionTime(); const locationName = entry.getAttribute('toLocation'); - LifecycleMetrics.onLifecycleCompleted(log, 'transition', + LifecycleMetrics.onLifecycleCompleted(log, getCopyLocationMetricsType(entry), locationName, Date.now() - Date.parse(transitionTime)); next(err); }); diff --git a/extensions/replication/tasks/CopyLocationTask.js b/extensions/replication/tasks/CopyLocationTask.js index b7d455e909..8ee326f5bc 100644 --- a/extensions/replication/tasks/CopyLocationTask.js +++ b/extensions/replication/tasks/CopyLocationTask.js @@ -16,7 +16,7 @@ const { MultipleBackendAbortMPUCommand, addContentLengthMiddleware, } = require('@scality/cloudserverclient'); -const { LifecycleMetrics } = require('../../lifecycle/LifecycleMetrics'); +const { LifecycleMetrics, getCopyLocationMetricsType } = require('../../lifecycle/LifecycleMetrics'); const ReplicationMetric = require('../ReplicationMetric'); const ReplicationMetrics = require('../ReplicationMetrics'); const { isRetryableMiddleware, TIMEOUT_MS } = require('../../../lib/clients/utils'); @@ -138,7 +138,7 @@ class CopyLocationTask extends BackbeatTask { const transitionTime = actionEntry.getAttribute('metrics.transitionTime') || objMD.getTransitionTime(); - LifecycleMetrics.onLifecycleStarted(log, 'transition', + LifecycleMetrics.onLifecycleStarted(log, getCopyLocationMetricsType(actionEntry), actionEntry.getAttribute('toLocation'), startTime - Date.parse(transitionTime)); diff --git a/tests/unit/lifecycle/LifecycleMetrics.spec.js b/tests/unit/lifecycle/LifecycleMetrics.spec.js index 2a6a973848..53e571a093 100644 --- a/tests/unit/lifecycle/LifecycleMetrics.spec.js +++ b/tests/unit/lifecycle/LifecycleMetrics.spec.js @@ -2,8 +2,10 @@ const assert = require('assert'); const sinon = require('sinon'); const { LifecycleMetrics, + getCopyLocationMetricsType, resetLifecycleScanMetricCleanupTimers, } = require('../../../extensions/lifecycle/LifecycleMetrics'); +const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const { ZenkoMetrics } = require('arsenal').metrics; describe('LifecycleMetrics', () => { @@ -20,6 +22,23 @@ describe('LifecycleMetrics', () => { sinon.restore(); }); + describe('getCopyLocationMetricsType', () => { + [ + ['localization', 'localization'], + ['lifecycle', 'transition'], + [undefined, 'transition'], + ].forEach(([origin, expected]) => { + it(`should report ${origin} actions as ${expected}`, () => { + const entry = ActionQueueEntry.create('copyLocation'); + if (origin !== undefined) { + entry.setAttribute('metrics', { origin }); + } + + assert.strictEqual(getCopyLocationMetricsType(entry), expected); + }); + }); + }); + describe('error handling', () => { it('should catch errors in onProcessBuckets', () => { const metric = ZenkoMetrics.getMetric('s3_lifecycle_latest_batch_start_time'); From 02739de209042c4a6b985f034150f06a3d6f4350 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:28:16 +0200 Subject: [PATCH 3/4] Resolve the account id of actions that only carry an owner The lifecycle conductor knows the account id of the bucket it is scanning and stamps it on the actions it publishes. The queue populator does not: it works off the oplog, where an object only carries its owner's canonical id, and resolving an account per entry would throttle the whole populator. So let the transition processor do the lookup, once per action and only when needed, the same way the garbage collector already does, and pass the result on to the garbage collection entry it emits. Issue: BB-814 --- .../LifecycleObjectTransitionProcessor.js | 81 +++++++++++- extensions/lifecycle/objectProcessor/task.js | 2 +- .../tasks/LifecycleUpdateTransitionTask.js | 52 +++++++- ...LifecycleObjectTransitionProcessor.spec.js | 118 ++++++++++++++++++ .../LifecycleUpdateTransitionTask.spec.js | 43 +++++++ tests/unit/mocks.js | 17 +++ 6 files changed, 307 insertions(+), 6 deletions(-) diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js index e390280660..aa0c31943f 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js +++ b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js @@ -1,6 +1,7 @@ 'use strict'; const assert = require('assert'); const async = require('async'); +const { errors } = require('arsenal'); const ColdStorageStatusQueueEntry = require('../../../lib/models/ColdStorageStatusQueueEntry'); const { LifecycleMetrics } = require('../LifecycleMetrics'); @@ -14,6 +15,9 @@ const { updateCircuitBreakerConfigForImplicitOutputQueue } = require('../../../l const { LifecycleRetriggerRestoreTask } = require('../tasks/LifecycleRetriggerRestoreTask'); const BackbeatProducer = require('../../../lib/BackbeatProducer'); const GarbageCollectorProducer = require('../../gc/GarbageCollectorProducer'); +const VaultClientWrapper = require('../../utils/VaultClientWrapper'); +const { AccountIdCache } = require('../../utils/AccountIdCache'); +const { authTypeAssumeRole } = require('../../../lib/constants'); class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor { @@ -44,9 +48,78 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor { * @param {Number} s3Config.port - s3 endpoint port * @param {String} [transport="http"] - transport method ("http" * or "https") + * @param {Object} [vaultAdminConfig] - vault admin endpoint, used to + * resolve canonical ids into account ids */ - constructor(zkConfig, kafkaConfig, lcConfig, s3Config, transport = 'http') { + constructor(zkConfig, kafkaConfig, lcConfig, s3Config, transport = 'http', + vaultAdminConfig = undefined) { super(zkConfig, kafkaConfig, lcConfig, s3Config, transport); + + // The transition processor is the only one receiving actions published + // without an account id (clean room localization), so it is the only + // one needing a vault client to resolve them. + const authConfig = this.getAuthConfig(this._lcConfig); + if (authConfig.type === authTypeAssumeRole && + (vaultAdminConfig || authConfig.vault)) { + this.vaultClientWrapper = new VaultClientWrapper( + `lifecycle:${this.getProcessorType()}`, + vaultAdminConfig, + authConfig, + this._log, + ); + this._accountIdCache = new AccountIdCache( + this._processConfig.concurrency); + } + } + + /** + * 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.vaultClientWrapper) { + 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); + }); } /** @@ -56,6 +129,7 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor { * @return {undefined} */ start(done) { + this.vaultClientWrapper?.init(); async.waterfall([ next => super.start(next), next => { @@ -226,8 +300,13 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor { ...super.getStateVars(), coldProducer: this._coldProducer, gcProducer: this._gcProducer, + getAccountId: this.getAccountId.bind(this), }; } + + isReady() { + return super.isReady() && (!this.vaultClientWrapper || this.vaultClientWrapper.tempCredentialsReady()); + } } module.exports = LifecycleObjectTransitionProcessor; diff --git a/extensions/lifecycle/objectProcessor/task.js b/extensions/lifecycle/objectProcessor/task.js index d09c2020aa..d7bd60ab47 100644 --- a/extensions/lifecycle/objectProcessor/task.js +++ b/extensions/lifecycle/objectProcessor/task.js @@ -32,7 +32,7 @@ let objectProcessor; switch (process.env.LIFECYCLE_OBJECT_PROCESSOR_TYPE) { case 'transition': objectProcessor = new LifecycleObjectTransitionProcessor( - zkConfig, kafkaConfig, lcConfig, s3Config, transport); + zkConfig, kafkaConfig, lcConfig, s3Config, transport, config.vaultAdmin); break; case 'expiration': // fallthrough default: diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index aa2d1af478..37e1337a9c 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -245,6 +245,43 @@ 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) { + return process.nextTick(cb); + } + + if (!owner) { + // Every publisher sets one or the other, so this is a malformed + // entry: log it, and let the task fail on its own further down + // rather than retrying something that cannot be fixed. + log.error('cannot resolve account id: entry has no account id nor 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 @@ -261,11 +298,18 @@ 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/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa51388517..e1da530108 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,121 @@ 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, + config.transport, + { host: 'localhost', port: 8600 }, + ); + log = { debug: () => {}, error: () => {} }; + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should skip the lookup when auth type is not assume role', done => { + assert.strictEqual(objectProcessor.vaultClientWrapper, undefined); + objectProcessor.getAccountId(ownerId, log, (err, id) => { + assert.ifError(err); + assert.strictEqual(id, undefined); + 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, + ); + assert.strictEqual(noVault.vaultClientWrapper, undefined); + // readiness must not wait on credentials that are never fetched + noVault._consumers = { isReady: () => true }; + assert.strictEqual(noVault.isReady(), true); + + noVault.getAccountId(ownerId, log, err => { + assert(err.InternalError); + done(); + }); + }); + }); }); 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), }; } } From aedd8492ef381f897c2b0ce97df29e29dffbb117 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:28:37 +0200 Subject: [PATCH 4/4] 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. The destination comes from the object metadata, which the source-side rewrite stamps as it prepares the entry; if it names a location we do not know, fall back to the first local one and log about it. This is unrelated to replicationInfo, which describes replication of a *local* object to remote sites, so the check sits before any replication condition. Localization is neither lifecycle nor CRR replication, so it gets its own action origin, and the legacy CRR byte metrics - which only make sense for replication to a remote site - skip it like they already skip lifecycle. Issue: BB-814 --- conf/locationConfig.json | 7 + extensions/replication/ReplicationMetric.js | 12 +- .../replication/ReplicationQueuePopulator.js | 153 ++++++++++- tests/unit/ReplicationMetric.js | 28 +- .../lifecycle/CircuitBreakerGroup.spec.js | 10 + .../ReplicationQueuePopulator.spec.js | 260 ++++++++++++++++++ 6 files changed, 447 insertions(+), 23 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/extensions/replication/ReplicationMetric.js b/extensions/replication/ReplicationMetric.js index 607e99ab5d..585cbb3e21 100644 --- a/extensions/replication/ReplicationMetric.js +++ b/extensions/replication/ReplicationMetric.js @@ -2,6 +2,10 @@ const { Logger } = require('werelogs'); const MetricsModel = require('../../lib/models/MetricsModel'); +// Flows which report their own metrics, and for which the CRR metrics +// below are meaningless: they are not replication to a CRR site. +const ORIGINS_WITH_OWN_METRICS = ['lifecycle', 'localization']; + /** * Legacy: consider converting replication metrics with * Prometheus-based {@link ReplicationMetrics} class @@ -46,9 +50,9 @@ class ReplicationMetric { return this; } - _isLifecycleAction() { + _hasOwnMetrics() { const { origin } = this._entry.getContext(); - return origin !== undefined && origin === 'lifecycle'; + return ORIGINS_WITH_OWN_METRICS.includes(origin); } _createProducerMessage() { @@ -65,8 +69,8 @@ class ReplicationMetric { } publish() { - // Lifecycle metrics not yet implemented. - if (this._isLifecycleAction()) { + // Lifecycle and localization metrics not yet implemented. + if (this._hasOwnMetrics()) { return undefined; } const message = this._createProducerMessage(); diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index 22c31ad0bb..3ec59ec47f 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -1,13 +1,26 @@ 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 { LifecycleMetrics, LOCALIZATION_TYPE } = require('../lifecycle/LifecycleMetrics'); +const config = require('../../lib/Config'); const locationsConfig = require('../../conf/locationConfig.json') || {}; const safeJsonParse = require('../../lib/util/safeJsonParse'); +const { getTransitionAttempt } = require('../../lib/util/transitionAttempt'); const { traceHeadersFromEntry } = require('arsenal/build/lib/tracing').kafka; +const { transitionTasksTopic } = config.extensions.lifecycle; + +// Where clean room objects are localized when their metadata does not name a +// usable target. Cold and source (isCRR) locations can never hold localized +// data, any other one is a valid local destination. +const defaultLocalLocation = Object.keys(locationsConfig).find( + name => !locationsConfig[name].isCold && !locationsConfig[name].isCRR); + class ReplicationQueuePopulator extends QueuePopulatorExtension { constructor(params) { super(params); @@ -73,6 +86,15 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (sanityCheckRes) { return; } + const locationConfig = locationsConfig[queueEntry.getDataStoreName()] || {}; + // 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._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 +103,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 +143,132 @@ 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 targetLocation = this._getLocalizationTarget(queueEntry, locations); + if (!targetLocation) { + return; + } + const transitionTime = new Date(entry.overheadFields?.commitTimestamp ?? Date.now()); + const action = ReplicationAPI.createCopyLocationAction({ + bucketName: bucket, + objectKey, + owner: queueEntry.getOwnerId(), + versionId: value.versionId ? encode(value.versionId) : undefined, + eTag: `"${queueEntry.getContentMd5()}"`, + lastModified: queueEntry.getLastModified(), + toLocation: targetLocation, + originLabel: 'localization', + fromLocation: queueEntry.getDataStoreName(), + contentLength, + resultsTopic: transitionTasksTopic, + transitionTime: transitionTime.toISOString(), + attempt: getTransitionAttempt(queueEntry.getUserMetadata()), + }); + // '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(), + }); + + LifecycleMetrics.onLifecycleTriggered(this.log, 'queuePopulator', + LOCALIZATION_TYPE, targetLocation, Date.now() - transitionTime.getTime()); + + this.log.trace('publishing object localization entry', { entry: queueEntry.getLogInfo() }); + this.publish(ReplicationAPI.getDataMoverTopic(), + `${bucket}/${objectKey}`, + action.toKafkaMessage(), + undefined, + traceHeadersFromEntry(value)); + } + + /** + * Local location the object data must be copied to. + * + * It is named in the source location entry itself, next to the bucket and + * role the copy needs: the rewrite pipeline resolves it from the bucket + * when it synthesizes that entry, which keeps this populator -a single + * threaded oplog reader- from having to look the bucket up per object. + * + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @param {Object[]} locations - object data locations + * @return {String|undefined} target location, undefined if there is none + */ + _getLocalizationTarget(queueEntry, locations) { + const { targetLocation } = locations[0]; + if (locationsConfig[targetLocation]) { + return targetLocation; + } + // Either the object predates the rewrite pipeline naming a target, or + // the location was deleted since the metadata was written. Neither is + // recoverable here, so fall back to the default location: localizing + // elsewhere beats leaving the data on the source forever. + if (!defaultLocalLocation) { + this.log.error('invalid localization target and no local location ' + + 'to fall back to, skipping localization', { + method: 'ReplicationQueuePopulator._getLocalizationTarget', + ...queueEntry.getLogInfo(), + targetLocation, + }); + return undefined; + } + this.log.error('invalid localization target in object metadata', { + method: 'ReplicationQueuePopulator._getLocalizationTarget', + ...queueEntry.getLogInfo(), + targetLocation, + fallbackLocation: defaultLocalLocation, + }); + return defaultLocalLocation; + } + /** * 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/tests/unit/ReplicationMetric.js b/tests/unit/ReplicationMetric.js index bffbae9033..a11b85ee14 100644 --- a/tests/unit/ReplicationMetric.js +++ b/tests/unit/ReplicationMetric.js @@ -56,27 +56,25 @@ describe('ReplicationMetric', () => { .forEach(key => assert.strictEqual(data[key], mock[key])); }); - it('::_isLifecycleAction should return false by default', () => { + it('::_hasOwnMetrics should return false by default', () => { metric.withEntry(entry); - assert.strictEqual(metric._isLifecycleAction(), false); + assert.strictEqual(metric._hasOwnMetrics(), false); }); - it('::_isLifecycleAction should return true when origin is lifecycle', - () => { - entry.setAttribute('contextInfo', { - origin: 'lifecycle', + ['lifecycle', 'localization'].forEach(origin => { + it(`::_hasOwnMetrics should return true when origin is ${origin}`, + () => { + entry.setAttribute('contextInfo', { origin }); + metric.withEntry(entry); + assert.strictEqual(metric._hasOwnMetrics(), true); }); - metric.withEntry(entry); - assert.strictEqual(metric._isLifecycleAction(), true); - }); - it('::publish should not send data to topic if lifecycle task', () => { - entry.setAttribute('contextInfo', { - origin: 'lifecycle', + it(`::publish should not send data to topic for a ${origin} action`, () => { + entry.setAttribute('contextInfo', { origin }); + metric.withEntry(entry); + metric.publish(); + assert.strictEqual(sentMessages.length, 0); }); - metric.withEntry(entry); - metric.publish(); - assert.strictEqual(sentMessages.length, 0); }); it('::publish should send data to topic', () => { diff --git a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js index 56e825d455..c8c8737b6f 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/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index cc384b6950..8628d95795 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -1,8 +1,14 @@ 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 { LifecycleMetrics } = + require('../../../extensions/lifecycle/LifecycleMetrics'); +const config = require('../../../lib/Config'); const fakeLogger = require('../../utils/fakeLogger'); @@ -382,3 +388,257 @@ 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'; + // location named in the object metadata by the rewrite pipeline + const TARGET_LOCATION = 'us-east-2'; + // first non-cold, non-CRR location: used when the target is unusable + const LOCAL_LOCATION = 'us-east-1'; + const RESULTS_TOPIC = config.extensions.lifecycle.transitionTasksTopic; + const VERSION_ID = '98477724999464999999RG001 1.30.12'; + const VERSIONED_KEY = `a-test-key\u0000${VERSION_ID}`; + + let params; + let rqp; + let triggeredMetric; + + 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', + bucket: 'test-bucket-source', + role: 'arn:aws:iam::123456789012:role/clean-room-read', + targetLocation: TARGET_LOCATION, + }], + ...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, + }, + logger: fakeLogger, + metricsHandler: { + bytes: sinon.spy(), + objects: sinon.spy(), + }, + }; + rqp = new RecordingQueuePopulatorMock(params); + triggeredMetric = sinon.stub(LifecycleMetrics, 'onLifecycleTriggered'); + }); + + afterEach(() => { + triggeredMetric.restore(); + }); + + 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, TARGET_LOCATION); + assert.strictEqual(action.resultsTopic, RESULTS_TOPIC); + assert.strictEqual(action.contextInfo.ruleType, 'transition'); + assert.strictEqual(action.contextInfo.origin, 'localization'); + assert.strictEqual(action.metrics.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 report the transition as triggered', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + sinon.assert.calledOnceWithExactly(triggeredMetric, rqp.log, + 'queuePopulator', 'localization', TARGET_LOCATION, + sinon.match.number); + 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(triggeredMetric); + 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(triggeredMetric); + }); + + // the target may have been deleted since the metadata was written, and + // objects written before the pipeline named one carry no target at all + [ + ['an unknown target', 'a-deleted-location'], + ['no target', undefined], + ].forEach(([desc, targetLocation]) => { + it(`should localize to the default location with ${desc}`, () => { + const errorSpy = sinon.spy(rqp.log, 'error'); + const value = makeValue({ + location: [{ + key: 'some-data-key', + size: 128, + start: 0, + dataStoreName: CRR_LOCATION, + dataStoreETag: '1:d41d8cd98f00b204e9800118ecf8427e', + targetLocation, + }], + }); + rqp._filterKeyOp(makeEntry(value)); + + assert.strictEqual(rqp.published.length, 1); + const action = JSON.parse(rqp.published[0].message); + assert.strictEqual(action.toLocation, LOCAL_LOCATION); + sinon.assert.calledOnce(errorSpy); + errorSpy.restore(); + }); + }); + + it('should skip localization when there is no local location', () => { + sinon.stub(rqp, '_getLocalizationTarget').returns(undefined); + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 0); + sinon.assert.notCalled(triggeredMetric); + }); +});