diff --git a/extensions/notification/deliveryWorker/DeliveryKafkaProducer.js b/extensions/notification/deliveryWorker/DeliveryKafkaProducer.js new file mode 100644 index 000000000..529296ca4 --- /dev/null +++ b/extensions/notification/deliveryWorker/DeliveryKafkaProducer.js @@ -0,0 +1,43 @@ +const joi = require('joi'); + +const KafkaProducer = require('../destination/KafkaProducer'); + +/** + * Producer used by the delivery worker to publish notifications to an + * external destination. + * + * It behaves like the notification KafkaProducer, with an added bound on + * how long librdkafka keeps retrying a message before it expires. The + * delivery worker holds the consumer offset until the delivery report is + * received, so an unbounded retry would block the offset forever. + */ +class DeliveryKafkaProducer extends KafkaProducer { + + getConfigJoi() { + return super.getConfigJoi() + .append({ deliveryTimeoutMs: joi.number() }); + } + + getClientId() { + return 'NotificationDeliveryProducer'; + } + + setFromConfig(joiResult) { + super.setFromConfig(joiResult); + this._deliveryTimeoutMs = joiResult.deliveryTimeoutMs; + } + + get topicConfig() { + const base = super.topicConfig; + if (this._deliveryTimeoutMs === undefined) { + return base; + } + return { + ...base, + 'message.timeout.ms': this._deliveryTimeoutMs, + }; + } + +} + +module.exports = DeliveryKafkaProducer; diff --git a/extensions/notification/deliveryWorker/DeliveryProducerPool.js b/extensions/notification/deliveryWorker/DeliveryProducerPool.js new file mode 100644 index 000000000..c2cb9f077 --- /dev/null +++ b/extensions/notification/deliveryWorker/DeliveryProducerPool.js @@ -0,0 +1,376 @@ +const async = require('async'); +const { ZenkoMetrics } = require('arsenal').metrics; +const errors = require('arsenal').errors; + +const DeliveryKafkaProducer = require('./DeliveryKafkaProducer'); + +const lanesGauge = ZenkoMetrics.createGauge({ + name: 's3_notification_delivery_worker_lanes', + help: 'Number of notification deliveries currently in flight', +}); + +const producersGauge = ZenkoMetrics.createGauge({ + name: 's3_notification_delivery_worker_producers', + help: 'Number of open producers per external destination endpoint', + labelNames: ['endpoint'], +}); + +/** + * A producer to one external destination, along with the bookkeeping the + * pool needs to decide when it can be closed. + */ +class PooledProducer { + /** + * @constructor + * @param {Object} params - constructor params + * @param {string} params.destinationId - destination id (resource name) + * @param {string} params.endpoint - kafka hosts string of the destination + * @param {function} params.onSendStarted - called when a delivery starts + * @param {function} params.onSendFinished - called when a delivery ends + * @param {Logger} params.logger - logger object + */ + constructor(params) { + this.destinationId = params.destinationId; + this.endpoint = params.endpoint; + this.producer = null; + this.ready = false; + this.lastUsed = Date.now(); + this.inFlight = 0; + this.waiters = []; + this._onSendStarted = params.onSendStarted; + this._onSendFinished = params.onSendFinished; + this._log = params.logger; + } + + attach(producer) { + this.producer = producer; + } + + markReady() { + this.ready = true; + this.lastUsed = Date.now(); + } + + addWaiter(cb) { + this.waiters.push(cb); + } + + /** + * Hand this producer, or the error that prevented it from connecting, to + * everyone that asked for it while it was connecting + * + * @param {Error} [err] - error that prevented the producer from connecting + * @return {undefined} + */ + flushWaiters(err) { + const waiters = this.waiters; + this.waiters = []; + waiters.forEach(cb => (err ? cb(err) : cb(null, this))); + } + + /** + * True when this producer is holding no delivery, so closing it cannot + * lose a delivery report that a consumer offset is waiting on + * + * @return {boolean} whether the producer can be closed + */ + isIdle() { + return this.ready && this.inFlight === 0; + } + + /** + * Send messages and keep track of the delivery being in flight + * + * @param {Object[]} messages - messages to send + * @param {function} cb - callback called on the delivery report + * @return {undefined} + */ + send(messages, cb) { + this.inFlight++; + this.lastUsed = Date.now(); + this._onSendStarted(); + this.producer.send(messages, err => { + this.inFlight--; + this.lastUsed = Date.now(); + this._onSendFinished(); + cb(err); + }); + } + + close(cb) { + const done = cb || (() => {}); + if (!this.producer) { + return process.nextTick(done); + } + return this.producer.close(err => { + if (err) { + this._log.error('error closing producer', { + method: 'PooledProducer.close', + destinationId: this.destinationId, + endpoint: this.endpoint, + error: err.message, + }); + } + done(); + }); + } +} + +/** + * Pool of producers to external destinations, keyed by destination id. + * + * Producers are created on demand and closed once they have been idle for + * producerIdleMs, so that a worker consuming a shared topic only holds + * connections to the destinations it actually delivers to. + */ +class DeliveryProducerPool { + /** + * @constructor + * @param {Object} params - constructor params + * @param {Object} params.destinationsById - destination configurations + * keyed by destination id (resource name) + * @param {Object} params.deliveryPoolConfig - delivery pool configuration + * @param {number} params.deliveryPoolConfig.deliveryTimeoutMs - time after + * which librdkafka expires a message that could not be delivered + * @param {number} params.deliveryPoolConfig.producerIdleMs - time after + * which an unused producer is closed + * @param {number} params.deliveryPoolConfig.maxProducers - maximum number + * of producers kept open at once + * @param {Logger} params.logger - logger object + */ + constructor(params) { + const { deliveryTimeoutMs, producerIdleMs, maxProducers } = params.deliveryPoolConfig; + this._destinationsById = params.destinationsById; + this._deliveryTimeoutMs = deliveryTimeoutMs; + this._producerIdleMs = producerIdleMs; + this._maxProducers = maxProducers; + this._log = params.logger; + // destination id -> PooledProducer + this._producers = new Map(); + // endpoints ever seen, to reset their gauge when they drop out + this._knownEndpoints = new Set(); + this._inFlight = 0; + this._reapTimer = null; + this._closed = false; + } + + /** + * Start the periodic reaping of idle producers + * + * @return {undefined} + */ + start() { + if (this._reapTimer) { + return; + } + this._reapTimer = setInterval(() => this._reapIdleProducers(), + Math.max(1, Math.floor(this._producerIdleMs / 2))); + // do not keep the process alive just for the reaper + if (this._reapTimer.unref) { + this._reapTimer.unref(); + } + } + + /** + * Get a producer for the given destination, creating and connecting it + * if needed. Concurrent calls made while a producer is connecting are + * queued and served with the same producer once it is ready. + * + * @param {string} destinationId - destination id (resource name) + * @param {function} done - callback: done(err, producer), where producer + * exposes send(messages, cb) + * @return {undefined} + */ + get(destinationId, done) { + if (this._closed) { + return process.nextTick(() => done(errors.InternalError.customizeDescription( + 'delivery producer pool is closed'))); + } + const existing = this._producers.get(destinationId); + if (existing) { + if (existing.ready) { + existing.lastUsed = Date.now(); + return process.nextTick(() => done(null, existing)); + } + existing.addWaiter(done); + return undefined; + } + const destConfig = this._destinationsById[destinationId]; + if (!destConfig) { + return process.nextTick(() => done(errors.InternalError.customizeDescription( + `no destination configured for "${destinationId}"`))); + } + this._evictIfAtCapacity(); + const { host, port } = destConfig; + const entry = new PooledProducer({ + destinationId, + endpoint: port ? `${host}:${port}` : host, + onSendStarted: () => this._onSendStarted(), + onSendFinished: () => this._onSendFinished(), + logger: this._log, + }); + entry.addWaiter(done); + this._producers.set(destinationId, entry); + this._connect(entry, destConfig); + return undefined; + } + + _connect(entry, destConfig) { + const { topic, pollIntervalMs, auth, requiredAcks, compressionType } = destConfig; + const producer = new DeliveryKafkaProducer({ + kafka: { hosts: entry.endpoint }, + topic, + pollIntervalMs, + auth, + compressionType, + requiredAcks, + deliveryTimeoutMs: this._deliveryTimeoutMs, + }); + entry.attach(producer); + producer.once('error', err => { + this._log.error('error connecting producer to external destination', { + method: 'DeliveryProducerPool._connect', + destinationId: entry.destinationId, + endpoint: entry.endpoint, + topic, + error: err.message, + }); + // forget it, so that the next entry for this destination retries + // instead of waiting on a producer that will never be ready + if (this._producers.get(entry.destinationId) === entry) { + this._producers.delete(entry.destinationId); + } + entry.flushWaiters(err); + }); + producer.once('ready', () => { + producer.removeAllListeners('error'); + // BackbeatProducer emits 'error' from the delivery report path, + // an unhandled 'error' event would take down the process + producer.on('error', err => { + this._log.error('error from delivery producer', { + method: 'DeliveryProducerPool._connect', + destinationId: entry.destinationId, + endpoint: entry.endpoint, + topic, + error: err.message, + }); + }); + entry.markReady(); + this._updateProducersGauge(); + this._log.info('opened producer to external destination', { + method: 'DeliveryProducerPool._connect', + destinationId: entry.destinationId, + endpoint: entry.endpoint, + topic, + }); + entry.flushWaiters(null); + }); + } + + _onSendStarted() { + this._inFlight++; + lanesGauge.set(this._inFlight); + } + + _onSendFinished() { + this._inFlight--; + lanesGauge.set(this._inFlight); + } + + /** + * Close producers that have been idle for longer than producerIdleMs. + * A producer with deliveries in flight is never closed, its delivery + * reports are still needed to release the consumer offsets. + * + * @return {undefined} + */ + _reapIdleProducers() { + const now = Date.now(); + this._producers.forEach((entry, destinationId) => { + if (!entry.isIdle() || now - entry.lastUsed < this._producerIdleMs) { + return; + } + // remove from the map before closing, so that a get() racing with + // the close creates a new producer instead of reusing this one + this._producers.delete(destinationId); + this._updateProducersGauge(); + this._log.info('closing idle producer', { + method: 'DeliveryProducerPool._reapIdleProducers', + destinationId, + endpoint: entry.endpoint, + idleMs: now - entry.lastUsed, + }); + entry.close(); + }); + } + + /** + * Make room for a new producer when the pool is at capacity, by closing + * the least recently used idle one. If every producer is busy the cap is + * exceeded rather than dropping deliveries in flight. + * + * @return {undefined} + */ + _evictIfAtCapacity() { + if (this._producers.size < this._maxProducers) { + return; + } + let lru = null; + this._producers.forEach(entry => { + if (!entry.isIdle()) { + return; + } + if (lru === null || entry.lastUsed < lru.lastUsed) { + lru = entry; + } + }); + if (lru === null) { + this._log.warn('producer pool is at capacity and all producers are busy, ' + + 'temporarily exceeding the limit', { + method: 'DeliveryProducerPool._evictIfAtCapacity', + maxProducers: this._maxProducers, + producers: this._producers.size, + }); + return; + } + this._producers.delete(lru.destinationId); + this._updateProducersGauge(); + this._log.info('evicting least recently used producer', { + method: 'DeliveryProducerPool._evictIfAtCapacity', + destinationId: lru.destinationId, + endpoint: lru.endpoint, + }); + lru.close(); + } + + _updateProducersGauge() { + const counts = new Map(); + this._producers.forEach(entry => { + this._knownEndpoints.add(entry.endpoint); + counts.set(entry.endpoint, (counts.get(entry.endpoint) || 0) + 1); + }); + this._knownEndpoints.forEach(endpoint => { + producersGauge.set({ endpoint }, counts.get(endpoint) || 0); + }); + } + + /** + * Close every producer in the pool + * + * @param {function} done - callback + * @return {undefined} + */ + closeAll(done) { + this._closed = true; + if (this._reapTimer) { + clearInterval(this._reapTimer); + this._reapTimer = null; + } + const entries = [...this._producers.values()]; + this._producers.clear(); + this._updateProducersGauge(); + return async.each(entries, (entry, next) => entry.close(next), () => done()); + } +} + +module.exports = DeliveryProducerPool; diff --git a/extensions/notification/deliveryWorker/DeliveryWorker.js b/extensions/notification/deliveryWorker/DeliveryWorker.js new file mode 100644 index 000000000..1d8ea09a5 --- /dev/null +++ b/extensions/notification/deliveryWorker/DeliveryWorker.js @@ -0,0 +1,333 @@ +'use strict'; + +const { EventEmitter } = require('events'); +const Logger = require('werelogs').Logger; +const async = require('async'); +const { CODES } = require('node-rdkafka'); +const { ZenkoMetrics } = require('arsenal').metrics; + +const BackbeatConsumer = require('../../../lib/BackbeatConsumer'); +const messageUtil = require('../utils/message'); +const DeliveryProducerPool = require('./DeliveryProducerPool'); + +// target label used when the entry could not be parsed, so no destination +// is known for it +const UNKNOWN_TARGET = 'unknown'; + +const deliveredEvents = ZenkoMetrics.createCounter({ + name: 's3_notification_delivery_worker_delivered_total', + help: 'Total number of notifications delivered to an external destination', + labelNames: ['target'], +}); + +const droppedEvents = ZenkoMetrics.createCounter({ + name: 's3_notification_delivery_worker_dropped_total', + help: 'Total number of notifications dropped without being delivered', + labelNames: ['target', 'reason'], +}); + +const deliveryDelay = ZenkoMetrics.createHistogram({ + name: 's3_notification_delivery_worker_delivery_delay_seconds', + help: 'Time between sending a notification and receiving its delivery report', + labelNames: ['target', 'status'], + buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], +}); + +function onDelivered(target) { + deliveredEvents.inc({ target }); +} + +function onDropped(target, reason) { + droppedEvents.inc({ target, reason }); +} + +function observeDelay(target, status, delay) { + deliveryDelay.observe({ target, status }, delay); +} + +class DeliveryWorker extends EventEmitter { + /** + * Create a delivery worker, consuming a shared delivery topic and + * dispatching each entry to the external destination named by the entry + * itself. + * + * Unlike the per-destination queue processor, one worker serves every + * destination: the destination id and the notification configuration id + * are carried by the record, so no bucket notification configuration + * lookup is needed here. + * + * @constructor + * @param {Object} kafkaConfig - kafka configuration object + * @param {string} kafkaConfig.hosts - list of kafka brokers + * as "host:port[,host:port...]" + * @param {Object} notifConfig - notification configuration object + * @param {Object[]} notifConfig.destinations - destination configurations + * @param {Object} notifConfig.deliveryPool - delivery pool configuration + * @param {String} notifConfig.deliveryPool.topic - delivery topic name + * @param {String} notifConfig.deliveryPool.groupId - kafka consumer group + * id, shared by every worker of the pool + * @param {number} notifConfig.deliveryPool.concurrency - how many + * notifications can be in flight at once + * @param {number} notifConfig.deliveryPool.maxQueued - how many + * notifications can be queued for processing + */ + constructor(kafkaConfig, notifConfig) { + super(); + this.kafkaConfig = kafkaConfig; + this.notifConfig = notifConfig; + this.deliveryPoolConfig = notifConfig.deliveryPool; + this._destinationsById = {}; + (notifConfig.destinations || []).forEach(destConfig => { + this._destinationsById[destConfig.resource] = destConfig; + }); + this._consumer = null; + this._producerPool = null; + + this.logger = new Logger('Backbeat:Notification:DeliveryWorker'); + } + + /** + * Compute the ordering key of a consumed entry. + * + * The default ordering of BackbeatConsumer is by kafka key, which would + * serialize every notification of a whole destination. Ordering per + * object keeps the per-object ordering guarantee while letting objects of + * the same destination be delivered in parallel. + * + * The parsed entry is stashed on the entry object, which is the same + * object later handed to processKafkaEntry, so the payload is parsed once. + * + * @param {object} ctx - task context pushed by BackbeatConsumer + * @return {string|undefined} ordering key, or undefined to leave the + * entry unordered + */ + _orderBy(ctx) { + const entry = ctx && ctx.entry; + if (!entry) { + return undefined; + } + let parsed; + try { + parsed = JSON.parse(entry.value); + } catch { + // leave it unordered, processKafkaEntry counts the drop + return undefined; + } + entry._notifEntry = parsed; + return `${parsed.destinationId}|${parsed.bucket}/${parsed.key}`; + } + + /** + * Start the producer pool and the kafka consumer. Emits a 'ready' event + * when the consumer is ready. + * + * @param {object} [options] options object + * @param {boolean} [options.disableConsumer] - true to disable startup of + * the consumer (for testing: one has to call processKafkaEntry() + * explicitly) + * @param {function} done callback + * @return {undefined} + */ + start(options, done) { + this._producerPool = new DeliveryProducerPool({ + destinationsById: this._destinationsById, + deliveryPoolConfig: this.deliveryPoolConfig, + logger: this.logger, + }); + this._producerPool.start(); + async.series([ + next => { + if (options && options.disableConsumer) { + this.emit('ready'); + return process.nextTick(next); + } + const { topic, groupId, concurrency, maxQueued } = this.deliveryPoolConfig; + this._consumer = new BackbeatConsumer({ + kafka: { + hosts: this.kafkaConfig.hosts, + site: this.kafkaConfig.site, + compressionType: this.kafkaConfig.compressionType, + requiredAcks: this.kafkaConfig.requiredAcks, + }, + topic, + groupId, + concurrency, + maxQueued, + // librdkafka defaults to 'latest': a worker joining with a + // fresh group would skip everything already in the topic + fromOffset: 'earliest', + queueProcessor: this.processKafkaEntry.bind(this), + orderByFunc: ctx => this._orderBy(ctx), + }); + this._consumer.on('error', err => { + this.logger.error('error starting notification delivery consumer', + { method: 'DeliveryWorker.start', error: err.message }); + // crash if got error at startup + if (!this.isReady()) { + return next(err); + } + return undefined; + }); + this._consumer.on('ready', () => { + this._consumer.subscribe(); + this.logger.info('delivery worker is ready to consume ' + + 'notification entries'); + this.emit('ready'); + return next(); + }); + return undefined; + }, + ], err => { + if (err) { + this.logger.error('error starting notification delivery worker', + { method: 'DeliveryWorker.start', error: err.message }); + return done(err); + } + return done(); + }); + } + + /** + * Stop the kafka consumer and close every pooled producer + * + * @param {function} done - callback + * @return {undefined} + */ + stop(done) { + async.series([ + next => { + if (this._consumer) { + return this._consumer.close(next); + } + return process.nextTick(next); + }, + next => { + if (this._producerPool) { + return this._producerPool.closeAll(next); + } + return process.nextTick(next); + }, + ], err => done(err)); + } + + /** + * Process a kafka entry: deliver it to the external destination named by + * the entry. + * + * The callback is held until the delivery report is received, so that the + * consumer offset is only committed once the notification has left the + * process. A delivery failure is counted and the entry is dropped: the + * callback is never called with an error, which the consumer would report + * as a consumer level error. + * + * @param {object} kafkaEntry - entry consumed from the delivery topic + * @param {function} done - callback function + * @return {undefined} + */ + processKafkaEntry(kafkaEntry, done) { + let parsed = kafkaEntry._notifEntry; + if (!parsed) { + try { + parsed = JSON.parse(kafkaEntry.value); + } catch (error) { + this.logger.error('error parsing JSON entry', { + method: 'DeliveryWorker.processKafkaEntry', + error: error.message, + }); + onDropped(UNKNOWN_TARGET, 'parse_error'); + return done(); + } + } + const { destinationId, bucket, key } = parsed; + const destConfig = this._destinationsById[destinationId]; + if (!destConfig) { + this.logger.warn('no destination configured for entry, dropping', { + method: 'DeliveryWorker.processKafkaEntry', + destinationId, + bucket, + key, + }); + onDropped(destinationId || UNKNOWN_TARGET, 'unknown_destination'); + return done(); + } + return this._producerPool.get(destinationId, (err, producer) => { + if (err) { + this.logger.error('could not get a producer for destination, dropping', { + method: 'DeliveryWorker.processKafkaEntry', + destinationId, + bucket, + key, + error: err.message, + }); + onDropped(destinationId, 'producer_error'); + return done(); + } + const message = messageUtil.transformToSpec(parsed); + const msg = { + // for Kafka keyed partitioning, to map a particular bucket + // and key to a partition + key: `${bucket}/${key}`, + message: JSON.stringify(message), + }; + const startTime = Date.now(); + this.logger.debug('sending message to external destination', { + method: 'DeliveryWorker.processKafkaEntry', + destinationId, + bucket, + key, + eventType: parsed.eventType, + }); + // one entry per send call: BackbeatProducer aggregates delivery + // reports per send, batching would conflate outcomes of entries + // owned by different consumer offsets + return producer.send([msg], sendErr => { + const delay = (Date.now() - startTime) / 1000; + if (sendErr) { + const reason = sendErr.code === CODES.ERRORS.ERR__MSG_TIMED_OUT ? + 'delivery_timeout' : 'delivery_error'; + this.logger.error('error delivering notification to external destination', { + method: 'DeliveryWorker.processKafkaEntry', + destinationId, + bucket, + key, + reason, + error: sendErr.message, + }); + observeDelay(destinationId, 'failure', delay); + onDropped(destinationId, reason); + return done(); + } + observeDelay(destinationId, 'success', delay); + onDelivered(destinationId); + return done(); + }); + }); + } + + /** + * Checks if the delivery worker is ready to consume + * + * @returns {boolean} is delivery worker ready + */ + isReady() { + return !!(this._consumer && this._consumer.isReady()); + } + + /** + * Handle ProbeServer metrics + * + * @param {http.HTTPServerResponse} res - HTTP Response to respond with + * @param {Logger} log - Logger + * @returns {undefined} + */ + async handleMetrics(res, log) { + log.debug('metrics requested'); + res.writeHead(200, { + 'Content-Type': ZenkoMetrics.asPrometheusContentType(), + }); + const metrics = await ZenkoMetrics.asPrometheus(); + res.end(metrics); + } +} + +module.exports = DeliveryWorker; diff --git a/extensions/notification/deliveryWorker/probeConfig.js b/extensions/notification/deliveryWorker/probeConfig.js new file mode 100644 index 000000000..441dfc18c --- /dev/null +++ b/extensions/notification/deliveryWorker/probeConfig.js @@ -0,0 +1,46 @@ +// Several delivery workers can run from a single rendered config file, on +// one host or inside one container, and only one process can bind a given +// port. The deployment hands each worker process its own port through this +// environment variable, which wins over the configured one. +const DELIVERY_POOL_PROBE_PORT_ENV = 'DELIVERY_POOL_PROBE_PORT'; + +const MAX_PORT = 65535; + +/** + * Resolve the probe server configuration of a delivery worker + * + * @param {Object} deliveryPoolConfig - delivery pool configuration + * @param {Object} [env] - environment to read the port override from + * @param {Logger} [logger] - logger object + * @return {Object|undefined} probe server configuration, or undefined when + * no probe server is configured + */ +function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { + const probeServer = deliveryPoolConfig && deliveryPoolConfig.probeServer; + if (!probeServer) { + return undefined; + } + const rawPort = (env || {})[DELIVERY_POOL_PROBE_PORT_ENV]; + if (rawPort === undefined || `${rawPort}`.trim() === '') { + return probeServer; + } + const trimmedPort = `${rawPort}`.trim(); + const port = /^\d+$/.test(trimmedPort) ? Number.parseInt(trimmedPort, 10) : NaN; + if (!Number.isInteger(port) || port <= 0 || port > MAX_PORT) { + if (logger) { + logger.warn('ignoring invalid probe server port from the environment', { + method: 'resolveProbeServerConfig', + envVar: DELIVERY_POOL_PROBE_PORT_ENV, + value: rawPort, + port: probeServer.port, + }); + } + return probeServer; + } + return { ...probeServer, port }; +} + +module.exports = { + DELIVERY_POOL_PROBE_PORT_ENV, + resolveProbeServerConfig, +}; diff --git a/extensions/notification/deliveryWorker/task.js b/extensions/notification/deliveryWorker/task.js new file mode 100644 index 000000000..8e6df7b60 --- /dev/null +++ b/extensions/notification/deliveryWorker/task.js @@ -0,0 +1,98 @@ +'use strict'; +const assert = require('assert'); +const { errors, jsutil } = require('arsenal'); +const async = require('async'); +const werelogs = require('werelogs'); +const { + DEFAULT_LIVE_ROUTE, + DEFAULT_READY_ROUTE, + DEFAULT_METRICS_ROUTE, +} = require('arsenal').network.probe.ProbeServer; +const { sendSuccess, sendError } = require('arsenal').network.probe.Utils; +const DeliveryWorker = require('./DeliveryWorker'); +const { resolveProbeServerConfig } = require('./probeConfig'); +const { startProbeServer } = require('../../../lib/util/probe'); + +const config = require('../../../lib/Config'); +const kafkaConfig = config.kafka; +const notifConfig = config.extensions.notification; + +const log = new werelogs.Logger('Backbeat:NotificationDeliveryWorker:task'); +werelogs.configure({ + level: config.log.logLevel, + dump: config.log.dumpLevel, +}); + +assert(notifConfig && notifConfig.deliveryPool && notifConfig.deliveryPool.enabled, + 'delivery worker requires extensions.notification.deliveryPool.enabled ' + + 'to be set'); + +// no destination argument: the destination and the notification +// configuration id are carried by each record of the delivery topic +const deliveryWorker = new DeliveryWorker(kafkaConfig, notifConfig); + +/** + * Handle ProbeServer liveness check + * + * @param {http.HTTPServerResponse} res - HTTP Response to respond with + * @param {Logger} log - Logger + * @returns {undefined} + */ +function handleLiveness(res, log) { + if (deliveryWorker.isReady()) { + sendSuccess(res, log); + } else { + log.error('Notification Delivery Worker is not ready'); + sendError(res, log, errors.ServiceUnavailable, 'unhealthy'); + } +} + +const probeServerConfig = resolveProbeServerConfig( + notifConfig.deliveryPool, process.env, log); + +async.series([ + next => deliveryWorker.start(null, next), + next => startProbeServer(probeServerConfig, jsutil.once((err, probeServer) => { + if (err) { + // a worker that cannot serve its probe routes still delivers + // notifications, so keep going rather than taking the process + // down: workers sharing a config file also share a port, and + // only the first of them can bind it + log.error('probe server not started, continuing without it', { + error: err.message, + port: probeServerConfig && probeServerConfig.port, + }); + return next(); + } + if (probeServer !== undefined) { + // following the same pattern as other extensions, where liveness + // and readiness are handled by the same handler + probeServer.addHandler([DEFAULT_LIVE_ROUTE, DEFAULT_READY_ROUTE], handleLiveness); + probeServer.addHandler(DEFAULT_METRICS_ROUTE, + (res, log) => deliveryWorker.handleMetrics(res, log) + ); + } + return next(); + })) +], err => { + if (err) { + log.error('error starting notification delivery worker task', { + method: 'notification.task.deliveryWorker', + error: err, + }); + process.emit('SIGTERM'); + } +}); + +process.on('SIGTERM', () => { + log.info('received SIGTERM, exiting'); + deliveryWorker.stop(error => { + if (error) { + log.error('failed to exit properly', { + error, + }); + process.exit(1); + } + process.exit(0); + }); +}); diff --git a/package.json b/package.json index 871194c0c..0c94c5a54 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,9 @@ "lint_md": "mdlint $(git ls-files '[^bucket-scanner/]*.md')", "start": "node bin/backbeat.js", "notification_populator": "BACKBEAT_QUEUEPOPULATOR_EXTENSIONS=notification node bin/queuePopulator.js", - "notification_processor": "node extensions/notification/queueProcessor/task.js" + "notification_processor": "node extensions/notification/queueProcessor/task.js", + "notification_delivery_worker": "node extensions/notification/deliveryWorker/task.js", + "notification_delivery_replay": "node bin/notificationDeliveryReplay.js" }, "repository": { "type": "git", diff --git a/tests/unit/notification/DeliveryProducerPool.spec.js b/tests/unit/notification/DeliveryProducerPool.spec.js new file mode 100644 index 000000000..6c920e440 --- /dev/null +++ b/tests/unit/notification/DeliveryProducerPool.spec.js @@ -0,0 +1,298 @@ +const assert = require('assert'); +const sinon = require('sinon'); + +const FakeLogger = require('../../utils/fakeLogger'); + +const DeliveryProducerPool = require( + '../../../extensions/notification/deliveryWorker/DeliveryProducerPool'); +const DeliveryKafkaProducer = require( + '../../../extensions/notification/deliveryWorker/DeliveryKafkaProducer'); + +const destinationsById = { + destA: { + resource: 'destA', + type: 'kafka', + host: 'external-kafka-host', + port: 9092, + topic: 'topic-a', + pollIntervalMs: 1000, + requiredAcks: 1, + compressionType: 'none', + }, + destB: { + resource: 'destB', + type: 'kafka', + host: 'other-kafka-host', + port: 9092, + topic: 'topic-b', + }, + destC: { + resource: 'destC', + type: 'kafka', + host: 'third-kafka-host', + topic: 'topic-c', + }, +}; + +const deliveryPoolConfig = { + deliveryTimeoutMs: 30000, + producerIdleMs: 1000, + maxProducers: 50, +}; + +/** + * Build a pool whose producers become ready asynchronously without ever + * reaching a broker + * @param {object} [overrides] - deliveryPoolConfig overrides + * @return {DeliveryProducerPool} pool under test + */ +function makePool(overrides) { + return new DeliveryProducerPool({ + destinationsById, + deliveryPoolConfig: { ...deliveryPoolConfig, ...overrides }, + logger: FakeLogger, + }); +} + +describe('notification DeliveryProducerPool', () => { + let connectStub; + let closeStub; + + beforeEach(() => { + // emit 'ready' rather than connecting to a broker + connectStub = sinon.stub(DeliveryKafkaProducer.prototype, 'connect') + .callsFake(function connect() { + setTimeout(() => this.emit('ready'), 10); + }); + closeStub = sinon.stub(DeliveryKafkaProducer.prototype, 'close') + .callsFake(cb => process.nextTick(cb)); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should bound message retries with the configured delivery timeout', done => { + const pool = makePool(); + pool.get('destA', (err, producer) => { + assert.ifError(err); + const { topicConfig } = producer.producer; + assert.strictEqual(topicConfig['message.timeout.ms'], 30000); + // the inherited topic config is preserved + assert.strictEqual(topicConfig['request.required.acks'], 1); + assert.strictEqual(topicConfig['request.timeout.ms'], 5000); + done(); + }); + }); + + it('should configure the producer from the destination config', done => { + const pool = makePool(); + pool.get('destA', (err, producer) => { + assert.ifError(err); + assert.strictEqual(producer.producer._kafkaHosts, 'external-kafka-host:9092'); + assert.strictEqual(producer.producer._topic, 'topic-a'); + assert.strictEqual(producer.producer._pollIntervalMs, 1000); + assert.strictEqual(producer.producer._compressionType, 'none'); + assert.strictEqual(producer.producer._requiredAcks, 1); + done(); + }); + }); + + it('should use the bare host when the destination has no port', done => { + const pool = makePool(); + pool.get('destC', (err, producer) => { + assert.ifError(err); + assert.strictEqual(producer.producer._kafkaHosts, 'third-kafka-host'); + done(); + }); + }); + + it('should create a single producer for concurrent gets while connecting', done => { + const pool = makePool(); + const handles = []; + const collect = (err, producer) => { + assert.ifError(err); + handles.push(producer); + if (handles.length === 3) { + assert.strictEqual(connectStub.callCount, 1); + assert.strictEqual(handles[0], handles[1]); + assert.strictEqual(handles[1], handles[2]); + assert.strictEqual(pool._producers.size, 1); + done(); + } + }; + pool.get('destA', collect); + pool.get('destA', collect); + pool.get('destA', collect); + }); + + it('should reuse a ready producer', done => { + const pool = makePool(); + pool.get('destA', (err, first) => { + assert.ifError(err); + pool.get('destA', (err2, second) => { + assert.ifError(err2); + assert.strictEqual(first, second); + assert.strictEqual(connectStub.callCount, 1); + done(); + }); + }); + }); + + it('should fail every waiter and forget the producer when connecting fails', done => { + connectStub.restore(); + const connectError = new Error('cannot reach broker'); + sinon.stub(DeliveryKafkaProducer.prototype, 'connect') + .callsFake(function connect() { + setTimeout(() => this.emit('error', connectError), 10); + }); + const pool = makePool(); + let failures = 0; + const expectError = err => { + assert.strictEqual(err, connectError); + failures++; + if (failures === 2) { + // forgotten, so the next entry retries instead of waiting + // on a producer that will never be ready + assert.strictEqual(pool._producers.size, 0); + done(); + } + }; + pool.get('destA', expectError); + pool.get('destA', expectError); + }); + + it('should fail for a destination that is not configured', done => { + const pool = makePool(); + pool.get('goneDestId', err => { + assert(err); + assert.strictEqual(connectStub.callCount, 0); + done(); + }); + }); + + it('should track deliveries in flight around a send', done => { + const pool = makePool(); + let deliveryReportCb = null; + pool.get('destA', (err, producer) => { + assert.ifError(err); + sinon.stub(producer.producer, 'send').callsFake((messages, cb) => { + deliveryReportCb = cb; + }); + producer.send([{ key: 'k', message: '{}' }], sendErr => { + assert.ifError(sendErr); + assert.strictEqual(producer.inFlight, 0); + assert.strictEqual(pool._inFlight, 0); + done(); + }); + assert.strictEqual(producer.inFlight, 1); + assert.strictEqual(pool._inFlight, 1); + deliveryReportCb(); + }); + }); + + describe('reaping idle producers', () => { + it('should close a producer that has been idle for too long', done => { + const pool = makePool(); + pool.get('destA', err => { + assert.ifError(err); + const entry = pool._producers.get('destA'); + entry.lastUsed = Date.now() - 5000; + pool._reapIdleProducers(); + // removed from the map before the close completes, so a get + // racing the close gets a new producer + assert.strictEqual(pool._producers.size, 0); + assert.strictEqual(closeStub.callCount, 1); + done(); + }); + }); + + it('should keep a producer that still has deliveries in flight', done => { + const pool = makePool(); + pool.get('destA', err => { + assert.ifError(err); + const entry = pool._producers.get('destA'); + entry.lastUsed = Date.now() - 5000; + entry.inFlight = 1; + pool._reapIdleProducers(); + assert.strictEqual(pool._producers.size, 1); + assert.strictEqual(closeStub.callCount, 0); + done(); + }); + }); + + it('should keep a recently used producer', done => { + const pool = makePool(); + pool.get('destA', err => { + assert.ifError(err); + pool._reapIdleProducers(); + assert.strictEqual(pool._producers.size, 1); + assert.strictEqual(closeStub.callCount, 0); + done(); + }); + }); + }); + + describe('capacity', () => { + it('should evict the least recently used idle producer', done => { + const pool = makePool({ maxProducers: 2 }); + pool.get('destA', errA => { + assert.ifError(errA); + pool.get('destB', errB => { + assert.ifError(errB); + // make destA the least recently used + pool._producers.get('destA').lastUsed = Date.now() - 5000; + pool.get('destC', errC => { + assert.ifError(errC); + assert.strictEqual(pool._producers.size, 2); + assert.strictEqual(pool._producers.has('destA'), false); + assert.strictEqual(pool._producers.has('destB'), true); + assert.strictEqual(pool._producers.has('destC'), true); + assert.strictEqual(closeStub.callCount, 1); + done(); + }); + }); + }); + }); + + it('should exceed the cap rather than evict a busy producer', done => { + const pool = makePool({ maxProducers: 2 }); + pool.get('destA', errA => { + assert.ifError(errA); + pool.get('destB', errB => { + assert.ifError(errB); + pool._producers.get('destA').inFlight = 1; + pool._producers.get('destB').inFlight = 1; + pool.get('destC', errC => { + assert.ifError(errC); + assert.strictEqual(pool._producers.size, 3); + assert.strictEqual(closeStub.callCount, 0); + done(); + }); + }); + }); + }); + }); + + it('should close every producer on closeAll', done => { + const pool = makePool(); + pool.start(); + pool.get('destA', errA => { + assert.ifError(errA); + pool.get('destB', errB => { + assert.ifError(errB); + assert.strictEqual(pool._producers.size, 2); + pool.closeAll(() => { + assert.strictEqual(pool._producers.size, 0); + assert.strictEqual(closeStub.callCount, 2); + assert.strictEqual(pool._reapTimer, null); + // a closed pool hands out no more producers + pool.get('destA', err => { + assert(err); + done(); + }); + }); + }); + }); + }); +}); diff --git a/tests/unit/notification/DeliveryWorker.spec.js b/tests/unit/notification/DeliveryWorker.spec.js new file mode 100644 index 000000000..3fe9e00b0 --- /dev/null +++ b/tests/unit/notification/DeliveryWorker.spec.js @@ -0,0 +1,364 @@ +const assert = require('assert'); +const sinon = require('sinon'); +const { ZenkoMetrics } = require('arsenal').metrics; + +const FakeLogger = require('../../utils/fakeLogger'); + +const DeliveryWorker = require( + '../../../extensions/notification/deliveryWorker/DeliveryWorker'); +const { DELIVERY_POOL_PROBE_PORT_ENV, resolveProbeServerConfig } = require( + '../../../extensions/notification/deliveryWorker/probeConfig'); + +const DELIVERED_METRIC = 's3_notification_delivery_worker_delivered_total'; +const DROPPED_METRIC = 's3_notification_delivery_worker_dropped_total'; +const DELAY_METRIC = 's3_notification_delivery_worker_delivery_delay_seconds'; + +const kafkaConfig = { + hosts: 'internal-kafka-host:9092', +}; + +const notifConfig = { + destinations: [ + { + resource: 'destId', + type: 'kafka', + host: 'external-kafka-host', + port: 9092, + topic: 'dest-topic', + }, + ], + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + deliveryTimeoutMs: 30000, + producerIdleMs: 300000, + maxProducers: 50, + concurrency: 1000, + maxQueued: 1000, + }, +}; + +const notifRecord = { + destinationId: 'destId', + configurationId: 'config-1', + bucket: 'mybucket', + key: 'mykey', + eventType: 's3:ObjectCreated:Put', + dateTime: '2024-08-02T09:19:43.991Z', + region: 'us-east-1', + size: 42, +}; + +function makeEntry(value) { + return { + topic: 'delivery-topic', + partition: 0, + offset: 42, + key: Buffer.from('destId'), + value: typeof value === 'string' ? value : JSON.stringify(value), + }; +} + +/** + * Read the value of a labelled counter, 0 if not observed yet + * @param {string} name - metric name + * @param {object} labels - labels to match + * @return {Promise} current counter value + */ +async function counterValue(name, labels) { + const data = await ZenkoMetrics.getMetric(name).get(); + const entry = data.values.find(value => Object.entries(labels) + .every(([label, expected]) => value.labels[label] === expected)); + return entry ? entry.value : 0; +} + +/** + * Read how many observations a labelled histogram received + * @param {object} labels - labels to match + * @return {Promise} number of observations + */ +async function delayObservations(labels) { + const data = await ZenkoMetrics.getMetric(DELAY_METRIC).get(); + const entry = data.values.find(value => + value.metricName === `${DELAY_METRIC}_count` && + Object.entries(labels).every(([label, expected]) => value.labels[label] === expected)); + return entry ? entry.value : 0; +} + +/** + * Build a producer pool stub + * @param {function} sendImpl - implementation of producer.send(messages, cb) + * @param {Error} [getError] - error to fail pool.get() with + * @return {object} pool stub, with "send" exposing the send stub + */ +function fakePool(sendImpl, getError) { + const send = sinon.stub().callsFake(sendImpl); + return { + send, + start: sinon.stub(), + closeAll: sinon.stub().callsFake(cb => cb()), + get: sinon.stub().callsFake((destinationId, cb) => process.nextTick( + () => (getError ? cb(getError) : cb(null, { send })))), + }; +} + +describe('notification DeliveryWorker', () => { + let worker; + + beforeEach(() => { + worker = new DeliveryWorker(kafkaConfig, notifConfig); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should hold the callback until the delivery report is received', done => { + let deliveryReportCb = null; + const pool = fakePool((messages, cb) => { + deliveryReportCb = cb; + }); + worker._producerPool = pool; + + let doneCalled = false; + worker.processKafkaEntry(makeEntry(notifRecord), err => { + assert.ifError(err); + doneCalled = true; + }); + + setTimeout(() => { + assert(pool.send.calledOnce); + assert.strictEqual(doneCalled, false, + 'callback must not be called before the delivery report'); + deliveryReportCb(); + setImmediate(() => { + assert.strictEqual(doneCalled, true); + done(); + }); + }, 50); + }); + + it('should send one record per send call, keyed by bucket and object key', done => { + const pool = fakePool((messages, cb) => cb()); + worker._producerPool = pool; + + worker.processKafkaEntry(makeEntry(notifRecord), err => { + assert.ifError(err); + assert(pool.get.calledOnceWith('destId')); + const [messages] = pool.send.args[0]; + assert(Array.isArray(messages)); + assert.strictEqual(messages.length, 1); + assert.strictEqual(messages[0].key, 'mybucket/mykey'); + const message = JSON.parse(messages[0].message); + assert.strictEqual(message.Records.length, 1); + assert.strictEqual(message.Records[0].eventName, 's3:ObjectCreated:Put'); + // the configuration id rides in the payload, no config lookup + assert.strictEqual(message.Records[0].s3.configurationId, 'config-1'); + done(); + }); + }); + + it('should count a delivered notification and observe its delay', async () => { + const pool = fakePool((messages, cb) => cb()); + worker._producerPool = pool; + + const deliveredBefore = await counterValue(DELIVERED_METRIC, { target: 'destId' }); + const observedBefore = await delayObservations({ target: 'destId', status: 'success' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry(notifRecord), err => { + assert.ifError(err); + resolve(); + })); + + assert.strictEqual( + await counterValue(DELIVERED_METRIC, { target: 'destId' }), deliveredBefore + 1); + assert.strictEqual( + await delayObservations({ target: 'destId', status: 'success' }), observedBefore + 1); + }); + + it('should drop and not fail the task when the delivery report is an error', async () => { + const pool = fakePool((messages, cb) => cb(new Error('delivery error'))); + worker._producerPool = pool; + + const droppedBefore = await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'delivery_error' }); + const observedBefore = await delayObservations({ target: 'destId', status: 'failure' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry(notifRecord), (...args) => { + // never call back with an error, the consumer would emit + // a consumer level 'error' event for it + assert.strictEqual(args.length, 0); + resolve(); + })); + + assert.strictEqual(await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'delivery_error' }), droppedBefore + 1); + assert.strictEqual( + await delayObservations({ target: 'destId', status: 'failure' }), observedBefore + 1); + }); + + it('should drop with a delivery_timeout reason when the message expired', async () => { + const timeoutError = new Error('Local: Message timed out'); + // ERR__MSG_TIMED_OUT + timeoutError.code = -192; + const pool = fakePool((messages, cb) => cb(timeoutError)); + worker._producerPool = pool; + + const droppedBefore = await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'delivery_timeout' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry(notifRecord), err => { + assert.ifError(err); + resolve(); + })); + + assert.strictEqual(await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'delivery_timeout' }), droppedBefore + 1); + }); + + it('should drop an entry that is not valid JSON', async () => { + const pool = fakePool((messages, cb) => cb()); + worker._producerPool = pool; + + const droppedBefore = await counterValue(DROPPED_METRIC, + { target: 'unknown', reason: 'parse_error' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry('this is not json'), err => { + assert.ifError(err); + resolve(); + })); + + assert(pool.send.notCalled); + assert.strictEqual(await counterValue(DROPPED_METRIC, + { target: 'unknown', reason: 'parse_error' }), droppedBefore + 1); + }); + + it('should drop an entry for a destination that is not configured', async () => { + const pool = fakePool((messages, cb) => cb()); + worker._producerPool = pool; + + const droppedBefore = await counterValue(DROPPED_METRIC, + { target: 'goneDestId', reason: 'unknown_destination' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry({ ...notifRecord, destinationId: 'goneDestId' }), err => { + assert.ifError(err); + resolve(); + })); + + assert(pool.get.notCalled); + assert(pool.send.notCalled); + assert.strictEqual(await counterValue(DROPPED_METRIC, + { target: 'goneDestId', reason: 'unknown_destination' }), droppedBefore + 1); + }); + + it('should drop an entry when no producer can be obtained', async () => { + const pool = fakePool((messages, cb) => cb(), new Error('connect failed')); + worker._producerPool = pool; + + const droppedBefore = await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'producer_error' }); + + await new Promise(resolve => worker.processKafkaEntry( + makeEntry(notifRecord), err => { + assert.ifError(err); + resolve(); + })); + + assert(pool.send.notCalled); + assert.strictEqual(await counterValue(DROPPED_METRIC, + { target: 'destId', reason: 'producer_error' }), droppedBefore + 1); + }); + + describe('ordering', () => { + it('should order by destination and object, not by the kafka key', () => { + const entry = makeEntry(notifRecord); + assert.strictEqual(worker._orderBy({ entry }), 'destId|mybucket/mykey'); + }); + + it('should stash the parsed entry so it is parsed only once', done => { + const entry = makeEntry(notifRecord); + worker._orderBy({ entry }); + assert.deepStrictEqual(entry._notifEntry, notifRecord); + + // the stash is what gets used: an unparseable value would + // otherwise be dropped + entry.value = 'this is not json'; + const pool = fakePool((messages, cb) => cb()); + worker._producerPool = pool; + worker.processKafkaEntry(entry, err => { + assert.ifError(err); + assert(pool.send.calledOnce); + assert.strictEqual(pool.send.args[0][0][0].key, 'mybucket/mykey'); + done(); + }); + }); + + it('should leave an unparseable entry unordered', () => { + const entry = makeEntry('this is not json'); + assert.strictEqual(worker._orderBy({ entry }), undefined); + assert.strictEqual(entry._notifEntry, undefined); + }); + }); + + describe('probe server config', () => { + const probeServer = { bindAddress: '0.0.0.0', port: 8900 }; + const withProbe = { ...notifConfig.deliveryPool, probeServer }; + + it('should keep the configured port when the environment is silent', () => { + assert.strictEqual(resolveProbeServerConfig(withProbe, {}), probeServer); + }); + + it('should let the environment give this worker its own port', () => { + const resolved = resolveProbeServerConfig(withProbe, + { [DELIVERY_POOL_PROBE_PORT_ENV]: '8902' }); + assert.strictEqual(resolved.port, 8902); + assert.strictEqual(resolved.bindAddress, '0.0.0.0'); + // the configured object is left alone + assert.strictEqual(probeServer.port, 8900); + }); + + it('should fall back to the configured port and warn for a bad override', () => { + ['notaport', '0', '70000', '8900abc'].forEach(value => { + const logger = { ...FakeLogger, warn: sinon.stub() }; + const resolved = resolveProbeServerConfig(withProbe, + { [DELIVERY_POOL_PROBE_PORT_ENV]: value }, logger); + assert.strictEqual(resolved.port, 8900, `for value "${value}"`); + assert(logger.warn.calledOnce, `no warning for value "${value}"`); + assert.strictEqual(logger.warn.args[0][1].value, value); + }); + }); + + it('should fall back quietly when the override is empty', () => { + ['', ' '].forEach(value => { + const logger = { ...FakeLogger, warn: sinon.stub() }; + const resolved = resolveProbeServerConfig(withProbe, + { [DELIVERY_POOL_PROBE_PORT_ENV]: value }, logger); + assert.strictEqual(resolved, probeServer, `for value "${value}"`); + assert(logger.warn.notCalled, `unexpected warning for value "${value}"`); + }); + }); + + it('should stay undefined when no probe server is configured', () => { + assert.strictEqual( + resolveProbeServerConfig(notifConfig.deliveryPool, {}), undefined); + assert.strictEqual(resolveProbeServerConfig(undefined, {}), undefined); + }); + }); + + describe('isReady', () => { + it('should not be ready without a consumer', () => { + assert.strictEqual(worker.isReady(), false); + }); + + it('should follow the consumer readiness', () => { + worker._consumer = { isReady: () => true }; + assert.strictEqual(worker.isReady(), true); + }); + }); +});