diff --git a/extensions/notification/NotificationConfigValidator.js b/extensions/notification/NotificationConfigValidator.js index f6854522e6..67d2939ba9 100644 --- a/extensions/notification/NotificationConfigValidator.js +++ b/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 diff --git a/extensions/notification/NotificationQueuePopulator.js b/extensions/notification/NotificationQueuePopulator.js index 30b084d5fb..4fe3f8d6e0 100644 --- a/extensions/notification/NotificationQueuePopulator.js +++ b/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; diff --git a/extensions/notification/utils/deliveryKey.js b/extensions/notification/utils/deliveryKey.js new file mode 100644 index 0000000000..4ea9fe6e4b --- /dev/null +++ b/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 }; diff --git a/tests/config.notification.json b/tests/config.notification.json index 5fc51ab1db..36a99d7f32 100644 --- a/tests/config.notification.json +++ b/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": {} }, { diff --git a/tests/unit/notification/NotificationConfigValidator.js b/tests/unit/notification/NotificationConfigValidator.js index b65e5a64e1..2829d06343 100644 --- a/tests/unit/notification/NotificationConfigValidator.js +++ b/tests/unit/notification/NotificationConfigValidator.js @@ -476,3 +476,108 @@ describe('NotificationConfigValidator ::', () => { }) ); }); + +describe('NotificationConfigValidator delivery pool ::', () => { + const destinationConfig = { + resource: 'resource', + type: 'kafka', + host: 'host', + port: 8000, + topic: 'topic', + }; + + it('should default the destination spread factor to 1', () => { + const config = notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [destinationConfig], + }); + assert.strictEqual(config.destinations[0].spreadFactor, 1); + }); + + it('should reject a spread factor below 1', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [{ ...destinationConfig, spreadFactor: 0 }], + })); + }); + + it('should reject a non integer spread factor', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [{ ...destinationConfig, spreadFactor: 1.5 }], + })); + }); + + it('should leave the delivery pool unset when it is not configured', () => { + const config = notificationConfigValidator(null, defaultExtConfig); + assert.strictEqual(config.deliveryPool, undefined); + }); + + it('should apply the delivery pool defaults', () => { + const config = notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: {}, + }); + assert.strictEqual(config.deliveryPool.enabled, false); + assert.strictEqual(config.deliveryPool.deliveryTimeoutMs, 30000); + assert.strictEqual(config.deliveryPool.producerIdleMs, 300000); + assert.strictEqual(config.deliveryPool.maxProducers, 50); + assert.strictEqual(config.deliveryPool.concurrency, 1000); + assert.strictEqual(config.deliveryPool.maxQueued, 1000); + }); + + it('should accept an enabled delivery pool with a topic and a group id', () => { + assert.doesNotThrow(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + }, + })); + }); + + it('should require a topic when the delivery pool is enabled', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + groupId: 'delivery-group', + }, + })); + }); + + it('should require a group id when the delivery pool is enabled', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + }, + })); + }); + + it('should reject a delivery timeout below the producer request timeout', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + deliveryTimeoutMs: 5000, + }, + })); + }); + + it('should reject a delivery timeout above the poll interval margin', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + deliveryTimeoutMs: 240001, + }, + })); + }); +}); diff --git a/tests/unit/notification/NotificationQueuePopulator.js b/tests/unit/notification/NotificationQueuePopulator.js index b38a16d678..95766630b4 100644 --- a/tests/unit/notification/NotificationQueuePopulator.js +++ b/tests/unit/notification/NotificationQueuePopulator.js @@ -741,3 +741,169 @@ describe('NotificationQueuePopulator with multiple rules ::', () => { }); }); }); + +describe('NotificationQueuePopulator with delivery pool ::', () => { + const deliveryTopic = notificationConfig.deliveryPool.topic; + const addressedConfig = { + bucket: 'example-bucket', + notificationConfiguration: { + queueConfig: [ + { + events: ['s3:ObjectCreated:Put'], + queueArn: 'arn:scality:bucketnotif:::destination1', + id: 'config-1', + filterRules: [], + }, + { + events: ['s3:ObjectCreated:Put'], + queueArn: 'arn:scality:bucketnotif:::destination2', + id: 'config-2', + filterRules: [], + }, + ], + }, + }; + const objectEntry = { + 'originOp': 's3:ObjectCreated:Put', + 'dataStoreName': 'metastore', + 'content-length': '100', + 'last-modified': '0000', + 'md-model-version': '1', + }; + let bnConfigManager; + let notificationQueuePopulator; + + beforeEach(() => { + bnConfigManager = new NotificationConfigManager({ + mongoConfig, + bucketMetastore: '__metastore', + maxCachedConfigs: 1000, + logger, + }); + sinon.stub(bnConfigManager, 'getConfig').returns(addressedConfig); + notificationQueuePopulator = new NotificationQueuePopulator({ + config: { + ...notificationConfig, + deliveryPool: { + ...notificationConfig.deliveryPool, + enabled: true, + }, + }, + bnConfigManager, + logger, + }); + notificationQueuePopulator._metricsStore = { + notifEvent: () => null, + }; + }); + + it('should publish one record per matching destination on the delivery topic', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.calledTwice); + assert.strictEqual(publishStub.getCall(0).args.at(0), deliveryTopic); + assert.strictEqual(publishStub.getCall(1).args.at(0), deliveryTopic); + }); + + it('should use the destination resource as key when the spread factor is 1', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert.strictEqual(publishStub.getCall(0).args.at(1), 'destination1'); + assert.strictEqual(publishStub.getCall(1).args.at(1), 'destination2'); + }); + + it('should address each record with its destination and configuration', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + const first = JSON.parse(publishStub.getCall(0).args.at(2)); + const second = JSON.parse(publishStub.getCall(1).args.at(2)); + assert.strictEqual(first.destinationId, 'destination1'); + assert.strictEqual(first.configurationId, 'config-1'); + assert.strictEqual(first.bucket, 'example-bucket'); + assert.strictEqual(first.key, 'example-key'); + assert.strictEqual(first.eventType, 's3:ObjectCreated:Put'); + assert.strictEqual(second.destinationId, 'destination2'); + assert.strictEqual(second.configurationId, 'config-2'); + }); + + it('should publish one record per destination even when destinations ' + + 'share an internal topic', async () => { + notificationQueuePopulator.notificationConfig = { + ...notificationQueuePopulator.notificationConfig, + destinations: notificationConfig.destinations.map(destination => ({ + ...destination, + internalTopic: 'custom-topic', + })), + }; + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.calledTwice); + assert.strictEqual(publishStub.getCall(0).args.at(0), deliveryTopic); + assert.strictEqual(publishStub.getCall(1).args.at(0), deliveryTopic); + const destinationIds = [0, 1].map(call => + JSON.parse(publishStub.getCall(call).args.at(2)).destinationId); + assert.deepStrictEqual(destinationIds, ['destination1', 'destination2']); + }); + + it('should spread a destination over its spread factor and keep the key ' + + 'stable for the same object', async () => { + notificationQueuePopulator.notificationConfig = { + ...notificationQueuePopulator.notificationConfig, + destinations: notificationConfig.destinations.map(destination => ({ + ...destination, + spreadFactor: 4, + })), + }; + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert.strictEqual(publishStub.callCount, 4); + const key = publishStub.getCall(0).args.at(1); + const [resource, index] = key.split('|'); + assert.strictEqual(resource, 'destination1'); + assert(Number(index) >= 0 && Number(index) < 4); + // the same object is always addressed to the same key, so the same + // partition, so the same delivery worker + assert.strictEqual(publishStub.getCall(2).args.at(1), key); + }); + + it('should not publish anything when no destination matches', async () => { + bnConfigManager.getConfig.returns({ + bucket: 'example-bucket', + notificationConfiguration: { + queueConfig: [ + { + events: ['s3:ObjectRemoved:Delete'], + queueArn: 'arn:scality:bucketnotif:::destination1', + id: 'config-1', + filterRules: [], + }, + ], + }, + }); + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.notCalled); + }); +}); diff --git a/tests/unit/notification/utils/deliveryKey.js b/tests/unit/notification/utils/deliveryKey.js new file mode 100644 index 0000000000..2437dd7041 --- /dev/null +++ b/tests/unit/notification/utils/deliveryKey.js @@ -0,0 +1,55 @@ +const assert = require('assert'); + +const { buildDeliveryKey } + = require('../../../../extensions/notification/utils/deliveryKey'); + +describe('deliveryKey ::', () => { + it('should use the destination resource when no spread factor is set', () => { + const key = buildDeliveryKey({ resource: 'destination1' }, + 'example-bucket', 'example-key'); + assert.strictEqual(key, 'destination1'); + }); + + it('should use the destination resource when the spread factor is 1', () => { + const key = buildDeliveryKey({ resource: 'destination1', spreadFactor: 1 }, + 'example-bucket', 'example-key'); + assert.strictEqual(key, 'destination1'); + }); + + it('should keep the key stable for the same bucket and object', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + const key = buildDeliveryKey(destination, 'example-bucket', 'example-key'); + for (let i = 0; i < 10; i++) { + assert.strictEqual( + buildDeliveryKey(destination, 'example-bucket', 'example-key'), key); + } + }); + + it('should keep the spread index within the spread factor bounds', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + for (let i = 0; i < 100; i++) { + const key = buildDeliveryKey(destination, 'example-bucket', `example-key-${i}`); + const [resource, index] = key.split('|'); + assert.strictEqual(resource, 'destination1'); + assert(Number.isInteger(Number(index))); + assert(Number(index) >= 0 && Number(index) < 4); + } + }); + + it('should spread the objects of a destination over several keys', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + const keys = new Set(); + for (let i = 0; i < 100; i++) { + keys.add(buildDeliveryKey(destination, 'example-bucket', `example-key-${i}`)); + } + assert(keys.size > 1); + }); + + it('should give two destinations different keys for the same object', () => { + const first = buildDeliveryKey({ resource: 'destination1', spreadFactor: 4 }, + 'example-bucket', 'example-key'); + const second = buildDeliveryKey({ resource: 'destination2', spreadFactor: 4 }, + 'example-bucket', 'example-key'); + assert.notStrictEqual(first, second); + }); +});