From 50c63e971ad83f9f479d99179f3cfc1a7efa27d1 Mon Sep 17 00:00:00 2001 From: Francois Ferrand Date: Mon, 24 Aug 2026 12:19:52 +0200 Subject: [PATCH] Reclaim local data when a localized version is deleted In a clean room the bucket location constraint is the isCRR source location, so once an object has been localized its metadata looks exactly like a transitioned object: the mongo processor was bailing out with "transitioned to another location", never deleting the metadata and leaving the local copy of the data behind forever. Detect that case and, after the metadata delete, publish a deleteData action for the local parts only. Versions which were never localized still point at the isCRR location and are left alone, as before. Issue: BB-815 --- .../mongoProcessor/MongoQueueProcessor.js | 96 ++++++++++++- .../mongoProcessor/mongoProcessorTask.js | 3 +- .../MongoQueueProcessor.spec.js | 126 ++++++++++++++++++ 3 files changed, 218 insertions(+), 7 deletions(-) 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(); + }); + }); +});