Retry the setup of an ingestion reader when it fails

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,14 +734,19 @@ 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
             this.log.debug('removed ingestion reader from logReadersUpdate', {
                 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);
         }

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 = {