[Fleet] Rollover data streams when package w/ TSDB setting changed is…

… installed (elastic#157869)

## Summary

Fixes elastic#157345

When a package with a changed `index.mode` or `source.mode` setting is
installed, Fleet will now automatically perform a rollover to ensure the
correct setting is present on the resulting backing index.

There is an issue with Elasticsearch wherein toggling these settings
back and forth will incur a backing index range overlap error. See
elastic/elasticsearch#96163.

To test
1. Install the `system` integration at version `1.28.0`
2. Create an integration policy for the `system` integration (a standard
default agent policy will do)
3. Enroll an agent in this policy, and allow it to ingest some data
4. Confirm that there are documents present in the
`metrics-system.cpu-default` data stream, and note its backing index via
Stack Management
5. Create a new `1.28.1` version of the `system` integration where
`elasticsearch.index_mode: time_series` is set and install it via
`elastic-package install --zip`
6. Confirm that a rollover occurs and the backing index for the
`metrics-system.cpu-default` data stream has been updated

### Checklist

Delete any items that are not applicable to this PR.

- [x] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios

---------

Co-authored-by: Kibana Machine <[email protected]>
(cherry picked from commit 22e3847)

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

@@ -6,7 +6,10 @@
  */
 
 import type { ElasticsearchClient, Logger } from '@kbn/core/server';
-import type { IndicesIndexSettings } from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
+import type {
+  IndicesIndexSettings,
+  MappingTypeMapping,
+} from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 
 import type { Field, Fields } from '../../fields/field';
 import type {
@@ -642,7 +645,20 @@ const updateExistingDataStream = async ({
   esClient: ElasticsearchClient;
   logger: Logger;
 }) => {
+  const existingDs = await esClient.indices.get({
+    index: dataStreamName,
+  });
+
+  const existingDsConfig = Object.values(existingDs);
+  const currentBackingIndexConfig = existingDsConfig.at(-1);
+
+  const currentIndexMode = currentBackingIndexConfig?.settings?.index?.mode;
+  // @ts-expect-error Property 'mode' does not exist on type 'MappingSourceField'
+  const currentSourceType = currentBackingIndexConfig.mappings?._source?.mode;
+
   let settings: IndicesIndexSettings;
+  let mappings: MappingTypeMapping;
+
   try {
     const simulateResult = await retryTransientEsErrors(() =>
       esClient.indices.simulateTemplate({
@@ -651,14 +667,17 @@ const updateExistingDataStream = async ({
     );
 
     settings = simulateResult.template.settings;
-    const mappings = simulateResult.template.mappings;
+    mappings = simulateResult.template.mappings;
+
     // for now, remove from object so as not to update stream or data stream properties of the index until type and name
     // are added in https://github.com/elastic/kibana/issues/66551.  namespace value we will continue
     // to skip updating and assume the value in the index mapping is correct
     if (mappings && mappings.properties) {
       delete mappings.properties.stream;
       delete mappings.properties.data_stream;
     }
+
+    logger.debug(`Updating mappings for ${dataStreamName}`);
     await retryTransientEsErrors(
       () =>
         esClient.indices.putMapping({
@@ -668,16 +687,38 @@ const updateExistingDataStream = async ({
         }),
       { logger }
     );
-    // if update fails, rollover data stream
+
+    // if update fails, rollover data stream and bail out
   } catch (err) {
+    logger.error(`Mappings update for ${dataStreamName} failed`);
+    logger.error(err);
+
     await rolloverDataStream(dataStreamName, esClient);
     return;
   }
+
+  // Trigger a rollover if the index mode or source type has changed
+  if (
+    currentIndexMode !== settings?.index?.mode ||
+    // @ts-expect-error Property 'mode' does not exist on type 'MappingSourceField'
+    currentSourceType !== mappings?._source?.mode
+  ) {
+    logger.info(
+      `Index mode or source type has changed for ${dataStreamName}, triggering a rollover`
+    );
+    await rolloverDataStream(dataStreamName, esClient);
+  }
+
   // update settings after mappings was successful to ensure
   // pointing to the new pipeline is safe
   // for now, only update the pipeline
-  if (!settings?.index?.default_pipeline) return;
+  if (!settings?.index?.default_pipeline) {
+    return;
+  }
+
   try {
+    logger.debug(`Updating settings for ${dataStreamName}`);
+
     await retryTransientEsErrors(
       () =>
         esClient.indices.putSettings({