Support limiting bucket number in connectors and mongodb 6

.github/workflows/alerts.yaml

@@ -81,4 +81,5 @@ jobs:
             namespace=zenko
             oplog_populator_job=artesca-data-backbeat-oplog-populator-headless
             oplogPopulatorChangeStreamLagThreshold=10
+            oplogPopulatorChangeStreamLimit=10
           github_token: ${{ secrets.GIT_ACCESS_TOKEN }}

.gitignore

@@ -9,9 +9,13 @@ pids
 
 # Coverage directory used by tools like istanbul
 coverage
+.nyc_output
 
 # Dependencies
 node_modules
 
 # Redis
 *.rdb
+
+# Python cache, created when refreshing the monitoring files
+__pycache__

extensions/oplogPopulator/OplogPopulator.js

@@ -11,6 +11,9 @@ const { ZenkoMetrics } = require('arsenal').metrics;
 const OplogPopulatorMetrics = require('./OplogPopulatorMetrics');
 const { OplogPopulatorConfigJoiSchema } = require('./OplogPopulatorConfigValidator');
 const { mongoJoi } = require('../../lib/config/configItems.joi');
+const ImmutableConnector = require('./allocationStrategy/ImmutableConnector');
+const RetainBucketsDecorator = require('./allocationStrategy/RetainBucketsDecorator');
+const LeastFullConnector = require('./allocationStrategy/LeastFullConnector');
 
 const paramsJoi = joi.object({
     config: OplogPopulatorConfigJoiSchema.required(),
@@ -51,7 +54,7 @@ class OplogPopulator {
         this._logger = params.logger;
         this._changeStreamWrapper = null;
         this._allocator = null;
-        this._connectorsManager  = null;
+        this._connectorsManager = null;
         // contains OplogPopulatorUtils class of each supported extension
         this._extHelpers = {};
         // MongoDB related
@@ -78,9 +81,9 @@ class OplogPopulator {
     async _setupMongoClient() {
         try {
             const client = await MongoClient.connect(this._mongoUrl, {
-                 replicaSet: this._replicaSet,
-                 useNewUrlParser: true,
-                 useUnifiedTopology: true,
+                replicaSet: this._replicaSet,
+                useNewUrlParser: true,
+                useUnifiedTopology: true,
             });
             // connect to metadata DB
             this._mongoClient = client.db(this._database, {
@@ -242,35 +245,58 @@ class OplogPopulator {
         this._changeStreamWrapper.start();
     }
 
+    _arePipelinesImmutable() {
+        return semver.gte(this._mongoVersion, constants.mongodbVersionWithImmutablePipelines);
+    }
+
+    async _initializeConnectorsManager() {
+        return this._connectorsManager.initializeConnectors();
+    }
+
     /**
      * Sets the OplogPopulator
      * @returns {Promise|undefined} undefined
      * @throws {InternalError}
      */
     async setup() {
-       try {
-           this._loadOplogHelperClasses();
-           this._connectorsManager = new ConnectorsManager({
-               nbConnectors: this._config.numberOfConnectors,
-               database: this._database,
-               mongoUrl: this._mongoUrl,
-               oplogTopic: this._config.topic,
-               cronRule: this._config.connectorsUpdateCronRule,
-               prefix: this._config.prefix,
-               heartbeatIntervalMs: this._config.heartbeatIntervalMs,
-               kafkaConnectHost: this._config.kafkaConnectHost,
-               kafkaConnectPort: this._config.kafkaConnectPort,
-               metricsHandler: this._metricsHandler,
-               logger: this._logger,
+        try {
+            this._loadOplogHelperClasses();
+            // initialize mongo client
+            await this._setupMongoClient();
+            this._allocationStrategy = this.initStrategy();
+            this._connectorsManager = new ConnectorsManager({
+                nbConnectors: this._config.numberOfConnectors,
+                database: this._database,
+                mongoUrl: this._mongoUrl,
+                oplogTopic: this._config.topic,
+                cronRule: this._config.connectorsUpdateCronRule,
+                prefix: this._config.prefix,
+                heartbeatIntervalMs: this._config.heartbeatIntervalMs,
+                kafkaConnectHost: this._config.kafkaConnectHost,
+                kafkaConnectPort: this._config.kafkaConnectPort,
+                metricsHandler: this._metricsHandler,
+                allocationStrategy: this._allocationStrategy,
+                logger: this._logger,
             });
-            await this._connectorsManager.initializeConnectors();
+            await this._initializeConnectorsManager();
             this._allocator = new Allocator({
                 connectorsManager: this._connectorsManager,
                 metricsHandler: this._metricsHandler,
+                allocationStrategy: this._allocationStrategy,
                 logger: this._logger,
             });
-            // initialize mongo client
-            await this._setupMongoClient();
+            // For now, we always use the RetainBucketsDecorator
+            // so, we map the events from the classes
+            this._connectorsManager.on('connector-updated', connector =>
+                this._allocationStrategy.onConnectorUpdatedOrDestroyed(connector));
+            this._allocator.on('bucket-removed', (bucket, connector) =>
+                this._allocationStrategy.onBucketRemoved(bucket, connector));
+            this._connectorsManager.on('connectors-reconciled', bucketsExceedingLimit => {
+                this._metricsHandler.onConnectorsReconciled(
+                    bucketsExceedingLimit,
+                    this._allocationStrategy.retainedBucketsCount,
+                );
+            });
             // get currently valid buckets from mongo
             const validBuckets = await this._getBackbeatEnabledBuckets();
             // listen to valid buckets
@@ -291,13 +317,48 @@ class OplogPopulator {
             this._logger.info('OplogPopulator setup complete', {
                 method: 'OplogPopulator.setup',
             });
-       } catch (err) {
+        } catch (err) {
             this._logger.error('An error occured when setting up the OplogPopulator', {
                 method: 'OplogPopulator.setup',
                 error: err.description || err.message,
             });
             throw errors.InternalError.customizeDescription(err.description);
-       }
+        }
+    }
+
+    /**
+     * Init the allocation strategy
+     * @returns {RetainBucketsDecorator} extended allocation strategy
+     * handling retained buckets
+     */
+    initStrategy() {
+        let strategy;
+        if (this._arePipelinesImmutable()) {
+            // In this case, mongodb does not support reusing a
+            // resume token from a different pipeline. In other
+            // words, we cannot alter an existing pipeline. In this
+            // case, the strategy is to allow a maximum of one
+            // bucket per kafka connector.
+            strategy = new ImmutableConnector({
+                logger: this._logger,
+                metricsHandler: this._metricsHandler,
+                connectorsManager: this._connectorsManager,
+            });
+        } else {
+            // In this case, we can have multiple buckets per
+            // kafka connector. However, we want to proactively
+            // ensure that the pipeline will be accepted by
+            // mongodb.
+            strategy = new LeastFullConnector({
+                logger: this._logger,
+                metricsHandler: this._metricsHandler,
+                connectorsManager: this._connectorsManager,
+            });
+        }
+        return new RetainBucketsDecorator(
+            strategy,
+            { logger: this._logger },
+        );
     }
 
     /**

extensions/oplogPopulator/OplogPopulatorMetrics.js

@@ -33,6 +33,14 @@
             help: 'Total number of buckets per connector',
             labelNames: ['connector'],
         });
+        this.bucketsExceedingLimit = ZenkoMetrics.createGauge({
+            name: 's3_oplog_populator_connector_buckets_exceeding_limit',
+            help: 'Total number of buckets exceeding the limit for all connectors',
+        });
+        this.retainedBuckets = ZenkoMetrics.createGauge({
+            name: 's3_oplog_populator_connector_retained_buckets',
+            help: 'Current number of buckets still listened to by immutable connectors despite intended removal',
+        });
         this.requestSize = ZenkoMetrics.createCounter({
             name: 's3_oplog_populator_connector_request_bytes_total',
             help: 'Total size of kafka connect request in bytes',
@@ -196,6 +204,26 @@
             });
         }
     }
+
+    /**
+     * updates metrics when the connectors are reconciled
+     * @param {number} bucketsExceedingLimit number of buckets above the limit
+     * for all connectors
+     * @param {number} retainedBuckets number of buckets still listened to by
+     * immutable connectors despite intended removal
+     * @returns {undefined}
+     */
+    onConnectorsReconciled(bucketsExceedingLimit, retainedBuckets) {
+        try {
+            this.bucketsExceedingLimit.set(bucketsExceedingLimit);
+            this.retainedBuckets.set(retainedBuckets);
+        } catch (error) {
+            this._logger.error('An error occured while pushing metric', {
+                method: 'OplogPopulatorMetrics.onConnectorsReconciled',
+                error: error.message,
+            });
+        }
+    }
 }
 
 module.exports = OplogPopulatorMetrics;

extensions/oplogPopulator/allocationStrategy/AllocationStrategy.js

@@ -4,21 +4,39 @@
 
     /**
      * @constructor
-     * @param {Logger} logger logger object
+     * @param {Object} params params
+     * @param {Logger} params.logger logger object
      */
-    constructor(logger) {
-        this._logger = logger;
+    constructor(params) {
+        this._logger = params.logger;
     }
 
     /**
-     * Get best connector for assigning a bucket
-     * @param {Connector[]} connectors available connectors
-     * @returns {Connector} connector
+     * Get best connector to assign a bucket to.
+     * If no connector is available, null is returned.
+     * @param {Array<Connector>} connectors connectors
+     * @param {String} bucket bucket name
+     * @returns {Connector | null} connector
      */
-    getConnector(connectors) { // eslint-disable-line no-unused-vars
+    getConnector(connectors, bucket) { // eslint-disable-line no-unused-vars
         throw errors.NotImplemented;
     }
 
+    /**
+     * Assess if a pipeline can be updated
+     * @returns {Boolean} true if the connector can be updated
+     */
+    canUpdate() {
+        throw errors.NotImplemented;
+    }
+
+    /**
+     * Getter for the maximum number of buckets per connector
+     * @returns {Number} maximum number of buckets per connector
+     */
+    get maximumBucketsPerConnector() {
+        throw errors.NotImplemented;
+    }
 }
 
 module.exports = AllocationStrategy;

extensions/oplogPopulator/allocationStrategy/ImmutableConnector.js

@@ -0,0 +1,40 @@
+const AllocationStrategy = require('./AllocationStrategy');
+
+/**
+ * @class ImmutableConnector
+ *
+ * @classdesc ImmutableConnector is an allocation
+ * strategy that always requires a new connector to be
+ * created for each bucket.
+ */
+class ImmutableConnector extends AllocationStrategy {
+    /**
+     * Get best connector to assign a bucket to.
+     * If no connector is available, null is returned.
+     * @param {Array<Connector>} connectors connectors
+     * @param {String} bucket bucket name
+     * @returns {Connector | null} connector
+     */
+    getConnector(connectors, bucket) { // eslint-disable-line no-unused-vars
+        return null;
+    }
+
+    /**
+     * Assess if a pipeline can be updated. With the immutable
+     * strategy, a connector cannot be updated.
+     * @returns {false} false
+     */
+    canUpdate() {
+        return false;
+    }
+
+    /**
+     * Getter for the maximum number of buckets per connector
+     * @returns {Number} maximum number of buckets per connector
+     */
+    get maximumBucketsPerConnector() {
+        return 1;
+    }
+}
+
+module.exports = ImmutableConnector;

extensions/oplogPopulator/allocationStrategy/LeastFullConnector.js

@@ -1,3 +1,4 @@
+const constants = require('../constants');
 const AllocationStrategy = require('./AllocationStrategy');
 
 /**
@@ -11,21 +12,37 @@ const AllocationStrategy = require('./AllocationStrategy');
 class LeastFullConnector extends AllocationStrategy {
 
     /**
-     * @constructor
-     * @param {Object} params params
-     * @param {Logger} params.logger logger object
+     * Get best connector to assign a bucket to.
+     * If no connector is available, null is returned.
+     * @param {Array<Connector>} connectors connectors
+     * @param {String} bucket bucket name
+     * @returns {Connector | null} connector
      */
-    constructor(params) {
-        super(params.logger);
+    getConnector(connectors, bucket) { // eslint-disable-line no-unused-vars
+        if (!connectors.length) {
+            return null;
+        }
+        const connector = connectors.reduce((prev, elt) => (elt.bucketCount < prev.bucketCount ? elt : prev));
+        if (connector.buckets.length >= this.maximumBucketsPerConnector) {
+            return null;
+        }
+        return connector;
     }
 
     /**
-     * Get best connector for assigning a bucket
-     * @param {Connector[]} connectors available connectors
-     * @returns {Connector} connector
+     * Assess if a pipeline can be updated.
+     * @returns {true} true
      */
-    getConnector(connectors) {
-        return connectors.reduce((prev, elt) => (elt.bucketCount < prev.bucketCount ? elt : prev));
+    canUpdate() {
+        return true;
+    }
+
+    /**
+     * Getter for the maximum number of buckets per connector
+     * @returns {Number} maximum number of buckets per connector
+     */
+    get maximumBucketsPerConnector() {
+        return constants.maxBucketsPerConnector;
     }
 }