Trigger localization of clean room objects

In a clean room, objects are created locally but their metadata still
points at the source cluster's location: the data itself has not been
copied over yet. Something has to notice those objects and ask for the
data to be pulled in.

The queue populator is the natural place for it, since bootstrap,
re-bootstrap and streamed updates all go through the same oplog. When an
object lands on a location flagged isCRR, publish a copyLocation action
on the data mover topic and let the existing data mover + transition
merge pipeline do the actual copy. The destination comes from the object
metadata, which the source-side rewrite stamps as it prepares the entry;
if it names a location we do not know, fall back to the first local one
and log about it.

This is unrelated to replicationInfo, which describes replication of a
*local* object to remote sites, so the check sits before any replication
condition.

Localization is neither lifecycle nor CRR replication, so it gets its
own action origin, and the legacy CRR byte metrics - which only make
sense for replication to a remote site - skip it like they already skip
lifecycle.

Issue: BB-814

Author: francoisferrand

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/replication/ReplicationMetric.js

@@ -2,6 +2,10 @@ const { Logger } = require('werelogs');
 
 const MetricsModel = require('../../lib/models/MetricsModel');
 
+// Flows which report their own metrics, and for which the CRR metrics
+// below are meaningless: they are not replication to a CRR site.
+const ORIGINS_WITH_OWN_METRICS = ['lifecycle', 'localization'];
+
 /**
  * Legacy: consider converting replication metrics with
  * Prometheus-based {@link ReplicationMetrics} class
@@ -46,9 +50,9 @@ class ReplicationMetric {
         return this;
     }
 
-    _isLifecycleAction() {
+    _hasOwnMetrics() {
         const { origin } = this._entry.getContext();
-        return origin !== undefined && origin === 'lifecycle';
+        return ORIGINS_WITH_OWN_METRICS.includes(origin);
     }
 
     _createProducerMessage() {
@@ -65,8 +69,8 @@ class ReplicationMetric {
     }
 
     publish() {
-        // Lifecycle metrics not yet implemented.
-        if (this._isLifecycleAction()) {
+        // Lifecycle and localization metrics not yet implemented.
+        if (this._hasOwnMetrics()) {
             return undefined;
         }
         const message = this._createProducerMessage();

extensions/replication/ReplicationQueuePopulator.js

@@ -1,13 +1,26 @@
 const { isMasterKey } = require('arsenal').versioning;
+const { encode } = require('arsenal').versioning.VersionID;
 const { usersBucket, mpuBucketPrefix } = require('arsenal').constants;
 
 const QueuePopulatorExtension =
           require('../../lib/queuePopulator/QueuePopulatorExtension');
 const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry');
+const ReplicationAPI = require('./ReplicationAPI');
+const { LifecycleMetrics, LOCALIZATION_TYPE } = require('../lifecycle/LifecycleMetrics');
+const config = require('../../lib/Config');
 const locationsConfig = require('../../conf/locationConfig.json') || {};
 const safeJsonParse = require('../../lib/util/safeJsonParse');
+const { getTransitionAttempt } = require('../../lib/util/transitionAttempt');
 const { traceHeadersFromEntry } = require('arsenal/build/lib/tracing').kafka;
 
+const { transitionTasksTopic } = config.extensions.lifecycle;
+
+// Where clean room objects are localized when their metadata does not name a
+// usable target. Cold and source (isCRR) locations can never hold localized
+// data, any other one is a valid local destination.
+const defaultLocalLocation = Object.keys(locationsConfig).find(
+    name => !locationsConfig[name].isCold && !locationsConfig[name].isCRR);
+
 class ReplicationQueuePopulator extends QueuePopulatorExtension {
     constructor(params) {
         super(params);
@@ -73,6 +86,15 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension {
         if (sanityCheckRes) {
             return;
         }
+        const locationConfig = locationsConfig[queueEntry.getDataStoreName()] || {};
+        // Clean room: the object data still lives on the source (isCRR)
+        // location and first needs to be localized. This is unrelated to
+        // replicationInfo, which tracks replication of a *local* object to
+        // remote sites, hence the check before any replication condition.
+        if (locationConfig.isCRR) {
+            this._publishLocalizationAction(entry, queueEntry, value);
+            return;
+        }
         // Allow a non-versioned object if being replicated from an NFS bucket.
         // Or if the master key is of a non versioned object
         if (!this._entryCanBeReplicated(queueEntry)) {
@@ -81,11 +103,8 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension {
         if (queueEntry.getReplicationStatus() !== 'PENDING') {
             return;
         }
-        const dataStoreName = queueEntry.getDataStoreName();
-        const isObjectCold = dataStoreName && locationsConfig[dataStoreName]
-            && locationsConfig[dataStoreName].isCold;
         // We do not replicate cold objects.
-        if (isObjectCold) {
+        if (locationConfig.isCold) {
             return;
         }
 
@@ -124,6 +143,132 @@ class ReplicationQueuePopulator extends QueuePopulatorExtension {
                      traceHeaders);
     }
 
+    /**
+     * Queue a copyLocation action for an object whose data still lives on the
+     * source (isCRR) location, so the data mover copies it to the local
+     * location and the transition processor merges the new location back into
+     * the object metadata.
+     *
+     * Duplicates are expected (and harmless): the same object may show up
+     * several times in the oplog, and the copy is idempotent.
+     *
+     * @param {Object} entry - raw metadata log entry
+     * @param {ObjectQueueEntry} queueEntry - parsed entry
+     * @param {Object} value - parsed entry metadata
+     * @return {undefined}
+     */
+    _publishLocalizationAction(entry, queueEntry, value) {
+        // Clean room buckets are versioned: the master key is repaired by the
+        // metadata layer once the version has been localized.
+        if (isMasterKey(queueEntry.getObjectVersionedKey())) {
+            return;
+        }
+        if (queueEntry.getIsDeleteMarker()) {
+            return;
+        }
+        const locations = queueEntry.getLocation();
+        if (!locations || locations.length === 0) {
+            // Empty objects hold no data, there is nothing to localize. Any
+            // other object without location information is inconsistent.
+            if (queueEntry.getContentLength() > 0) {
+                this.log.error('non-empty object without location, skipping localization', {
+                    method: 'ReplicationQueuePopulator._publishLocalizationAction',
+                    ...queueEntry.getLogInfo(),
+                    dataStoreName: queueEntry.getDataStoreName(),
+                    contentLength: queueEntry.getContentLength(),
+                });
+            }
+            return;
+        }
+
+        const bucket = queueEntry.getBucket();
+        const objectKey = queueEntry.getObjectKey();
+        const contentLength = queueEntry.getContentLength();
+        const targetLocation = this._getLocalizationTarget(queueEntry, locations);
+        if (!targetLocation) {
+            return;
+        }
+        const transitionTime = new Date(entry.overheadFields?.commitTimestamp ?? Date.now());
+        const action = ReplicationAPI.createCopyLocationAction({
+            bucketName: bucket,
+            objectKey,
+            owner: queueEntry.getOwnerId(),
+            versionId: value.versionId ? encode(value.versionId) : undefined,
+            eTag: `"${queueEntry.getContentMd5()}"`,
+            lastModified: queueEntry.getLastModified(),
+            toLocation: targetLocation,
+            originLabel: 'localization',
+            fromLocation: queueEntry.getDataStoreName(),
+            contentLength,
+            resultsTopic: transitionTasksTopic,
+            transitionTime: transitionTime.toISOString(),
+            attempt: getTransitionAttempt(queueEntry.getUserMetadata()),
+        });
+        // 'transition' is what the lifecycle transition processor dispatches
+        // on to pick up the copyLocation result.
+        action.addContext({
+            origin: 'localization',
+            ruleType: 'transition',
+            bucketName: bucket,
+            objectKey,
+            versionId: value.versionId,
+        });
+        action.setAttribute('source', {
+            bucket,
+            objectKey,
+            storageClass: queueEntry.getDataStoreName(),
+        });
+
+        LifecycleMetrics.onLifecycleTriggered(this.log, 'queuePopulator',
+            LOCALIZATION_TYPE, targetLocation, Date.now() - transitionTime.getTime());
+
+        this.log.trace('publishing object localization entry', { entry: queueEntry.getLogInfo() });
+        this.publish(ReplicationAPI.getDataMoverTopic(),
+                     `${bucket}/${objectKey}`,
+                     action.toKafkaMessage(),
+                     undefined,
+                     traceHeadersFromEntry(value));
+    }
+
+    /**
+     * Local location the object data must be copied to.
+     *
+     * It is named in the source location entry itself, next to the bucket and
+     * role the copy needs: the rewrite pipeline resolves it from the bucket
+     * when it synthesizes that entry, which keeps this populator -a single
+     * threaded oplog reader- from having to look the bucket up per object.
+     *
+     * @param {ObjectQueueEntry} queueEntry - parsed entry
+     * @param {Object[]} locations - object data locations
+     * @return {String|undefined} target location, undefined if there is none
+     */
+    _getLocalizationTarget(queueEntry, locations) {
+        const { targetLocation } = locations[0];
+        if (locationsConfig[targetLocation]) {
+            return targetLocation;
+        }
+        // Either the object predates the rewrite pipeline naming a target, or
+        // the location was deleted since the metadata was written. Neither is
+        // recoverable here, so fall back to the default location: localizing
+        // elsewhere beats leaving the data on the source forever.
+        if (!defaultLocalLocation) {
+            this.log.error('invalid localization target and no local location ' +
+                'to fall back to, skipping localization', {
+                method: 'ReplicationQueuePopulator._getLocalizationTarget',
+                ...queueEntry.getLogInfo(),
+                targetLocation,
+            });
+            return undefined;
+        }
+        this.log.error('invalid localization target in object metadata', {
+            method: 'ReplicationQueuePopulator._getLocalizationTarget',
+            ...queueEntry.getLogInfo(),
+            targetLocation,
+            fallbackLocation: defaultLocalLocation,
+        });
+        return defaultLocalLocation;
+    }
+
     /**
      * Filter if the entry is considered a valid master key entry.
      * There is a case where a single null entry looks like a master key and

tests/unit/ReplicationMetric.js

@@ -56,27 +56,25 @@ describe('ReplicationMetric', () => {
             .forEach(key => assert.strictEqual(data[key], mock[key]));
     });
 
-    it('::_isLifecycleAction should return false by default', () => {
+    it('::_hasOwnMetrics should return false by default', () => {
         metric.withEntry(entry);
-        assert.strictEqual(metric._isLifecycleAction(), false);
+        assert.strictEqual(metric._hasOwnMetrics(), false);
     });
 
-    it('::_isLifecycleAction should return true when origin is lifecycle',
-        () => {
-            entry.setAttribute('contextInfo', {
-                origin: 'lifecycle',
+    ['lifecycle', 'localization'].forEach(origin => {
+        it(`::_hasOwnMetrics should return true when origin is ${origin}`,
+            () => {
+                entry.setAttribute('contextInfo', { origin });
+                metric.withEntry(entry);
+                assert.strictEqual(metric._hasOwnMetrics(), true);
             });
-            metric.withEntry(entry);
-            assert.strictEqual(metric._isLifecycleAction(), true);
-        });
 
-    it('::publish should not send data to topic if lifecycle task', () => {
-        entry.setAttribute('contextInfo', {
-            origin: 'lifecycle',
+        it(`::publish should not send data to topic for a ${origin} action`, () => {
+            entry.setAttribute('contextInfo', { origin });
+            metric.withEntry(entry);
+            metric.publish();
+            assert.strictEqual(sentMessages.length, 0);
         });
-        metric.withEntry(entry);
-        metric.publish();
-        assert.strictEqual(sentMessages.length, 0);
     });
 
     it('::publish should send data to topic', () => {

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: [],