S3C-11127: publish addressed notification entries behind a delivery pool flag

extensions/notification/NotificationConfigValidator.js

@@ -77,6 +77,9 @@ const destinationSchema = joi.object({
         then: joi.forbidden(),
         otherwise: joi.string().default('none'),
     }),
+    // number of record keys the destination is spread over: raise it to let
+    // more than one delivery worker handle the destination in parallel
+    spreadFactor: joi.number().integer().min(1).default(1),
 });
 
 const joiSchema = joi.object({
@@ -89,6 +92,29 @@ const joiSchema = joi.object({
         concurrency: joi.number().greater(0).default(1000),
         maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
     }),
+    // single consumer group delivering to every destination, addressed by
+    // the record itself instead of by one topic per destination.
+    // deliveryTimeoutMs must stay above the producer request timeout (5000)
+    // and below kafka.maxPollIntervalMs minus a margin, otherwise a slow
+    // destination holds the partition past the poll deadline and the
+    // consumer is evicted.
+    deliveryPool: joi.object({
+        enabled: joi.boolean().default(false),
+        topic: joi.string().when('enabled', {
+            is: joi.boolean().valid(true).required(),
+            then: joi.required(),
+        }),
+        groupId: joi.string().when('enabled', {
+            is: joi.boolean().valid(true).required(),
+            then: joi.required(),
+        }),
+        deliveryTimeoutMs: joi.number().min(6000).max(240000).default(30000),
+        producerIdleMs: joi.number().greater(0).default(300000),
+        maxProducers: joi.number().greater(0).default(50),
+        concurrency: joi.number().greater(0).default(1000),
+        maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
+        probeServer: probeServerJoi.optional(),
+    }).optional(),
     destinations: joi.array().items(destinationSchema).default([]),
     // TODO: BB-625 reset to being required after supporting probeserver in S3C
     // for bucket notification proceses

extensions/notification/NotificationQueuePopulator.js

@@ -7,6 +7,7 @@ const VID_SEPERATOR = require('arsenal').versioning.VersioningConstants.VersionI
 const configUtil = require('./utils/config');
 const safeJsonParse = require('../../lib/util/safeJsonParse');
 const messageUtil = require('./utils/message');
+const { buildDeliveryKey } = require('./utils/deliveryKey');
 const notifConstants = require('./constants');
 const QueuePopulatorExtension =
     require('../../lib/queuePopulator/QueuePopulatorExtension');
@@ -204,6 +205,132 @@ class NotificationQueuePopulator extends QueuePopulatorExtension {
         return (value && value[notifConstants.eventMessageProperty.dateTime]) || null;
     }
 
+    /**
+     * Publish the entry on the internal topic of every destination it
+     * matches, at most once per internal topic
+     *
+     * @param {String} bucket - bucket
+     * @param {Object} config - bucket notification configuration
+     * @param {Object} ent - notification entry
+     * @param {Object} value - log entry object
+     * @return {undefined}
+     */
+    _publishLegacyEntries(bucket, config, ent, value) {
+        const { versionId, eventType } = ent;
+        const pushedToTopic = new Map();
+        // validate and push kafka message foreach destination topic
+        this.notificationConfig.destinations.forEach(destination => {
+            const topic = destination.internalTopic ||
+                this.notificationConfig.topic;
+            // avoid pushing a message multiple times to the
+            // same internal topic
+            if (pushedToTopic[topic]) {
+                return undefined;
+            }
+            // get destination specific notification config
+            const queueConfig = config.notificationConfiguration.queueConfig.filter(
+                c => c.queueArn.split(':').pop() === destination.resource
+            );
+            if (!queueConfig.length) {
+                // skip, if there is no config for the current
+                // destination resource
+                return undefined;
+            }
+            // pass only destination resource specific config to
+            // validate entry
+            const destConfig = {
+                bucket,
+                notificationConfiguration: {
+                    queueConfig,
+                },
+            };
+            const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent);
+            if (isValid) {
+                const message
+                    = messageUtil.addLogAttributes(value, ent);
+                this.log.info('publishing message', {
+                    method: 'NotificationQueuePopulator._processObjectEntry',
+                    bucket,
+                    key: message.key,
+                    versionId,
+                    eventType,
+                    eventTime: message.dateTime,
+                    matchingConfig,
+                });
+                this.publish(topic,
+                    // keeping all messages for same object
+                    // in the same partition to keep the order.
+                    // here we use the object name and not the
+                    // "_id" which also includes the versionId
+                    `${bucket}/${message.key}`,
+                    JSON.stringify(message));
+                // keep track of internal topics we have pushed to
+                pushedToTopic[topic] = true;
+            }
+            return undefined;
+        });
+    }
+
+    /**
+     * Publish one addressed entry per matching destination on the shared
+     * delivery topic. The destination is carried in the message instead of
+     * the topic, so the delivery pool knows where to send it without
+     * needing a topic per destination.
+     *
+     * @param {String} bucket - bucket
+     * @param {Object} config - bucket notification configuration
+     * @param {Object} ent - notification entry
+     * @param {Object} value - log entry object
+     * @return {undefined}
+     */
+    _publishAddressedEntries(bucket, config, ent, value) {
+        const { versionId, eventType } = ent;
+        const { deliveryPool } = this.notificationConfig;
+        this.notificationConfig.destinations.forEach(destination => {
+            // get destination specific notification config
+            const queueConfig = config.notificationConfiguration.queueConfig.filter(
+                c => c.queueArn.split(':').pop() === destination.resource
+            );
+            if (!queueConfig.length) {
+                // skip, if there is no config for the current
+                // destination resource
+                return undefined;
+            }
+            // pass only destination resource specific config to
+            // validate entry
+            const destConfig = {
+                bucket,
+                notificationConfiguration: {
+                    queueConfig,
+                },
+            };
+            const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent);
+            if (isValid) {
+                const message
+                    = messageUtil.addLogAttributes(value, ent);
+                message.destinationId = destination.resource;
+                message.configurationId = matchingConfig.id;
+                this.log.info('publishing addressed message', {
+                    method: 'NotificationQueuePopulator._publishAddressedEntries',
+                    bucket,
+                    key: message.key,
+                    versionId,
+                    eventType,
+                    eventTime: message.dateTime,
+                    destinationId: message.destinationId,
+                    matchingConfig,
+                });
+                this.publish(deliveryPool.topic,
+                    // the key keeps all messages for the same destination
+                    // and object in the same partition, so that a single
+                    // worker handles them in order
+                    buildDeliveryKey(destination, bucket, message.key),
+                    JSON.stringify(message));
+            }
+            return undefined;
+        });
+    }
+
     /**
      * Process object entry from the log
      *
@@ -250,58 +377,12 @@ class NotificationQueuePopulator extends QueuePopulatorExtension {
                     key,
                     eventType,
                 });
-                const pushedToTopic = new Map();
-                // validate and push kafka message foreach destination topic
-                this.notificationConfig.destinations.forEach(destination => {
-                    const topic = destination.internalTopic ||
-                        this.notificationConfig.topic;
-                    // avoid pushing a message multiple times to the
-                    // same internal topic
-                    if (pushedToTopic[topic]) {
-                        return undefined;
-                    }
-                    // get destination specific notification config
-                    const queueConfig = config.notificationConfiguration.queueConfig.filter(
-                        c => c.queueArn.split(':').pop() === destination.resource
-                    );
-                    if (!queueConfig.length) {
-                        // skip, if there is no config for the current
-                        // destination resource
-                        return undefined;
-                    }
-                    // pass only destination resource specific config to
-                    // validate entry
-                    const destConfig = {
-                        bucket,
-                        notificationConfiguration: {
-                            queueConfig,
-                        },
-                    };
-                    const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent);
-                    if (isValid) {
-                        const message
-                            = messageUtil.addLogAttributes(value, ent);
-                        this.log.info('publishing message', {
-                            method: 'NotificationQueuePopulator._processObjectEntry',
-                            bucket,
-                            key: message.key,
-                            versionId,
-                            eventType,
-                            eventTime: message.dateTime,
-                            matchingConfig,
-                        });
-                        this.publish(topic,
-                            // keeping all messages for same object
-                            // in the same partition to keep the order.
-                            // here we use the object name and not the
-                            // "_id" which also includes the versionId
-                            `${bucket}/${message.key}`,
-                            JSON.stringify(message));
-                        // keep track of internal topics we have pushed to
-                        pushedToTopic[topic] = true;
-                    }
-                    return undefined;
-                });
+                const { deliveryPool } = this.notificationConfig;
+                if (deliveryPool && deliveryPool.enabled) {
+                    this._publishAddressedEntries(bucket, config, ent, value);
+                } else {
+                    this._publishLegacyEntries(bucket, config, ent, value);
+                }
             }
             // skip if there is no bucket notification configuration
             return undefined;

extensions/notification/utils/deliveryKey.js

@@ -0,0 +1,25 @@
+const crypto = require('crypto');
+
+/**
+ * Builds the delivery-topic record key for an addressed notification.
+ * The key is stable across processes and reruns: same destination and
+ * object always map to the same key, hence the same partition.
+ *
+ * @param {Object} destination - destination config entry
+ * @param {String} bucket - bucket name
+ * @param {String} objectKey - object key
+ * @return {String} record key
+ */
+function buildDeliveryKey(destination, bucket, objectKey) {
+    const m = destination.spreadFactor || 1;
+    if (m <= 1) {
+        return destination.resource;
+    }
+    const h = crypto.createHash('md5')
+        .update(`${bucket}/${objectKey}`)
+        .digest()
+        .readUInt32BE(0);
+    return `${destination.resource}|${h % m}`;
+}
+
+module.exports = { buildDeliveryKey };

tests/config.notification.json

@@ -35,6 +35,11 @@
                 "groupId": "backbeat-bucket-notification-group",
                 "concurrency": 10
             },
+            "deliveryPool": {
+                "enabled": false,
+                "topic": "backbeat-bucket-notification-delivery",
+                "groupId": "backbeat-bucket-notification-delivery-group"
+            },
             "destinations": [
                 {
                     "resource": "destination1",
@@ -43,6 +48,7 @@
                     "port": 9092,
                     "topic": "destination-topic-1",
                     "internalTopic": "internal-notification-topic-destination1",
+                    "spreadFactor": 1,
                     "auth": {}
                 },
                 {