Pull replication of objects still on the source location

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": {}
   }
 }

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) {

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,
 };

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;

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:

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 => {

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;
     }
 

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,

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);
+        });
     }
 }
 

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();