Support per extension log level

bin/ingestion.js

@@ -33,8 +33,9 @@ const { connectionString, autoCreateNamespace, retries } = zkConfig;
 const RESUME_NODE = 'scheduledResume';
 
 const log = new werelogs.Logger('Backbeat:IngestionPopulator');
-werelogs.configure({ level: config.log.logLevel,
-    dump: config.log.dumpLevel });
+const ingestionLogConfig = ingestionExtConfigs.log ?? config.log;
+werelogs.configure({ level: ingestionLogConfig.logLevel,
+    dump: ingestionLogConfig.dumpLevel });
 
 let scheduler;
 let ingestionPopulator;
@@ -47,7 +48,8 @@ function getIngestionZkPath() {
 
 function queueBatch(ingestionPopulator, log) {
     log.debug('start queueing ingestion batch');
-    const maxRead = qpConfig.batchMaxRead;
+    // Extension-level batchMaxRead overrides the shared queuePopulator value.
+    const maxRead = ingestionExtConfigs.batchMaxRead ?? qpConfig.batchMaxRead;
     // apply updates to Ingestion Readers
     ingestionPopulator.applyUpdates();
     ingestionPopulator.processLogEntries({ maxRead }, err => {

extensions/ingestion/IngestionConfigValidator.js

@@ -1,5 +1,5 @@
 const joi = require('joi');
-const { probeServerJoi } = require('../../lib/config/configItems.joi');
+const { probeServerJoi, logJoiOptional } = require('../../lib/config/configItems.joi');
 
 const joiSchema = joi.object({
     auth: joi.object({
@@ -10,12 +10,15 @@ const joiSchema = joi.object({
     zookeeperPath: joi.string().required(),
     cronRule: joi.string().default('*/5 * * * * *'),
     maxParallelReaders: joi.number().greater(0).default(5),
+    batchMaxRead: joi.number().greater(0).optional(),
     sources: joi.array().required(),
     probeServer: probeServerJoi.default(),
     circuitBreaker: joi.object().optional(),
     processor: joi.object({
         circuitBreaker: joi.object().optional(),
     }).optional(),
+    producerParams: joi.object().unknown(true).default({}),
+    log: logJoiOptional,
 });
 
 function configValidator(backbeatConfig, extConfig) {

extensions/mongoProcessor/MongoProcessorConfigValidator.js

@@ -1,5 +1,5 @@
 const joi = require('joi');
-const { retryParamsJoi, probeServerJoi } = require('../../lib/config/configItems.joi');
+const { retryParamsJoi, probeServerJoi, logJoiOptional } = require('../../lib/config/configItems.joi');
 
 const { MAX_QUEUED_DEFAULT }  = require('../../lib/constants').backbeatConsumer;
 
@@ -11,6 +11,7 @@ const joiSchema = joi.object({
     maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
     probeServer: probeServerJoi.default(),
     circuitBreaker: joi.object().optional(),
+    log: logJoiOptional,
 });
 
 function configValidator(backbeatConfig, extConfig) {

extensions/mongoProcessor/mongoProcessorTask.js

@@ -23,8 +23,9 @@ const mongoProcessorConfig = config.extensions.mongoProcessor;
 const mongoClientConfig = config.queuePopulator.mongo;
 
 const log = new werelogs.Logger('Backbeat:MongoProcessor:task');
-werelogs.configure({ level: config.log.logLevel,
-    dump: config.log.dumpLevel });
+const mongoProcessorLogConfig = mongoProcessorConfig.log ?? config.log;
+werelogs.configure({ level: mongoProcessorLogConfig.logLevel,
+    dump: mongoProcessorLogConfig.dumpLevel });
 
 const mqp = new MongoQueueProcessor(kafkaConfig, mongoProcessorConfig,
     mongoClientConfig, mConfig);

lib/BackbeatProducer.js

@@ -58,6 +58,7 @@ class BackbeatProducer extends EventEmitter {
                 maxRequestSize: joi.number().default(KAFKA_PRODUCER_MESSAGE_MAX_BYTES),
                 compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
                 requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
+                producerParams: joi.object().unknown(true).default({}),
             }
         );
     }
@@ -75,7 +76,8 @@ class BackbeatProducer extends EventEmitter {
     }
 
     get producerConfig() {
-        const producerParams = {
+        const config = {
+            ...this._producerParams,
             'metadata.broker.list': this._kafkaHosts,
             'message.max.bytes': this._maxRequestSize,
             'dr_cb': true,
@@ -84,10 +86,10 @@ class BackbeatProducer extends EventEmitter {
         };
 
         if (process.env.RDKAFKA_DEBUG_LOGS) {
-            producerParams.debug = process.env.RDKAFKA_DEBUG_LOGS;
+            config.debug = process.env.RDKAFKA_DEBUG_LOGS;
         }
 
-        return producerParams;
+        return config;
     }
 
     get topicConfig() {
@@ -125,13 +127,15 @@ class BackbeatProducer extends EventEmitter {
             maxRequestSize,
             compressionType,
             requiredAcks,
+            producerParams,
         } = joiResult;
         this._kafkaHosts = kafka.hosts;
         this._topic = topic && withTopicPrefix(topic);
         this._pollIntervalMs = pollIntervalMs;
         this._maxRequestSize = maxRequestSize;
         this._compressionType = compressionType;
         this._requiredAcks = requiredAcks;
+        this._producerParams = producerParams;
     }
 
     connect() {

lib/config.joi.js

@@ -35,6 +35,7 @@ const joiSchema = joi.object({
         site: joi.string(),
         compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
         requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
+        producerParams: joi.object().unknown(true).default({}),
     },
     transport: transportJoi,
     s3: hostPortJoi.optional(),

lib/config/configItems.joi.js

@@ -51,6 +51,16 @@ const logJoi =
               dumpLevel: 'error',
           });
 
+// logJoi with no default :
+// Callers fall back to the global log config when this one is not configured
+const logJoiOptional =
+          joi.object({
+              logLevel: joi.alternatives()
+                  .try('error', 'warn', 'info', 'debug', 'trace').required(),
+              dumpLevel: joi.alternatives()
+                  .try('error', 'warn', 'info', 'debug', 'trace').required(),
+          }).optional();
+
 const adminCredsJoi = joi.object()
           .min(1)
           .pattern(/^[A-Za-z0-9]{20}$/, joi.string());
@@ -168,6 +178,7 @@ module.exports = {
     transportJoi,
     bootstrapListJoi,
     logJoi,
+    logJoiOptional,
     adminCredsJoi,
     authJoi,
     inheritedAuthJoi,

lib/queuePopulator/IngestionPopulator.js

@@ -173,6 +173,10 @@ class IngestionPopulator {
             maxRequestSize: this.kafkaConfig.maxRequestSize,
             compressionType: this.kafkaConfig.compressionType,
             requiredAcks: this.kafkaConfig.requiredAcks,
+            producerParams: {
+                ...this.kafkaConfig.producerParams,
+                ...this.ingestionConfig.producerParams, // Extension params override global params
+            },
             topic,
             pollIntervalMs: POLL_INTERVAL_MS,
         });

tests/unit/backbeatProducer.js

@@ -111,6 +111,39 @@ describe('backbeatProducer', () => {
             'custom-topic', [{ key: 'foo', message: 'bar' }], () => {});
     });
 
+    describe('producerParams', () => {
+        it('should include extra producerParams in producerConfig', () => {
+            const producer = new BackbeatProducer({
+                kafka,
+                producerParams: {
+                    'queue.buffering.max.kbytes': 1048576,
+                    'queue.buffering.max.messages': 200000,
+                },
+            });
+            const config = producer.producerConfig;
+            assert.strictEqual(config['queue.buffering.max.kbytes'], 1048576);
+            assert.strictEqual(config['queue.buffering.max.messages'], 200000);
+        });
+
+        it('should not let producerParams override critical built-in params', () => {
+            const producer = new BackbeatProducer({
+                kafka,
+                producerParams: {
+                    'metadata.broker.list': 'attacker:9092',
+                    'dr_cb': false,
+                },
+            });
+            const config = producer.producerConfig;
+            assert.strictEqual(config['metadata.broker.list'], kafka.hosts);
+            assert.strictEqual(config['dr_cb'], true);
+        });
+
+        it('should default to empty producerParams when not provided', () => {
+            const producer = new BackbeatProducer({ kafka });
+            assert.deepStrictEqual(producer._producerParams, {});
+        });
+    });
+
     afterEach(() => {
         process.env.KAFKA_TOPIC_PREFIX = '';
     });

tests/unit/ingestion/IngestionConfigValidator.spec.js

@@ -0,0 +1,63 @@
+'use strict';
+
+const assert = require('assert');
+const config = require('../../../lib/Config');
+const configValidator = require('../../../extensions/ingestion/IngestionConfigValidator');
+
+const baseExtConfig = {
+    auth: { type: 'service', account: 'test-account' },
+    topic: 'backbeat-ingestion',
+    zookeeperPath: '/test',
+    sources: [],
+    probeServer: { port: 4000 },
+};
+
+const qpBatchMaxRead = config.queuePopulator.batchMaxRead;
+
+describe('IngestionConfigValidator log override', () => {
+    it('should pass through log config when set', () => {
+        const validated = configValidator({}, {
+            ...baseExtConfig,
+            log: { logLevel: 'debug', dumpLevel: 'error' },
+        });
+        assert.deepStrictEqual(validated.log, { logLevel: 'debug', dumpLevel: 'error' });
+    });
+
+    it('should leave log undefined when not set, deferring to global config.log', () => {
+        const validated = configValidator({}, baseExtConfig);
+        assert.strictEqual(validated.log, undefined);
+    });
+
+    it('should reject a partial log config with missing dumpLevel', () => {
+        let err;
+        try {
+            configValidator({}, { ...baseExtConfig, log: { logLevel: 'debug' } });
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on partial log config');
+    });
+});
+
+describe('IngestionConfigValidator batchMaxRead fallback', () => {
+    it('should override queuePopulator.batchMaxRead when set in extension config', () => {
+        const validated = configValidator({}, { ...baseExtConfig, batchMaxRead: 500 });
+        assert.strictEqual(validated.batchMaxRead, 500);
+        assert.notStrictEqual(validated.batchMaxRead, qpBatchMaxRead);
+    });
+
+    it('should leave batchMaxRead undefined when not set, allowing fallback to queuePopulator.batchMaxRead', () => {
+        const validated = configValidator({}, baseExtConfig);
+        assert.strictEqual(validated.batchMaxRead, undefined);
+    });
+
+    it('should reject a non-positive batchMaxRead', () => {
+        let err;
+        try {
+            configValidator({}, { ...baseExtConfig, batchMaxRead: 0 });
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on batchMaxRead: 0');
+    });
+});

tests/unit/ingestion/IngestionPopulator.js

@@ -7,6 +7,7 @@ const config = require('../../../lib/Config');
 const IngestionPopulator =
     require('../../../lib/queuePopulator/IngestionPopulator');
 const IngestionReader = require('../../../lib/queuePopulator/IngestionReader');
+const BackbeatProducer = require('../../../lib/BackbeatProducer');
 const fakeLogger = require('../../utils/fakeLogger');
 
 const zkConfig = config.zookeeper;
@@ -354,4 +355,94 @@ describe('Ingestion Populator', () => {
             });
         });
     });
+
+    describe('_setupProducer producerParams merge', () => {
+        let capturedProducerParams;
+
+        beforeEach(() => {
+            sinon.stub(BackbeatProducer.prototype, 'setFromConfig').callsFake(function (cfg) {
+                capturedProducerParams = cfg.producerParams;
+                // Minimal instance state so producerConfig getter doesn't throw.
+                this._kafkaHosts = cfg.kafka.hosts;
+                this._topic = null;
+                this._pollIntervalMs = 2000;
+                this._maxRequestSize = 5000020;
+                this._compressionType = 'Zstd';
+                this._requiredAcks = -1;
+                this._producerParams = cfg.producerParams || {};
+            });
+        });
+
+        afterEach(() => {
+            sinon.restore();
+            capturedProducerParams = undefined;
+        });
+
+        it('should pass merged producerParams : extension overrides global', () => {
+            const globalParams = {
+                'queue.buffering.max.kbytes': 1048576,
+                'queue.buffering.max.ms': 100,
+            };
+            const extParams = {
+                'queue.buffering.max.messages': 200000,
+                'queue.buffering.max.ms': 500,
+            };
+
+            const populator = new IngestionPopulator(
+                null,
+                zkConfig,
+                { ...kafkaConfig, producerParams: globalParams },
+                qpConfig,
+                mConfig,
+                rConfig,
+                { ...ingestionConfig, producerParams: extParams },
+                s3Config
+            );
+
+            populator._setupProducer(() => {});
+
+            assert.strictEqual(capturedProducerParams['queue.buffering.max.kbytes'], 1048576);
+            assert.strictEqual(capturedProducerParams['queue.buffering.max.messages'], 200000);
+            assert.strictEqual(capturedProducerParams['queue.buffering.max.ms'], 500,
+                'extension producerParams should override global kafka.producerParams');
+        });
+
+        it('should work when only global kafka.producerParams are set', () => {
+            const globalParams = { 'queue.buffering.max.kbytes': 524288 };
+
+            const populator = new IngestionPopulator(
+                null,
+                zkConfig,
+                { ...kafkaConfig, producerParams: globalParams },
+                qpConfig,
+                mConfig,
+                rConfig,
+                ingestionConfig,
+                s3Config
+            );
+
+            populator._setupProducer(() => {});
+
+            assert.strictEqual(capturedProducerParams['queue.buffering.max.kbytes'], 524288);
+        });
+
+        it('should work when only extension producerParams are set', () => {
+            const extParams = { 'queue.buffering.max.messages': 100000 };
+
+            const populator = new IngestionPopulator(
+                null,
+                zkConfig,
+                kafkaConfig,
+                qpConfig,
+                mConfig,
+                rConfig,
+                { ...ingestionConfig, producerParams: extParams },
+                s3Config
+            );
+
+            populator._setupProducer(() => {});
+
+            assert.strictEqual(capturedProducerParams['queue.buffering.max.messages'], 100000);
+        });
+    });
 });