Skip non-localized entries in replication and lifecycle

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

docker-entrypoint.sh

@@ -154,6 +154,16 @@ if [[ "$EXTENSIONS_REPLICATION_DEST_BOOTSTRAPLIST" ]]; then
     fi
 fi
 
+# Clean room: localize objects whose data still lives on the source (isCRR)
+# location. Setting the target location enables the trigger.
+if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.toLocation=\"$EXTENSIONS_REPLICATION_LOCALIZATION_TO_LOCATION\""
+fi
+
+if [[ "$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC" ]]; then
+    JQ_FILTERS_CONFIG="$JQ_FILTERS_CONFIG | .extensions.replication.localization.resultsTopic=\"$EXTENSIONS_REPLICATION_LOCALIZATION_RESULTS_TOPIC\""
+fi
+
 # START Retry config
 
 # AWS_S3

extensions/lifecycle/LifecycleConfigValidator.js

@@ -71,6 +71,7 @@ const joiSchema = joi.object({
         concurrency: joi.number().greater(0).default(10),
         maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
         probeServer: probeServerJoi.default(),
+        vaultAdmin: hostPortJoi,
         circuitBreaker: joi.object().optional(),
     },
     coldStorageArchiveTopicPrefix: joi.string().default('cold-archive-req-'),

extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js

@@ -2,11 +2,15 @@
 
 const { EventEmitter } = require('events');
 const Logger = require('werelogs').Logger;
+const { errors } = require('arsenal');
 
 const BackbeatConsumerManager = require('../../../lib/BackbeatConsumerManager');
 const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry');
 const ClientManager = require('../../../lib/clients/ClientManager');
 const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
+const VaultClientWrapper = require('../../utils/VaultClientWrapper');
+const { AccountIdCache } = require('../../utils/AccountIdCache');
+const { authTypeAssumeRole } = require('../../../lib/constants');
 
 const logIdFromType = {
     'object-processor': 'Backbeat:Lifecycle:ObjectProcessor',
@@ -61,9 +65,81 @@ class LifecycleObjectProcessor extends EventEmitter {
             transport,
         }, this._log);
 
+        this.vaultClientWrapper = new VaultClientWrapper(
+            `lifecycle:${this.getProcessorType()}`,
+            this._processConfig.vaultAdmin,
+            this.getAuthConfig(this._lcConfig),
+            this._log,
+        );
+        this._accountIdCache = new AccountIdCache(
+            this._processConfig.concurrency);
+
         this.retryWrapper = new BackbeatTask(this._processConfig.retry);
     }
 
+    /**
+     * Whether this processor can resolve canonical ids through Vault. Only the
+     * transition processor receives actions published without an account id
+     * (clean room localization), and only it is configured with a Vault admin
+     * endpoint - so leave the Vault client alone everywhere else.
+     * @return {Boolean} true if account id lookups are available
+     */
+    _accountIdLookupEnabled() {
+        const authConfig = this.getAuthConfig(this._lcConfig);
+        return authConfig.type === authTypeAssumeRole &&
+            !!(this._processConfig.vaultAdmin || authConfig.vault);
+    }
+
+    /**
+     * 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._accountIdLookupEnabled()) {
+            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);
+        });
+    }
+
     getProcessorType() {
         return 'object-processor';
     }
@@ -130,6 +206,9 @@ class LifecycleObjectProcessor extends EventEmitter {
     start(done) {
         this.clientManager.initSTSConfig();
         this.clientManager.initCredentialsManager();
+        if (this._accountIdLookupEnabled()) {
+            this.vaultClientWrapper.init();
+        }
         this._setupConsumers(done);
     }
 
@@ -225,12 +304,15 @@ class LifecycleObjectProcessor extends EventEmitter {
                 this.clientManager.getBackbeatClient.bind(this.clientManager),
             getBackbeatMetadataProxy:
                 this.clientManager.getBackbeatMetadataProxy.bind(this.clientManager),
+            getAccountId: this.getAccountId.bind(this),
             logger: this._log,
         };
     }
 
     isReady() {
-        return this._consumers && this._consumers.isReady();
+        return this._consumers && this._consumers.isReady() &&
+            (!this._accountIdLookupEnabled() ||
+                this.vaultClientWrapper.tempCredentialsReady());
     }
 }
 

extensions/lifecycle/tasks/LifecycleTask.js

@@ -31,6 +31,8 @@ const errorTransitionInProgress = errors.InternalError.
     customizeDescription('transition is currently in progress');
 const errorTransitionColdObject = errors.InternalError.
     customizeDescription('transitioning a cold object is forbidden');
+const errorTransitionNonLocalizedObject = errors.InternalError.
+    customizeDescription('transitioning a non-localized object is forbidden');
 const errorObjectTemporarilyRestored = errors.InternalError.
     customizeDescription('object temporarily restored');
 const errorReplicationInProgress = errors.InternalError.
@@ -1270,12 +1272,21 @@ class LifecycleTask extends BackbeatTask {
                     return next(errorReplicationInProgress);
                 }
                 const dataStoreName = objectMD.getDataStoreName();
-                const isObjectCold = dataStoreName && locationsConfig[dataStoreName]
-                    && locationsConfig[dataStoreName].isCold;
+                const locationConfig = (dataStoreName
+                    && locationsConfig[dataStoreName]) || {};
                 // We do not transition cold objects
-                if (isObjectCold) {
+                if (locationConfig.isCold) {
                     return next(errorTransitionColdObject);
                 }
+                // Clean room: the object data still lives on the source
+                // (isCRR) location. Localization is the only valid transition
+                // out of such a location, and it is triggered by its own path
+                // (the queue populator), not by lifecycle rules. The version
+                // is simply re-evaluated on a later scan, once localization
+                // has completed.
+                if (locationConfig.isCRR) {
+                    return next(errorTransitionNonLocalizedObject);
+                }
                 // If transition is in progress, do not re-publish entry
                 // to data-mover or cold-archive topic.
                 if (objectMD.getTransitionInProgress()) {

extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js

@@ -252,6 +252,36 @@ class LifecycleUpdateTransitionTask extends BackbeatTask {
         ], done);
     }
 
+    /**
+     * Actions published by the lifecycle conductor carry the account id;
+     * those published by the queue populator (clean room localization) only
+     * know the object owner's canonical id. Resolve it once, up-front, so the
+     * rest of the task - and the garbage collection entry it emits - can use
+     * `target.accountId` as usual.
+     * @param {ActionQueueEntry} entry - action entry to execute
+     * @param {Logger} log - logger instance
+     * @param {Function} cb - callback function
+     * @return {undefined}
+     */
+    _resolveAccountId(entry, log, cb) {
+        const { accountId, owner } = this.getTargetAttribute(entry);
+        if (accountId || !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,17 @@ 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/ReplicationConfigValidator.js

@@ -139,6 +139,13 @@ const joiSchema = joi.object({
         probeServer: probeServerPerSite,
     }).optional(),
     objectSizeMetrics: joi.array().items(joi.number()).default(OBJECT_SIZE_METRICS),
+    // Clean room: localization of objects whose data still lives on the source
+    // (isCRR) location. Enabled by setting `toLocation`.
+    localization: joi.object({
+        toLocation: joi.string().required(),
+        resultsTopic: joi.string()
+            .default('backbeat-lifecycle-transition-tasks'),
+    }).optional(),
 });
 
 /**