Merge pull request #15386 from backstage/search/elastic-stability-improvements

[Search] Improve stability in the ES search indexer

.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.

.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.

.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.

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);
+  });
 });

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<any>;
+  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<void> {
@@ -115,7 +122,6 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
   async index(documents: IndexableDocument[]): Promise<void> {
     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<void> {
-    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);
     });