-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Bennett Buchanan
committed
May 26, 2018
1 parent
92eb9ac
commit 7dfdd46
Showing
6 changed files
with
235 additions
and
52 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| 'use strict'; // eslint-disable-line strict | ||
|
|
||
| const Logger = require('werelogs').Logger; | ||
| const redisClient = require('../../replication/utils/getRedisClient')(); | ||
|
|
||
| const FailedCRRProducer = require('./FailedCRRProducer'); | ||
| const BackbeatConsumer = require('../../../lib/BackbeatConsumer'); | ||
| const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); | ||
| const redisKeys = require('../constants').redisKeys; | ||
| const config = require('../../../conf/Config'); | ||
|
|
||
| // BackbeatConsumer constant defaults | ||
| const CONSUMER_FETCH_MAX_BYTES = 5000020; | ||
| const CONCURRENCY = 10; | ||
|
|
||
| class FailedCRRConsumer { | ||
| /** | ||
| * Create the retry consumer. | ||
| */ | ||
| constructor() { | ||
| this._kafkaConfig = config.kafka; | ||
| this._topic = config.extensions.replication.replicationFailedTopic; | ||
| this.logger = new Logger('Backbeat:FailedCRRConsumer'); | ||
| this._failedCRRProducer = new FailedCRRProducer(this.kafkaConfig); | ||
| this._backbeatTask = new BackbeatTask(); | ||
| } | ||
|
|
||
| /** | ||
| * Start the retry consumer by subscribing to the retry kafka topic. Setup | ||
| * the failed CRR producer for pushing any failed redis operations back to | ||
| * the queue. | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| start(cb) { | ||
| const consumer = new BackbeatConsumer({ | ||
| kafka: { hosts: this._kafkaConfig.hosts }, | ||
| topic: this._topic, | ||
| groupId: 'backbeat-retry-group', | ||
| concurrency: CONCURRENCY, | ||
| queueProcessor: this.processKafkaEntry.bind(this), | ||
| fetchMaxBytes: CONSUMER_FETCH_MAX_BYTES, | ||
| }); | ||
| consumer.on('error', () => {}); | ||
| consumer.on('ready', () => { | ||
| consumer.subscribe(); | ||
| this.logger.info('retry consumer is ready to consume entries'); | ||
| }); | ||
| return this._failedCRRProducer.setupProducer(err => { | ||
| if (err) { | ||
| this.logger.error('could not setup producer', { | ||
| method: 'FailedCRRConsumer.processKafkaEntry', | ||
| error: err, | ||
| }); | ||
| return cb(err); | ||
| } | ||
| return cb(); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Process an entry from the retry topic, and set the data in a Redis hash. | ||
| * @param {Object} kafkaEntry - The entry from the retry topic | ||
| * @param {function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| processKafkaEntry(kafkaEntry, cb) { | ||
| const log = this.logger.newRequestLogger(); | ||
| let data; | ||
| try { | ||
| data = JSON.parse(kafkaEntry.value); | ||
| } catch (err) { | ||
| log.error('error processing retry entry', { | ||
| method: 'FailedCRRConsumer.processKafkaEntry', | ||
| error: err, | ||
| }); | ||
| log.end(); | ||
| return cb(); | ||
| } | ||
| return this._setRedisHash(data, kafkaEntry, log, cb); | ||
| } | ||
|
|
||
| /** | ||
| * Attempt to set the Redis hash, using an exponential backoff should the | ||
| * key set fail. If the backoff time is exceeded, push the entry back into | ||
| * the retry entry topic for a later reattempt. | ||
| * @param {Object} data - The field and value for the Redis hash | ||
| * @param {Object} kafkaEntry - The entry from the retry topic | ||
| * @param {Werelogs} log - The werelogs logger | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| _setRedisHash(data, kafkaEntry, log, cb) { | ||
| this._backbeatTask.retry({ | ||
| actionDesc: 'set redis key', | ||
| logFields: {}, | ||
| actionFunc: done => this._setRedisHashOnce(data, log, done), | ||
| shouldRetryFunc: err => err.retryable, | ||
| log, | ||
| }, err => { | ||
| if (err && err.retryable === true) { | ||
| log.info('publishing entry back into the kafka queue'); | ||
| const entry = Buffer.from(kafkaEntry.value).toString(); | ||
| return this._failedCRRProducer.publishFailedCRREntry(entry, cb); | ||
| } | ||
| log.info('successfully set redis key'); | ||
| return cb(); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Attempt to set the Redis hash. | ||
| * @param {Object} data - The field and value for the Redis hash | ||
| * @param {Werelogs} log - The werelogs logger | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| _setRedisHashOnce(data, log, cb) { | ||
| const cmds = ['hmset', redisKeys.failedCRR, [data.field, data.value]]; | ||
| return redisClient.batch([cmds], (err, res) => { | ||
| if (err) { | ||
| return cb({ retryable: true }); | ||
| } | ||
| const [cmdErr] = res[0]; | ||
| return cb({ retryable: cmdErr !== null }); | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| module.exports = FailedCRRConsumer; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| 'use strict'; // eslint-disable-line strict | ||
|
|
||
| const { Logger } = require('werelogs'); | ||
|
|
||
| const BackbeatProducer = require('../../../lib/BackbeatProducer'); | ||
| const config = require('../../../conf/Config'); | ||
|
|
||
| class FailedCRRProducer { | ||
| /** | ||
| * Create the retry producer. | ||
| */ | ||
| constructor() { | ||
| this._kafkaConfig = config.kafka; | ||
| this._topic = config.extensions.replication.replicationFailedTopic; | ||
| this._producer = null; | ||
| this._log = new Logger('Backbeat:FailedCRRProducer'); | ||
| } | ||
|
|
||
| /** | ||
| * Set up the retry producer. | ||
| * @param {function} [cb] - Optional callback called when startup | ||
| * is complete | ||
| * @return {undefined} | ||
| */ | ||
| setupProducer(cb) { | ||
| const producer = new BackbeatProducer({ | ||
| kafka: { hosts: this._kafkaConfig.hosts }, | ||
| topic: this._topic, | ||
| }); | ||
| producer.once('error', () => {}); | ||
| producer.once('ready', () => { | ||
| producer.removeAllListeners('error'); | ||
| producer.on('error', err => | ||
| this._log.error('error from backbeat producer', { | ||
| error: err, | ||
| })); | ||
| this._producer = producer; | ||
| if (cb) { | ||
| return cb(); | ||
| } | ||
| return undefined; | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Publish the given message to the retry Kafka topic. | ||
| * @param {String} message - The message to publish | ||
| * @param {Function} cb - The callback function | ||
| * @return {undefined} | ||
| */ | ||
| publishFailedCRREntry(message, cb) { | ||
| this._producer.send([{ message }], err => { | ||
| if (err) { | ||
| this._log.trace('error publishing retry entry'); | ||
| } | ||
| return cb(); | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| module.exports = FailedCRRProducer; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters