diff --git a/extensions/mongoProcessor/MongoQueueProcessor.js b/extensions/mongoProcessor/MongoQueueProcessor.js index cc2843f07..d4ffdcb7b 100644 --- a/extensions/mongoProcessor/MongoQueueProcessor.js +++ b/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 diff --git a/extensions/mongoProcessor/mongoProcessorTask.js b/extensions/mongoProcessor/mongoProcessorTask.js index e0ccf0258..5b0e3412b 100644 --- a/extensions/mongoProcessor/mongoProcessorTask.js +++ b/extensions/mongoProcessor/mongoProcessorTask.js @@ -21,6 +21,7 @@ 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; @@ -28,7 +29,7 @@ werelogs.configure({ level: mongoProcessorLogConfig.logLevel, dump: mongoProcessorLogConfig.dumpLevel }); const mqp = new MongoQueueProcessor(kafkaConfig, mongoProcessorConfig, - mongoClientConfig, mConfig); + mongoClientConfig, mConfig, gcConfig); /** * Handle ProbeServer liveness check diff --git a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js b/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js index 4ed3b85a1..45331f643 100644 --- a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js +++ b/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(); + }); + }); +});