diff --git a/bin/notificationDeliveryReplay.js b/bin/notificationDeliveryReplay.js new file mode 100644 index 0000000000..6e73f6178b --- /dev/null +++ b/bin/notificationDeliveryReplay.js @@ -0,0 +1,65 @@ +'use strict'; + +/** + * Cutover replay: drains whatever the old per-destination queue + * processors left undelivered in the internal notification topics into + * the delivery topic, as addressed records. + * + * Run it once, with the queue populator and every notification queue + * processor stopped, before starting the delivery worker pool. It exits + * 0 once every destination is drained, non-zero otherwise. It writes no + * consumer group offset, so a failed run can simply be rerun, and going + * back to the old pipeline stays possible. + * + * Usage: node bin/notificationDeliveryReplay.js [replayId] + * The optional replayId only names the throwaway consumer group. + */ + +const async = require('async'); +const werelogs = require('werelogs'); + +const config = require('../lib/Config'); +const kafkaConfig = config.kafka; +const notifConfig = config.extensions.notification; +const mongoConfig = config.queuePopulator.mongo; +const zkConfig = config.zookeeper; +const DeliveryTopicDrainer = + require('../extensions/notification/deliveryWorker/DeliveryTopicDrainer'); + +const log = new werelogs.Logger('Backbeat:Notification:DeliveryReplay'); + +werelogs.configure({ + level: config.log.logLevel, + dump: config.log.dumpLevel, +}); + +const replayId = process.argv[2] || String(Date.now()); + +const drainer = new DeliveryTopicDrainer({ + kafkaConfig, + mongoConfig, + zkConfig, + notifConfig, + replayId, + logger: log, +}); + +async.waterfall([ + next => drainer.start(err => next(err)), + next => drainer.run(next), +], (err, totals) => { + if (err) { + log.error('notification delivery replay failed', { + method: 'notificationDeliveryReplay', + error: err.message, + }); + return drainer.stop(() => process.exit(1)); + } + log.info('notification delivery replay done', totals); + return drainer.stop(() => process.exit(0)); +}); + +process.on('SIGTERM', () => { + log.info('received SIGTERM, exiting'); + drainer.stop(() => process.exit(1)); +}); diff --git a/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js b/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js new file mode 100644 index 0000000000..4aa6c14905 --- /dev/null +++ b/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js @@ -0,0 +1,834 @@ +'use strict'; + +const async = require('async'); +const { KafkaConsumer } = require('node-rdkafka'); +const Logger = require('werelogs').Logger; +const { errors, jsutil } = require('arsenal'); + +const BackbeatProducer = require('../../../lib/BackbeatProducer'); +const { withTopicPrefix } = require('../../../lib/util/topic'); +const NotificationConfigManager = require('../NotificationConfigManager'); +const configUtil = require('../utils/config'); +const { buildDeliveryKey } = require('../utils/deliveryKey'); + +const CONNECT_TIMEOUT_MS = 30000; +const METADATA_TIMEOUT_MS = 10000; +const WATERMARK_TIMEOUT_MS = 10000; +const COMMITTED_TIMEOUT_MS = 10000; +const CONSUME_BATCH_SIZE = 100; +const EMPTY_BATCH_SLEEP_MS = 500; +const MAX_EMPTY_BATCHES = 120; +const MATCH_CONCURRENCY = 10; + +/** + * @class DeliveryTopicDrainer + * + * @classdesc Drains the internal notification topics of the old + * per-destination pipeline into the delivery topic consumed by the + * delivery worker pool, as addressed records (old payload plus + * destinationId and configurationId). + * + * It is meant to run once during cutover, with the queue populator and + * every queue processor stopped, so that the topics it reads are not + * moving. It reads the committed offsets of the old consumer groups but + * never writes them: a rerun starts from the same place and produces + * duplicates, and a rollback to the old pipeline resumes exactly where + * it stopped. Duplicates are acceptable on the delivery topic, gaps are + * not, so any error (an unreadable bucket configuration, a failed + * produce) aborts the drain instead of stepping over the record. + */ +class DeliveryTopicDrainer { + + /** + * @constructor + * @param {Object} params - params object + * @param {Object} params.kafkaConfig - kafka configuration object + * @param {String} params.kafkaConfig.hosts - list of kafka brokers + * @param {Object} [params.mongoConfig] - mongodb connection config, + * when absent the zookeeper config backend is used + * @param {Object} [params.zkConfig] - zookeeper configuration object + * @param {Object} params.notifConfig - notification configuration + * object, the same "extensions.notification" block the populator + * and the queue processors read + * @param {String} [params.replayId] - suffix of the throwaway + * consumer group, defaults to the process id + * @param {Object} [params.consumer] - node-rdkafka consumer, for tests + * @param {Object} [params.producer] - BackbeatProducer, for tests + * @param {Object} [params.bnConfigManager] - config manager, for tests + * @param {Number} [params.batchSize] - records per consume call + * @param {Number} [params.emptyBatchSleepMs] - pause before rechecking + * partition positions after an empty batch + * @param {Number} [params.maxEmptyBatches] - how many consecutive + * fruitless batches to accept before giving up on a partition + * @param {Number} [params.matchConcurrency] - how many records of a + * batch to match against bucket configurations in parallel + * @param {Object} [params.logger] - werelogs logger + */ + constructor(params) { + const { + kafkaConfig, mongoConfig, zkConfig, notifConfig, replayId, + consumer, producer, bnConfigManager, batchSize, + emptyBatchSleepMs, maxEmptyBatches, matchConcurrency, logger, + } = params; + this.kafkaConfig = kafkaConfig; + this.mongoConfig = mongoConfig; + this.zkConfig = zkConfig; + this.notifConfig = notifConfig; + this.replayId = replayId || String(process.pid); + this.groupId = `bn-replay-${this.replayId}`; + + this.bnConfigManager = bnConfigManager || null; + this._consumer = consumer || null; + this._producer = producer || null; + // injected clients belong to the caller, we neither connect nor + // close them + this._ownsConsumer = !consumer; + this._ownsProducer = !producer; + + this._batchSize = batchSize || CONSUME_BATCH_SIZE; + this._emptyBatchSleepMs = emptyBatchSleepMs === undefined ? + EMPTY_BATCH_SLEEP_MS : emptyBatchSleepMs; + this._maxEmptyBatches = maxEmptyBatches || MAX_EMPTY_BATCHES; + this._matchConcurrency = matchConcurrency || MATCH_CONCURRENCY; + + this.totals = { drained: 0, produced: 0, skipped: 0 }; + this.logger = logger || + new Logger('Backbeat:Notification:DeliveryTopicDrainer'); + } + + /** + * Resolves where a partition has to be drained from, out of the old + * consumer group offset and the partition watermarks. + * + * A group that never committed on the partition, or that committed + * below the low watermark because retention already dropped those + * records, restarts at the low watermark: that is the earliest thing + * we can still deliver. + * + * @param {Object} bounds - partition bounds + * @param {Number} [bounds.committedOffset] - offset committed by the + * old consumer group, negative or missing when there is none + * @param {Number} bounds.lowOffset - low watermark + * @param {Number} bounds.highOffset - high watermark, the offset the + * next record would get + * @return {Object} object with the resolved startOffset, a skip flag + * telling there is nothing to drain, and a reason to log + */ + static resolveStartOffset(bounds) { + const { committedOffset, lowOffset, highOffset } = bounds; + let startOffset = committedOffset; + let reason = 'resuming at old consumer group offset'; + if (typeof startOffset !== 'number' || + !Number.isFinite(startOffset) || startOffset < 0) { + startOffset = lowOffset; + reason = 'old consumer group has no committed offset'; + } else if (startOffset < lowOffset) { + startOffset = lowOffset; + reason = 'old consumer group offset is below the low watermark, ' + + 'records were already dropped by retention'; + } + if (startOffset >= highOffset) { + return { startOffset, skip: true, reason: 'nothing left to drain' }; + } + return { startOffset, skip: false, reason }; + } + + /** + * Initializes the config manager, the delivery topic producer and the + * consumer used to read the old topics + * + * @param {Function} done - callback + * @return {undefined} + */ + start(done) { + async.series([ + next => this._setupNotificationConfigManager(next), + next => this._setupProducer(next), + next => this._setupConsumer(next), + ], err => { + if (err) { + this.logger.error('error starting delivery topic drainer', { + method: 'DeliveryTopicDrainer.start', + error: err.message, + }); + return done(err); + } + return done(); + }); + } + + /** + * Initializes the NotificationConfigManager + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupNotificationConfigManager(done) { + if (this.bnConfigManager) { + return process.nextTick(done); + } + try { + this.bnConfigManager = new NotificationConfigManager({ + mongoConfig: this.mongoConfig, + bucketMetastore: this.notifConfig.bucketMetastore, + maxCachedConfigs: this.notifConfig.maxCachedConfigs, + zkConfig: this.zkConfig, + zkPath: this.notifConfig.zookeeperPath, + zkConcurrency: this.notifConfig.zookeeperOpConcurrency, + logger: this.logger, + }); + return this.bnConfigManager.setup(done); + } catch (err) { + return done(err); + } + } + + /** + * Creates the single producer shared by every destination. It is + * bound to the delivery topic and applies the topic prefix itself. + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupProducer(done) { + if (this._producer) { + return process.nextTick(done); + } + const topic = this._getDeliveryTopic(); + if (!topic) { + return process.nextTick(() => done(errors.InternalError + .customizeDescription( + 'missing extensions.notification.deliveryPool.topic'))); + } + const doneOnce = jsutil.once(done); + this._producer = new BackbeatProducer({ + kafka: { hosts: this.kafkaConfig.hosts }, + topic, + compressionType: this.kafkaConfig.compressionType, + requiredAcks: this.kafkaConfig.requiredAcks, + }); + this._producer.on('error', err => { + this.logger.error('error with delivery topic producer', { + method: 'DeliveryTopicDrainer._setupProducer', + error: err.message, + }); + doneOnce(err); + }); + this._producer.once('ready', () => doneOnce()); + return undefined; + } + + /** + * Creates the consumer that reads the old topics. It uses a throwaway + * group and has every form of offset writing disabled: the old groups + * must be left exactly as the queue processors left them. + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupConsumer(done) { + if (this._consumer) { + return process.nextTick(done); + } + this._consumer = new KafkaConsumer({ + 'metadata.broker.list': this.kafkaConfig.hosts, + 'group.id': this.groupId, + 'enable.auto.commit': false, + 'enable.auto.offset.store': false, + 'allow.auto.create.topics': false, + 'metadata.max.age.ms': 5000, + }, {}); + this._consumer.on('event.error', err => + this.logger.error('rdkafka.error', { err })); + return this._consumer.connect({ timeout: CONNECT_TIMEOUT_MS }, err => { + if (err) { + this.logger.error('error connecting replay consumer', { + method: 'DeliveryTopicDrainer._setupConsumer', + groupId: this.groupId, + error: err.message, + }); + return done(err); + } + return done(); + }); + } + + _getDeliveryTopic() { + return this.notifConfig.deliveryPool && + this.notifConfig.deliveryPool.topic; + } + + /** + * Drains every configured destination, one after the other + * + * @param {Function} done - callback: done(err, totals) + * @return {undefined} + */ + run(done) { + const destinations = this.notifConfig.destinations || []; + this.logger.info('starting notification delivery replay', { + destinations: destinations.map(d => d.resource), + deliveryTopic: this._getDeliveryTopic(), + replayGroupId: this.groupId, + }); + return async.eachSeries(destinations, (destination, next) => + this.drainDestination(destination, next), err => { + if (err) { + this.logger.error('notification delivery replay failed', { + method: 'DeliveryTopicDrainer.run', + error: err.message, + totals: this.totals, + }); + return done(err); + } + this.logger.info('notification delivery replay complete', + this.totals); + return done(null, this.totals); + }); + } + + /** + * Drains the old topic of a single destination + * + * @param {Object} destination - destination config entry + * @param {Function} done - callback + * @return {undefined} + */ + drainDestination(destination, done) { + const groupIdPrefix = this.notifConfig.queueProcessor && + this.notifConfig.queueProcessor.groupId; + if (!groupIdPrefix) { + return done(errors.InternalError.customizeDescription( + 'missing extensions.notification.queueProcessor.groupId')); + } + const internalTopic = destination.internalTopic || + this.notifConfig.topic; + if (!internalTopic) { + return done(errors.InternalError.customizeDescription( + `no internal topic configured for ${destination.resource}`)); + } + // raw rdkafka consumption does not apply the topic prefix, unlike + // BackbeatConsumer, so it is applied here + const oldTopic = withTopicPrefix(internalTopic); + const oldGroup = `${groupIdPrefix}-${destination.resource}`; + return async.waterfall([ + next => this._getPartitions(oldTopic, next), + (partitions, next) => + this._getDrainPlan(oldTopic, oldGroup, partitions, next), + (plan, next) => + this._drainPartitions(destination, oldTopic, oldGroup, plan, + next), + ], done); + } + + /** + * Lists the partitions of an old topic + * + * @param {String} oldTopic - prefixed topic name + * @param {Function} done - callback: done(err, partitions) + * @return {undefined} + */ + _getPartitions(oldTopic, done) { + return this._consumer.getMetadata({ + topic: oldTopic, + timeout: METADATA_TIMEOUT_MS, + }, (err, metadata) => { + if (err) { + this.logger.error('error getting metadata for old topic', { + method: 'DeliveryTopicDrainer._getPartitions', + oldTopic, + errorCode: err, + }); + return done(errors.InternalError); + } + const topicMd = metadata.topics.find(t => t.name === oldTopic); + if (!topicMd || topicMd.partitions.length === 0) { + this.logger.info('old topic has no partitions, nothing to ' + + 'drain', { oldTopic }); + return done(null, []); + } + return done(null, topicMd.partitions.map(p => p.id)); + }); + } + + /** + * Builds the per partition drain plan: where to start, and where the + * head was when the drain started. The head is captured once so that + * the drain has a fixed target and terminates. + * + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Number[]} partitions - partition ids + * @param {Function} done - callback: done(err, plan) + * @return {undefined} + */ + _getDrainPlan(oldTopic, oldGroup, partitions, done) { + if (partitions.length === 0) { + return process.nextTick(() => done(null, [])); + } + return async.waterfall([ + next => this._readCommittedOffsets(oldTopic, oldGroup, partitions, + next), + (committedOffsets, next) => async.mapSeries(partitions, + (partition, partitionDone) => this._planPartition(oldTopic, + partition, committedOffsets[partition], partitionDone), + next), + ], (err, plan) => { + if (err) { + return done(err); + } + plan.forEach(entry => this._logPlanEntry(oldTopic, oldGroup, entry)); + return done(null, plan.filter(entry => !entry.skip)); + }); + } + + /** + * Captures the watermarks of one partition once, and resolves where + * the drain starts from + * + * @param {String} oldTopic - prefixed topic name + * @param {Number} partition - partition id + * @param {Number} [committedOffset] - old consumer group offset + * @param {Function} done - callback: done(err, planEntry) + * @return {undefined} + */ + _planPartition(oldTopic, partition, committedOffset, done) { + return this._consumer.queryWatermarkOffsets(oldTopic, partition, + WATERMARK_TIMEOUT_MS, (err, offsets) => { + if (err) { + this.logger.error('error getting watermark offsets', { + method: 'DeliveryTopicDrainer._planPartition', + oldTopic, + partition, + errorCode: err, + }); + return done(errors.InternalError); + } + const { lowOffset, highOffset } = offsets; + const resolved = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset, + lowOffset, + highOffset, + }); + return done(null, Object.assign({ + partition, + committedOffset, + lowOffset, + headOffset: highOffset, + }, resolved)); + }); + } + + _logPlanEntry(oldTopic, oldGroup, entry) { + const info = { + oldTopic, + oldGroup, + partition: entry.partition, + committedOffset: entry.committedOffset, + lowOffset: entry.lowOffset, + headOffset: entry.headOffset, + startOffset: entry.startOffset, + reason: entry.reason, + }; + if (typeof entry.committedOffset === 'number' && + entry.committedOffset >= 0 && + entry.committedOffset < entry.lowOffset) { + this.logger.warn('records were lost before the replay', info); + } else { + this.logger.info('resolved replay start offset', info); + } + } + + /** + * Reads the offsets committed by the old consumer group. + * + * In node-rdkafka 2.18 committed() is scoped to the group.id of the + * client it is called on, so a client configured with the old group id + * is the only way to see those offsets. That client only ever calls + * committed(): it never subscribes, assigns or commits, so the old + * group keeps its offsets and a rollback stays possible. + * + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Number[]} partitions - partition ids + * @param {Function} done - callback: done(err, offsetsByPartition) + * @return {undefined} + */ + _readCommittedOffsets(oldTopic, oldGroup, partitions, done) { + const toppars = partitions.map(partition => ({ + topic: oldTopic, + partition, + })); + return this._withOffsetReader(oldGroup, (reader, next) => + reader.committed(toppars, COMMITTED_TIMEOUT_MS, next), + (err, committedToppars) => { + if (err) { + this.logger.error('error reading committed offsets of ' + + 'the old consumer group', { + method: 'DeliveryTopicDrainer._readCommittedOffsets', + oldTopic, + oldGroup, + error: err.message, + }); + return done(errors.InternalError.customizeDescription( + err.message)); + } + const offsets = {}; + (committedToppars || []).forEach(tp => { + offsets[tp.partition] = tp.offset; + }); + return done(null, offsets); + }); + } + + /** + * Runs fn against a consumer bound to the given group id. When a + * consumer was injected it is reused as is, otherwise a short lived + * one is connected and disconnected around the call. + * + * @param {String} groupId - consumer group id to read offsets of + * @param {Function} fn - fn(reader, cb) + * @param {Function} done - callback: done(err, result) + * @return {undefined} + */ + _withOffsetReader(groupId, fn, done) { + if (!this._ownsConsumer) { + return fn(this._consumer, done); + } + const reader = new KafkaConsumer({ + 'metadata.broker.list': this.kafkaConfig.hosts, + 'group.id': groupId, + 'enable.auto.commit': false, + 'enable.auto.offset.store': false, + }, {}); + return reader.connect({ timeout: CONNECT_TIMEOUT_MS }, connectErr => { + if (connectErr) { + return done(connectErr); + } + return fn(reader, (err, result) => + reader.disconnect(() => done(err, result))); + }); + } + + /** + * Assigns the partitions to drain at their start offsets and consumes + * them until each one reaches the head captured in the plan + * + * @param {Object} destination - destination config entry + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Object[]} plan - per partition drain plan + * @param {Function} done - callback + * @return {undefined} + */ + _drainPartitions(destination, oldTopic, oldGroup, plan, done) { + if (plan.length === 0) { + this.logger.info('nothing to drain for destination', { + destination: destination.resource, + oldTopic, + oldGroup, + }); + return process.nextTick(done); + } + const state = new Map(); + plan.forEach(p => state.set(p.partition, { + partition: p.partition, + startOffset: p.startOffset, + headOffset: p.headOffset, + lastOffset: -1, + position: -1, + drained: 0, + produced: 0, + skipped: 0, + })); + this._consumer.assign(plan.map(p => ({ + topic: oldTopic, + partition: p.partition, + offset: p.startOffset, + }))); + let emptyBatches = 0; + const loop = () => { + const pending = this._pendingPartitions(state); + if (pending.length === 0) { + return this._finishDestination(destination, oldTopic, oldGroup, + state, done); + } + if (emptyBatches >= this._maxEmptyBatches) { + this.logger.error('gave up waiting for partitions to reach ' + + 'their head offset', { + method: 'DeliveryTopicDrainer._drainPartitions', + oldTopic, + oldGroup, + pending: pending.map(s => ({ + partition: s.partition, + lastOffset: s.lastOffset, + headOffset: s.headOffset, + })), + }); + return done(errors.InternalError.customizeDescription( + `replay stalled on topic ${oldTopic}`)); + } + return this._drainBatch(destination, oldTopic, state, + (err, consumed) => { + if (err) { + return done(err); + } + emptyBatches = consumed > 0 ? 0 : emptyBatches + 1; + return setImmediate(loop); + }); + }; + return loop(); + } + + _pendingPartitions(state) { + return [...state.values()].filter(s => { + if (s.lastOffset >= s.headOffset - 1) { + return false; + } + // position() is the offset the consumer would read next, it + // moves past records that consume() never returns + return !(s.position >= 0 && s.position >= s.headOffset); + }); + } + + _finishDestination(destination, oldTopic, oldGroup, state, done) { + // destinations sharing one internal topic are drained from + // different offsets, so drop the assignment and its fetch queue + // before the next one assigns the same partitions + this._consumer.unassign(); + state.forEach(s => { + this.logger.info('drained partition', { + oldTopic, + oldGroup, + partition: s.partition, + startOffset: s.startOffset, + headOffset: s.headOffset, + drained: s.drained, + produced: s.produced, + skipped: s.skipped, + }); + }); + this.logger.info('drained destination', { + destination: destination.resource, + oldTopic, + oldGroup, + partitions: state.size, + }); + return done(); + } + + /** + * Consumes one batch and produces every matching record of it to the + * delivery topic. The next batch is only consumed once the delivery + * reports of this one are in, which bounds memory and keeps the + * counters honest. + * + * @param {Object} destination - destination config entry + * @param {String} oldTopic - prefixed topic name + * @param {Map} state - per partition drain state + * @param {Function} done - callback: done(err, consumedCount) + * @return {undefined} + */ + _drainBatch(destination, oldTopic, state, done) { + return this._consumer.consume(this._batchSize, (err, records) => { + if (err) { + this.logger.error('error consuming from old topic', { + method: 'DeliveryTopicDrainer._drainBatch', + oldTopic, + errorCode: err, + }); + return done(errors.InternalError); + } + if (!records || records.length === 0) { + return this._refreshPositions(oldTopic, state, done); + } + return this._processBatch(destination, records, state, batchErr => { + if (batchErr) { + return done(batchErr); + } + return done(null, records.length); + }); + }); + } + + /** + * Waits a little then rereads the consumer positions, so that a + * partition whose remaining offsets yield no record still gets to the + * head instead of looping forever + * + * @param {String} oldTopic - prefixed topic name + * @param {Map} state - per partition drain state + * @param {Function} done - callback: done(err, consumedCount) + * @return {undefined} + */ + _refreshPositions(oldTopic, state, done) { + return setTimeout(() => { + let positions; + try { + positions = this._consumer.position() || []; + } catch (err) { + this.logger.error('error reading consumer positions', { + method: 'DeliveryTopicDrainer._refreshPositions', + oldTopic, + error: err.message, + }); + return done(errors.InternalError.customizeDescription( + err.message)); + } + positions.filter(p => p.topic === oldTopic).forEach(p => { + const partitionState = state.get(p.partition); + if (partitionState && typeof p.offset === 'number' && + p.offset >= 0) { + partitionState.position = p.offset; + } + }); + return done(null, 0); + }, this._emptyBatchSleepMs); + } + + _processBatch(destination, records, state, done) { + return async.mapLimit(records, this._matchConcurrency, + (record, next) => this._matchRecord(destination, record, next), + (err, messages) => { + if (err) { + return done(err); + } + const toProduce = []; + records.forEach((record, i) => { + const partitionState = state.get(record.partition); + if (partitionState) { + partitionState.drained++; + if (record.offset > partitionState.lastOffset) { + partitionState.lastOffset = record.offset; + } + } + this.totals.drained++; + if (messages[i]) { + toProduce.push(messages[i]); + this.totals.produced++; + if (partitionState) { + partitionState.produced++; + } + } else { + this.totals.skipped++; + if (partitionState) { + partitionState.skipped++; + } + } + }); + if (toProduce.length === 0) { + return process.nextTick(done); + } + return this._producer.send(toProduce, sendErr => { + if (sendErr) { + this.logger.error('error producing to delivery topic', { + method: 'DeliveryTopicDrainer._processBatch', + destination: destination.resource, + error: sendErr.message, + }); + return done(sendErr); + } + return done(); + }); + }); + } + + /** + * Rematches one old record against the bucket notification + * configuration, the way the queue processor of this destination + * would have, and turns it into an addressed delivery record. + * + * Old records carry no configurationId, it is only known once the + * matching rule is found again here. + * + * @param {Object} destination - destination config entry + * @param {Object} record - kafka record from the old topic + * @param {Function} done - callback: done(err, message), message being + * null when the record has nothing to deliver for this destination + * @return {undefined} + */ + _matchRecord(destination, record, done) { + let entry; + try { + entry = JSON.parse(record.value); + } catch (err) { + this.logger.error('error parsing JSON entry, skipping record', { + method: 'DeliveryTopicDrainer._matchRecord', + partition: record.partition, + offset: record.offset, + error: err.message, + }); + return process.nextTick(() => done(null, null)); + } + const { bucket, key } = entry; + return this.bnConfigManager.getConfig(bucket, (err, bnConfig) => { + if (err) { + // a record we cannot match is a record we cannot decide + // about, and skipping it would be a gap, so give up and + // let the operator rerun the replay + this.logger.error('error getting notification configuration', { + method: 'DeliveryTopicDrainer._matchRecord', + bucket, + key, + error: err.message, + }); + return done(err); + } + if (!bnConfig || Object.keys(bnConfig).length === 0 || + !bnConfig.notificationConfiguration) { + return done(null, null); + } + const queueConfig = + bnConfig.notificationConfiguration.queueConfig.filter( + c => c.queueArn.split(':').pop() === destination.resource); + if (!queueConfig.length) { + return done(null, null); + } + const destConfig = { + bucket, + notificationConfiguration: { queueConfig }, + }; + const { isValid, matchingConfig } = + configUtil.validateEntry(destConfig, entry); + if (!isValid) { + return done(null, null); + } + entry.destinationId = destination.resource; + entry.configurationId = matchingConfig.id; + return done(null, { + // the populator's publish() path url-encodes record keys, + // producing here goes straight through BackbeatProducer so + // the same encoding has to be applied, otherwise records of + // one object would land on two different partitions + key: encodeURIComponent( + buildDeliveryKey(destination, bucket, key)), + message: JSON.stringify(entry), + }); + }); + } + + /** + * Disconnects the clients this drainer created + * + * @param {Function} done - callback + * @return {undefined} + */ + stop(done) { + return async.series([ + next => { + if (!this._consumer || !this._ownsConsumer) { + return next(); + } + return this._consumer.disconnect(() => next()); + }, + next => { + if (!this._producer || !this._ownsProducer) { + return next(); + } + return this._producer.close(() => next()); + }, + ], () => done()); + } +} + +module.exports = DeliveryTopicDrainer; 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/unit/notification/DeliveryTopicDrainer.spec.js b/tests/unit/notification/DeliveryTopicDrainer.spec.js new file mode 100644 index 0000000000..6525f098b9 --- /dev/null +++ b/tests/unit/notification/DeliveryTopicDrainer.spec.js @@ -0,0 +1,638 @@ +const assert = require('assert'); +const sinon = require('sinon'); + +const DeliveryTopicDrainer = require( + '../../../extensions/notification/deliveryWorker/DeliveryTopicDrainer'); +const { buildDeliveryKey } = require( + '../../../extensions/notification/utils/deliveryKey'); + +const OLD_TOPIC = 'bucket-notification'; +const DESTINATION = { resource: 'destId', type: 'kafka', host: 'external-host' }; + +const notifConfig = { + topic: OLD_TOPIC, + queueProcessor: { + groupId: 'backbeat-bucket-notification-group', + }, + deliveryPool: { + topic: 'bucket-notification-delivery', + }, + destinations: [DESTINATION], +}; + +const bucketConfig = { + bucket: 'mybucket', + notificationConfiguration: { + queueConfig: [ + { + id: 'config-1', + queueArn: 'arn:scality:bucketnotif:::destId', + events: ['s3:ObjectCreated:*'], + }, + ], + }, +}; + +function makeEntry(overrides) { + return Object.assign({ + bucket: 'mybucket', + key: 'obj1', + eventType: 's3:ObjectCreated:Put', + value: '{}', + }, overrides); +} + +function makeRecord(topic, partition, offset, entry) { + return { + topic, + partition, + offset, + key: Buffer.from(`${entry.bucket}/${entry.key}`), + value: Buffer.from(JSON.stringify(entry)), + }; +} + +/** + * Stubs the bits of a node-rdkafka consumer the drainer uses. Offset + * writing methods are spies so that tests can assert the drainer never + * touches the offsets of the old consumer groups. + * + * @param {Object} params - stub params + * @param {Object} params.watermarks - partition id to { lowOffset, highOffset } + * @param {Object} params.committed - partition id to committed offset + * @param {Array[]} params.batches - batches returned by consume(), in order + * @param {Object[]} [params.positions] - what position() returns + * @return {Object} consumer stub + */ +function makeConsumer(params) { + const { watermarks, committed, batches, positions } = params; + const consumer = { + consumeCalls: 0, + unassignCalls: 0, + assigned: null, + committedArgs: null, + onConsume: null, + commit: sinon.spy(), + commitSync: sinon.spy(), + commitMessage: sinon.spy(), + subscribe: sinon.spy(), + getMetadata: (opts, cb) => process.nextTick(() => cb(null, { + topics: [{ + name: opts.topic, + partitions: Object.keys(watermarks) + .map(id => ({ id: Number(id) })), + }], + })), + committed: (toppars, timeout, cb) => { + consumer.committedArgs = { toppars, timeout }; + return process.nextTick(() => cb(null, toppars.map(tp => ({ + topic: tp.topic, + partition: tp.partition, + offset: committed[tp.partition], + })))); + }, + queryWatermarkOffsets: (topic, partition, timeout, cb) => + process.nextTick(() => cb(null, watermarks[partition])), + assign: assignments => { + consumer.assigned = assignments; + }, + unassign: () => { + consumer.unassignCalls++; + }, + consume: (count, cb) => { + consumer.consumeCalls++; + if (consumer.onConsume) { + consumer.onConsume(); + } + const batch = batches.length > 0 ? batches.shift() : []; + return process.nextTick(() => cb(null, batch)); + }, + position: () => positions || [], + }; + return consumer; +} + +function makeProducer(sent, onSend) { + return { + send: (entries, cb) => { + sent.push(...entries); + if (onSend) { + return onSend(entries, cb); + } + return process.nextTick(cb); + }, + }; +} + +function makeDrainer(params) { + return new DeliveryTopicDrainer(Object.assign({ + kafkaConfig: { hosts: 'localhost:9092' }, + notifConfig, + emptyBatchSleepMs: 0, + batchSize: 10, + }, params)); +} + +describe('notification DeliveryTopicDrainer', () => { + describe('resolveStartOffset', () => { + it('should start at the low watermark when the old group never ' + + 'committed', () => { + // -1001 is what librdkafka reports for an unset offset + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: -1001, + lowOffset: 4, + highOffset: 20, + }); + assert.strictEqual(res.startOffset, 4); + assert.strictEqual(res.skip, false); + }); + + it('should start at the low watermark when there is no committed ' + + 'offset at all', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: undefined, + lowOffset: 0, + highOffset: 3, + }); + assert.strictEqual(res.startOffset, 0); + assert.strictEqual(res.skip, false); + }); + + it('should start at the low watermark when retention dropped the ' + + 'committed offset', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 3, + lowOffset: 10, + highOffset: 20, + }); + assert.strictEqual(res.startOffset, 10); + assert.strictEqual(res.skip, false); + }); + + it('should resume at the committed offset', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 7, + lowOffset: 0, + highOffset: 12, + }); + assert.strictEqual(res.startOffset, 7); + assert.strictEqual(res.skip, false); + }); + + it('should skip a partition the old group fully consumed', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 5, + lowOffset: 0, + highOffset: 5, + }); + assert.strictEqual(res.skip, true); + }); + + it('should skip an empty partition', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: -1001, + lowOffset: 0, + highOffset: 0, + }); + assert.strictEqual(res.skip, true); + }); + }); + + describe('drainDestination', () => { + it('should read the committed offsets of the old destination group ' + + 'without ever writing them', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + const readOffsets = sinon.spy(drainer, '_readCommittedOffsets'); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert(readOffsets.calledOnce); + assert.strictEqual(readOffsets.args[0][0], OLD_TOPIC); + assert.strictEqual(readOffsets.args[0][1], + 'backbeat-bucket-notification-group-destId'); + assert.deepStrictEqual(consumer.committedArgs.toppars, + [{ topic: OLD_TOPIC, partition: 0 }]); + assert(consumer.commit.notCalled); + assert(consumer.commitSync.notCalled); + assert(consumer.commitMessage.notCalled); + assert(consumer.subscribe.notCalled); + done(); + }); + }); + + it('should assign the partitions left to drain at their start ' + + 'offsets', done => { + const consumer = makeConsumer({ + watermarks: { + 0: { lowOffset: 0, highOffset: 3 }, + 1: { lowOffset: 8, highOffset: 20 }, + 2: { lowOffset: 0, highOffset: 5 }, + }, + // partition 1 committed below the low watermark, partition 2 + // was fully consumed by the old queue processor + committed: { 0: 1, 1: 2, 2: 5 }, + batches: [[ + makeRecord(OLD_TOPIC, 0, 2, makeEntry()), + makeRecord(OLD_TOPIC, 1, 19, makeEntry()), + ]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.deepStrictEqual(consumer.assigned, [ + { topic: OLD_TOPIC, partition: 0, offset: 1 }, + { topic: OLD_TOPIC, partition: 1, offset: 8 }, + ]); + done(); + }); + }); + + it('should address matching records and key them with the encoded ' + + 'delivery key', done => { + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 1); + assert.strictEqual(sent[0].key, encodeURIComponent( + buildDeliveryKey(DESTINATION, 'mybucket', 'obj1'))); + assert.strictEqual(sent[0].key, 'destId'); + const message = JSON.parse(sent[0].message); + assert.strictEqual(message.destinationId, 'destId'); + assert.strictEqual(message.configurationId, 'config-1'); + // the old payload is carried over untouched + assert.strictEqual(message.bucket, 'mybucket'); + assert.strictEqual(message.key, 'obj1'); + assert.strictEqual(message.eventType, 's3:ObjectCreated:Put'); + assert.deepStrictEqual(drainer.totals, + { drained: 1, produced: 1, skipped: 0 }); + done(); + }); + }); + + it('should url-encode a spread delivery key', done => { + const spreadDestination = Object.assign({ spreadFactor: 4 }, + DESTINATION); + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(spreadDestination, err => { + assert.ifError(err); + const rawKey = buildDeliveryKey(spreadDestination, 'mybucket', + 'obj1'); + assert(rawKey.includes('|')); + assert.strictEqual(sent[0].key, encodeURIComponent(rawKey)); + assert(sent[0].key.includes('%7C')); + done(); + }); + }); + + it('should skip records that no longer match a configuration', done => { + const sent = []; + const entries = [ + // no configuration at all for this bucket + makeEntry({ bucket: 'noconfig' }), + // configuration targets another destination + makeEntry({ bucket: 'otherdest' }), + // event type is not subscribed to + makeEntry({ eventType: 's3:ObjectRemoved:Delete' }), + // deletion placeholder, no event type + makeEntry({ eventType: undefined }), + makeEntry({ key: 'obj5' }), + ]; + const configs = { + mybucket: bucketConfig, + noconfig: undefined, + otherdest: { + bucket: 'otherdest', + notificationConfiguration: { + queueConfig: [{ + id: 'other', + queueArn: 'arn:scality:bucketnotif:::otherDestId', + events: ['s3:ObjectCreated:*'], + }], + }, + }, + }; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 5 } }, + committed: { 0: 0 }, + batches: [entries.map((entry, i) => + makeRecord(OLD_TOPIC, 0, i, entry))], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { + getConfig: (bucket, cb) => cb(null, configs[bucket]), + }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 1); + assert.strictEqual(JSON.parse(sent[0].message).key, 'obj5'); + assert.deepStrictEqual(drainer.totals, + { drained: 5, produced: 1, skipped: 4 }); + done(); + }); + }); + + it('should skip a record whose payload is not valid JSON', done => { + const sent = []; + const record = makeRecord(OLD_TOPIC, 0, 0, makeEntry()); + record.value = Buffer.from('{not json'); + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[record]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 0); + assert.strictEqual(drainer.totals.skipped, 1); + done(); + }); + }); + + it('should consume the next batch only once the previous one is ' + + 'acked', done => { + const events = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 3 } }, + committed: { 0: 0 }, + batches: [ + [ + makeRecord(OLD_TOPIC, 0, 0, makeEntry()), + makeRecord(OLD_TOPIC, 0, 1, makeEntry()), + ], + [makeRecord(OLD_TOPIC, 0, 2, makeEntry())], + ], + }); + consumer.onConsume = () => events.push('consume'); + const producer = makeProducer([], (entries, cb) => { + events.push('send'); + // a delivery report is not immediate, the drainer has to + // wait for it before pulling more records + return setTimeout(() => { + events.push('ack'); + cb(); + }, 10); + }); + const drainer = makeDrainer({ + consumer, + producer, + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.deepStrictEqual(events, [ + 'consume', 'send', 'ack', + 'consume', 'send', 'ack', + ]); + done(); + }); + }); + + it('should stop at the head offset captured when the drain started', + done => { + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 2 } }, + committed: { 0: 0 }, + batches: [ + [ + makeRecord(OLD_TOPIC, 0, 0, makeEntry()), + makeRecord(OLD_TOPIC, 0, 1, makeEntry()), + ], + // would be returned if the drainer kept consuming + [makeRecord(OLD_TOPIC, 0, 2, makeEntry())], + ], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(consumer.consumeCalls, 1); + assert.strictEqual(sent.length, 2); + done(); + }); + }); + + it('should finish a partition whose remaining offsets yield no ' + + 'record', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 4 } }, + committed: { 0: 2 }, + batches: [], + positions: [{ topic: OLD_TOPIC, partition: 0, offset: 4 }], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(drainer.totals.drained, 0); + done(); + }); + }); + + it('should give up when a partition never reaches its head offset', + done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 4 } }, + committed: { 0: 0 }, + batches: [], + positions: [{ topic: OLD_TOPIC, partition: 0, offset: 1 }], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + maxEmptyBatches: 3, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(consumer.consumeCalls, 3); + done(); + }); + }); + + it('should abort when a bucket configuration cannot be read', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { + getConfig: (b, cb) => cb(new Error('mongo is down')), + }, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(err.message, 'mongo is down'); + done(); + }); + }); + + it('should abort when producing to the delivery topic fails', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const producer = makeProducer([], (entries, cb) => + process.nextTick(() => cb(new Error('delivery report error')))); + const drainer = makeDrainer({ + consumer, + producer, + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(err.message, 'delivery report error'); + done(); + }); + }); + + it('should read the old topic through its prefixed name', done => { + process.env.KAFKA_TOPIC_PREFIX = 'pfx-'; + const prefixedTopic = `pfx-${OLD_TOPIC}`; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(prefixedTopic, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + delete process.env.KAFKA_TOPIC_PREFIX; + assert.ifError(err); + assert.strictEqual(consumer.committedArgs.toppars[0].topic, + prefixedTopic); + assert.strictEqual(consumer.assigned[0].topic, prefixedTopic); + done(); + }); + }); + + it('should use the destination internal topic when it has one', + done => { + const destination = Object.assign({ internalTopic: 'dest-topic' }, + DESTINATION); + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord('dest-topic', 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(destination, err => { + assert.ifError(err); + assert.strictEqual(consumer.assigned[0].topic, 'dest-topic'); + done(); + }); + }); + + it('should not assign anything when the old topic does not exist', + done => { + const consumer = makeConsumer({ + watermarks: {}, + committed: {}, + batches: [], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(consumer.assigned, null); + assert.strictEqual(consumer.consumeCalls, 0); + done(); + }); + }); + }); + + describe('run', () => { + it('should drain every destination and report the totals', done => { + const destinations = [ + { resource: 'destId', type: 'kafka', host: 'h' }, + { resource: 'otherDestId', type: 'kafka', host: 'h', + internalTopic: 'other-topic' }, + ]; + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [ + [makeRecord(OLD_TOPIC, 0, 0, makeEntry())], + [makeRecord('other-topic', 0, 0, makeEntry())], + ], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + notifConfig: Object.assign({}, notifConfig, { destinations }), + }); + drainer.run((err, totals) => { + assert.ifError(err); + // the second destination is not in the bucket configuration + assert.deepStrictEqual(totals, + { drained: 2, produced: 1, skipped: 1 }); + assert.strictEqual(sent.length, 1); + // each destination drops its assignment when it is done + assert.strictEqual(consumer.unassignCalls, 2); + done(); + }); + }); + }); +});