Reclaim local data when a localized version is deleted

extensions/mongoProcessor/MongoQueueProcessor.js

@@ -13,6 +13,7 @@ const { extractVersionId } = require('../../lib/util/versioning');
 
 const Config = require('../../lib/Config');
 const BackbeatConsumer = require('../../lib/BackbeatConsumer');
+const ActionQueueEntry = require('../../lib/models/ActionQueueEntry');
 const QueueEntry = require('../../lib/models/QueueEntry');
 const DeleteOpQueueEntry = require('../../lib/models/DeleteOpQueueEntry');
 const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry');
@@ -22,6 +23,8 @@ const { metricsExtension, metricsTypeCompleted, metricsTypePendingOnly } =
 const getContentType = require('./utils/contentTypeHelper');
 const BucketMemState = require('./utils/BucketMemState');
 const MongoProcessorMetrics = require('./MongoProcessorMetrics');
+const GarbageCollectorProducer = require('../gc/GarbageCollectorProducer');
+const { isCRRLocation, filterOutCRRLocations } = require('../../lib/util/locations');
 
 // batch metrics by location and send to kafka metrics topic every 5 seconds
 const METRIC_REPORT_INTERVAL_MS = process.env.CI === 'true' ? 1000 : 5000;
@@ -64,13 +67,17 @@ class MongoQueueProcessor {
      * @param {number} [mongoProcessorConfig.concurrency] - consumer concurrency
      * @param {Object} mongoClientConfig - config for connecting to mongo
      * @param {Object} mConfig - metrics config
+     * @param {Object} [gcConfig] - garbage collector config, required to reclaim
+     *   the data of localized objects
      */
-    constructor(kafkaConfig, mongoProcessorConfig, mongoClientConfig, mConfig) {
+    constructor(kafkaConfig, mongoProcessorConfig, mongoClientConfig, mConfig, gcConfig) {
         this.kafkaConfig = kafkaConfig;
         this.mongoProcessorConfig = mongoProcessorConfig;
         this.mongoClientConfig = mongoClientConfig;
         this._mConfig = mConfig;
+        this._gcConfig = gcConfig;
 
+        this._gcProducer = null;
         this._consumer = null;
         this._bootstrapList = null;
         this.logger = new Logger('Backbeat:Ingestion:MongoProcessor');
@@ -119,6 +126,16 @@ class MongoQueueProcessor {
                 }
                 return next(err);
             }),
+            next => {
+                if (!this._gcConfig) {
+                    this.logger.info('no garbage collector configured', {
+                        method: 'MongoQueueProcessor.start',
+                    });
+                    return next();
+                }
+                this._gcProducer = new GarbageCollectorProducer();
+                return this._gcProducer.setupProducer(next);
+            },
         ], error => {
             if (error) {
                 this.logger.fatal('error starting mongo queue processor');
@@ -377,16 +394,24 @@ class MongoQueueProcessor {
         async.waterfall([
             cb => this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb),
             (zenkoObjMd, cb) => {
+                // In a clean room the bucket location constraint is the (isCRR) source
+                // location: once an object has been localized its data does not live there
+                // anymore, so it looks exactly like a transitioned object. It must still be
+                // deleted, and the local copy of the data reclaimed.
+                const localizedLocations = isCRRLocation(location) ?
+                    filterOutCRRLocations(zenkoObjMd.location) : [];
+
                 // 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 ||
+                if (localizedLocations.length === 0 && (
+                    zenkoObjMd.dataStoreName !== location ||
                     zenkoObjMd.location?.length !== 1 ||
                     zenkoObjMd.location[0].dataStoreName !== location ||
                     zenkoObjMd.location[0].key !== key ||
-                    (zenkoObjMd.location[0].dataStoreVersionId || 'null') !== encode(entryVersionId)
+                    (zenkoObjMd.location[0].dataStoreVersionId || 'null') !== encode(entryVersionId))
                 ) {
                     log.end().info('ignore delete entry, transitioned to another location', {
                         entry: sourceEntry.getLogInfo(),
@@ -395,9 +420,9 @@ class MongoQueueProcessor {
                     return done();
                 }
 
-                return cb(null, zenkoObjMd);
+                return cb(null, zenkoObjMd, localizedLocations);
             },
-            (zenkoObjMd, cb) => {
+            (zenkoObjMd, localizedLocations, cb) => {
                 const options = {};
 
                 // Calling deleteObject with empty options to use deleteObjectNoVer which is used
@@ -418,7 +443,19 @@ class MongoQueueProcessor {
                     options.doesNotNeedOpogUpdate = true;
                 }
 
-                return this._mongoClient.deleteObject(bucket, key, options, log, cb);
+                return this._mongoClient.deleteObject(bucket, key, options, log, err => {
+                    if (err) {
+                        return cb(err);
+                    }
+                    // The data copied locally is not referenced by anything anymore: reclaim
+                    // it. Doing it after the metadata delete means a failure here leaks the
+                    // data, instead of losing it.
+                    if (localizedLocations.length > 0) {
+                        this._garbageCollectData(log, sourceEntry, zenkoObjMd, versionId,
+                            localizedLocations);
+                    }
+                    return cb();
+                });
             },
         ], err => {
             if (err?.is.NoSuchKey) {
@@ -448,6 +485,53 @@ class MongoQueueProcessor {
         });
     }
 
+    /**
+     * Publish a garbage collection entry to reclaim the data of an object which has just
+     * been deleted from mongo.
+     * @param {Logger.newRequestLogger} log - request logger object
+     * @param {DeleteOpQueueEntry} sourceEntry - delete object entry
+     * @param {Object} zenkoObjMd - metadata of the object which was deleted
+     * @param {string} versionId - decoded version id of the object which was deleted
+     * @param {Object[]} locations - location parts to delete
+     * @return {undefined}
+     */
+    _garbageCollectData(log, sourceEntry, zenkoObjMd, versionId, locations) {
+        const bucket = sourceEntry.getBucket();
+        const key = sourceEntry.getObjectKey();
+
+        if (!this._gcProducer) {
+            log.error('no garbage collector configured, cannot reclaim the data', {
+                bucket,
+                objectKey: key,
+                versionId,
+            });
+            return;
+        }
+
+        // The version is already gone, so no lastModified is passed along: the GC must not
+        // send any conditional header, only the locations matter here.
+        const gcEntry = ActionQueueEntry.create('deleteData')
+            .addContext({
+                origin: 'localization',
+                ruleType: 'transition',
+                reqId: log.getSerializedUids(),
+                bucketName: bucket,
+                objectKey: key,
+                versionId: versionId ? VersionID.encode(versionId) : undefined,
+                eTag: zenkoObjMd['content-md5'],
+            })
+            .setAttribute('source', {
+                bucket,
+                objectKey: key,
+                storageClass: zenkoObjMd.dataStoreName,
+            })
+            .setAttribute('serviceName', 'md-ingestion')
+            .setAttribute('target.owner', zenkoObjMd['owner-id'])
+            .setAttribute('target.locations', locations);
+
+        this._gcProducer.publishActionEntry(gcEntry);
+    }
+
     /**
      * Process an object entry
      * @param {Logger.newRequestLogger} log - request logger object

extensions/mongoProcessor/mongoProcessorTask.js

@@ -21,14 +21,15 @@ const mongoProcessorConfig = config.extensions.mongoProcessor;
 // TODO: consider whether we would want a separate mongo config
 // for the consumer side
 const mongoClientConfig = config.queuePopulator.mongo;
+const gcConfig = config.extensions.gc;
 
 const log = new werelogs.Logger('Backbeat:MongoProcessor:task');
 const mongoProcessorLogConfig = mongoProcessorConfig.log ?? config.log;
 werelogs.configure({ level: mongoProcessorLogConfig.logLevel,
     dump: mongoProcessorLogConfig.dumpLevel });
 
 const mqp = new MongoQueueProcessor(kafkaConfig, mongoProcessorConfig,
-    mongoClientConfig, mConfig);
+    mongoClientConfig, mConfig, gcConfig);
 
 /**
  * Handle ProbeServer liveness check

tests/unit/mongoProcessor/MongoQueueProcessor.spec.js

@@ -1,9 +1,20 @@
 const assert = require('assert');
+const sinon = require('sinon');
+
+const { VersionID, VersioningConstants } = require('arsenal').versioning;
 
 const MongoQueueProcessor =
     require('../../../extensions/mongoProcessor/MongoQueueProcessor');
 const ObjectQueueEntry =
     require('../../../lib/models/ObjectQueueEntry');
+const DeleteOpQueueEntry =
+    require('../../../lib/models/DeleteOpQueueEntry');
+
+const VID_SEP = VersioningConstants.VersionId.Separator;
+
+// see conf/locationConfig.json
+const CRR_LOCATION = 'location-crr-source';
+const LOCAL_LOCATION = 'us-east-1';
 
 function _makeProcessor(bootstrapList) {
     const proc = Object.create(MongoQueueProcessor.prototype);
@@ -234,3 +245,118 @@ describe('MongoQueueProcessor._updateReplicationInfo', () => {
         assert.strictEqual(bySite['cloud-b'].status, 'PENDING');
     });
 });
+
+describe('MongoQueueProcessor._processDeleteOpQueueEntry', () => {
+    const bucket = 'cleanroom-bucket';
+    const objectKey = 'docs/report.pdf';
+    const versionId = '98765432109876999999RG001  1';
+    const encodedVersionId = VersionID.encode(versionId);
+
+    function _makeLog() {
+        const log = {
+            debug: () => {},
+            info: () => {},
+            error: () => {},
+            warn: () => {},
+            getSerializedUids: () => 'req-uid',
+        };
+        log.end = () => log;
+        return log;
+    }
+
+    function _makeDeleteProcessor(zenkoObjMd, gcProducer) {
+        const proc = Object.create(MongoQueueProcessor.prototype);
+        proc.logger = { debug: () => {} };
+        proc._gcProducer = gcProducer;
+        proc._mongoClient = {
+            deleteObject: sinon.stub().callsFake((b, k, opts, log, cb) => cb()),
+        };
+        proc._getZenkoObjectMetadata =
+            sinon.stub().callsFake((log, entry, vid, cb) => cb(null, zenkoObjMd));
+        proc._produceMetricCompletionEntry = () => {};
+        proc._normalizePendingMetric = () => {};
+        return proc;
+    }
+
+    function _makeEntry() {
+        return new DeleteOpQueueEntry(bucket, `${objectKey}${VID_SEP}${versionId}`, {});
+    }
+
+    it('deletes a localized version and publishes its local data for GC', done => {
+        const zenkoObjMd = {
+            'dataStoreName': LOCAL_LOCATION,
+            'owner-id': 'owner-canonical-id',
+            'content-md5': 'etag-value',
+            'location': [{ dataStoreName: LOCAL_LOCATION, key: 'local-data-key' }],
+        };
+        const gcProducer = { publishActionEntry: sinon.stub() };
+        const proc = _makeDeleteProcessor(zenkoObjMd, gcProducer);
+
+        proc._processDeleteOpQueueEntry(_makeLog(), _makeEntry(), CRR_LOCATION, {}, err => {
+            assert.ifError(err);
+            assert.strictEqual(proc._mongoClient.deleteObject.callCount, 1);
+            assert.strictEqual(gcProducer.publishActionEntry.callCount, 1);
+
+            const gcEntry = gcProducer.publishActionEntry.firstCall.args[0];
+            assert.strictEqual(gcEntry.getActionType(), 'deleteData');
+            assert.deepStrictEqual(gcEntry.getAttribute('target.locations'),
+                zenkoObjMd.location);
+            assert.strictEqual(gcEntry.getAttribute('target.owner'), 'owner-canonical-id');
+            assert.strictEqual(gcEntry.getAttribute('serviceName'), 'md-ingestion');
+            // the version is gone: no conditional header must be sent by the GC
+            assert.strictEqual(gcEntry.getAttribute('source').lastModified, undefined);
+            assert.strictEqual(gcEntry.getContextAttribute('versionId'), encodedVersionId);
+            done();
+        });
+    });
+
+    it('does not publish anything for a version which was never localized', done => {
+        const zenkoObjMd = {
+            dataStoreName: CRR_LOCATION,
+            location: [{
+                dataStoreName: CRR_LOCATION,
+                key: objectKey,
+                dataStoreVersionId: encodedVersionId,
+            }],
+        };
+        const gcProducer = { publishActionEntry: sinon.stub() };
+        const proc = _makeDeleteProcessor(zenkoObjMd, gcProducer);
+
+        proc._processDeleteOpQueueEntry(_makeLog(), _makeEntry(), CRR_LOCATION, {}, err => {
+            assert.ifError(err);
+            assert.strictEqual(proc._mongoClient.deleteObject.callCount, 1);
+            assert.strictEqual(gcProducer.publishActionEntry.callCount, 0);
+            done();
+        });
+    });
+
+    it('still ignores an object transitioned outside of a clean room', done => {
+        const zenkoObjMd = {
+            dataStoreName: 'location-dmf-v1',
+            location: [{ dataStoreName: 'location-dmf-v1', key: 'cold-key' }],
+        };
+        const gcProducer = { publishActionEntry: sinon.stub() };
+        const proc = _makeDeleteProcessor(zenkoObjMd, gcProducer);
+
+        proc._processDeleteOpQueueEntry(_makeLog(), _makeEntry(), LOCAL_LOCATION, {}, err => {
+            assert.ifError(err);
+            assert.strictEqual(proc._mongoClient.deleteObject.callCount, 0);
+            assert.strictEqual(gcProducer.publishActionEntry.callCount, 0);
+            done();
+        });
+    });
+
+    it('deletes the metadata even when no garbage collector is configured', done => {
+        const zenkoObjMd = {
+            dataStoreName: LOCAL_LOCATION,
+            location: [{ dataStoreName: LOCAL_LOCATION, key: 'local-data-key' }],
+        };
+        const proc = _makeDeleteProcessor(zenkoObjMd, null);
+
+        proc._processDeleteOpQueueEntry(_makeLog(), _makeEntry(), CRR_LOCATION, {}, err => {
+            assert.ifError(err);
+            assert.strictEqual(proc._mongoClient.deleteObject.callCount, 1);
+            done();
+        });
+    });
+});