[8.x] [Fleet] Fix upgrade with a large number of stream backing indic…

…es (#201272) (#201801)

# Backport

This will backport the following commits from `main` to `8.x`:
- [[Fleet] Fix upgrade with a large number of stream backing indices
(#201272)](#201272)

<!--- Backport version: 9.4.3 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Nicolas
Chaulet","email":"[email protected]"},"sourceCommit":{"committedDate":"2024-11-26T14:32:13Z","message":"[Fleet]
Fix upgrade with a large number of stream backing indices
(#201272)","sha":"97318c9f537bd6e9fecb35e40250aa14df325b81","branchLabelMapping":{"^v9.0.0$":"main","^v8.18.0$":"8.x","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["bug","release_note:skip","Team:Fleet","v9.0.0","backport:prev-minor","v8.17.0"],"title":"[Fleet]
Fix upgrade with a large number of stream backing
indices","number":201272,"url":"https://github.com/elastic/kibana/pull/201272","mergeCommit":{"message":"[Fleet]
Fix upgrade with a large number of stream backing indices
(#201272)","sha":"97318c9f537bd6e9fecb35e40250aa14df325b81"}},"sourceBranch":"main","suggestedTargetBranches":["8.17"],"targetPullRequestStates":[{"branch":"main","label":"v9.0.0","branchLabelMappingKey":"^v9.0.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/201272","number":201272,"mergeCommit":{"message":"[Fleet]
Fix upgrade with a large number of stream backing indices
(#201272)","sha":"97318c9f537bd6e9fecb35e40250aa14df325b81"}},{"branch":"8.17","label":"v8.17.0","branchLabelMappingKey":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"}]}]
BACKPORT-->

Co-authored-by: Nicolas Chaulet <[email protected]>

x-pack/plugins/fleet/server/services/epm/elasticsearch/template/template.test.ts

@@ -1870,9 +1870,19 @@ describe('EPM template', () => {
 
     it('should fill constant keywords from previous mappings', async () => {
       const esClient = elasticsearchServiceMock.createElasticsearchClient();
+
       esClient.indices.getDataStream.mockResponse({
-        data_streams: [{ name: 'test-constant.keyword-default' }],
+        data_streams: [
+          {
+            name: 'test-constant.keyword-default',
+            indices: [
+              { index_name: '.ds-test-constant.keyword-default-0001' },
+              { index_name: '.ds-test-constant.keyword-default-0002' },
+            ],
+          },
+        ],
       } as any);
+
       esClient.indices.get.mockResponse({
         'test-constant.keyword-default': {
           mappings: {
@@ -1912,6 +1922,9 @@ describe('EPM template', () => {
           } as any,
         },
       ]);
+      expect(esClient.indices.get).toBeCalledWith({
+        index: '.ds-test-constant.keyword-default-0002',
+      });
       const putMappingsCalls = esClient.indices.putMapping.mock.calls;
       expect(putMappingsCalls).toHaveLength(1);
       expect(putMappingsCalls[0][0]).toEqual({

x-pack/plugins/fleet/server/services/epm/elasticsearch/template/template.ts

@@ -61,6 +61,7 @@ export interface CurrentDataStream {
   dataStreamName: string;
   replicated: boolean;
   indexTemplate: IndexTemplate;
+  currentWriteIndex: string;
 }
 
 const DEFAULT_IGNORE_ABOVE = 1024;
@@ -944,6 +945,7 @@ const getDataStreams = async (
     dataStreamName: dataStream.name,
     replicated: dataStream.replicated,
     indexTemplate,
+    currentWriteIndex: dataStream.indices?.at(-1)?.index_name,
   }));
 };
 
@@ -979,6 +981,7 @@ const updateAllDataStreams = async (
       return updateExistingDataStream({
         esClient,
         logger,
+        currentWriteIndex: templateEntry.currentWriteIndex,
         dataStreamName: templateEntry.dataStreamName,
         options,
       });
@@ -992,11 +995,13 @@ const updateAllDataStreams = async (
 
 const updateExistingDataStream = async ({
   dataStreamName,
+  currentWriteIndex,
   esClient,
   logger,
   options,
 }: {
   dataStreamName: string;
+  currentWriteIndex: string;
   esClient: ElasticsearchClient;
   logger: Logger;
   options?: {
@@ -1005,7 +1010,7 @@ const updateExistingDataStream = async ({
   };
 }) => {
   const existingDs = await esClient.indices.get({
-    index: dataStreamName,
+    index: currentWriteIndex,
   });
 
   const existingDsConfig = Object.values(existingDs);