diff --git a/conf/locationConfig.json b/conf/locationConfig.json index 8ba3dc334..dceb31dda 100644 --- a/conf/locationConfig.json +++ b/conf/locationConfig.json @@ -42,5 +42,12 @@ "legacyAwsBehavior": false, "isCold": true, "details": {} + }, + "location-crr-source": { + "type": "scality", + "objectId": "location-crr-source", + "legacyAwsBehavior": false, + "isCRR": true, + "details": {} } } diff --git a/extensions/gc/tasks/GarbageCollectorTask.js b/extensions/gc/tasks/GarbageCollectorTask.js index dae27c580..e745652a0 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/LifecycleMetrics.js b/extensions/lifecycle/LifecycleMetrics.js index bde431553..020a3c87a 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/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js index e39028066..aa0c31943 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 d09c2020a..d7bd60ab4 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/LifecycleColdStatusArchiveTask.js b/extensions/lifecycle/tasks/LifecycleColdStatusArchiveTask.js index 8cea3c62b..f81eece66 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 8768d577f..95b27bf6a 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 8a8311840..ff3bcf9df 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 4f57c33c7..37e1337a9 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -5,7 +5,12 @@ 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, + 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) { @@ -200,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); }); @@ -232,26 +235,53 @@ 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); }, ], 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 @@ -268,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/extensions/replication/ReplicationMetric.js b/extensions/replication/ReplicationMetric.js index 607e99ab5..585cbb3e2 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 22c31ad0b..3ec59ec47 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/extensions/replication/tasks/CopyLocationTask.js b/extensions/replication/tasks/CopyLocationTask.js index b7d455e90..8ee326f5b 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/lib/util/transitionAttempt.js b/lib/util/transitionAttempt.js new file mode 100644 index 000000000..51d8260e6 --- /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, +}; diff --git a/tests/unit/ReplicationMetric.js b/tests/unit/ReplicationMetric.js index bffbae903..a11b85ee1 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 56e825d45..c8c8737b6 100644 --- a/tests/unit/lifecycle/CircuitBreakerGroup.spec.js +++ b/tests/unit/lifecycle/CircuitBreakerGroup.spec.js @@ -436,6 +436,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], @@ -493,6 +498,11 @@ describe('extractBucketProcessorCircuitBreakerConfigs', () => { '${location}', 'location-dmf-v1', ), + formatProbeConfig( + topicSpecificLocationTemplateProbe, + '${location}', + 'location-crr-source', + ), ], }, global: [], diff --git a/tests/unit/lifecycle/LifecycleMetrics.spec.js b/tests/unit/lifecycle/LifecycleMetrics.spec.js index 2a6a97384..53e571a09 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'); diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa5138851..e1da53010 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 61748e96e..bebdc9c57 100644 --- a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js +++ b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js @@ -169,4 +169,47 @@ describe('LifecycleUpdateTransitionTask', () => { done(); }); }); + + // clean room localization actions are published by the queue populator, + // which only knows the object owner's canonical id + describe('account id resolution', () => { + it('should not look up the account id when the entry has one', done => { + actionEntry.setAttribute('target.accountId', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 0); + done(); + }); + }); + + it('should resolve the account id from the owner canonical id', done => { + objectProcessor.setAccountId('some-canonical-id', '000000000042'); + actionEntry.setAttribute('target.owner', 'some-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert.ifError(err); + assert.strictEqual(objectProcessor.accountIdLookups, 1); + assert.strictEqual( + actionEntry.getAttribute('target.accountId'), + '000000000042'); + // the garbage collection entry must not resolve it again + const receivedGcEntry = gcProducer.getReceivedEntry(); + assert.strictEqual( + receivedGcEntry.getAttribute('target.accountId'), + '000000000042'); + done(); + }); + }); + + it('should fail the entry when the account id cannot be resolved', + done => { + actionEntry.setAttribute('target.owner', 'unknown-canonical-id'); + task.processActionEntry(actionEntry, err => { + assert(err); + assert.strictEqual( + backbeatMetadataProxyClient.getReceivedMd(), null); + done(); + }); + }); + }); }); diff --git a/tests/unit/mocks.js b/tests/unit/mocks.js index 2b9096f88..99ebb0871 100644 --- a/tests/unit/mocks.js +++ b/tests/unit/mocks.js @@ -1,4 +1,5 @@ const assert = require('assert'); +const { errors } = require('arsenal'); const { ObjectMD } = require('arsenal').models; class GarbageCollectorProducerMock { @@ -157,6 +158,21 @@ class ProcessorMock { this.coldProducer = coldProducer; this._gcConfig = gcConfig; this.logger = logger; + this.accountIds = {}; + this.accountIdLookups = 0; + } + + setAccountId(ownerId, accountId) { + this.accountIds[ownerId] = accountId; + } + + getAccountId(ownerId, log, cb) { + this.accountIdLookups += 1; + const accountId = this.accountIds[ownerId]; + if (!accountId) { + return process.nextTick(cb, errors.NoSuchEntity); + } + return process.nextTick(cb, null, accountId); } getStateVars() { @@ -170,6 +186,7 @@ class ProcessorMock { getBackbeatClient: () => this.backbeatClient, getBackbeatMetadataProxy: () => this.backbeatMetadataProxy, getS3Client: () => this.s3Client, + getAccountId: this.getAccountId.bind(this), }; } } diff --git a/tests/unit/replication/ReplicationQueuePopulator.spec.js b/tests/unit/replication/ReplicationQueuePopulator.spec.js index cc384b695..8628d9579 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); + }); +});