KafkaNotificationDestination.js

const NotificationDestination = require('./NotificationDestination');
const KafkaProducer = require('./KafkaProducer');
const { ZenkoMetrics } = require('arsenal').metrics;

const notificationSize = ZenkoMetrics.createCounter({
    name: 's3_notification_queue_processor_notification_size_total',
    help: 'Total size of the notifications that were sent',
    labelNames: ['target'],
});

const deliveryLag = ZenkoMetrics.createHistogram({
    name: 's3_notification_queue_processor_delivery_delay_seconds',
    help: 'Difference between the time a notification is sent and when it gets delivered',
    labelNames: ['target', 'status'],
    buckets: [0.001, 0.01, 1, 10, 100, 1000, 10000],
});

function onNotificationSent(target, status, messages, delay) {
    if (status === 'success') {
        // Update total size of notifications sent
        const messageSize = new TextEncoder().encode(JSON.stringify(messages)).length;
        notificationSize.inc({
            target,
        }, messageSize);
    }
    if (delay) {
        // update notification delivery delay
        deliveryLag.observe({
            target,
            status,
        }, delay);
    }
}

class KafkaNotificationDestination extends NotificationDestination {
    /**
     * @constructor
     * @param {Object} params - constructor params
     * @param {Object} params.destConfig - destination-specific
     *   configuration object
     * @param {Logger} params.logger - logger object
     */
    constructor(params) {
        super(params);
        this._notificationProducer = null;
    }

    _setupProducer(done) {
        const { host, port, topic, pollIntervalMs, auth, requiredAcks, compressionType } = this._destinationConfig;
        let kafkaHost = host;
        if (port) {
            kafkaHost = `${host}:${port}`;
        }
        const producer = new KafkaProducer({
            kafka: { hosts: kafkaHost },
            topic,
            pollIntervalMs,
            auth,
            compressionType,
            requiredAcks,
        });
        producer.once('error', done);
        producer.once('ready', () => {
            producer.removeAllListeners('error');
            producer.on('error', err => {
                this._log.error('error from kafka producer', {
                    topic,
                    error: err,
                });
            });
            this._notificationProducer = producer;
            done();
        });
    }

    /**
     * Init/setup/authenticate the notification client
     *
     * @param {function} done - callback
     * @return {undefined}
     */
    init(done) {
        this._setupProducer(err => {
            if (err) {
                this._log.info('error setting up kafka notif destination',
                    { error: err.message });
                done(err);
            } else {
                done();
            }
        });
    }

    /**
     * Process entry in the sub-class and send it
     *
     * @param {Object[]} messages - message to be sent
     * @param {function} done - callback when delivery report has been received
     * @return {undefined}
     */
    send(messages, done) {
        const starTime = Date.now();
        // Trust boundary: strip any trace headers before producing to the
        // external customer Kafka destination. There's no ingress-layer
        // strip on the Kafka protocol (unlike HTTP via nginx), so the only
        // place to drop headers is at the producer. Keeps everything else
        // on the message intact.
        const safeMessages = Array.isArray(messages)
            ? messages.map(m => {
                if (m && m.headers) {
                    // eslint-disable-next-line no-unused-vars
                    const { headers, ...rest } = m;
                    return rest;
                }
                return m;
            })
            : messages;
        this._notificationProducer.send(safeMessages, error => {
            if (error) {
                const { host, topic } = this._destinationConfig;
                this._log.error('error in message delivery to external Kafka destination', {
                    method: 'KafkaNotificationDestination.send',
                    host,
                    topic,
                    error: error.message,
                });
                onNotificationSent(
                    this._destinationConfig.resource,
                    'failure',
                    messages,
                    null,
                );
                return done();
            }
            const delay = (Date.now() - starTime) / 1000;
            onNotificationSent(
                this._destinationConfig.resource,
                'success',
                messages,
                delay,
            );
            return done();
        });
    }

    /**
     * Stop the notification client
     *
     * @param {function} done - callback
     * @return {undefined}
     */
    stop(done) {
        if (!this._notificationProducer) {
            return setImmediate(done);
        }
        return this._notificationProducer.close(done);
    }
}

module.exports = KafkaNotificationDestination;