Add a D/R mode to the mongo-processor

extensions/mongoProcessor/MongoProcessorConfigValidator.js

@@ -1,12 +1,17 @@
 const joi = require('joi');
-const { retryParamsJoi, probeServerJoi, logJoiOptional } = require('../../lib/config/configItems.joi');
+const { modes, defaultMode } = require('./modes');
+const { authJoi, retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } =
+    require('../../lib/config/configItems.joi');
 const { extensionConfigValidator } = require('../../lib/config/extensionConfigValidator');
 
 const { MAX_QUEUED_DEFAULT }  = require('../../lib/constants').backbeatConsumer;
 
 const joiSchema = joi.object({
     topic: joi.string().required(),
     groupId: joi.string().required(),
+    mode: joi.string().valid(...Object.keys(modes)).default(defaultMode),
+    mongodb: mongoJoi.when('mode', { is: 'dr', then: joi.required() }),
+    auth: authJoi.when('mode', { is: 'dr', then: joi.required() }),
     retry: retryParamsJoi,
     concurrency: joi.number().greater(0).default(1),
     maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),

extensions/mongoProcessor/MongoQueueProcessor.js

@@ -4,11 +4,10 @@ const async = require('async');
 
 const Logger = require('werelogs').Logger;
 const errors = require('arsenal').errors;
-const { replicationBackends, emptyFileMd5 } = require('arsenal').constants;
+const { replicationBackends } = require('arsenal').constants;
 const MongoClient = require('arsenal').storage
     .metadata.mongoclient.MongoClientInterface;
 const { ObjectMD, ReplicationConfiguration } = require('arsenal').models;
-const { VersionID } = require('arsenal').versioning;
 const { extractVersionId } = require('../../lib/util/versioning');
 
 const Config = require('../../lib/Config');
@@ -19,7 +18,7 @@ const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry');
 const MetricsProducer = require('../../lib/MetricsProducer');
 const { metricsExtension, metricsTypeCompleted, metricsTypePendingOnly } =
     require('../ingestion/constants');
-const getContentType = require('./utils/contentTypeHelper');
+const { modes, defaultMode } = require('./modes');
 const BucketMemState = require('./utils/BucketMemState');
 const MongoProcessorMetrics = require('./MongoProcessorMetrics');
 
@@ -75,6 +74,8 @@ class MongoQueueProcessor {
         this._bootstrapList = null;
         this.logger = new Logger('Backbeat:Ingestion:MongoProcessor');
         this.mongoClientConfig.logger = this.logger;
+        this._mode =
+            new modes[mongoProcessorConfig.mode ?? defaultMode]();
         this._mongoClient = new MongoClient(this.mongoClientConfig);
         this._bucketMemState = new BucketMemState(Config);
 
@@ -236,78 +237,6 @@ class MongoQueueProcessor {
         });
     }
 
-    /**
-     * Update ingested entry metadata fields: owner-id, owner-display-name
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {BucketInfo} bucketInfo - bucket info object
-     * @return {undefined}
-     */
-    _updateOwnerMD(entry, bucketInfo) {
-        // zenko bucket owner information is being set on ingested md
-        entry.setOwnerDisplayName(bucketInfo.getOwnerDisplayName());
-        entry.setOwnerId(bucketInfo.getOwner());
-    }
-
-    /**
-     * Update ingested entry metadata fields: dataStoreName
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {string} location - owner details
-     * @return {undefined}
-     */
-    _updateObjectDataStoreName(entry, location) {
-        entry.setDataStoreName(location);
-    }
-
-    /**
-     * Update ingested entry metadata location field. Each location change
-     * includes: key, dataStoreName, dataStoreType, dataStoreVersionId
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {string} zenkoLocation - zenko storage location name
-     * @return {undefined}
-     */
-    _updateLocations(entry, zenkoLocation) {
-        const locations = entry.getLocation();
-        // if version id is undefined, we have a single null object.
-        // To hold reference to this null object, we need to encode "null"
-        // as its dataStoreVersionId
-        const dataStoreVersionId = entry.getVersionId() ?
-            entry.getEncodedVersionId() : 'null';
-        let zenkoDataLocations;
-        if (!locations || locations.length === 0) {
-            zenkoDataLocations = [{
-                key: entry.getObjectKey(),
-                size: 0,
-                start: 0,
-                dataStoreName: zenkoLocation,
-                dataStoreType: 'aws_s3',
-                dataStoreETag: `1:${emptyFileMd5}`,
-                dataStoreVersionId,
-            }];
-        } else {
-            zenkoDataLocations = [{
-                key: entry.getObjectKey(),
-                size: entry.getContentLength(),
-                start: 0,
-                dataStoreName: zenkoLocation,
-                dataStoreType: 'aws_s3',
-                dataStoreETag: `1:${entry.getContentMd5()}`,
-                dataStoreVersionId,
-            }];
-        }
-        entry.setLocation(zenkoDataLocations);
-    }
-
-    /**
-     * Update acl info on ingested object MD
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @return {undefined}
-     */
-    _updateAcl(entry) {
-        // reset acl info
-        const objectMDModel = new ObjectMD();
-        entry.setAcl(objectMDModel.getAcl());
-    }
-
     /**
      * Update replication info on ingested object MD to match Zenko defined
      * replication info.
@@ -367,29 +296,17 @@ class MongoQueueProcessor {
         const key = sourceEntry.getObjectKey();
         const entryVersionId = extractVersionId(sourceEntry.getObjectVersionedKey());
 
-        // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object.
-        // This should happen only for restored objects : in all other situations, both the source
-        // and ingested objects should have the same version id (and no x-amz-meta-scal-version-id
-        // metadata).
         const scalVersionId = sourceEntry.getOverheadField('x-amz-meta-scal-version-id');
-        const versionId = scalVersionId ? VersionID.decode(scalVersionId) : entryVersionId;
+        const versionId =
+            this._mode.resolveVersionId(scalVersionId, entryVersionId);
 
         this.logger.debug('processing object delete', { bucket, key, versionId });
 
         async.waterfall([
             cb => this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb),
             (zenkoObjMd, cb) => {
-                // Skip if the object is in a different location, i.e. when the delete was caused
-                // by restored-object expiration or transition. It works because the dataStoreName
-                // is updated before actually sending the object to GC to effectively delete the
-                // data.
-                const encode = versionId => (versionId ? VersionID.encode(versionId) : 'null');
-                if (zenkoObjMd.dataStoreName !== location ||
-                    zenkoObjMd.location?.length !== 1 ||
-                    zenkoObjMd.location[0].dataStoreName !== location ||
-                    zenkoObjMd.location[0].key !== key ||
-                    (zenkoObjMd.location[0].dataStoreVersionId || 'null') !== encode(entryVersionId)
-                ) {
+                if (!this._mode.shouldProcessDelete(zenkoObjMd, location, key,
+                                                   entryVersionId)) {
                     log.end().info('ignore delete entry, transitioned to another location', {
                         entry: sourceEntry.getLogInfo(),
                         location,
@@ -467,19 +384,12 @@ class MongoQueueProcessor {
         this.logger.debug('processing object metadata', { bucket, key, scalVersionId });
 
         const maybeGetZenkoObjectMetadata = cb => {
-            // NOTE: ZenkoObjMD is used for updating replication info, as well as validating the
-            // `x-amz-meta-scal-version-id` header of restored objects. If the Zenko bucket does
-            // not have repInfo set and the header is not set, then we can skip fetching.
-            const bucketRepInfo = bucketInfo.getReplicationConfiguration();
-            if (!scalVersionId && !bucketRepInfo?.rules?.some(r => r.enabled)) {
+            if (!this._mode.needsExistingMetadata(sourceEntry, bucketInfo)) {
                 return cb();
             }
 
-            // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object.
-            // This should happen only for restored objects : in all other situations, both the source
-            // and ingested objects should have the same version id (and not x-amz-meta-scal-version-id
-            // metadata).
-            const versionId = scalVersionId ? VersionID.decode(scalVersionId) : sourceEntry.getVersionId();
+            const versionId = this._mode.resolveVersionId(scalVersionId,
+                sourceEntry.getVersionId());
             return this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb);
         };
 
@@ -494,7 +404,8 @@ class MongoQueueProcessor {
                 return done(err);
             }
 
-            const content = getContentType(sourceEntry, zenkoObjMd);
+            const content =
+                this._mode.getChangedContent(sourceEntry, zenkoObjMd);
             if (content.length === 0) {
                 this._normalizePendingMetric(location);
                 log.end().debug('skipping duplicate entry', {
@@ -507,16 +418,10 @@ class MongoQueueProcessor {
             }
 
             if (zenkoObjMd) {
-                // Keep existing metadata fields, only need to update the tags
-                const tags = sourceEntry.getTags();
-                sourceEntry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign
-                sourceEntry.setTags(tags);
+                this._mode.mergeExistingMetadata(sourceEntry, zenkoObjMd);
             } else {
-                // Update necessary metadata fields before saving to Zenko MongoDB
-                this._updateOwnerMD(sourceEntry, bucketInfo);
-                this._updateObjectDataStoreName(sourceEntry, location);
-                this._updateLocations(sourceEntry, location);
-                this._updateAcl(sourceEntry);
+                this._mode.applyNewObjectMetadata(sourceEntry, location,
+                                                 bucketInfo);
             }
 
             // Try to update replication info, if applicable

extensions/mongoProcessor/modes/DRMode.js

@@ -0,0 +1,133 @@
+'use strict';
+
+const { ObjectMD } = require('arsenal').models;
+
+const locations = require('../../../lib/util/locations');
+const ProcessorMode = require('./ProcessorMode');
+const getContentType = require('../utils/contentTypeHelper');
+
+class DRMode extends ProcessorMode {
+    /**
+     * The stored document tells a first write from an update: a source insert
+     * is redelivered on replay and overlaps the bootstrap dump.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {boolean} true if the stored document is needed
+     */
+    needsExistingMetadata(entry, bucketInfo) { // eslint-disable-line no-unused-vars
+        return true;
+    }
+
+    /**
+     * The ingestion diff only covers tags; a replicated object can also change
+     * its object-lock state and, until localized, its placement.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo
+     * @return {Array} array of ReplicationInfo Content Type
+     */
+    getChangedContent(entry, zenkoObjMd) {
+        const content = getContentType(entry, zenkoObjMd);
+        if (!zenkoObjMd || content.length !== 0) {
+            return content;
+        }
+
+        return this._hasMutableChange(entry, zenkoObjMd) ? ['METADATA'] : [];
+    }
+
+    /**
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {boolean} true if the entry changes mutable metadata
+     */
+    _hasMutableChange(entry, zenkoObjMd) {
+        return entry.getRetentionMode() !== zenkoObjMd.retentionMode ||
+            entry.getRetentionDate() !== zenkoObjMd.retentionDate ||
+            entry.getLegalHold() !== !!zenkoObjMd.legalHold ||
+            (this._isNotLocalized(zenkoObjMd) &&
+                entry.getDataStoreName() !== zenkoObjMd.dataStoreName);
+    }
+
+    /**
+     * A version whose data still lives on the remote site.
+     *
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {boolean} true if the stored version is not localized
+     */
+    _isNotLocalized(zenkoObjMd) {
+        return locations.isCRRLocation(zenkoObjMd.dataStoreName);
+    }
+
+    /**
+     * The source-side pipeline shaped everything this object keeps. ACLs are
+     * not replicated, so they are reset as they are for an ingested object.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} location - zenko storage location name
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {undefined}
+     */
+    applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars
+        entry.setAcl(new ObjectMD().getAcl());
+    }
+
+    /**
+     * An update brings tags and object-lock state, cleared values included. A
+     * localized version keeps the placement the copy engine gave it, one still
+     * on the remote site takes the entry's.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {undefined}
+     */
+    mergeExistingMetadata(entry, zenkoObjMd) {
+        const tags = entry.getTags();
+        const retentionMode = entry.getRetentionMode();
+        const retentionDate = entry.getRetentionDate();
+        const legalHold = entry.getLegalHold();
+        const notLocalized = this._isNotLocalized(zenkoObjMd);
+        const dataStoreName = entry.getDataStoreName();
+        const location = entry.getLocation();
+
+        entry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign
+
+        entry.setTags(tags);
+        entry.setRetentionMode(retentionMode);
+        entry.setRetentionDate(retentionDate);
+        entry.setLegalHold(legalHold);
+
+        if (notLocalized) {
+            entry.setDataStoreName(dataStoreName);
+            entry.setLocation(location);
+        }
+    }
+
+    /**
+     * Version ids are identical on both sides, so the entry's is authoritative:
+     * a scal version id names a version of another system entirely.
+     *
+     * @param {string|undefined} scalVersionId - encoded scal version id
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {string|undefined} version id to act on
+     */
+    resolveVersionId(scalVersionId, versionId) {
+        return versionId;
+    }
+
+    /**
+     * A replicated object's location legitimately differs, so the ingestion
+     * guard would ignore every deletion.
+     *
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @param {string} location - zenko storage location name
+     * @param {string} key - object key
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {boolean} true if the object should be deleted
+     */
+    shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars
+        return true;
+    }
+}
+
+module.exports = DRMode;