From 868282d4489092e3ecd9792a60451d3dbf8b673b Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:13:13 +0200 Subject: [PATCH 1/6] S3C-11127: add pooled kafka producers for notification delivery The delivery worker consumes a topic shared by every destination, so it needs a producer per destination created on demand rather than the single producer of the per-destination queue processor. DeliveryKafkaProducer bounds how long librdkafka retries a message, since the worker holds the consumer offset until the delivery report arrives and an unbounded retry would block that offset forever. DeliveryProducerPool creates producers on demand, serves concurrent requests made during connect from a single producer, and closes producers once idle. A producer with deliveries in flight is never closed: its delivery reports still release consumer offsets. --- .../deliveryWorker/DeliveryKafkaProducer.js | 43 ++ .../deliveryWorker/DeliveryProducerPool.js | 376 ++++++++++++++++++ 2 files changed, 419 insertions(+) create mode 100644 extensions/notification/deliveryWorker/DeliveryKafkaProducer.js create mode 100644 extensions/notification/deliveryWorker/DeliveryProducerPool.js 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; From 18967aa87732872dc1d1fb2c3c236675a4644c8e Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:13:22 +0200 Subject: [PATCH 2/6] S3C-11127: add the notification delivery worker One worker serves every destination: the destination id and the notification configuration id ride in each record, so no bucket notification configuration lookup and no mongo or zookeeper connection is needed here. Two details are load bearing. The consumer callback is held until the delivery report arrives, so an offset is only committed once the notification has left the process, and a delivery failure is counted and dropped rather than passed back as an error, which the consumer would raise as a consumer level error. Ordering is by destination and object instead of the default kafka key, which would serialize a whole destination behind one lane. The consumer reads from the earliest offset: librdkafka defaults to latest, and a worker joining with a fresh group would skip everything already sitting in the topic. --- .../deliveryWorker/DeliveryWorker.js | 333 ++++++++++++++++++ .../notification/deliveryWorker/task.js | 87 +++++ package.json | 4 +- 3 files changed, 423 insertions(+), 1 deletion(-) create mode 100644 extensions/notification/deliveryWorker/DeliveryWorker.js create mode 100644 extensions/notification/deliveryWorker/task.js 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/task.js b/extensions/notification/deliveryWorker/task.js new file mode 100644 index 000000000..ecb815001 --- /dev/null +++ b/extensions/notification/deliveryWorker/task.js @@ -0,0 +1,87 @@ +'use strict'; +const assert = require('assert'); +const { errors } = 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 { 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'); + } +} + +async.series([ + next => deliveryWorker.start(null, next), + next => startProbeServer(notifConfig.deliveryPool.probeServer, (err, probeServer) => { + if (err) { + log.error('error starting probe server', { error: err }); + return next(err); + } + 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", From 60180fa3d1104b97a3da464858345eb10650f089 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:13:29 +0200 Subject: [PATCH 3/6] S3C-11127: add unit tests for the notification delivery worker Covers the commit timing (the callback waits for the delivery report), the drop reasons and their counters, the ordering key and the parse-once stash, and for the pool the single producer per destination during connect, idle reaping that spares busy producers, capacity eviction, and the message timeout landing in the producer topic config. --- .../notification/DeliveryProducerPool.spec.js | 298 +++++++++++++++++ .../unit/notification/DeliveryWorker.spec.js | 315 ++++++++++++++++++ 2 files changed, 613 insertions(+) create mode 100644 tests/unit/notification/DeliveryProducerPool.spec.js create mode 100644 tests/unit/notification/DeliveryWorker.spec.js 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..7d6895779 --- /dev/null +++ b/tests/unit/notification/DeliveryWorker.spec.js @@ -0,0 +1,315 @@ +const assert = require('assert'); +const sinon = require('sinon'); +const { ZenkoMetrics } = require('arsenal').metrics; + +const DeliveryWorker = require( + '../../../extensions/notification/deliveryWorker/DeliveryWorker'); + +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('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); + }); + }); +}); From 5f8f3db94ddb6bf3c81e985cac8b7dad92b301f1 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:18:00 +0200 Subject: [PATCH 4/6] S3C-11127: let each delivery worker bind its own probe port Several 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, so the DELIVERY_POOL_PROBE_PORT environment variable now wins over the configured port and the deployment can hand each worker process its own. A probe server that fails to bind no longer takes the worker down either. A worker that cannot serve its probe routes still delivers notifications, and dying instead would turn a port clash into a crash loop. --- .../deliveryWorker/probeConfig.js | 46 +++++++++++++++++++ .../notification/deliveryWorker/task.js | 21 +++++++-- .../unit/notification/DeliveryWorker.spec.js | 36 +++++++++++++++ 3 files changed, 98 insertions(+), 5 deletions(-) create mode 100644 extensions/notification/deliveryWorker/probeConfig.js 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 index ecb815001..8e6df7b60 100644 --- a/extensions/notification/deliveryWorker/task.js +++ b/extensions/notification/deliveryWorker/task.js @@ -1,6 +1,6 @@ 'use strict'; const assert = require('assert'); -const { errors } = require('arsenal'); +const { errors, jsutil } = require('arsenal'); const async = require('async'); const werelogs = require('werelogs'); const { @@ -10,6 +10,7 @@ const { } = 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'); @@ -46,12 +47,22 @@ function handleLiveness(res, log) { } } +const probeServerConfig = resolveProbeServerConfig( + notifConfig.deliveryPool, process.env, log); + async.series([ next => deliveryWorker.start(null, next), - next => startProbeServer(notifConfig.deliveryPool.probeServer, (err, probeServer) => { + next => startProbeServer(probeServerConfig, jsutil.once((err, probeServer) => { if (err) { - log.error('error starting probe server', { error: err }); - return next(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 @@ -62,7 +73,7 @@ async.series([ ); } return next(); - }) + })) ], err => { if (err) { log.error('error starting notification delivery worker task', { diff --git a/tests/unit/notification/DeliveryWorker.spec.js b/tests/unit/notification/DeliveryWorker.spec.js index 7d6895779..a6da380bc 100644 --- a/tests/unit/notification/DeliveryWorker.spec.js +++ b/tests/unit/notification/DeliveryWorker.spec.js @@ -2,8 +2,12 @@ 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'; @@ -302,6 +306,38 @@ describe('notification DeliveryWorker', () => { }); }); + 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 for a bad override', () => { + ['', ' ', 'notaport', '0', '70000', '8900abc'].forEach(value => { + const resolved = resolveProbeServerConfig(withProbe, + { [DELIVERY_POOL_PROBE_PORT_ENV]: value }, FakeLogger); + assert.strictEqual(resolved.port, 8900, `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); From 28dc8f4186c4f9e05f3c4ce339b96bb8893a0ed1 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:20:22 +0200 Subject: [PATCH 5/6] S3C-11127: rename the probe port override to DELIVERY_PROBE_PORT Aligns the environment variable with the name the federation supervisord template exports per worker program. Behaviour is unchanged: a valid integer replaces the configured probe server port and the rest of the probe server config is kept, anything else falls back to the configured port with a warning. --- .../deliveryWorker/probeConfig.js | 8 +++---- .../unit/notification/DeliveryWorker.spec.js | 23 +++++++++++++++---- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/extensions/notification/deliveryWorker/probeConfig.js b/extensions/notification/deliveryWorker/probeConfig.js index 441dfc18c..363675d91 100644 --- a/extensions/notification/deliveryWorker/probeConfig.js +++ b/extensions/notification/deliveryWorker/probeConfig.js @@ -2,7 +2,7 @@ // 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 DELIVERY_PROBE_PORT_ENV = 'DELIVERY_PROBE_PORT'; const MAX_PORT = 65535; @@ -20,7 +20,7 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { if (!probeServer) { return undefined; } - const rawPort = (env || {})[DELIVERY_POOL_PROBE_PORT_ENV]; + const rawPort = (env || {})[DELIVERY_PROBE_PORT_ENV]; if (rawPort === undefined || `${rawPort}`.trim() === '') { return probeServer; } @@ -30,7 +30,7 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { if (logger) { logger.warn('ignoring invalid probe server port from the environment', { method: 'resolveProbeServerConfig', - envVar: DELIVERY_POOL_PROBE_PORT_ENV, + envVar: DELIVERY_PROBE_PORT_ENV, value: rawPort, port: probeServer.port, }); @@ -41,6 +41,6 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { } module.exports = { - DELIVERY_POOL_PROBE_PORT_ENV, + DELIVERY_PROBE_PORT_ENV, resolveProbeServerConfig, }; diff --git a/tests/unit/notification/DeliveryWorker.spec.js b/tests/unit/notification/DeliveryWorker.spec.js index a6da380bc..bd7c829a6 100644 --- a/tests/unit/notification/DeliveryWorker.spec.js +++ b/tests/unit/notification/DeliveryWorker.spec.js @@ -6,7 +6,7 @@ const FakeLogger = require('../../utils/fakeLogger'); const DeliveryWorker = require( '../../../extensions/notification/deliveryWorker/DeliveryWorker'); -const { DELIVERY_POOL_PROBE_PORT_ENV, resolveProbeServerConfig } = require( +const { DELIVERY_PROBE_PORT_ENV, resolveProbeServerConfig } = require( '../../../extensions/notification/deliveryWorker/probeConfig'); const DELIVERED_METRIC = 's3_notification_delivery_worker_delivered_total'; @@ -316,18 +316,31 @@ describe('notification DeliveryWorker', () => { it('should let the environment give this worker its own port', () => { const resolved = resolveProbeServerConfig(withProbe, - { [DELIVERY_POOL_PROBE_PORT_ENV]: '8902' }); + { [DELIVERY_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 for a bad override', () => { - ['', ' ', 'notaport', '0', '70000', '8900abc'].forEach(value => { + 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 }, FakeLogger); + { [DELIVERY_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_PROBE_PORT_ENV]: value }, logger); + assert.strictEqual(resolved, probeServer, `for value "${value}"`); + assert(logger.warn.notCalled, `unexpected warning for value "${value}"`); }); }); From 5403bde405737ac9ec5024ff22afc0b764576264 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:27:00 +0200 Subject: [PATCH 6/6] S3C-11127: revert the probe port override to DELIVERY_POOL_PROBE_PORT Restores the environment variable name the federation supervisord template exports and the contract pins. The resolution itself is unchanged: a valid integer replaces the configured probe server port and the rest of the probe server config is kept, anything else falls back to the configured port, with a warning when the value was non-empty. --- extensions/notification/deliveryWorker/probeConfig.js | 8 ++++---- tests/unit/notification/DeliveryWorker.spec.js | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/extensions/notification/deliveryWorker/probeConfig.js b/extensions/notification/deliveryWorker/probeConfig.js index 363675d91..441dfc18c 100644 --- a/extensions/notification/deliveryWorker/probeConfig.js +++ b/extensions/notification/deliveryWorker/probeConfig.js @@ -2,7 +2,7 @@ // 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_PROBE_PORT_ENV = 'DELIVERY_PROBE_PORT'; +const DELIVERY_POOL_PROBE_PORT_ENV = 'DELIVERY_POOL_PROBE_PORT'; const MAX_PORT = 65535; @@ -20,7 +20,7 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { if (!probeServer) { return undefined; } - const rawPort = (env || {})[DELIVERY_PROBE_PORT_ENV]; + const rawPort = (env || {})[DELIVERY_POOL_PROBE_PORT_ENV]; if (rawPort === undefined || `${rawPort}`.trim() === '') { return probeServer; } @@ -30,7 +30,7 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { if (logger) { logger.warn('ignoring invalid probe server port from the environment', { method: 'resolveProbeServerConfig', - envVar: DELIVERY_PROBE_PORT_ENV, + envVar: DELIVERY_POOL_PROBE_PORT_ENV, value: rawPort, port: probeServer.port, }); @@ -41,6 +41,6 @@ function resolveProbeServerConfig(deliveryPoolConfig, env, logger) { } module.exports = { - DELIVERY_PROBE_PORT_ENV, + DELIVERY_POOL_PROBE_PORT_ENV, resolveProbeServerConfig, }; diff --git a/tests/unit/notification/DeliveryWorker.spec.js b/tests/unit/notification/DeliveryWorker.spec.js index bd7c829a6..3fe9e00b0 100644 --- a/tests/unit/notification/DeliveryWorker.spec.js +++ b/tests/unit/notification/DeliveryWorker.spec.js @@ -6,7 +6,7 @@ const FakeLogger = require('../../utils/fakeLogger'); const DeliveryWorker = require( '../../../extensions/notification/deliveryWorker/DeliveryWorker'); -const { DELIVERY_PROBE_PORT_ENV, resolveProbeServerConfig } = require( +const { DELIVERY_POOL_PROBE_PORT_ENV, resolveProbeServerConfig } = require( '../../../extensions/notification/deliveryWorker/probeConfig'); const DELIVERED_METRIC = 's3_notification_delivery_worker_delivered_total'; @@ -316,7 +316,7 @@ describe('notification DeliveryWorker', () => { it('should let the environment give this worker its own port', () => { const resolved = resolveProbeServerConfig(withProbe, - { [DELIVERY_PROBE_PORT_ENV]: '8902' }); + { [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 @@ -327,7 +327,7 @@ describe('notification DeliveryWorker', () => { ['notaport', '0', '70000', '8900abc'].forEach(value => { const logger = { ...FakeLogger, warn: sinon.stub() }; const resolved = resolveProbeServerConfig(withProbe, - { [DELIVERY_PROBE_PORT_ENV]: value }, logger); + { [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); @@ -338,7 +338,7 @@ describe('notification DeliveryWorker', () => { ['', ' '].forEach(value => { const logger = { ...FakeLogger, warn: sinon.stub() }; const resolved = resolveProbeServerConfig(withProbe, - { [DELIVERY_PROBE_PORT_ENV]: value }, logger); + { [DELIVERY_POOL_PROBE_PORT_ENV]: value }, logger); assert.strictEqual(resolved, probeServer, `for value "${value}"`); assert(logger.warn.notCalled, `unexpected warning for value "${value}"`); });