diff --git a/bin/ingestion.js b/bin/ingestion.js index d7ea6a16c6..6bd785fbf9 100644 --- a/bin/ingestion.js +++ b/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 => { diff --git a/extensions/ingestion/IngestionConfigValidator.js b/extensions/ingestion/IngestionConfigValidator.js index b1a9b85caa..7ba9951048 100644 --- a/extensions/ingestion/IngestionConfigValidator.js +++ b/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) { diff --git a/extensions/mongoProcessor/MongoProcessorConfigValidator.js b/extensions/mongoProcessor/MongoProcessorConfigValidator.js index 3479357a1b..8bb075f9c4 100644 --- a/extensions/mongoProcessor/MongoProcessorConfigValidator.js +++ b/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) { diff --git a/extensions/mongoProcessor/mongoProcessorTask.js b/extensions/mongoProcessor/mongoProcessorTask.js index 696e013252..e0ccf02581 100644 --- a/extensions/mongoProcessor/mongoProcessorTask.js +++ b/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); diff --git a/lib/BackbeatProducer.js b/lib/BackbeatProducer.js index 493924605e..807de84adf 100644 --- a/lib/BackbeatProducer.js +++ b/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,6 +127,7 @@ class BackbeatProducer extends EventEmitter { maxRequestSize, compressionType, requiredAcks, + producerParams, } = joiResult; this._kafkaHosts = kafka.hosts; this._topic = topic && withTopicPrefix(topic); @@ -132,6 +135,7 @@ class BackbeatProducer extends EventEmitter { this._maxRequestSize = maxRequestSize; this._compressionType = compressionType; this._requiredAcks = requiredAcks; + this._producerParams = producerParams; } connect() { diff --git a/lib/config.joi.js b/lib/config.joi.js index 2ea7a94169..378a5d7657 100644 --- a/lib/config.joi.js +++ b/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(), diff --git a/lib/config/configItems.joi.js b/lib/config/configItems.joi.js index 4efb886b9a..5e333b08c4 100644 --- a/lib/config/configItems.joi.js +++ b/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, diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index c9e1407c83..0b917cb8f0 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/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, }); diff --git a/tests/unit/backbeatProducer.js b/tests/unit/backbeatProducer.js index 793e2c8564..119e219753 100644 --- a/tests/unit/backbeatProducer.js +++ b/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 = ''; }); diff --git a/tests/unit/ingestion/IngestionConfigValidator.spec.js b/tests/unit/ingestion/IngestionConfigValidator.spec.js new file mode 100644 index 0000000000..ae2ef4f4b0 --- /dev/null +++ b/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'); + }); +}); diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index 270700b5db..ad10e19049 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/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); + }); + }); }); diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js new file mode 100644 index 0000000000..dca9309349 --- /dev/null +++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js @@ -0,0 +1,35 @@ +'use strict'; + +const assert = require('assert'); +const configValidator = require('../../../extensions/mongoProcessor/MongoProcessorConfigValidator'); + +const baseConfig = { + topic: 'backbeat-ingestion', + groupId: 'backbeat-ingestion-group', + probeServer: { port: 4000 }, +}; + +describe('MongoProcessorConfigValidator log override', () => { + it('should pass through log config when set', () => { + const validated = configValidator({}, { + ...baseConfig, + log: { logLevel: 'warn', dumpLevel: 'error' }, + }); + assert.deepStrictEqual(validated.log, { logLevel: 'warn', dumpLevel: 'error' }); + }); + + it('should leave log undefined when not set, deferring to global config.log', () => { + const validated = configValidator({}, baseConfig); + assert.strictEqual(validated.log, undefined); + }); + + it('should reject a partial log config with missing dumpLevel', () => { + let err; + try { + configValidator({}, { ...baseConfig, log: { logLevel: 'warn' } }); + } catch (e) { + err = e; + } + assert(err, 'expected configValidator to throw on partial log config'); + }); +});