Trigger localization of clean room objects

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,62 @@ 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);
     }
 
+    /**
+     * 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);
+        }
+
+        // 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 +187,9 @@ class LifecycleObjectProcessor extends EventEmitter {
     start(done) {
         this.clientManager.initSTSConfig();
         this.clientManager.initCredentialsManager();
+        if (this.getAuthConfig(this._lcConfig).type === authTypeAssumeRole) {
+            this.vaultClientWrapper.init();
+        }
         this._setupConsumers(done);
     }
 
@@ -225,12 +285,14 @@ 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.vaultClientWrapper.tempCredentialsReady();
     }
 }
 

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(),
 });
 
 /**

extensions/replication/ReplicationQueuePopulator.js

@@ -1,18 +1,25 @@
 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 locationsConfig = require('../../conf/locationConfig.json') || {};
 const safeJsonParse = require('../../lib/util/safeJsonParse');
 const { traceHeadersFromEntry } = require('arsenal/build/lib/tracing').kafka;
 
+const TRANSITION_ATTEMPT_MD = 'x-amz-meta-scal-s3-transition-attempt';
+
 class ReplicationQueuePopulator extends QueuePopulatorExtension {
     constructor(params) {
         super(params);
         this.repConfig = params.config;
         this.metricsHandler = params.metricsHandler;
+        // Clean room: when set, objects whose data still lives on the source
+        // (isCRR) location are queued for localization instead of replication.
+        this.localizationConfig = params.config.localization;
     }
 
     filter(entry) {
@@ -73,6 +80,17 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension {
         if (sanityCheckRes) {
             return;
         }
+        const dataStoreName = queueEntry.getDataStoreName();
+        const locationConfig = (dataStoreName && locationsConfig[dataStoreName])
+            || {};
+        // 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.localizationConfig) {
+            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 +99,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 +139,119 @@ 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 action = ReplicationAPI.createCopyLocationAction({
+            bucketName: bucket,
+            objectKey,
+            owner: queueEntry.getOwnerId(),
+            versionId: value.versionId ? encode(value.versionId) : undefined,
+            eTag: `"${queueEntry.getContentMd5()}"`,
+            lastModified: queueEntry.getLastModified(),
+            toLocation: this.localizationConfig.toLocation,
+            originLabel: 'localization',
+            fromLocation: queueEntry.getDataStoreName(),
+            contentLength,
+            resultsTopic: this.localizationConfig.resultsTopic,
+            transitionTime: new Date(
+                entry.overheadFields?.commitTimestamp ?? Date.now()
+            ).toISOString(),
+            attempt: this._getTransitionAttempt(queueEntry),
+        });
+        // '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(),
+        });
+
+        this.metricsHandler.localizationBytes(
+            entry.logReader.getMetricLabels(),
+            contentLength
+        );
+        this.metricsHandler.localizationObjects(
+            entry.logReader.getMetricLabels()
+        );
+
+        this.log.trace('publishing object localization entry',
+                       { entry: queueEntry.getLogInfo() });
+        this.publish(ReplicationAPI.getDataMoverTopic(),
+                     `${bucket}/${objectKey}`,
+                     action.toKafkaMessage(),
+                     undefined,
+                     traceHeadersFromEntry(value));
+    }
+
+    /**
+     * Number of times the data mover already tried to copy this object. The
+     * transition processor bumps the counter on failure, which produces a new
+     * oplog entry and re-triggers the copy.
+     * @param {ObjectQueueEntry} queueEntry - parsed entry
+     * @return {Number|undefined} attempt count, if any
+     */
+    _getTransitionAttempt(queueEntry) {
+        const umd = queueEntry.getUserMetadata();
+        if (!umd) {
+            return undefined;
+        }
+        const { error, result } = safeJsonParse(umd);
+        if (error) {
+            return undefined;
+        }
+        const attempt = Number.parseInt(result[TRANSITION_ATTEMPT_MD], 10);
+        return Number.isInteger(attempt) ? attempt : undefined;
+    }
+
     /**
      * 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