diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index ec2849a40..84cf17440 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -613,13 +613,26 @@ class IngestionPopulator { const newReaders = this.logReadersUpdate; this.logReadersUpdate = []; async.each(newReaders, (logReader, cb) => logReader.setup(err => { + const zenkoBucket = logReader.getTargetZenkoBucketName(); + // The reader is in neither list while its setup is in flight, so + // its source may have been removed or replaced in the meantime. It + // must then neither be activated nor retried. + if (this._ingestionSources[zenkoBucket] !== logReader) { + return cb(); + } if (err) { // if setup fails for a log reader, don't add it to `logReaders` // log the error and continue setting up others - this.log.fatal('error setting up log reader', { + this.log.error('error setting up log reader, retrying later', { method: 'IngestionPopulator._setupUpdatedReaders', + zenkoBucket, + location: logReader.getLocationConstraint(), error: err, }); + // Setup failures are usually transient (source unreachable, + // invalid credentials...): queue the reader again so that it + // is retried on the next cycle. + this.logReadersUpdate.push(logReader); } else { this.logReaders.push(logReader); } @@ -721,6 +734,12 @@ class IngestionPopulator { * @return {undefined} */ _closeLogState(key) { + // Unregister the source first: a reader whose setup is in flight is in + // neither list, and leaving its entry here would let it register itself + // once the setup completes. + const reader = this._ingestionSources[key]; + delete this._ingestionSources[key]; + if (this._checkAndRemoveLogReader(key, this.logReadersUpdate)) { // if removed from `this.logReadersUpdate`, zookeeper setup has not // been performed yet @@ -728,7 +747,6 @@ class IngestionPopulator { method: 'IngestionPopulator._closeLogState', bucket: key, }); - delete this._ingestionSources[key]; } if (this._checkAndRemoveLogReader(key, this.logReaders)) { // if removed from `this.logReaders`, we must cleanup zookeeper @@ -739,8 +757,6 @@ class IngestionPopulator { }); // we should first validate this reader is not currently processing // a batch of entries before removing zookeeper state - const reader = this._ingestionSources[key]; - delete this._ingestionSources[key]; const path = `${this.ingestionConfig.zookeeperPath}/${key}`; this._removeReaderState(reader, path); } diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index ad10e1904..9abb717cb 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/tests/unit/ingestion/IngestionPopulator.js @@ -278,6 +278,322 @@ describe('Ingestion Populator', () => { }); }); + describe('_setupUpdatedReaders', () => { + const FAILING_BUCKET = 'failing-zenko-bucket'; + const WORKING_BUCKET = 'working-zenko-bucket'; + + /** + * @param {string} zenkoBucket - target zenko bucket of the reader + * @param {Error|null} setupError - error to fail `setup` with + * @return {object} the stubbed reader + */ + function createLogReaderMock(zenkoBucket, setupError) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + logReader.setup.yieldsAsync(setupError); + return logReader; + } + + beforeEach(() => { + ip.logReaders = []; + ip.logReadersUpdate = []; + }); + + it('should activate a log reader once its setup succeeds', done => { + const logReaderMock = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[WORKING_BUCKET] = logReaderMock; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, [logReaderMock]); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should queue a log reader again when its setup fails', done => { + const logReaderMock = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + ip._ingestionSources[FAILING_BUCKET] = logReaderMock; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, [logReaderMock]); + done(); + }); + }); + + it('should not queue a log reader again when its setup fails and ' + + 'its source is no longer configured', done => { + const logReaderMock = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + delete ip._ingestionSources[FAILING_BUCKET]; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should not queue a log reader again when its setup fails and ' + + 'its source has been registered with another reader', done => { + const staleReader = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + const currentReader = createLogReaderMock(FAILING_BUCKET, null); + ip._ingestionSources[FAILING_BUCKET] = currentReader; + ip.logReadersUpdate = [staleReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should not activate a log reader when its setup succeeds and ' + + 'its source is no longer configured', done => { + const logReaderMock = createLogReaderMock(WORKING_BUCKET, null); + delete ip._ingestionSources[WORKING_BUCKET]; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should not activate a log reader when its setup succeeds and ' + + 'its source has been registered with another reader', done => { + const staleReader = createLogReaderMock(WORKING_BUCKET, null); + const currentReader = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[WORKING_BUCKET] = currentReader; + ip.logReadersUpdate = [staleReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should keep setting up other log readers when one fails', done => { + const failingReader = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + const workingReader = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[FAILING_BUCKET] = failingReader; + ip._ingestionSources[WORKING_BUCKET] = workingReader; + ip.logReadersUpdate = [failingReader, workingReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, [workingReader]); + assert.deepStrictEqual(ip.logReadersUpdate, [failingReader]); + done(); + }); + }); + }); + + describe('removing a source while its setup is in flight', () => { + const REMOVED_BUCKET = 'removed-zenko-bucket'; + + // `IngestionPopulatorMock` stubs out `_closeLogState`, so a real + // populator is needed to exercise the removal path. + let populator; + + beforeEach(() => { + populator = new IngestionPopulator( + zkConfig, + kafkaConfig, + qpConfig, + mConfig, + rConfig, + ingestionConfig, + s3Config, + ); + }); + + /** + * Build a reader whose `setup` stays pending until it is released, to + * hold the reader in neither `logReaders` nor `logReadersUpdate`. + * + * @param {string} zenkoBucket - target zenko bucket of the reader + * @param {Error|null} setupError - error to fail `setup` with + * @return {object} the stubbed reader and the release function + */ + function createPendingLogReaderMock(zenkoBucket, setupError) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + + let setupCb = null; + logReader.setup.callsFake(cb => { + setupCb = cb; + }); + + return { logReader, release: () => setupCb(setupError) }; + } + + it('should not queue a log reader again when its source is removed ' + + 'while its setup fails', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, errors.InternalError); + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReadersUpdate, []); + done(); + }); + + populator._closeLogState(REMOVED_BUCKET); + assert.strictEqual(populator._ingestionSources[REMOVED_BUCKET], + undefined); + + release(); + }); + + it('should not activate a log reader when its source is removed ' + + 'while its setup succeeds', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, null); + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReaders, []); + assert.deepStrictEqual(populator.logReadersUpdate, []); + done(); + }); + + populator._closeLogState(REMOVED_BUCKET); + release(); + }); + + it('should not activate a log reader when its source is replaced ' + + 'while its setup succeeds', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, null); + const freshReader = sinon.createStubInstance(IngestionReader); + freshReader.getTargetZenkoBucketName.returns(REMOVED_BUCKET); + + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReaders, []); + assert.deepStrictEqual(populator.logReadersUpdate, []); + // the reader registered in the meantime is left untouched + assert.strictEqual(populator._ingestionSources[REMOVED_BUCKET], + freshReader); + done(); + }); + + // the source is removed, then configured again before the setup + // of the first reader completes + populator._closeLogState(REMOVED_BUCKET); + populator._ingestionSources[REMOVED_BUCKET] = freshReader; + + release(); + }); + }); + + describe('_closeLogState', () => { + const ACTIVE_BUCKET = 'active-zenko-bucket'; + const PENDING_BUCKET = 'pending-zenko-bucket'; + + // `IngestionPopulatorMock` stubs out `_closeLogState`, so a real + // populator is needed to exercise it. + let populator; + let removeReaderState; + + beforeEach(() => { + populator = new IngestionPopulator( + zkConfig, + kafkaConfig, + qpConfig, + mConfig, + rConfig, + ingestionConfig, + s3Config, + ); + // `_removeReaderState` polls the reader on a timer that would + // outlive the test + removeReaderState = sinon.stub(populator, '_removeReaderState'); + }); + + /** + * @param {string} zenkoBucket - target zenko bucket of the reader + * @return {object} the stubbed reader + */ + function createReaderMock(zenkoBucket) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + return logReader; + } + + it('should unregister a source whose reader is still pending setup', + () => { + const logReader = createReaderMock(PENDING_BUCKET); + populator._ingestionSources[PENDING_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._closeLogState(PENDING_BUCKET); + + assert.deepStrictEqual(populator.logReadersUpdate, []); + assert.strictEqual(populator._ingestionSources[PENDING_BUCKET], + undefined); + // zookeeper state is only created once the setup succeeded + assert.strictEqual(removeReaderState.called, false); + }); + + it('should unregister an active source and clean its zookeeper state', + () => { + const logReader = createReaderMock(ACTIVE_BUCKET); + populator._ingestionSources[ACTIVE_BUCKET] = logReader; + populator.logReaders = [logReader]; + + populator._closeLogState(ACTIVE_BUCKET); + + assert.deepStrictEqual(populator.logReaders, []); + assert.strictEqual(populator._ingestionSources[ACTIVE_BUCKET], + undefined); + // the reader is read before being unregistered, as + // `_removeReaderState` polls it until its batch completes + assert.strictEqual(removeReaderState.calledOnce, true); + assert.strictEqual(removeReaderState.firstCall.args[0], logReader); + }); + + it('should unregister a source whose setup is still in flight', () => { + const logReader = createReaderMock(PENDING_BUCKET); + populator._ingestionSources[PENDING_BUCKET] = logReader; + + populator._closeLogState(PENDING_BUCKET); + + assert.strictEqual(populator._ingestionSources[PENDING_BUCKET], + undefined); + assert.strictEqual(removeReaderState.called, false); + }); + + it('should not throw when the source is unknown', () => { + populator._closeLogState('never-configured-bucket'); + + assert.deepStrictEqual(populator._ingestionSources, {}); + assert.strictEqual(removeReaderState.called, false); + }); + }); + describe('_processLogReaderEntries', () => { it('should skip when previous batch currently in progress', () => { const logReaderMock = {