Merge pull request #14938 from backstage/search/resilient-indices

[Search] Preserve existing index if latest index run contains no documents
2022-12-15 15:24:11 +01:00
parent 3a958300e0 d75d73d33b
commit e2c798e31c
13 changed files with 199 additions and 10 deletions
@@ -0,0 +1,12 @@
+---
+'@backstage/plugin-search-backend-module-pg': minor
+---
+
+Added the option to pass a logger to `PgSearchEngine` during instantiation. You may do so as follows:
+
+```diff
+const searchEngine = await PgSearchEngine.fromConfig(env.config, {
+  database: env.database,
+ logger: env.logger,
+});
+```
@@ -0,0 +1,7 @@
+---
+'@backstage/plugin-search-backend-module-elasticsearch': minor
+'@backstage/plugin-search-backend-module-pg': minor
+'@backstage/plugin-search-backend-node': minor
+---
+
+The search engine now better handles the case when it receives 0 documents at index-time. Prior to this change, the indexer would replace any existing index with an empty index, effectively deleting it. Now instead, a warning is logged, and any existing index is left alone (preserving the index from the last successful indexing attempt).
@@ -254,6 +254,7 @@ export class ElasticSearchSearchEngine implements SearchEngine {

  async getIndexer(type: string) {
    const alias = this.constructSearchAlias(type);
+    const indexerLogger = this.logger.child({ documentType: type });

    const indexer = new ElasticSearchSearchEngineIndexer({
      type,
@@ -261,13 +262,13 @@ export class ElasticSearchSearchEngine implements SearchEngine {
      indexSeparator: this.indexSeparator,
      alias,
      elasticSearchClientWrapper: this.elasticSearchClientWrapper,
-      logger: this.logger,
+      logger: indexerLogger,
      batchSize: this.batchSize,
    });

    // Attempt cleanup upon failure.
    indexer.on('error', async e => {
-      this.logger.error(`Failed to index documents for type ${type}`, e);
+      indexerLogger.error(`Failed to index documents for type ${type}`, e);
      let cleanupError: Error | undefined;

      // In some cases, a failure may have occurred before the indexer was able
@@ -298,11 +299,13 @@ export class ElasticSearchSearchEngine implements SearchEngine {
      });

      if (cleanupError) {
-        this.logger.error(
+        indexerLogger.error(
          `Unable to clean up elastic index ${indexer.indexName}: ${cleanupError}`,
        );
      } else {
-        this.logger.info(`Removed partial, failed index ${indexer.indexName}`);
+        indexerLogger.info(
+          `Removed partial, failed index ${indexer.indexName}`,
+        );
      }
    });

@@ -176,6 +176,30 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    expect(deleteSpy).toHaveBeenCalled();
  });

+  it('handles when no documents are received', async () => {
+    await TestPipeline.fromIndexer(indexer).withDocuments([]).execute();
+
+    // Older indices should have been queried for.
+    expect(catSpy).toHaveBeenCalled();
+
+    // A new index should have been created.
+    expect(createSpy).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        path: expect.stringContaining('some-type-index__'),
+      }),
+    );
+
+    // No documents should have been sent
+    expect(bulkSpy).not.toHaveBeenCalled();
+
+    // Alias should not have been rotated.
+    expect(aliasesSpy).not.toHaveBeenCalled();
+
+    // Old index should not be cleaned up.
+    expect(deleteSpy).not.toHaveBeenCalled();
+  });
+
  it('handles bulk and batching during indexing', async () => {
    const documents = range(550).map(i => ({
      title: `Hello World ${i}`,
@@ -131,6 +131,24 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
    // Wait for the bulk helper to finish processing.
    const result = await this.bulkResult;

+    // Warn that no documents were indexed, early return so that alias swapping
+    // does not occur, and clean up the empty index we just created.
+    if (this.processed === 0) {
+      this.logger.warn(
+        `Index for ${this.type} was not ${
+          this.removableIndices.length ? 'replaced' : 'created'
+        }: indexer received 0 documents`,
+      );
+      try {
+        await this.elasticSearchClientWrapper.deleteIndex({
+          index: this.indexName,
+        });
+      } catch (error) {
+        this.logger.error(`Unable to clean up elastic index: ${error}`);
+      }
+      return;
+    }
+
    // Rotate main alias upon completion. Apply permanent secondary alias so
    // stale indices can be referenced for deletion in case initial attempt
    // fails. Allow errors to bubble up so that we can clean up the created index.
@@ -8,6 +8,7 @@ import { Config } from '@backstage/config';
 import { IndexableDocument } from '@backstage/plugin-search-common';
 import { IndexableResultSet } from '@backstage/plugin-search-common';
 import { Knex } from 'knex';
+import { Logger } from 'winston';
 import { PluginDatabaseManager } from '@backstage/backend-common';
 import { SearchEngine } from '@backstage/plugin-search-common';
 import { SearchQuery } from '@backstage/plugin-search-common';
@@ -84,11 +85,12 @@ export interface DocumentResultRow {
 // @public (undocumented)
 export class PgSearchEngine implements SearchEngine {
  // @deprecated
-  constructor(databaseStore: DatabaseStore, config: Config);
+  constructor(databaseStore: DatabaseStore, config: Config, logger?: Logger);
  // @deprecated (undocumented)
  static from(options: {
    database: PluginDatabaseManager;
    config: Config;
+    logger?: Logger;
  }): Promise<PgSearchEngine>;
  // (undocumented)
  static fromConfig(
@@ -126,6 +128,7 @@ export type PgSearchEngineIndexerOptions = {
  batchSize: number;
  type: string;
  databaseStore: DatabaseStore;
+  logger?: Logger;
 };

 // @public
@@ -144,6 +147,7 @@ export type PgSearchHighlightOptions = {
 // @public
 export type PgSearchOptions = {
  database: PluginDatabaseManager;
+  logger?: Logger;
 };

 // @public (undocumented)
@@ -29,7 +29,8 @@
    "@backstage/plugin-search-common": "workspace:^",
    "knex": "^2.0.0",
    "lodash": "^4.17.21",
-    "uuid": "^8.3.2"
+    "uuid": "^8.3.2",
+    "winston": "^3.2.1"
  },
  "devDependencies": {
    "@backstage/backend-test-utils": "workspace:^",
@@ -28,6 +28,7 @@ import {
  PgSearchQuery,
 } from '../database';
 import { v4 as uuid } from 'uuid';
+import { Logger } from 'winston';
 import { Config } from '@backstage/config';

 /**
@@ -62,6 +63,7 @@ export type PgSearchQueryTranslator = (
 */
 export type PgSearchOptions = {
  database: PluginDatabaseManager;
+  logger?: Logger;
 };

 /**
@@ -82,12 +84,17 @@ export type PgSearchHighlightOptions = {

 /** @public */
 export class PgSearchEngine implements SearchEngine {
+  private readonly logger?: Logger;
  private readonly highlightOptions: PgSearchHighlightOptions;

  /**
   * @deprecated This will be marked as private in a future release, please us fromConfig instead
   */
-  constructor(private readonly databaseStore: DatabaseStore, config: Config) {
+  constructor(
+    private readonly databaseStore: DatabaseStore,
+    config: Config,
+    logger?: Logger,
+  ) {
    const uuidTag = uuid();
    const highlightConfig = config.getOptionalConfig(
      'search.pg.highlightOptions',
@@ -107,6 +114,7 @@ export class PgSearchEngine implements SearchEngine {
        highlightConfig?.getOptionalString('fragmentDelimiter') ?? ' ... ',
    };
    this.highlightOptions = highlightOptions;
+    this.logger = logger;
  }

  /**
@@ -115,10 +123,12 @@ export class PgSearchEngine implements SearchEngine {
  static async from(options: {
    database: PluginDatabaseManager;
    config: Config;
+    logger?: Logger;
  }): Promise<PgSearchEngine> {
    return new PgSearchEngine(
      await DatabaseDocumentStore.create(options.database),
      options.config,
+      options.logger,
    );
  }

@@ -126,6 +136,7 @@ export class PgSearchEngine implements SearchEngine {
    return new PgSearchEngine(
      await DatabaseDocumentStore.create(options.database),
      config,
+      options.logger,
    );
  }

@@ -170,6 +181,7 @@ export class PgSearchEngine implements SearchEngine {
      batchSize: 1000,
      type,
      databaseStore: this.databaseStore,
+      logger: this.logger?.child({ documentType: type }),
    });
  }

@@ -81,6 +81,15 @@ describe('PgSearchEngineIndexer', () => {
    expect(database.completeInsert).toHaveBeenCalledWith(tx, 'my-type');
  });

+  it('should rollback transaction if no documents indexed', async () => {
+    await TestPipeline.fromIndexer(indexer).withDocuments([]).execute();
+
+    expect(database.getTransaction).toHaveBeenCalledTimes(1);
+    expect(database.insertDocuments).not.toHaveBeenCalled();
+    expect(database.completeInsert).not.toHaveBeenCalled();
+    expect(tx.rollback).toHaveBeenCalled();
+  });
+
  it('should close out stream and bubble up error on prepare', async () => {
    const expectedError = new Error('Prepare error');
    const documents = [
@@ -14,9 +14,11 @@
 * limitations under the License.
 */

+import { getVoidLogger } from '@backstage/backend-common';
 import { BatchSearchEngineIndexer } from '@backstage/plugin-search-backend-node';
 import { IndexableDocument } from '@backstage/plugin-search-common';
 import { Knex } from 'knex';
+import { Logger } from 'winston';
 import { DatabaseStore } from '../database';

 /** @public */
@@ -24,18 +26,22 @@ export type PgSearchEngineIndexerOptions = {
  batchSize: number;
  type: string;
  databaseStore: DatabaseStore;
+  logger?: Logger;
 };

 /** @public */
 export class PgSearchEngineIndexer extends BatchSearchEngineIndexer {
+  private logger: Logger;
  private store: DatabaseStore;
  private type: string;
  private tx: Knex.Transaction | undefined;
+  private numRecords = 0;

  constructor(options: PgSearchEngineIndexerOptions) {
    super({ batchSize: options.batchSize });
    this.store = options.databaseStore;
    this.type = options.type;
+    this.logger = options.logger || getVoidLogger();
  }

  async initialize(): Promise<void> {
@@ -51,6 +57,8 @@ export class PgSearchEngineIndexer extends BatchSearchEngineIndexer {
  }

  async index(documents: IndexableDocument[]): Promise<void> {
+    this.numRecords += documents.length;
+
    try {
      await this.store.insertDocuments(this.tx!, this.type, documents);
    } catch (e) {
@@ -62,6 +70,17 @@ export class PgSearchEngineIndexer extends BatchSearchEngineIndexer {
  }

  async finalize(): Promise<void> {
+    // If no documents were indexed, rollback the transaction, log a warning,
+    // and do not continue. This ensures that collators that return empty sets
+    // of documents do not cause the index to be deleted.
+    if (this.numRecords === 0) {
+      this.logger.warn(
+        `Index for ${this.type} was not replaced: indexer received 0 documents`,
+      );
+      this.tx!.rollback!();
+      return;
+    }
+
    // Attempt to complete and commit the transaction.
    try {
      await this.store.completeInsert(this.tx!, this.type);
@@ -1026,7 +1026,7 @@ describe('LunrSearchEngine', () => {

      // Get the indexer and invoke its close handler.
      await inspectableSearchEngine.getIndexer('test-index');
-      const onClose = indexerMock.on.mock.calls[0][1] as Function;
+      const onClose = indexerMock.on.mock.calls[1][1] as Function;
      onClose();

      // Ensure mocked methods were called.
@@ -1044,6 +1044,60 @@ describe('LunrSearchEngine', () => {
        'new-location': doc,
      });
    });
+
+    it('should not replace index or docs if no docs were indexed', async () => {
+      // Set up an inspectable search engine to pre-set some data.
+      const doc = { title: 'A doc', text: 'test', location: 'some-location' };
+      const inspectableSearchEngine = new LunrSearchEngineForTests({
+        logger: getVoidLogger(),
+      });
+      inspectableSearchEngine.setDocStore({ 'existing-location': doc });
+
+      // Mock methods called by close handler (resolving no documents)
+      indexerMock.buildIndex.mockReturnValueOnce('expected-index');
+      indexerMock.getDocumentStore.mockReturnValueOnce({});
+
+      // Get the indexer and invoke its close handler.
+      await inspectableSearchEngine.getIndexer('test-index');
+      const onClose = indexerMock.on.mock.calls[1][1] as Function;
+      onClose();
+
+      // Ensure buildIndex method was not called.
+      expect(indexerMock.buildIndex).not.toHaveBeenCalled();
+
+      // Ensure pre-existing documents still exist in the store
+      expect(inspectableSearchEngine.getDocStore()).toStrictEqual({
+        'existing-location': doc,
+      });
+    });
+
+    it('should not replace index or docs if an error was thrown', async () => {
+      // Set up an inspectable search engine to pre-set some data.
+      const doc = { title: 'A doc', text: 'test', location: 'some-location' };
+      const inspectableSearchEngine = new LunrSearchEngineForTests({
+        logger: getVoidLogger(),
+      });
+      inspectableSearchEngine.setDocStore({ 'existing-location': doc });
+
+      // Mock methods called by close handler (resolving no documents)
+      indexerMock.buildIndex.mockReturnValueOnce('expected-index');
+      indexerMock.getDocumentStore.mockReturnValueOnce({});
+
+      // Get the indexer and invoke its close handler after firing an error.
+      await inspectableSearchEngine.getIndexer('test-index');
+      const onError = indexerMock.on.mock.calls[0][1] as Function;
+      const onClose = indexerMock.on.mock.calls[1][1] as Function;
+      onError(new Error('Some collator error'));
+      onClose();
+
+      // Ensure buildIndex method was not called.
+      expect(indexerMock.buildIndex).not.toHaveBeenCalled();
+
+      // Ensure pre-existing documents still exist in the store
+      expect(inspectableSearchEngine.getDocStore()).toStrictEqual({
+        'existing-location': doc,
+      });
+    });
  });
 });

@@ -153,12 +153,37 @@ export class LunrSearchEngine implements SearchEngine {

  async getIndexer(type: string) {
    const indexer = new LunrSearchEngineIndexer();
+    const indexerLogger = this.logger.child({ documentType: type });
+    let errorThrown: Error | undefined;
+
+    indexer.on('error', err => {
+      errorThrown = err;
+    });

    indexer.on('close', () => {
      // Once the stream is closed, build the index and store the documents in
      // memory for later retrieval.
-      this.lunrIndices[type] = indexer.buildIndex();
-      this.docStore = { ...this.docStore, ...indexer.getDocumentStore() };
+      const newDocuments = indexer.getDocumentStore();
+      const docStoreExists = this.lunrIndices[type] !== undefined;
+      const documentsIndexed = Object.keys(newDocuments).length;
+
+      // Do not set the index if there was an error or if no documents were
+      // indexed. This ensures search continues to work for an index, even in
+      // case of transient issues in underlying collators.
+      if (!errorThrown && documentsIndexed > 0) {
+        this.lunrIndices[type] = indexer.buildIndex();
+        this.docStore = { ...this.docStore, ...newDocuments };
+      } else {
+        indexerLogger.warn(
+          `Index for ${type} was not ${
+            docStoreExists ? 'replaced' : 'created'
+          }: ${
+            errorThrown
+              ? 'an error was encountered'
+              : 'indexer received 0 documents'
+          }`,
+        );
+      }
    });

    return indexer;
@@ -7558,6 +7558,7 @@ __metadata:
    knex: ^2.0.0
    lodash: ^4.17.21
    uuid: ^8.3.2
+    winston: ^3.2.1
  languageName: unknown
  linkType: soft