From 9975ecac3607437e2925ffb869d638aa17cce784 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:27:36 +0200 Subject: [PATCH 1/6] 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 pull replication work needs to read as well. Issue: BB-814 --- extensions/gc/tasks/GarbageCollectorTask.js | 5 ++--- .../tasks/LifecycleColdStatusArchiveTask.js | 5 ++--- .../LifecycleResetTransitionInProgressTask.js | 5 ++--- extensions/lifecycle/tasks/LifecycleTask.js | 11 ++-------- .../tasks/LifecycleUpdateTransitionTask.js | 22 ++++++------------- lib/util/transitionAttempt.js | 20 +++++++++++++++++ tests/unit/lifecycle/LifecycleTask.spec.js | 8 +++---- 7 files changed, 39 insertions(+), 37 deletions(-) create mode 100644 lib/util/transitionAttempt.js diff --git a/extensions/gc/tasks/GarbageCollectorTask.js b/extensions/gc/tasks/GarbageCollectorTask.js index dae27c5807..60f2bd3d5a 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 { TRANSITION_ATTEMPT_MD } = require('../../../lib/util/transitionAttempt'); /** @typedef { import('../GarbageCollector.js') } GarbageCollector */ class GarbageCollectorTask extends BackbeatTask { @@ -291,9 +292,7 @@ class GarbageCollectorTask extends BackbeatTask { .setAmzStorageClass(newLocation) .setOriginOp('s3:LifecycleTransition') .setTransitionInProgress(false) - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }); + .setUserMetadata({ [TRANSITION_ATTEMPT_MD]: undefined }); 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..9e09dbdff2 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 { TRANSITION_ATTEMPT_MD } = require('../../../lib/util/transitionAttempt'); class SkipMdUpdateError extends Error {} @@ -116,9 +117,7 @@ class LifecycleColdStatusArchiveTask extends LifecycleUpdateTransitionTask { .setAmzStorageClass(coldLocation) .setTransitionInProgress(false) .setOriginOp('s3:LifecycleTransition') - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }); + .setUserMetadata({ [TRANSITION_ATTEMPT_MD]: undefined }); } this._putMetadata(entry, objectMD, log, err => { diff --git a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js index 8768d577fe..61e88b74ac 100644 --- a/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js +++ b/extensions/lifecycle/tasks/LifecycleResetTransitionInProgressTask.js @@ -1,6 +1,7 @@ 'use strict'; const { LifecycleRequeueTask } = require('./LifecycleRequeueTask'); +const { TRANSITION_ATTEMPT_MD } = 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_, - }); + md.setUserMetadata({ [TRANSITION_ATTEMPT_MD]: try_ }); return true; } diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js index 8a83118404..6f263317fc 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); const entry = ReplicationAPI.createCopyLocationAction({ bucketName: params.bucket, diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 4f57c33c7f..06ef34dd7e 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -6,6 +6,10 @@ const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry'); const ObjectMD = require('arsenal').models.ObjectMD; const { LifecycleMetrics } = require('../LifecycleMetrics'); +const { + TRANSITION_ATTEMPT_MD, + getTransitionAttempt, +} = require('../../../lib/util/transitionAttempt'); /** @typedef { import('../objectProcessor/LifecycleObjectProcessor.js') } LifecycleObjectProcessor */ class LifecycleUpdateTransitionTask extends BackbeatTask { @@ -71,9 +75,7 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { .setDataStoreName(newLocationName) .setAmzStorageClass(newLocationName) .setOriginOp('s3:LifecycleTransition') - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': undefined, - }) + .setUserMetadata({ [TRANSITION_ATTEMPT_MD]: undefined }) .setTransitionInProgress(false); } @@ -232,20 +234,10 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { next(err, objMD); }), (objMD, next) => { - const userMDStr = objMD.getUserMetadata() || '{}'; - const userMD = JSON.parse(userMDStr); - - let tryCount = userMD['x-amz-meta-scal-s3-transition-attempt']; - if (tryCount === undefined) { - tryCount = 1; - } else { - tryCount = parseInt(tryCount, 10) + 1; - } + const tryCount = (getTransitionAttempt(objMD) || 0) + 1; objMD.setTransitionInProgress(false) - .setUserMetadata({ - 'x-amz-meta-scal-s3-transition-attempt': tryCount, - }); + .setUserMetadata({ [TRANSITION_ATTEMPT_MD]: 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..605331aeea --- /dev/null +++ b/lib/util/transitionAttempt.js @@ -0,0 +1,20 @@ +// Kept in the user metadata so it survives across processes: the transition +// processor bumps it on failure, and clears it once the object reached its new +// location. +const TRANSITION_ATTEMPT_MD = 'x-amz-meta-scal-s3-transition-attempt'; + +/** + * Read the transition attempt count of an object. + * @param {ObjectMD} objMD - object metadata + * @return {Number|undefined} attempt count, or undefined if the object was + * never transitioned + */ +function getTransitionAttempt(objMD) { + const attempt = Number.parseInt(objMD.getValue()[TRANSITION_ATTEMPT_MD], 10); + return Number.isInteger(attempt) ? attempt : undefined; +} + +module.exports = { + TRANSITION_ATTEMPT_MD, + getTransitionAttempt, +}; diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe147..c039f73ebc 100644 --- a/tests/unit/lifecycle/LifecycleTask.spec.js +++ b/tests/unit/lifecycle/LifecycleTask.spec.js @@ -2374,7 +2374,7 @@ describe('lifecycle task helper methods', () => { getDataStoreName: () => 'local-site', getDataStoreVersionId: () => 'version-123', getContentLength: () => 1024, - getUserMetadata: () => null, + getValue: () => ({}), }; sinon.stub(lifecycleTask, '_canUnconditionallyGarbageCollect').returns(true); @@ -2415,7 +2415,7 @@ describe('lifecycle task helper methods', () => { getDataStoreName: () => 'local-site', getDataStoreVersionId: () => 'version-123', getContentLength: () => 1024, - getUserMetadata: () => JSON.stringify({ + getValue: () => ({ 'x-amz-meta-scal-s3-transition-attempt': '3' }), }; @@ -2434,7 +2434,7 @@ describe('lifecycle task helper methods', () => { getDataStoreName: () => 'aws-location', getDataStoreVersionId: () => null, getContentLength: () => 2048, - getUserMetadata: () => null, + getValue: () => ({}), getLocation: () => [{ name: 'aws-location', dataStoreVersionId: null }], }; @@ -2456,7 +2456,7 @@ describe('lifecycle task helper methods', () => { getDataStoreName: () => 'aws-location', getDataStoreVersionId: () => null, getContentLength: () => 2048, - getUserMetadata: () => null, + getValue: () => ({}), getLocation: () => [{ name: 'aws-location', dataStoreVersionId: null }], }; From 20575f6a116cc9a7dd30ca0e4afd132123308bdd Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 23:38:34 +0200 Subject: [PATCH 2/6] Report copyLocation results under the metrics type of their flow copyLocation actions were all reported as transitions, which is about to stop being true: pull replication reuses the same pipeline but times a different thing, the delay from an object's metadata landing locally to its data being copied, over a different population of objects. Folding both under type=transition would give quantiles that describe neither, and would dilute the LifecycleLatency alert, which groups by type. 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 | 15 +++++++++++++++ .../tasks/LifecycleUpdateTransitionTask.js | 4 ++-- .../replication/tasks/CopyLocationTask.js | 4 ++-- tests/unit/lifecycle/LifecycleMetrics.spec.js | 19 +++++++++++++++++++ 4 files changed, 38 insertions(+), 4 deletions(-) diff --git a/extensions/lifecycle/LifecycleMetrics.js b/extensions/lifecycle/LifecycleMetrics.js index bde4315530..5363a8d9c4 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 PULL_REPLICATION_TYPE = 'pullReplication'; + // 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,21 @@ class LifecycleMetrics { } } +/** + * Metrics type of a copyLocation action. + * @param {ActionQueueEntry} actionEntry - copyLocation action + * @return {string} metrics type + */ +function getCopyLocationMetricsType(actionEntry) { + return actionEntry.getAttribute('metrics.origin') === PULL_REPLICATION_TYPE ? + PULL_REPLICATION_TYPE : TRANSITION_TYPE; +} + module.exports = { DEFAULT_SCAN_METRIC_RETENTION_S, LifecycleMetrics, LIFECYCLE_MARKER_METRICS_LOCATION, + PULL_REPLICATION_TYPE, + getCopyLocationMetricsType, resetLifecycleScanMetricCleanupTimers, }; diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js index 06ef34dd7e..435e3ad360 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 { TRANSITION_ATTEMPT_MD, getTransitionAttempt, @@ -202,7 +202,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..d024bb128f 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', () => { + [ + ['pullReplication', 'pullReplication'], + ['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 d88229985d5e0865cfecf930956e04872f8da4c7 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:28:16 +0200 Subject: [PATCH 3/6] 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 | 71 +++++++++++++- extensions/lifecycle/objectProcessor/task.js | 2 +- .../tasks/LifecycleUpdateTransitionTask.js | 52 +++++++++- ...LifecycleObjectTransitionProcessor.spec.js | 95 +++++++++++++++++++ .../LifecycleUpdateTransitionTask.spec.js | 43 +++++++++ tests/unit/mocks.js | 17 ++++ 6 files changed, 274 insertions(+), 6 deletions(-) diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js index e390280660..16733e82de 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,68 @@ 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); + + const authConfig = this.getAuthConfig(this._lcConfig); + if (authConfig.type === authTypeAssumeRole) { + 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 (pull replication) 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.vaultClientWrapper) { + log.debug('skipping: not assume role auth type'); + return process.nextTick(cb); + } + + // 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 +119,7 @@ class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor { * @return {undefined} */ start(done) { + this.vaultClientWrapper?.init(); async.waterfall([ next => super.start(next), next => { @@ -226,8 +290,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 435e3ad360..30f860a248 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -244,6 +244,43 @@ class LifecycleUpdateTransitionTask extends BackbeatTask { ], done); } + /** + * Actions published by the lifecycle conductor carry the account id; those + * published by the queue populator (pull replication) 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 @@ -260,11 +297,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..38ab73b531 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,98 @@ 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 hold readiness back without assume role auth', () => { + objectProcessor._consumers = { isReady: () => true }; + assert.strictEqual(objectProcessor.isReady(), true); + }); + }); }); diff --git a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js index 61748e96e3..d2d0473e5b 100644 --- a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js +++ b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js @@ -169,4 +169,47 @@ describe('LifecycleUpdateTransitionTask', () => { done(); }); }); + + // pull replication 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 080dbdfde8a7063a7041557eef46530be503576e Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Tue, 25 Aug 2026 22:28:37 +0200 Subject: [PATCH 4/6] Trigger pull replication of objects still on the source 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. Pulling data in 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 | 10 +- .../replication/ReplicationQueuePopulator.js | 142 ++++++++- tests/unit/ReplicationMetric.js | 24 +- .../lifecycle/CircuitBreakerGroup.spec.js | 10 + .../ReplicationQueuePopulator.spec.js | 276 ++++++++++++++++++ 6 files changed, 439 insertions(+), 30 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..86d0931706 100644 --- a/extensions/replication/ReplicationMetric.js +++ b/extensions/replication/ReplicationMetric.js @@ -46,11 +46,6 @@ class ReplicationMetric { return this; } - _isLifecycleAction() { - const { origin } = this._entry.getContext(); - return origin !== undefined && origin === 'lifecycle'; - } - _createProducerMessage() { const { bucket, key, version } = this._entry.getAttribute('target'); const metricsModel = new MetricsModel() @@ -65,8 +60,9 @@ class ReplicationMetric { } publish() { - // Lifecycle metrics not yet implemented. - if (this._isLifecycleAction()) { + // Metrics API/routes only support CRR + const { origin } = this._entry.getContext(); + if (['lifecycle', 'pullReplication'].includes(origin)) { return undefined; } const message = this._createProducerMessage(); diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index 22c31ad0bb..bc97cb7383 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -1,11 +1,16 @@ 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, PULL_REPLICATION_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; class ReplicationQueuePopulator extends QueuePopulatorExtension { @@ -13,6 +18,12 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { super(params); this.repConfig = params.config; this.metricsHandler = params.metricsHandler; + this.transitionTasksTopic = config.extensions?.lifecycle?.transitionTasksTopic; + + // Where the data is fetched to when the object metadata does not name a usable + // target: the first location that can actually hold a local copy. + this.defaultLocalLocation = Object.keys(locationsConfig).find( + name => !locationsConfig[name].isCold && !locationsConfig[name].isCRR); } filter(entry) { @@ -73,6 +84,14 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { if (sanityCheckRes) { return; } + const locationConfig = locationsConfig[queueEntry.getDataStoreName()] || {}; + // Data still on the source location has to be fetched first. This is + // unrelated to replicationInfo, which tracks replication of a *local* + // object to remote sites, hence the check before any of its conditions. + if (locationConfig.isCRR && this.transitionTasksTopic) { + this._publishPullReplicationAction(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 +100,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 +140,124 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { traceHeaders); } + /** + * Queue a copyLocation action for an object whose data still lives on the + * source location: the data mover copies it over, and the transition + * processor merges the new location 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} + */ + _publishPullReplicationAction(entry, queueEntry, value) { + // Those buckets are versioned, and the metadata layer will repair the + // master key once the version has been copied over. + 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 fetch. Any other + // object without location information is inconsistent. + if (queueEntry.getContentLength() > 0) { + this.log.error('non-empty object without location, skipping pull replication', { + method: 'ReplicationQueuePopulator._publishPullReplicationAction', + ...queueEntry.getLogInfo(), + dataStoreName: queueEntry.getDataStoreName(), + contentLength: queueEntry.getContentLength(), + }); + } + return; + } + + const bucket = queueEntry.getBucket(); + const objectKey = queueEntry.getObjectKey(); + const contentLength = queueEntry.getContentLength(); + const targetLocation = this._getPullReplicationTarget(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: PULL_REPLICATION_TYPE, + fromLocation: queueEntry.getDataStoreName(), + contentLength, + resultsTopic: this.transitionTasksTopic, + transitionTime: transitionTime.toISOString(), + attempt: getTransitionAttempt(queueEntry), + }); + // 'transition' is what the lifecycle transition processor dispatches + // on to pick up the copyLocation result. + action.addContext({ + origin: PULL_REPLICATION_TYPE, + ruleType: 'transition', + bucketName: bucket, + objectKey, + versionId: value.versionId, + }); + action.setAttribute('source', { + bucket, + objectKey, + storageClass: queueEntry.getDataStoreName(), + }); + + LifecycleMetrics.onLifecycleTriggered(this.log, 'queuePopulator', + PULL_REPLICATION_TYPE, targetLocation, Date.now() - transitionTime.getTime()); + + this.log.trace('publishing pull replication entry', { entry: queueEntry.getLogInfo() }); + this.publish(ReplicationAPI.getDataMoverTopic(), + `${bucket}/${objectKey}`, + action.toKafkaMessage(), + undefined, + traceHeadersFromEntry(value)); + } + + /** + * Local location the object data must be copied to. + * @param {ObjectQueueEntry} queueEntry - parsed entry + * @param {Object[]} locations - object data locations + * @return {String|undefined} target location, undefined if there is none + */ + _getPullReplicationTarget(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. Neither is recoverable here, and + // copying the data elsewhere beats leaving it on the source forever. + if (!this.defaultLocalLocation) { + this.log.error('invalid target location and no local location ' + + 'to fall back to, skipping pull replication', { + method: 'ReplicationQueuePopulator._getPullReplicationTarget', + ...queueEntry.getLogInfo(), + targetLocation, + }); + return undefined; + } + this.log.error('invalid target location in object metadata', { + method: 'ReplicationQueuePopulator._getPullReplicationTarget', + ...queueEntry.getLogInfo(), + targetLocation, + fallbackLocation: this.defaultLocalLocation, + }); + return this.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..f6098b3feb 100644 --- a/tests/unit/ReplicationMetric.js +++ b/tests/unit/ReplicationMetric.js @@ -56,27 +56,13 @@ describe('ReplicationMetric', () => { .forEach(key => assert.strictEqual(data[key], mock[key])); }); - it('::_isLifecycleAction should return false by default', () => { - metric.withEntry(entry); - assert.strictEqual(metric._isLifecycleAction(), false); - }); - - it('::_isLifecycleAction should return true when origin is lifecycle', - () => { - entry.setAttribute('contextInfo', { - origin: 'lifecycle', - }); + ['lifecycle', 'pullReplication'].forEach(origin => { + it(`::publish should not send data to topic for a ${origin} action`, () => { + entry.setAttribute('contextInfo', { origin }); metric.withEntry(entry); - assert.strictEqual(metric._isLifecycleAction(), true); + metric.publish(); + assert.strictEqual(sentMessages.length, 0); }); - - it('::publish should not send data to topic if lifecycle task', () => { - entry.setAttribute('contextInfo', { - origin: 'lifecycle', - }); - 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..5204ea3cd1 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,273 @@ describe('replication queue populator', () => { assert.deepStrictEqual(rqp.getState(), {}); }); }); + +/** + * Records every published message, whatever the topic, so pull replication + * 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: pull replication', () => { + 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/source-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(() => { + sinon.restore(); + }); + + it('should publish a copyLocation action for an object still on the source', () => { + 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, 'pullReplication'); + assert.strictEqual(action.metrics.origin, 'pullReplication'); + 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', 'pullReplication', TARGET_LOCATION, + sinon.match.number); + sinon.assert.notCalled(params.metricsHandler.objects); + }); + + // pull replication 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 pulled, and behave as before. + it('should not pull 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 pull 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); + }); + + it('should not pull anything when lifecycle is not configured', () => { + sinon.replace(config, 'extensions', { + ...config.extensions, + lifecycle: undefined, + }); + const populator = new RecordingQueuePopulatorMock(params); + + populator._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual( + populator.published.filter( + p => p.topic === ReplicationAPI.getDataMoverTopic()).length, + 0); + 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 pull 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 pull replication when there is no local location', () => { + sinon.stub(rqp, '_getPullReplicationTarget').returns(undefined); + rqp._filterKeyOp(makeEntry(makeValue())); + + assert.strictEqual(rqp.published.length, 0); + sinon.assert.notCalled(triggeredMetric); + }); +}); From ce4ef0020577dc20d8c59c372bf9c4e5c81615ef Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Fri, 4 Sep 2026 07:42:48 +0200 Subject: [PATCH 5/6] Report pulled objects on the replication metrics The queue populator counted the objects it queued for replication, and the data mover reported the bytes it copied once they arrived, but nothing counted the objects queued for pull replication: a growing backlog was indistinguishable from an idle cluster. Count those too, and tell the two flows apart with the direction the data is moving. 'origin' cannot do it: it names the extensions loaded in the populator, and is the same either way. Their partition label is left empty, as the populator batches its publishes and only learns the partition long after queuing; an empty label and a missing one are the same series to prometheus. Issue: BB-814 --- extensions/replication/ReplicationMetrics.js | 17 +++++----- .../replication/ReplicationQueuePopulator.js | 28 +++++++++++----- extensions/replication/constants.js | 4 +++ lib/queuePopulator/QueuePopulator.js | 9 ++++-- .../ReplicationQueuePopulator.spec.js | 32 +++++++++++++++++-- 5 files changed, 69 insertions(+), 21 deletions(-) diff --git a/extensions/replication/ReplicationMetrics.js b/extensions/replication/ReplicationMetrics.js index ec9ae1863d..15a1d6707d 100644 --- a/extensions/replication/ReplicationMetrics.js +++ b/extensions/replication/ReplicationMetrics.js @@ -83,17 +83,18 @@ class ReplicationMetrics extends ZenkoMetrics { const fromLocationType = _getReplicationEndpointType(fromLocation); const toLocationType = _getReplicationEndpointType(toLocation); - replicationQueuedTotal.inc({ + const labels = { origin: originLabel, fromLocation, fromLocationType, - toLocation, toLocationType, partition, - }); + toLocation, toLocationType, + // Empty when queued by the populator, which batches its publishes + // and only learns their partition much later. + partition: partition ?? '', + }; - replicationQueuedBytes.inc({ - origin: originLabel, - fromLocation, fromLocationType, - toLocation, toLocationType, partition, - }, Number.parseInt(contentLength, 10)); + replicationQueuedTotal.inc(labels); + + replicationQueuedBytes.inc(labels, Number.parseInt(contentLength, 10)); } static onReplicationProcessed(originLabel, fromLocation, toLocation, diff --git a/extensions/replication/ReplicationQueuePopulator.js b/extensions/replication/ReplicationQueuePopulator.js index bc97cb7383..f4a841360d 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -6,12 +6,14 @@ const QueuePopulatorExtension = require('../../lib/queuePopulator/QueuePopulatorExtension'); const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry'); const ReplicationAPI = require('./ReplicationAPI'); +const ReplicationMetrics = require('./ReplicationMetrics'); const { LifecycleMetrics, PULL_REPLICATION_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 { replicationDirections } = require('./constants'); class ReplicationQueuePopulator extends QueuePopulatorExtension { constructor(params) { @@ -117,14 +119,12 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { this._incrementMetrics(backend.site, bytes); }); - // TODO: replication specific metrics go here - this.metricsHandler.bytes( - entry.logReader.getMetricLabels(), - bytes - ); - this.metricsHandler.objects( - entry.logReader.getMetricLabels() - ); + const metricLabels = { + ...entry.logReader.getMetricLabels(), + direction: replicationDirections.push, + }; + this.metricsHandler.bytes(metricLabels, bytes); + this.metricsHandler.objects(metricLabels); const publishedEntry = Object.assign({}, entry); delete publishedEntry.logReader; @@ -224,6 +224,18 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension { action.toKafkaMessage(), undefined, traceHeadersFromEntry(value)); + + // Counterpart of the bytes the data mover reports once the copy + // completed, to measure the pull replication backlog. + ReplicationMetrics.onReplicationQueued(PULL_REPLICATION_TYPE, + queueEntry.getDataStoreName(), targetLocation, contentLength); + + const metricLabels = { + ...entry.logReader.getMetricLabels(), + direction: replicationDirections.pull, + }; + this.metricsHandler.bytes(metricLabels, contentLength); + this.metricsHandler.objects(metricLabels); } /** diff --git a/extensions/replication/constants.js b/extensions/replication/constants.js index 300fa8d003..934e414e36 100644 --- a/extensions/replication/constants.js +++ b/extensions/replication/constants.js @@ -35,6 +35,10 @@ const constants = { failedCRR: testIsOn ? 'test:bb:crr:failed' : 'bb:crr:failed', }, replicationBackends: ['aws_s3', 'azure', 'gcp'], + replicationDirections: { + push: 'push', + pull: 'pull', + }, replicationStages: { sourceDataRead: 'ReplicationSourceDataRead', destinationDataWrite: 'ReplicationDestinationDataWrite', diff --git a/lib/queuePopulator/QueuePopulator.js b/lib/queuePopulator/QueuePopulator.js index 8b63e3d055..9c38f0e10f 100644 --- a/lib/queuePopulator/QueuePopulator.js +++ b/lib/queuePopulator/QueuePopulator.js @@ -62,16 +62,21 @@ const messageMetrics = ZenkoMetrics.createCounter({ labelNames: [...metricLabels, 'publishStatus'], }); +// Which way the data is moving: 'push' out to a remote site, or 'pull' in from +// a source cluster. Not to be confused with 'origin', which names the +// extensions loaded in the populator and is the same for both. +const DIRECTION_LABEL = 'direction'; + const objectMetrics = ZenkoMetrics.createCounter({ name: 's3_replication_populator_objects_total', help: 'Total objects queued for replication', - labelNames: metricLabels, + labelNames: [...metricLabels, DIRECTION_LABEL], }); const byteMetrics = ZenkoMetrics.createCounter({ name: 's3_replication_populator_bytes_total', help: 'Total number of bytes queued for replication not including metadata', - labelNames: metricLabels, + labelNames: [...metricLabels, DIRECTION_LABEL], }); const notificationEvent = ZenkoMetrics.createCounter({ diff --git a/tests/unit/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index 5204ea3cd1..9b1f9a34dd 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -6,6 +6,7 @@ const { encode } = require('arsenal').versioning.VersionID; const ReplicationQueuePopulator = require('../../../extensions/replication/ReplicationQueuePopulator'); const ReplicationAPI = require('../../../extensions/replication/ReplicationAPI'); +const ReplicationMetrics = require('../../../extensions/replication/ReplicationMetrics'); const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics'); const config = require('../../../lib/Config'); @@ -240,14 +241,15 @@ describe('replication queue populator', () => { rqp._filterKeyOp(entry); + const expectedLabels = { ...labels, direction: 'push' }; sinon.assert.calledOnceWithExactly( params.metricsHandler.bytes, - labels, + expectedLabels, 128 ); sinon.assert.calledOnceWithExactly( params.metricsHandler.objects, - labels + expectedLabels ); }); @@ -419,6 +421,7 @@ describe('replication queue populator: pull replication', () => { let params; let rqp; let triggeredMetric; + let queuedMetric; function makeValue(overrides = {}) { return JSON.stringify({ @@ -462,6 +465,7 @@ describe('replication queue populator: pull replication', () => { }; rqp = new RecordingQueuePopulatorMock(params); triggeredMetric = sinon.stub(LifecycleMetrics, 'onLifecycleTriggered'); + queuedMetric = sinon.stub(ReplicationMetrics, 'onReplicationQueued'); }); afterEach(() => { @@ -510,7 +514,29 @@ describe('replication queue populator: pull replication', () => { sinon.assert.calledOnceWithExactly(triggeredMetric, rqp.log, 'queuePopulator', 'pullReplication', TARGET_LOCATION, sinon.match.number); - sinon.assert.notCalled(params.metricsHandler.objects); + }); + + it('should count the object on the populator metrics as pulled', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + const expectedLabels = { ...labels, direction: 'pull' }; + sinon.assert.calledOnceWithExactly(params.metricsHandler.objects, + expectedLabels); + sinon.assert.calledOnceWithExactly(params.metricsHandler.bytes, + expectedLabels, 128); + }); + + it('should report the object as queued for pull replication', () => { + rqp._filterKeyOp(makeEntry(makeValue())); + + sinon.assert.calledOnceWithExactly(queuedMetric, 'pullReplication', + CRR_LOCATION, TARGET_LOCATION, 128); + }); + + it('should not report an object it skipped as queued', () => { + rqp._filterKeyOp(makeEntry(makeValue({ isDeleteMarker: true }))); + + sinon.assert.notCalled(queuedMetric); }); // pull replication is about where the data lives, forward replication is From 8b96b6f9b62e0853763a4eb71f8bc9e6952249e0 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Fri, 4 Sep 2026 23:21:05 +0200 Subject: [PATCH 6/6] Keep the configuration of the extensions a populator does not run BACKBEAT_QUEUEPOPULATOR_EXTENSIONS selects the extensions a queue populator runs, but it was applied by removing the others from the configuration altogether. Anything reading another extension's settings -a topic they share, for instance- then got nothing, in the populator processes only. Select them where they are loaded instead, and leave the configuration alone. Issue: BB-814 --- lib/Config.js | 11 ---------- lib/queuePopulator/QueuePopulator.js | 17 ++++++++++++++- tests/unit/QueuePopulator.spec.js | 31 ++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 12 deletions(-) diff --git a/lib/Config.js b/lib/Config.js index 09416610f6..8b13bf8c1c 100644 --- a/lib/Config.js +++ b/lib/Config.js @@ -132,17 +132,6 @@ class Config extends EventEmitter { parsedConfig.internalCertFilePaths); } - // Overwrite extension configs if configured - // We can specify the list of extensions that should be handled by this - // instance of the queuePopulator - const configuredExtensions = process.env.BACKBEAT_QUEUEPOPULATOR_EXTENSIONS; - if (configuredExtensions) { - const allowedExtensions = configuredExtensions.split(','); - const filteredExtensions = Object.entries(parsedConfig.extensions) - .filter(entry => allowedExtensions.includes(entry[0])); - parsedConfig.extensions = Object.fromEntries(filteredExtensions); - } - // config is validated, safe to assign directly to the config object Object.assign(this, parsedConfig); diff --git a/lib/queuePopulator/QueuePopulator.js b/lib/queuePopulator/QueuePopulator.js index 9c38f0e10f..44a5a77b78 100644 --- a/lib/queuePopulator/QueuePopulator.js +++ b/lib/queuePopulator/QueuePopulator.js @@ -164,7 +164,7 @@ class QueuePopulator { this.mConfig = mConfig; this.rConfig = rConfig; this.vConfig = vConfig; - this.extConfigs = extConfigs; + this.extConfigs = QueuePopulator._filterConfiguredExtensions(extConfigs); this.log = new Logger('Backbeat:QueuePopulator'); @@ -183,6 +183,21 @@ class QueuePopulator { this._circuitBreaker = new CircuitBreaker(qpConfig.circuitBreaker); } + /** + * Return the extensions this populator instance runs. + * @param {Object} extConfigs - configuration of every configured extension + * @return {Object} configuration of the extensions to run here + */ + static _filterConfiguredExtensions(extConfigs) { + const configured = process.env.BACKBEAT_QUEUEPOPULATOR_EXTENSIONS; + if (!configured) { + return extConfigs; + } + const names = configured.split(','); + return Object.fromEntries( + Object.entries(extConfigs).filter(([name]) => names.includes(name))); + } + /** * Open the queue populator * diff --git a/tests/unit/QueuePopulator.spec.js b/tests/unit/QueuePopulator.spec.js index 4ab032fddf..955de2b3a7 100644 --- a/tests/unit/QueuePopulator.spec.js +++ b/tests/unit/QueuePopulator.spec.js @@ -361,4 +361,35 @@ describe('QueuePopulator', () => { }); }); }); + + describe('configured extensions', () => { + const extConfigs = { replication: {}, lifecycle: {}, notification: {} }; + + function buildPopulator() { + return new QueuePopulator({}, {}, { logSource: 'bucketd' }, + null, null, null, null, extConfigs); + } + + afterEach(() => { + delete process.env.BACKBEAT_QUEUEPOPULATOR_EXTENSIONS; + }); + + it('should run every extension by default', () => { + assert.deepStrictEqual(Object.keys(buildPopulator().extConfigs), + ['replication', 'lifecycle', 'notification']); + }); + + it('should only run the extensions it is configured with', () => { + process.env.BACKBEAT_QUEUEPOPULATOR_EXTENSIONS = 'replication'; + assert.deepStrictEqual(Object.keys(buildPopulator().extConfigs), + ['replication']); + }); + + it('should leave the configuration of the other extensions alone', () => { + process.env.BACKBEAT_QUEUEPOPULATOR_EXTENSIONS = 'replication'; + buildPopulator(); + assert.deepStrictEqual(Object.keys(extConfigs), + ['replication', 'lifecycle', 'notification']); + }); + }); });