From 56633804ddcfcffe6c2bd324823d84f42f85a188 Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Thu, 22 Dec 2022 11:11:28 +0100 Subject: [PATCH 1/3] Gracefully handle client-level (vs. response-level) errors Signed-off-by: Eric Peterson --- .changeset/search-mele-kalikimaka.md | 5 ++++ .../ElasticSearchSearchEngineIndexer.test.ts | 29 +++++++++++++++++++ .../ElasticSearchSearchEngineIndexer.ts | 11 +++++++ 3 files changed, 45 insertions(+) create mode 100644 .changeset/search-mele-kalikimaka.md diff --git a/.changeset/search-mele-kalikimaka.md b/.changeset/search-mele-kalikimaka.md new file mode 100644 index 0000000000..4f1faff02f --- /dev/null +++ b/.changeset/search-mele-kalikimaka.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-search-backend-module-elasticsearch': patch +--- + +Fixed a bug that could cause the backstage backend to unexpectedly terminate when client errors were encountered during the indexing process. diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts index 92193976b4..94b59fcf42 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts @@ -250,4 +250,33 @@ describe('ElasticSearchSearchEngineIndexer', () => { // Final deletion shouldn't be called. expect(deleteSpy).not.toHaveBeenCalled(); }); + + it('handles bulk client rejection', async () => { + // Given an ES client wrapper that rejects an error + const expectedError = new Error('HTTP Timeout'); + const mockClientWrapper = ElasticSearchClientWrapper.fromClientOptions({ + node: 'http://localhost:9200', + Connection: mock.getConnection(), + }); + mockClientWrapper.bulk = jest.fn().mockRejectedValue(expectedError); + + // And a search engine indexer that uses that client wrapper + indexer = new ElasticSearchSearchEngineIndexer({ + type: 'some-type', + indexPrefix: '', + indexSeparator: '-index__', + alias: 'some-type-index__search', + logger: getVoidLogger(), + elasticSearchClientWrapper: mockClientWrapper, + batchSize: 1000, + }); + + // When the indexer is run in the test pipeline + const { error } = await TestPipeline.fromIndexer(indexer) + .withDocuments([{ title: 'a', location: 'a', text: '/a' }]) + .execute(); + + // Then the pipeline should have received the expected error + expect(error).toBe(expectedError); + }); }); diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts index 29f20ea650..3f842976e5 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts @@ -60,6 +60,7 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { private readonly sourceStream: Readable; private readonly elasticSearchClientWrapper: ElasticSearchClientWrapper; private bulkResult: Promise; + private bulkClientError?: Error; constructor(options: ElasticSearchSearchEngineIndexerOptions) { super({ batchSize: options.batchSize }); @@ -94,6 +95,11 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { }, refreshOnCompletion: that.indexName, }); + + // Safely catch errors thrown by the bulk helper client, e.g. HTTP timeouts + this.bulkResult.catch(e => { + this.bulkClientError = e; + }); } async initialize(): Promise { @@ -197,6 +203,11 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { * backpressure in other parts of the indexing pipeline. */ private isReady(): Promise { + // Early exit if the underlying ES client encountered an error. + if (this.bulkClientError) { + return Promise.reject(this.bulkClientError); + } + return new Promise(resolve => { const interval = setInterval(() => { if (this.received === this.processed) { From aa33a0689448a7e98185068f4c30aa7ecd0cc6d1 Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Thu, 22 Dec 2022 12:49:09 +0100 Subject: [PATCH 2/3] Optimize indexer throughput Signed-off-by: Eric Peterson --- .changeset/search-pressure-back.md | 5 +++++ .../engines/ElasticSearchSearchEngineIndexer.ts | 15 ++++++++++++--- 2 files changed, 17 insertions(+), 3 deletions(-) create mode 100644 .changeset/search-pressure-back.md diff --git a/.changeset/search-pressure-back.md b/.changeset/search-pressure-back.md new file mode 100644 index 0000000000..3dd36456a7 --- /dev/null +++ b/.changeset/search-pressure-back.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-search-backend-module-elasticsearch': patch +--- + +Improved index throughput by optimizing when and how many documents were made available to the bulk client. diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts index 3f842976e5..d1534c53a7 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts @@ -45,7 +45,6 @@ function duration(startTimestamp: [number, number]): string { * @public */ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { - private received: number = 0; private processed: number = 0; private removableIndices: string[] = []; @@ -59,11 +58,13 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { private readonly logger: Logger; private readonly sourceStream: Readable; private readonly elasticSearchClientWrapper: ElasticSearchClientWrapper; + private configuredBatchSize: number; private bulkResult: Promise; private bulkClientError?: Error; constructor(options: ElasticSearchSearchEngineIndexerOptions) { super({ batchSize: options.batchSize }); + this.configuredBatchSize = options.batchSize; this.logger = options.logger.child({ documentType: options.type }); this.startTimestamp = process.hrtime(); this.type = options.type; @@ -121,7 +122,6 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { async index(documents: IndexableDocument[]): Promise { await this.isReady(); documents.forEach(document => { - this.received++; this.sourceStream.push(document); }); } @@ -208,9 +208,18 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { return Promise.reject(this.bulkClientError); } + // Optimization: if the stream that ES reads from has fewer docs queued + // than the configured batch size, continue early to allow more docs to be + // queued + if (this.sourceStream.readableLength < this.configuredBatchSize) { + return Promise.resolve(); + } + + // Otherwise, continue periodically checking the stream queue to see if + // ES has consumed the documents and continue when it's ready for more. return new Promise(resolve => { const interval = setInterval(() => { - if (this.received === this.processed) { + if (this.sourceStream.readableLength < this.configuredBatchSize) { clearInterval(interval); resolve(); } From 1e1a9fe979802f96160e54e0b1bd1271dae156df Mon Sep 17 00:00:00 2001 From: Eric Peterson Date: Thu, 22 Dec 2022 14:09:49 +0100 Subject: [PATCH 3/3] Prevent the readiness check from looping endlessly Signed-off-by: Eric Peterson --- .changeset/search-break-loop.md | 5 +++++ .../ElasticSearchSearchEngineIndexer.ts | 20 +++++++++++++++++-- 2 files changed, 23 insertions(+), 2 deletions(-) create mode 100644 .changeset/search-break-loop.md diff --git a/.changeset/search-break-loop.md b/.changeset/search-break-loop.md new file mode 100644 index 0000000000..ce65645900 --- /dev/null +++ b/.changeset/search-break-loop.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-search-backend-module-elasticsearch': patch +--- + +Fixed a bug that could cause an indexing process to silently fail, timeout, and accumulate stale indices. diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts index d1534c53a7..a9d15c421e 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts @@ -217,11 +217,27 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { // Otherwise, continue periodically checking the stream queue to see if // ES has consumed the documents and continue when it's ready for more. - return new Promise(resolve => { + return new Promise((isReady, abort) => { + let streamLengthChecks = 0; const interval = setInterval(() => { + streamLengthChecks++; + if (this.sourceStream.readableLength < this.configuredBatchSize) { clearInterval(interval); - resolve(); + isReady(); + } + + // Do not allow this interval to loop endlessly; anything longer than 5 + // minutes likely indicates an unrecoverable error in ES; direct the + // user to inspect ES logs for more clues and abort in order to allow + // the index to be cleaned up. + if (streamLengthChecks >= 6000) { + clearInterval(interval); + abort( + new Error( + 'Exceeded 5 minutes waiting for elastic to be ready to accept more documents. Check the elastic logs for possible problems.', + ), + ); } }, 50); });