Merge pull request #24682 from leboncoin/fix-search-es-stale-indices-2

search-backend-module-elasticsearch: ensure all stale indices are deleted
2024-06-12 17:41:13 +02:00
parent 871d088fb2 5eba6fcd79
commit f7fbfab796
6 changed files with 173 additions and 71 deletions
@@ -0,0 +1,14 @@
+---
+'@backstage/plugin-search-backend-module-elasticsearch': minor
+---
+
+**BREAKING**: The ElasticSearch indexer will now delete stale indices matching the indexer's pattern.
+The method `getAliases` of `ElasticSearchClientWrapper` has been deprecated and might be removed in future releases.
+
+An indexer using the `some-type-index__*` pattern will remove indices matching this pattern after indexation
+to prevent stale indices leading to shards exhaustion.
+
+Before upgrading ensure that the index pattern doesn't match indices that are not managed by Backstage
+and thus shouldn't be deleted.
+
+Note: The ElasticSearch indexer already uses wildcards patterns to remove aliases on these indices.
@@ -153,7 +153,7 @@ export class ElasticSearchClientWrapper {
  static fromClientOptions(
    options: ElasticSearchClientOptions,
  ): ElasticSearchClientWrapper;
-  // (undocumented)
+  // @deprecated (undocumented)
  getAliases(options: {
    aliases: string[];
  }):
@@ -166,6 +166,12 @@ export class ElasticSearchClientWrapper {
    | TransportRequestPromise<ApiResponse<boolean, unknown>>
    | TransportRequestPromise_2<ApiResponse_2<boolean, unknown>>;
  // (undocumented)
+  listIndices(options: {
+    index: string;
+  }):
+    | TransportRequestPromise<ApiResponse<Record<string, any>, unknown>>
+    | TransportRequestPromise_2<ApiResponse_2<Record<string, any>, unknown>>;
+  // (undocumented)
  putIndexTemplate(
    template: ElasticSearchCustomIndexTemplate,
  ):
@@ -39,6 +39,7 @@ jest.mock('@elastic/elasticsearch', () => ({
        .mockImplementation(async args => ({ client: 'es', args })),
    },
    indices: {
+      get: jest.fn().mockImplementation(async args => ({ client: 'es', args })),
      create: jest
        .fn()
        .mockImplementation(async args => ({ client: 'es', args })),
@@ -75,6 +76,7 @@ jest.mock('@opensearch-project/opensearch', () => ({
        .mockImplementation(async args => ({ client: 'os', args })),
    },
    indices: {
+      get: jest.fn().mockImplementation(async args => ({ client: 'os', args })),
      create: jest
        .fn()
        .mockImplementation(async args => ({ client: 'os', args })),
@@ -154,6 +156,16 @@ describe('ElasticSearchClientWrapper', () => {
      expect(result.args).toStrictEqual(indexTemplate);
    });

+    it('indexList', async () => {
+      const wrapper = ElasticSearchClientWrapper.fromClientOptions(esOptions);
+
+      const input = { index: 'xyz-*' };
+      const result = (await wrapper.listIndices(input)) as any;
+
+      expect(result.client).toBe('es');
+      expect(result.args).toStrictEqual(input);
+    });
+
    it('indexExists', async () => {
      const wrapper = ElasticSearchClientWrapper.fromClientOptions(esOptions);

@@ -282,6 +294,16 @@ describe('ElasticSearchClientWrapper', () => {
      expect(result.args).toStrictEqual(indexTemplate);
    });

+    it('indexList', async () => {
+      const wrapper = ElasticSearchClientWrapper.fromClientOptions(osOptions);
+
+      const input = { index: 'xyz-*' };
+      const result = (await wrapper.listIndices(input)) as any;
+
+      expect(result.client).toBe('os');
+      expect(result.args).toStrictEqual(input);
+    });
+
    it('indexExists', async () => {
      const wrapper = ElasticSearchClientWrapper.fromClientOptions(osOptions);

@@ -141,6 +141,18 @@ export class ElasticSearchClientWrapper {
    throw new Error('No client defined');
  }

+  listIndices(options: { index: string }) {
+    if (this.openSearchClient) {
+      return this.openSearchClient.indices.get(options);
+    }
+
+    if (this.elasticSearchClient) {
+      return this.elasticSearchClient.indices.get(options);
+    }
+
+    throw new Error('No client defined');
+  }
+
  indexExists(options: { index: string | string[] }) {
    if (this.openSearchClient) {
      return this.openSearchClient.indices.exists(options);
@@ -165,18 +177,9 @@ export class ElasticSearchClientWrapper {
    throw new Error('No client defined');
  }

-  createIndex(options: { index: string }) {
-    if (this.openSearchClient) {
-      return this.openSearchClient.indices.create(options);
-    }
-
-    if (this.elasticSearchClient) {
-      return this.elasticSearchClient.indices.create(options);
-    }
-
-    throw new Error('No client defined');
-  }
-
+  /**
+   * @deprecated unused by the ElasticSearch Engine, will be removed in the future
+   */
  getAliases(options: { aliases: string[] }) {
    const { aliases } = options;

@@ -197,6 +200,18 @@ export class ElasticSearchClientWrapper {
    throw new Error('No client defined');
  }

+  createIndex(options: { index: string }) {
+    if (this.openSearchClient) {
+      return this.openSearchClient.indices.create(options);
+    }
+
+    if (this.elasticSearchClient) {
+      return this.elasticSearchClient.indices.create(options);
+    }
+
+    throw new Error('No client defined');
+  }
+
  updateAliases(options: { actions: ElasticSearchAliasAction[] }) {
    const filteredActions = options.actions.filter(Boolean);

@@ -30,7 +30,7 @@ const clientWrapper = ElasticSearchClientWrapper.fromClientOptions({
 describe('ElasticSearchSearchEngineIndexer', () => {
  let indexer: ElasticSearchSearchEngineIndexer;
  let bulkSpy: jest.Mock;
-  let catSpy: jest.Mock;
+  let getSpy: jest.Mock;
  let createSpy: jest.Mock;
  let aliasesSpy: jest.Mock;
  let deleteSpy: jest.Mock;
@@ -68,30 +68,24 @@ describe('ElasticSearchSearchEngineIndexer', () => {
      refreshSpy,
    );

-    catSpy = jest.fn().mockReturnValue([
-      {
-        alias: 'some-type-index__search',
-        index: 'some-type-index__123tobedeleted',
-        filter: '-',
-        'routing.index': '-',
-        'routing.search': '-',
-        is_write_index: '-',
+    getSpy = jest.fn().mockReturnValue({
+      'some-type-index__123tobedeleted': {
+        aliases: {},
+        mappings: {},
+        settings: {},
      },
-      {
-        alias: 'some-type-index__search_removable',
-        index: 'some-type-index__456tobedeleted',
-        filter: '-',
-        'routing.index': '-',
-        'routing.search': '-',
-        is_write_index: '-',
+      'some-type-index__456tobedeleted': {
+        aliases: {},
+        mappings: {},
+        settings: {},
      },
-    ]);
+    });
    mock.add(
      {
        method: 'GET',
-        path: '/_cat/aliases/some-type-index__search%2Csome-type-index__search_removable',
+        path: '/some-type-index__*',
      },
-      catSpy,
+      getSpy,
    );

    createSpy = jest.fn().mockReturnValue({
@@ -120,7 +114,7 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    mock.add(
      {
        method: 'DELETE',
-        path: '/some-type-index__123tobedeleted%2Csome-type-index__456tobedeleted',
+        path: '/*',
      },
      deleteSpy,
    );
@@ -143,7 +137,7 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    await TestPipeline.fromIndexer(indexer).withDocuments(documents).execute();

    // Older indices should have been queried for.
-    expect(catSpy).toHaveBeenCalled();
+    expect(getSpy).toHaveBeenCalled();

    // A new index should have been created.
    const createdIndex = createSpy.mock.calls[0][0].path.slice(1);
@@ -163,27 +157,22 @@ describe('ElasticSearchSearchEngineIndexer', () => {
      remove: { index: 'some-type-index__*', alias: 'some-type-index__search' },
    });
    expect(aliasActions[1]).toStrictEqual({
-      add: {
-        indices: [
-          'some-type-index__123tobedeleted',
-          'some-type-index__456tobedeleted',
-        ],
-        alias: 'some-type-index__search_removable',
-      },
-    });
-    expect(aliasActions[2]).toStrictEqual({
      add: { index: createdIndex, alias: 'some-type-index__search' },
    });

    // Old index should be cleaned up.
-    expect(deleteSpy).toHaveBeenCalled();
+    expect(deleteSpy).toHaveBeenCalledWith(
+      expect.objectContaining({
+        path: '/some-type-index__123tobedeleted%2Csome-type-index__456tobedeleted',
+      }),
+    );
  });

  it('handles when no documents are received', async () => {
    await TestPipeline.fromIndexer(indexer).withDocuments([]).execute();

    // Older indices should have been queried for.
-    expect(catSpy).toHaveBeenCalled();
+    expect(getSpy).toHaveBeenCalled();

    // A new index should have been created.
    expect(createSpy).toHaveBeenNthCalledWith(
@@ -200,7 +189,11 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    expect(aliasesSpy).not.toHaveBeenCalled();

    // Old index should not be cleaned up.
-    expect(deleteSpy).not.toHaveBeenCalled();
+    expect(deleteSpy).toHaveBeenCalledWith(
+      expect.objectContaining({
+        path: expect.not.stringContaining('tobedeleted'),
+      }),
+    );
  });

  it('handles bulk and batching during indexing', async () => {
@@ -236,17 +229,17 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    ];

    // Update initial alias cat to return nothing.
-    catSpy = jest.fn().mockReturnValue([]);
+    getSpy = jest.fn().mockReturnValue({});
    mock.clear({
      method: 'GET',
-      path: '/_cat/aliases/some-type-index__search%2Csome-type-index__search_removable',
+      path: '/some-type-index__*',
    });
    mock.add(
      {
        method: 'GET',
-        path: '/_cat/aliases/some-type-index__search%2Csome-type-index__search_removable',
+        path: '/some-type-index__*',
      },
-      catSpy,
+      getSpy,
    );

    await TestPipeline.fromIndexer(indexer).withDocuments(documents).execute();
@@ -255,6 +248,47 @@ describe('ElasticSearchSearchEngineIndexer', () => {
    expect(deleteSpy).not.toHaveBeenCalled();
  });

+  it('split index deletion in chunks', async () => {
+    // Generate 200 existing indices
+    const currentIndices = Array.from(
+      new Array(200),
+      (_, i) => `some-type-index__old${i}`,
+    );
+
+    const indicesResponse = currentIndices.reduce((acc, curr) => {
+      return {
+        ...acc,
+        [curr]: { mappings: {}, aliases: {} },
+      };
+    }, {});
+
+    getSpy = jest.fn().mockReturnValue(indicesResponse);
+    mock.clear({
+      method: 'GET',
+      path: '/some-type-index__*',
+    });
+    mock.add(
+      {
+        method: 'GET',
+        path: '/some-type-index__*',
+      },
+      getSpy,
+    );
+
+    await TestPipeline.fromIndexer(indexer)
+      .withDocuments([
+        {
+          title: 'testTerm',
+          text: 'testText',
+          location: 'test/location',
+        },
+      ])
+      .execute();
+
+    // Delete endpoint should have been called for 4 chunks of 50 (200 indices)
+    expect(deleteSpy).toHaveBeenCalledTimes(4);
+  });
+
  it('handles bulk client rejection', async () => {
    // Given an ES client wrapper that rejects an error
    const expectedError = new Error('HTTP Timeout');
@@ -56,7 +56,6 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
  private readonly indexPrefix: string;
  private readonly indexSeparator: string;
  private readonly alias: string;
-  private readonly removableAlias: string;
  private readonly logger: Logger | LoggerService;
  private readonly sourceStream: Readable;
  private readonly elasticSearchClientWrapper: ElasticSearchClientWrapper;
@@ -74,7 +73,6 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
    this.indexSeparator = options.indexSeparator;
    this.indexName = this.constructIndexName(`${Date.now()}`);
    this.alias = options.alias;
-    this.removableAlias = `${this.alias}_removable`;
    this.elasticSearchClientWrapper = options.elasticSearchClientWrapper;

    // The ES client bulk helper supports stream-based indexing, but we have to
@@ -108,13 +106,13 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
  async initialize(): Promise<void> {
    this.logger.info(`Started indexing documents for index ${this.type}`);

-    const aliases = await this.elasticSearchClientWrapper.getAliases({
-      aliases: [this.alias, this.removableAlias],
+    const indices = await this.elasticSearchClientWrapper.listIndices({
+      index: this.constructIndexName('*'),
    });

-    this.removableIndices = [
-      ...new Set(aliases.body.map((r: Record<string, any>) => r.index)),
-    ] as string[];
+    for (const key of Object.keys(indices.body)) {
+      this.removableIndices.push(key);
+    }

    await this.elasticSearchClientWrapper.createIndex({
      index: this.indexName,
@@ -171,14 +169,6 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
        {
          remove: { index: this.constructIndexName('*'), alias: this.alias },
        },
-        this.removableIndices.length
-          ? {
-              add: {
-                indices: this.removableIndices,
-                alias: this.removableAlias,
-              },
-            }
-          : undefined,
        {
          add: { index: this.indexName, alias: this.alias },
        },
@@ -191,12 +181,33 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer {
      this.logger.info('Removing stale search indices', {
        removableIndices: this.removableIndices,
      });
-      try {
-        await this.elasticSearchClientWrapper.deleteIndex({
-          index: this.removableIndices,
-        });
-      } catch (e) {
-        this.logger.warn(`Failed to remove stale search indices: ${e}`);
+
+      // Split the array into chunks of up to 50 indices to handle the case
+      // where we need to delete a lot of stalled indices
+      const chunks = this.removableIndices.reduce(
+        (resultArray, item, index) => {
+          const chunkIndex = Math.floor(index / 50);
+
+          if (!resultArray[chunkIndex]) {
+            resultArray[chunkIndex] = []; // start a new chunk
+          }
+
+          resultArray[chunkIndex].push(item);
+
+          return resultArray;
+        },
+        [] as string[][],
+      );
+
+      // Call deleteIndex for each chunk
+      for (const chunk of chunks) {
+        try {
+          await this.elasticSearchClientWrapper.deleteIndex({
+            index: chunk,
+          });
+        } catch (e) {
+          this.logger.warn(`Failed to remove stale search indices: ${e}`);
+        }
      }
    }
  }