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/.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/.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.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..a9d15c421e 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,10 +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; @@ -94,6 +96,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 { @@ -115,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); }); } @@ -197,11 +203,41 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { * backpressure in other parts of the indexing pipeline. */ private isReady(): Promise { - return new Promise(resolve => { + // Early exit if the underlying ES client encountered an error. + if (this.bulkClientError) { + 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((isReady, abort) => { + let streamLengthChecks = 0; const interval = setInterval(() => { - if (this.received === this.processed) { + 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); });