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..60f2bd3d5 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/LifecycleMetrics.js b/extensions/lifecycle/LifecycleMetrics.js index bde431553..5363a8d9c 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/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js index e39028066..16733e82d 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 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..9e09dbdff 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 8768d577f..61e88b74a 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 8a8311840..6f263317f 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 4f57c33c7..30f860a24 100644 --- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js +++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js @@ -5,7 +5,11 @@ 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, +} = 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); } @@ -200,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); }); @@ -232,26 +234,53 @@ 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); }, ], 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 @@ -268,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/extensions/replication/ReplicationMetric.js b/extensions/replication/ReplicationMetric.js index 607e99ab5..86d093170 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/ReplicationMetrics.js b/extensions/replication/ReplicationMetrics.js index ec9ae1863..15a1d6707 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 22c31ad0b..52443734c 100644 --- a/extensions/replication/ReplicationQueuePopulator.js +++ b/extensions/replication/ReplicationQueuePopulator.js @@ -1,12 +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 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'); + +const { transitionTasksTopic } = config.extensions.lifecycle; + +// 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. +const defaultLocalLocation = Object.keys(locationsConfig).find( + name => !locationsConfig[name].isCold && !locationsConfig[name].isCRR); class ReplicationQueuePopulator extends QueuePopulatorExtension { constructor(params) { @@ -73,6 +87,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._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 +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; } @@ -101,14 +120,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; @@ -124,6 +141,136 @@ 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: 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)); + + // 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); + } + + /** + * 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 (!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: 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/constants.js b/extensions/replication/constants.js index 300fa8d00..934e414e3 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/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/queuePopulator/QueuePopulator.js b/lib/queuePopulator/QueuePopulator.js index 8b63e3d05..9c38f0e10 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/lib/util/transitionAttempt.js b/lib/util/transitionAttempt.js new file mode 100644 index 000000000..605331aee --- /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/ReplicationMetric.js b/tests/unit/ReplicationMetric.js index bffbae903..f6098b3fe 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 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..d024bb128 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'); diff --git a/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js b/tests/unit/lifecycle/LifecycleObjectTransitionProcessor.spec.js index aa5138851..38ab73b53 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/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js index 7150bbe14..c039f73eb 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 }], }; diff --git a/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js b/tests/unit/lifecycle/LifecycleUpdateTransitionTask.spec.js index 61748e96e..d2d0473e5 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 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..0ed2f4f6f 100644 --- a/tests/unit/replication/ReplicationQueuePopulator.spec.js +++ b/tests/unit/replication/ReplicationQueuePopulator.spec.js @@ -1,8 +1,15 @@ 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 ReplicationMetrics = require('../../../extensions/replication/ReplicationMetrics'); +const { LifecycleMetrics } = + require('../../../extensions/lifecycle/LifecycleMetrics'); +const config = require('../../../lib/Config'); const fakeLogger = require('../../utils/fakeLogger'); @@ -234,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 ); }); @@ -382,3 +390,282 @@ 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; + let queuedMetric; + + 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'); + queuedMetric = sinon.stub(ReplicationMetrics, 'onReplicationQueued'); + }); + + afterEach(() => { + triggeredMetric.restore(); + queuedMetric.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); + }); + + 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 + // 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); + }); + + // 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); + }); +});