Use sourceCheckIfSizeGreaterThanMB for stripping locations

delthas · delthas · commit d4ad94ae832e · 2026-02-24T19:35:20.000+01:00
Issue: BB-491
diff --git a/conf/config.json b/conf/config.json
@@ -304,7 +304,6 @@
             "kafkaConnectHost": "127.0.0.1",
             "kafkaConnectPort": 8083,
             "numberOfConnectors": 1,
-            "locationStrippingThreshold": 500,
             "probeServer": {
                 "bindAddress": "0.0.0.0",
                 "port": 8556
diff --git a/extensions/oplogPopulator/OplogPopulator.js b/extensions/oplogPopulator/OplogPopulator.js
@@ -335,7 +335,7 @@ class OplogPopulator {
         if (this._config.numberOfConnectors === 0) {
             // If the number of connector is set to 0, then we
             // use a single connector to listen to the whole DB.
-            pipelineFactory = new WildcardPipelineFactory(this._config.locationStrippingThreshold);
+            pipelineFactory = new WildcardPipelineFactory(this._config.locationStrippingBytesThreshold);
             strategy = new UniqueConnector({
                 logger: this._logger,
             });
@@ -345,7 +345,7 @@ class OplogPopulator {
             // words, we cannot alter an existing pipeline. In this
             // case, the strategy is to allow a maximum of one
             // bucket per kafka connector.
-            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingThreshold);
+            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingBytesThreshold);
             strategy = new ImmutableConnector({
                 logger: this._logger,
             });
@@ -354,7 +354,7 @@ class OplogPopulator {
             // kafka connector. However, we want to proactively
             // ensure that the pipeline will be accepted by
             // mongodb.
-            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingThreshold);
+            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingBytesThreshold);
             strategy = new LeastFullConnector({
                 logger: this._logger,
             });
diff --git a/extensions/oplogPopulator/OplogPopulatorConfigValidator.js b/extensions/oplogPopulator/OplogPopulatorConfigValidator.js
@@ -6,7 +6,7 @@ const joiSchema = joi.object({
     kafkaConnectHost: joi.string().required(),
     kafkaConnectPort: joi.number().required(),
     numberOfConnectors: joi.number().required().min(0),
-    locationStrippingThreshold: joi.number().default(5),
+    locationStrippingBytesThreshold: joi.number().min(0).default(0),
     prefix: joi.string().optional(),
     probeServer: probeServerJoi.default(),
     connectorsUpdateCronRule: joi.string().default('*/1 * * * * *'),
diff --git a/extensions/oplogPopulator/OplogPopulatorTask.js b/extensions/oplogPopulator/OplogPopulatorTask.js
@@ -20,6 +20,12 @@ werelogs.configure({ level: config.log.logLevel,
 
 const mongoConfig = config.queuePopulator.mongo;
 const oplogPopulatorConfig = config.extensions.oplogPopulator;
+const replicationQPConfig = config.extensions.replication &&
+    config.extensions.replication.queueProcessor;
+if (replicationQPConfig && replicationQPConfig.sourceCheckIfSizeGreaterThanMB) {
+    oplogPopulatorConfig.locationStrippingBytesThreshold =
+        replicationQPConfig.sourceCheckIfSizeGreaterThanMB * 1024 * 1024;
+}
 
 const activeExtensions = [
     'notification',
diff --git a/extensions/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js b/extensions/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js
@@ -10,10 +10,10 @@ const PipelineFactory = require('./PipelineFactory');
 class MultipleBucketsPipelineFactory extends PipelineFactory {
     /**
      * @constructor
-     * @param {number} locationStrippingThreshold threshold for stripping location data
+     * @param {number} locationStrippingBytesThreshold threshold for stripping location data
      */
-    constructor(locationStrippingThreshold) {
-        super(locationStrippingThreshold);
+    constructor(locationStrippingBytesThreshold) {
+        super(locationStrippingBytesThreshold);
         // getPipeline is used standalone later, make sure its `this` reference binds to us.
         this.getPipeline = this.getPipeline.bind(this);
     }
diff --git a/extensions/oplogPopulator/pipeline/PipelineFactory.js b/extensions/oplogPopulator/pipeline/PipelineFactory.js
@@ -9,10 +9,10 @@ const { wildCardForAllBuckets } = require('../constants');
 class PipelineFactory {
     /**
      * @constructor
-     * @param {number} locationStrippingThreshold threshold for stripping location data
+     * @param {number} locationStrippingBytesThreshold threshold for stripping location data
      */
-    constructor(locationStrippingThreshold) {
-        this._locationStrippingThreshold = locationStrippingThreshold;
+    constructor(locationStrippingBytesThreshold) {
+        this._locationStrippingBytesThreshold = locationStrippingBytesThreshold;
     }
 
     /**
@@ -71,31 +71,35 @@ class PipelineFactory {
         const pipeline = [
             stage,
         ];
-        if (this._locationStrippingThreshold >= 0) {
+        if (this._locationStrippingBytesThreshold > 0) {
             pipeline.push({
                 $set: {
-                    'fullDocument.value.location':
-                        this._locationStrippingExpression('fullDocument.value.location'),
-                    'updateDescription.updatedFields.value.location':
-                        this._locationStrippingExpression('updateDescription.updatedFields.value.location'),
+                    'fullDocument.value.location': {
+                        $cond: {
+                            if: { $gte: [
+                                '$fullDocument.value.content-length',
+                                this._locationStrippingBytesThreshold,
+                            ] },
+                            then: '$$REMOVE',
+                            else: '$fullDocument.value.location',
+                        },
+                    },
+                    'updateDescription.updatedFields.value.location': {
+                        $cond: {
+                            if: { $gte: [
+                                '$updateDescription.updatedFields.value.content-length',
+                                this._locationStrippingBytesThreshold,
+                            ] },
+                            then: '$$REMOVE',
+                            else: '$updateDescription.updatedFields.value.location',
+                        },
+                    },
                 }
             });
         }
         return JSON.stringify(pipeline);
     }
 
-    _locationStrippingExpression(field) {
-        return {
-            $switch: {
-                branches: [
-                    { case: { $not: [{ $isArray: `$${field}` }] }, then: '$$REMOVE' },
-                    { case: { $gte: [{ $size: `$${field}` }, this._locationStrippingThreshold] }, then: '$$REMOVE' },
-                ],
-                default: `$${field}`,
-            }
-        };
-    }
-
     /**
      * Makes connector pipeline stage, to then be used by getPipeline.
      * @param {string[] | undefined} buckets buckets assigned to this connector
diff --git a/extensions/oplogPopulator/pipeline/WildcardPipelineFactory.js b/extensions/oplogPopulator/pipeline/WildcardPipelineFactory.js
@@ -11,10 +11,10 @@ const PipelineFactory = require('./PipelineFactory');
 class WildcardPipelineFactory extends PipelineFactory {
     /**
      * @constructor
-     * @param {number} locationStrippingThreshold threshold for stripping location data
+     * @param {number} locationStrippingBytesThreshold threshold for stripping location data
      */
-    constructor(locationStrippingThreshold) {
-        super(locationStrippingThreshold);
+    constructor(locationStrippingBytesThreshold) {
+        super(locationStrippingBytesThreshold);
         // getPipeline is used standalone later, make sure its `this` reference binds to us.
         this.getPipeline = this.getPipeline.bind(this);
     }
diff --git a/tests/unit/oplogPopulator/OplogPopulatorConfigValidator.js b/tests/unit/oplogPopulator/OplogPopulatorConfigValidator.js
@@ -13,15 +13,15 @@ const defaultConfig = {
 };
 
 describe('OplogPopulatorConfigValidator', () => {
-    describe('locationStrippingThreshold validation', () => {
+    describe('locationStrippingBytesThreshold validation', () => {
         it('should accept valid threshold', () => {
             const config = {
                 ...defaultConfig,
-                locationStrippingThreshold: 50,
+                locationStrippingBytesThreshold: 100 * 1024 * 1024,
             };
             const result = OplogPopulatorConfigJoiSchema.validate(config);
             assert.ifError(result.error);
-            assert.strictEqual(result.value.locationStrippingThreshold, 50);
+            assert.strictEqual(result.value.locationStrippingBytesThreshold, 100 * 1024 * 1024);
         });
     });
 });
diff --git a/tests/unit/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js b/tests/unit/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js
@@ -4,7 +4,8 @@ const MultipleBucketsPipelineFactory =
 const { constants } = require('arsenal');
 
 describe('MultipleBucketsPipelineFactory', () => {
-    const multipleBucketsPipelineFactory = new MultipleBucketsPipelineFactory(200);
+    const thresholdBytes = 100 * 1024 * 1024;
+    const multipleBucketsPipelineFactory = new MultipleBucketsPipelineFactory(thresholdBytes);
 
     describe('isValid', () => {
         it('should detect a valid list of buckets', () => {
@@ -52,9 +53,20 @@ describe('MultipleBucketsPipelineFactory', () => {
 
             assert.strictEqual(pipeline.length, 2);
             assert.deepStrictEqual(pipeline[0], {$match:{'ns.coll':{$in:['bucket1','bucket2']}}});
-            assert(pipeline[1].$set['fullDocument.value.location']);
-            assert(pipeline[1].$set['updateDescription.updatedFields.value.location']);
-            assert(result.includes('200'));
+            assert.deepStrictEqual(pipeline[1].$set['fullDocument.value.location'], {
+                $cond: {
+                    if: { $gte: ['$fullDocument.value.content-length', thresholdBytes] },
+                    then: '$$REMOVE',
+                    else: '$fullDocument.value.location',
+                },
+            });
+            assert.deepStrictEqual(pipeline[1].$set['updateDescription.updatedFields.value.location'], {
+                $cond: {
+                    if: { $gte: ['$updateDescription.updatedFields.value.content-length', thresholdBytes] },
+                    then: '$$REMOVE',
+                    else: '$updateDescription.updatedFields.value.location',
+                },
+            });
         });
     });
 
diff --git a/tests/unit/oplogPopulator/pipeline/WildcardPipelineFactory.js b/tests/unit/oplogPopulator/pipeline/WildcardPipelineFactory.js
@@ -3,7 +3,8 @@ const WildcardPipelineFactory = require('../../../../extensions/oplogPopulator/p
 const { constants } = require('arsenal');
 
 describe('WildcardPipelineFactory', () => {
-    const wildcardPipelineFactory = new WildcardPipelineFactory(200);
+    const thresholdBytes = 100 * 1024 * 1024;
+    const wildcardPipelineFactory = new WildcardPipelineFactory(thresholdBytes);
 
     describe('isValid', () => {
         it('should detect a wildcard', () => {
@@ -39,13 +40,24 @@ describe('WildcardPipelineFactory', () => {
 
             assert.strictEqual(pipeline.length, 2);
             assert.deepStrictEqual(pipeline[0], {$match:{'ns.coll':{$not:{$regex:'^(mpuShadowBucket|__).*'}}}});
-            assert(pipeline[1].$set['fullDocument.value.location']);
-            assert(pipeline[1].$set['updateDescription.updatedFields.value.location']);
-            assert(result.includes('200'));
+            assert.deepStrictEqual(pipeline[1].$set['fullDocument.value.location'], {
+                $cond: {
+                    if: { $gte: ['$fullDocument.value.content-length', thresholdBytes] },
+                    then: '$$REMOVE',
+                    else: '$fullDocument.value.location',
+                },
+            });
+            assert.deepStrictEqual(pipeline[1].$set['updateDescription.updatedFields.value.location'], {
+                $cond: {
+                    if: { $gte: ['$updateDescription.updatedFields.value.content-length', thresholdBytes] },
+                    then: '$$REMOVE',
+                    else: '$updateDescription.updatedFields.value.location',
+                },
+            });
         });
 
         it('should return the pipeline with buckets and no location stripping if disabled', () => {
-            const wildcardPipelineFactoryNoStripping = new WildcardPipelineFactory(-1);
+            const wildcardPipelineFactoryNoStripping = new WildcardPipelineFactory(0);
 
             const buckets = ['bucket1', 'bucket2'];
             const result = wildcardPipelineFactoryNoStripping.getPipeline(buckets);
