oplog: Strip locations when the array is large

This saves space on the Kafka side while preventing a refetch
on most (small) objects.

Issue: BB-491
Signed-off-by: Thomas Flament <thomas.flament@scality.com>

Author: delthas

conf/config.json

@@ -304,6 +304,7 @@
             "kafkaConnectHost": "127.0.0.1",
             "kafkaConnectPort": 8083,
             "numberOfConnectors": 1,
+            "locationStrippingThreshold": 5,
             "probeServer": {
                 "bindAddress": "0.0.0.0",
                 "port": 8556

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();
+            pipelineFactory = new WildcardPipelineFactory(this._config.locationStrippingThreshold);
             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();
+            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingThreshold);
             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();
+            pipelineFactory = new MultipleBucketsPipelineFactory(this._config.locationStrippingThreshold);
             strategy = new LeastFullConnector({
                 logger: this._logger,
             });

extensions/oplogPopulator/OplogPopulatorConfigValidator.js

@@ -6,6 +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),
     prefix: joi.string().optional(),
     probeServer: probeServerJoi.default(),
     connectorsUpdateCronRule: joi.string().default('*/1 * * * * *'),

extensions/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js

@@ -8,6 +8,16 @@ const PipelineFactory = require('./PipelineFactory');
  * given a list of buckets.
  */
 class MultipleBucketsPipelineFactory extends PipelineFactory {
+    /**
+     * @constructor
+     * @param {number} locationStrippingThreshold threshold for stripping location data
+     */
+    constructor(locationStrippingThreshold) {
+        super(locationStrippingThreshold);
+        // getPipeline is used standalone later, make sure its `this` reference binds to us.
+        this.getPipeline = this.getPipeline.bind(this);
+    }
+
     /**
      * Checks if an existing pipeline is valid against the current
      * factory.
@@ -22,24 +32,21 @@ class MultipleBucketsPipelineFactory extends PipelineFactory {
     }
 
     /**
-     * Makes new connector pipeline that includes
-     * buckets assigned to this connector.
+     * Makes connector pipeline stage, that includes buckets assigned to this connector.
      * @param {string[] | undefined} buckets buckets assigned to this connector
-     * @returns {string} new connector pipeline
+     * @returns {object} connector pipeline stage
      */
-    getPipeline(buckets) {
+    getPipelineStage(buckets) {
         if (!buckets || !buckets.length) {
-            return JSON.stringify([]);
+            return null;
         }
-        return JSON.stringify([
-            {
-                $match: {
-                    'ns.coll': {
-                        $in: buckets,
-                    }
+        return {
+            $match: {
+                'ns.coll': {
+                    $in: buckets,
                 }
             }
-        ]);
+        };
     }
 }
 

extensions/oplogPopulator/pipeline/PipelineFactory.js

@@ -7,6 +7,14 @@ const { wildCardForAllBuckets } = require('../constants');
  * @classdesc base class for the pipeline factories
  */
 class PipelineFactory {
+    /**
+     * @constructor
+     * @param {number} locationStrippingThreshold threshold for stripping location data
+     */
+    constructor(locationStrippingThreshold) {
+        this._locationStrippingThreshold = locationStrippingThreshold;
+    }
+
     /**
      * Checks if an existing pipeline is valid against the current
      * factory.
@@ -55,7 +63,40 @@ class PipelineFactory {
      * @param {string[] | undefined} buckets buckets assigned to this connector
      * @returns {string} new connector pipeline
      */
-    getPipeline(buckets) { // eslint-disable-line no-unused-vars
+    getPipeline(buckets) {
+        const stage = this.getPipelineStage(buckets);
+        if (!stage) {
+            return JSON.stringify([]);
+        }
+        const pipeline = [
+            stage,
+        ];
+        if (this._locationStrippingThreshold >= 0) {
+            // If a threshold is provided, strip the location array if the number of parts is larger than the threshold.
+            pipeline.push({
+                $set: {
+                    'fullDocument.value.location': {
+                        $cond: {
+                            if: {
+                                $gte: [
+                                    { $size: { $ifNull: ['$fullDocument.value.location', []] } },
+                                    this._locationStrippingThreshold
+                                ]
+                            },
+                        }
+                    }
+                }
+            });
+        }
+        return JSON.stringify(pipeline);
+    }
+
+    /**
+     * Makes connector pipeline stage, to then be used by getPipeline.
+     * @param {string[] | undefined} buckets buckets assigned to this connector
+     * @returns {object} connector pipeline stage
+     */
+    getPipelineStage(buckets) { // eslint-disable-line no-unused-vars
         throw errors.NotImplemented;
     }
 }

extensions/oplogPopulator/pipeline/WildcardPipelineFactory.js

@@ -9,6 +9,16 @@ const PipelineFactory = require('./PipelineFactory');
  * that listens to all buckets.
  */
 class WildcardPipelineFactory extends PipelineFactory {
+    /**
+     * @constructor
+     * @param {number} locationStrippingThreshold threshold for stripping location data
+     */
+    constructor(locationStrippingThreshold) {
+        super(locationStrippingThreshold);
+        // getPipeline is used standalone later, make sure its `this` reference binds to us.
+        this.getPipeline = this.getPipeline.bind(this);
+    }
+
     /**
      * Checks if an existing pipeline is valid against the current
      * factory.
@@ -23,23 +33,20 @@ class WildcardPipelineFactory extends PipelineFactory {
     }
 
     /**
-     * Create a pipeline for the connector, to listen to all
-     * non-special collections.
+     * Makes connector pipeline stage, to listen to all non-special collections.
      * @param {string[] | undefined} buckets buckets assigned to this connector
-     * @returns {string} new connector pipeline
+     * @returns {object} connector pipeline stage
      */
-    getPipeline(buckets) { // eslint-disable-line no-unused-vars
-        return JSON.stringify([
-            {
-                $match: {
-                    'ns.coll': {
-                        $not: {
-                            $regex: `^(${constants.mpuBucketPrefix}|__).*`,
-                        },
-                    }
+    getPipelineStage(buckets) { // eslint-disable-line no-unused-vars
+        return {
+            $match: {
+                'ns.coll': {
+                    $not: {
+                        $regex: `^(${constants.mpuBucketPrefix}|__).*`,
+                    },
                 }
             }
-        ]);
+        };
     }
 }
 

tests/unit/oplogPopulator/OplogPopulatorConfigValidator.js

@@ -0,0 +1,27 @@
+const assert = require('assert');
+const { OplogPopulatorConfigJoiSchema} = require('../../../extensions/oplogPopulator/OplogPopulatorConfigValidator');
+
+const defaultConfig = {
+    topic: 'backbeat-oplog',
+    kafkaConnectHost: '127.0.0.1',
+    kafkaConnectPort: 8083,
+    numberOfConnectors: 1,
+    probeServer: {
+        bindAddress: '0.0.0.0',
+        port: 8556,
+    },
+};
+
+describe('OplogPopulatorConfigValidator', () => {
+    describe('locationStrippingThreshold validation', () => {
+        it('should accept valid threshold', () => {
+            const config = {
+                ...defaultConfig,
+                locationStrippingThreshold: 50,
+            };
+            const result = OplogPopulatorConfigJoiSchema.validate(config);
+            assert.ifError(result.error);
+            assert.strictEqual(result.value.locationStrippingThreshold, 50);
+        });
+    });
+});

tests/unit/oplogPopulator/pipeline/MultipleBucketsPipelineFactory.js

@@ -4,7 +4,7 @@ const MultipleBucketsPipelineFactory =
 const { constants } = require('arsenal');
 
 describe('MultipleBucketsPipelineFactory', () => {
-    const multipleBucketsPipelineFactory = new MultipleBucketsPipelineFactory();
+    const multipleBucketsPipelineFactory = new MultipleBucketsPipelineFactory(200);
 
     describe('isValid', () => {
         it('should detect a valid list of buckets', () => {
@@ -45,10 +45,15 @@ describe('MultipleBucketsPipelineFactory', () => {
             assert.strictEqual(result, '[]');
         });
 
-        it('should return the pipeline with buckets', () => {
+        it('should return the pipeline with buckets and location stripping', () => {
             const buckets = ['bucket1', 'bucket2'];
             const result = multipleBucketsPipelineFactory.getPipeline(buckets);
-            assert.strictEqual(result, '[{"$match":{"ns.coll":{"$in":["bucket1","bucket2"]}}}]');
+            const pipeline = JSON.parse(result);
+
+            assert.strictEqual(pipeline.length, 2);
+            assert.deepStrictEqual(pipeline[0], {$match:{'ns.coll':{$in:['bucket1','bucket2']}}});
+            assert(pipeline[1].$set['fullDocument.value.location']);
+            assert(result.includes('200'));
         });
     });
 

tests/unit/oplogPopulator/pipeline/WildcardPipelineFactory.js

@@ -3,7 +3,7 @@ const WildcardPipelineFactory = require('../../../../extensions/oplogPopulator/p
 const { constants } = require('arsenal');
 
 describe('WildcardPipelineFactory', () => {
-    const wildcardPipelineFactory = new WildcardPipelineFactory();
+    const wildcardPipelineFactory = new WildcardPipelineFactory(200);
 
     describe('isValid', () => {
         it('should detect a wildcard', () => {
@@ -32,10 +32,25 @@ describe('WildcardPipelineFactory', () => {
     });
 
     describe('getPipeline', () => {
-        it('should return the pipeline with wildcard', () => {
+        it('should return the pipeline with buckets and location stripping', () => {
             const buckets = ['bucket1', 'bucket2'];
             const result = wildcardPipelineFactory.getPipeline(buckets);
-            assert.strictEqual(result, '[{"$match":{"ns.coll":{"$not":{"$regex":"^(mpuShadowBucket|__).*"}}}}]');
+            const pipeline = JSON.parse(result);
+
+            assert.strictEqual(pipeline.length, 2);
+            assert.deepStrictEqual(pipeline[0], {$match:{'ns.coll':{$not:{$regex:'^(mpuShadowBucket|__).*'}}}});
+            assert(pipeline[1].$set['fullDocument.value.location']);
+            assert(result.includes('200'));
+        });
+
+        it('should return the pipeline with buckets and no location stripping if disabled', () => {
+            const wildcardPipelineFactoryNoStripping = new WildcardPipelineFactory(-1);
+
+            const buckets = ['bucket1', 'bucket2'];
+            const result = wildcardPipelineFactoryNoStripping.getPipeline(buckets);
+            const pipeline = JSON.parse(result);
+
+            assert.strictEqual(pipeline.length, 1);
         });
     });