Move force merge from the downsmapling request to the ILM action (#135834)

In this PR we move the force-merge operation from the downsampling
request to the ILM action. 

Our goal is to decouple the downsampling operation from the force-merge
operation. With this change the downsampling request is responsible to
ensure that the downsampled index is refreshed and flushed but not to
force merge it.

We believe that most of the time this is not necessary, and executing
the force-merge operation unnecessarily can increase the load on the
cluster.

To preserve backwards compatibility we move the responsibility to
execute the existing force merge to the downsample ILM action and we
make it configurable. By default, it will run but a user can disable it
just as they can with a searchable snapshot.

```
"downsample": {
  "fixed_interval": "1h",
  "force_merge_index": false
}
```

**Update**

As a follow up of this PR, we pose the question is the force merge in
the downsample action intentional and useful? 

To answer this question, we extend time series telemetry. We define that
the force merge step in the downsample ILM action is useful, if this is
the only force merge step operation before a searchable snapshot.

Effectively, by this definition, we argue that the force merge in
downsampling is not an intentional operation the user has requested but
only the result of the implementation. We identify the biggest impact of
removing it to be a searchable snapshot, but if the searchable snapshot
performs its own force merge (and more performant force merge #133954)
then we could skip this operation in the downsample action altogether.

Fixes: #135618

Author: gmarouli

docs/changelog/135834.yaml

@@ -0,0 +1,6 @@
+pr: 135834
+summary: Move force merge from the downsampling request to the ILM action and allow
+  users to disable it.
+area: Downsampling
+type: enhancement
+issues: []

docs/reference/elasticsearch/index-lifecycle-actions/ilm-downsample.md

@@ -9,7 +9,7 @@ Phases allowed: hot, warm, cold.
 
 Aggregates a time series (TSDS) index and stores pre-computed statistical summaries (`min`, `max`, `sum`, and `value_count`) for each metric field grouped by a configured time interval. For example, a TSDS index that contains metrics sampled every 10 seconds can be downsampled to an hourly index. All documents within an hour interval are summarized and stored as a single document and stored in the downsample index.
 
-This action corresponds to the  [downsample API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-downsample).
+This action corresponds to the [downsample API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-downsample).
 
 The name of the resulting downsample index is `downsample-<original-index-name>-<random-uuid>`. If {{ilm-init}} performs the `downsample` action on a backing index for a data stream, the downsample index becomes a backing index for the same stream and the source index is deleted.
 
@@ -19,6 +19,8 @@ To use the `downsample` action in the `hot` phase, the `rollover` action **must*
 
 `fixed_interval`
 :   (Required, string) The [fixed time interval](docs-content://manage-data/lifecycle/rollup/understanding-groups.md#rollup-understanding-group-intervals) into which the data will be downsampled.
+`force_merge_index` {applies_to}`stack: ga 9.3`
+:   (Optional, boolean) When true, the downsampled index will be [force merged](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-forcemerge) to one [segment](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-segments). Defaults to `true`.
 
 
 ## Example [ilm-downsample-ex]
@@ -34,7 +36,8 @@ PUT _ilm/policy/datastream_policy
             "max_docs": 1
           },
           "downsample": {
-  	          "fixed_interval": "1h"
+  	          "fixed_interval": "1h",
+  	          "force_merge_index": false
   	      }
         }
       }

server/src/main/resources/transport/definitions/referable/ilm_downsample_force_merge.csv

@@ -0,0 +1 @@
+9196000

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-inference_update_trained_model_deployment_request_source,9195000
+ilm_downsample_force_merge,9196000

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/core/action/TimeSeriesUsageTransportActionIT.java

@@ -57,6 +57,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -130,6 +131,7 @@ public void testAction() throws Exception {
         var ilmRoundsMin = new AtomicInteger(Integer.MAX_VALUE);
         var ilmRoundsMax = new AtomicInteger(Integer.MIN_VALUE);
         Set<String> usedPolicies = new HashSet<>();
+        var forceMergeEnabled = new AtomicReference<IlmForceMergeInPolicies>();
 
         /*
          * We now add a number of simulated data streams to the cluster state. We mix different combinations of:
@@ -139,7 +141,7 @@ public void testAction() throws Exception {
          */
         updateClusterState(clusterState -> {
             Metadata.Builder metadataBuilder = Metadata.builder(clusterState.metadata());
-            addIlmPolicies(metadataBuilder);
+            forceMergeEnabled.set(addIlmPolicies(metadataBuilder));
 
             Map<String, DataStream> dataStreamMap = new HashMap<>();
             for (int dataStreamCount = 0; dataStreamCount < randomIntBetween(10, 100); dataStreamCount++) {
@@ -313,22 +315,31 @@ public void testAction() throws Exception {
                 ilmRoundsMin.get(),
                 ilmRoundsMax.get()
             );
+            var explicitlyEnabled = new AtomicInteger(0);
+            var explicitlyDisabled = new AtomicInteger(0);
+            var undefined = new AtomicInteger(0);
             Map<String, Object> phasesStats = (Map<String, Object>) ilmStats.get("phases_in_use");
             if (usedPolicies.contains(DOWNSAMPLING_IN_HOT_POLICY)) {
                 assertThat(phasesStats.get("hot"), equalTo(1));
+                updateForceMergeCounters(forceMergeEnabled.get().enabledInHot, explicitlyEnabled, explicitlyDisabled, undefined);
             } else {
                 assertThat(phasesStats.get("hot"), nullValue());
             }
             if (usedPolicies.contains(DOWNSAMPLING_IN_WARM_COLD_POLICY)) {
                 assertThat(phasesStats.get("warm"), equalTo(1));
+                updateForceMergeCounters(forceMergeEnabled.get().enabledInWarm, explicitlyEnabled, explicitlyDisabled, undefined);
                 assertThat(phasesStats.get("cold"), equalTo(1));
+                updateForceMergeCounters(forceMergeEnabled.get().enabledInCold, explicitlyEnabled, explicitlyDisabled, undefined);
             } else {
                 assertThat(phasesStats.get("warm"), nullValue());
                 assertThat(phasesStats.get("cold"), nullValue());
             }
-
+            Map<String, Object> forceMergeStats = (Map<String, Object>) ilmStats.get("force_merge");
+            assertThat((int) forceMergeStats.get("explicitly_enabled_count"), equalTo(explicitlyEnabled.get()));
+            assertThat(forceMergeStats.get("explicitly_disabled_count"), equalTo(explicitlyDisabled.get()));
+            assertThat(forceMergeStats.get("undefined_count"), equalTo(undefined.get()));
+            assertThat(forceMergeStats.get("undefined_force_merge_needed_count"), equalTo(0));
         }
-
     }
 
     @SuppressWarnings("unchecked")
@@ -383,6 +394,16 @@ private void assertDownsamplingStats(
         }
     }
 
+    private void updateForceMergeCounters(Boolean value, AtomicInteger enabled, AtomicInteger disabled, AtomicInteger undefined) {
+        if (value == null) {
+            undefined.incrementAndGet();
+        } else if (value) {
+            enabled.incrementAndGet();
+        } else {
+            disabled.incrementAndGet();
+        }
+    }
+
     private DataStreamLifecycle maybeCreateLifecycle(boolean isDownsampled, boolean hasDlm) {
         if (hasDlm == false) {
             return null;
@@ -462,25 +483,36 @@ private List<DataStreamLifecycle.DownsamplingRound> randomDownsamplingRounds() {
         return rounds;
     }
 
-    private void addIlmPolicies(Metadata.Builder metadataBuilder) {
+    private IlmForceMergeInPolicies addIlmPolicies(Metadata.Builder metadataBuilder) {
+        Boolean hotForceMergeEnabled = randomBoolean() ? randomBoolean() : null;
+        Boolean warmForceMergeEnabled = randomBoolean() ? randomBoolean() : null;
+        Boolean coldForceMergeEnabled = randomBoolean() ? randomBoolean() : null;
         List<LifecyclePolicy> policies = List.of(
             new LifecyclePolicy(
                 DOWNSAMPLING_IN_HOT_POLICY,
                 Map.of(
                     "hot",
-                    new Phase("hot", TimeValue.ZERO, Map.of("downsample", new DownsampleAction(DateHistogramInterval.MINUTE, null)))
+                    new Phase(
+                        "hot",
+                        TimeValue.ZERO,
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.MINUTE, null, hotForceMergeEnabled))
+                    )
                 )
             ),
             new LifecyclePolicy(
                 DOWNSAMPLING_IN_WARM_COLD_POLICY,
                 Map.of(
                     "warm",
-                    new Phase("warm", TimeValue.ZERO, Map.of("downsample", new DownsampleAction(DateHistogramInterval.HOUR, null))),
+                    new Phase(
+                        "warm",
+                        TimeValue.ZERO,
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.HOUR, null, warmForceMergeEnabled))
+                    ),
                     "cold",
                     new Phase(
                         "cold",
                         TimeValue.timeValueDays(3),
-                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.DAY, null))
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.DAY, null, coldForceMergeEnabled))
                     )
                 )
             ),
@@ -498,8 +530,11 @@ private void addIlmPolicies(Metadata.Builder metadataBuilder) {
             );
         IndexLifecycleMetadata newMetadata = new IndexLifecycleMetadata(policyMetadata, OperationMode.RUNNING);
         metadataBuilder.putCustom(IndexLifecycleMetadata.TYPE, newMetadata);
+        return new IlmForceMergeInPolicies(hotForceMergeEnabled, warmForceMergeEnabled, coldForceMergeEnabled);
     }
 
+    private record IlmForceMergeInPolicies(Boolean enabledInHot, Boolean enabledInWarm, Boolean enabledInCold) {}
+
     private static String randomIlmPolicy(DownsampledBy downsampledBy, boolean ovewrittenDlm) {
         if (downsampledBy == DownsampledBy.ILM || (downsampledBy == DownsampledBy.DLM && ovewrittenDlm)) {
             return randomFrom(DOWNSAMPLING_IN_HOT_POLICY, DOWNSAMPLING_IN_WARM_COLD_POLICY);

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/TimeSeriesUsageTransportAction.java

@@ -25,10 +25,14 @@
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.datastreams.TimeSeriesFeatureSetUsage;
 import org.elasticsearch.xpack.core.ilm.DownsampleAction;
+import org.elasticsearch.xpack.core.ilm.ForceMergeAction;
 import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
+import org.elasticsearch.xpack.core.ilm.LifecycleAction;
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
 import org.elasticsearch.xpack.core.ilm.LifecyclePolicyMetadata;
 import org.elasticsearch.xpack.core.ilm.Phase;
+import org.elasticsearch.xpack.core.ilm.SearchableSnapshotAction;
+import org.elasticsearch.xpack.core.ilm.TimeseriesLifecycleType;
 
 import java.util.HashMap;
 import java.util.LongSummaryStatistics;
@@ -128,7 +132,7 @@ protected void localClusterStateOperation(
                 tsDataStreamCount,
                 tsIndexCount,
                 ilmStats.getDownsamplingStats(),
-                ilmStats.getIlmPolicyStats(),
+                ilmStats.calculateIlmPolicyStats(),
                 dlmStats.getDownsamplingStats(),
                 indicesByInterval
             )
@@ -167,27 +171,79 @@ TimeSeriesFeatureSetUsage.DownsamplingFeatureStats getDownsamplingStats() {
         }
     }
 
+    /**
+     * Tracks the ILM policies currently in use by time series data streams.
+     */
     static class IlmDownsamplingStatsTracker extends DownsamplingStatsTracker {
         private final Map<String, Map<String, Phase>> policies = new HashMap<>();
 
         void trackPolicy(LifecyclePolicy ilmPolicy) {
             policies.putIfAbsent(ilmPolicy.getName(), ilmPolicy.getPhases());
         }
 
-        Map<String, Long> getIlmPolicyStats() {
+        /**
+         * Calculates ILM-policy-specific statistics that help us get a better understanding on the phases that use downsampling and on
+         * how the force merge step in the downsample action is used. More specifically, for downsampling we are tracking:
+         * - if users explicitly enabled or disabled force-merge after downsampling.
+         * - if the force merge could be skipped with minimal impact, when the force merge flag is undefined.
+         * @return a IlmPolicyStats record that contains these counters.
+         */
+        TimeSeriesFeatureSetUsage.IlmPolicyStats calculateIlmPolicyStats() {
             if (policies.isEmpty()) {
-                return Map.of();
+                return TimeSeriesFeatureSetUsage.IlmPolicyStats.EMPTY;
             }
+            long forceMergeExplicitlyEnabledCounter = 0;
+            long forceMergeExplicitlyDisabledCounter = 0;
+            long forceMergeDefaultCounter = 0;
+            long downsampledForceMergeNeededCounter = 0; // Meaning it's followed by a searchable snapshot with force-merge index false
             Map<String, Long> downsamplingPhases = new HashMap<>();
             for (String ilmPolicy : policies.keySet()) {
-                for (Phase phase : policies.get(ilmPolicy).values()) {
-                    if (phase.getActions().containsKey(DownsampleAction.NAME)) {
-                        Long current = downsamplingPhases.computeIfAbsent(phase.getName(), ignored -> 0L);
-                        downsamplingPhases.put(phase.getName(), current + 1);
+                Map<String, Phase> phases = policies.get(ilmPolicy);
+                boolean downsampledForceMergeNeeded = false;
+                for (String phase : TimeseriesLifecycleType.ORDERED_VALID_PHASES) {
+                    if (phases.containsKey(phase) == false) {
+                        continue;
+                    }
+                    Map<String, LifecycleAction> actions = phases.get(phase).getActions();
+                    if (actions.containsKey(DownsampleAction.NAME)) {
+                        // count the phase used
+                        Long current = downsamplingPhases.computeIfAbsent(phase, ignored -> 0L);
+                        downsamplingPhases.put(phase, current + 1);
+                        // Count force merge
+                        DownsampleAction downsampleAction = (DownsampleAction) actions.get(DownsampleAction.NAME);
+                        if (downsampleAction.forceMergeIndex() == null) {
+                            forceMergeDefaultCounter++;
+                            downsampledForceMergeNeeded = true; // this default force merge could be needed depending on the following steps
+                        } else if (downsampleAction.forceMergeIndex()) {
+                            forceMergeExplicitlyEnabledCounter++;
+                        } else {
+                            forceMergeExplicitlyDisabledCounter++;
+                        }
+                    }
+
+                    // If there is an explicit force merge action, we could consider the downsampling force merge redundant.
+                    if (actions.containsKey(ForceMergeAction.NAME)) {
+                        downsampledForceMergeNeeded = false;
+                    }
+                    if (downsampledForceMergeNeeded && actions.containsKey(SearchableSnapshotAction.NAME)) {
+                        SearchableSnapshotAction searchableSnapshotAction = (SearchableSnapshotAction) actions.get(
+                            SearchableSnapshotAction.NAME
+                        );
+                        if (searchableSnapshotAction.isForceMergeIndex() == false) {
+                            // If there was a searchable snapshot with force index false, then the downsample force merge has impact
+                            downsampledForceMergeNeededCounter++;
+                        }
+                        downsampledForceMergeNeeded = false;
                     }
                 }
             }
-            return downsamplingPhases;
+            return new TimeSeriesFeatureSetUsage.IlmPolicyStats(
+                downsamplingPhases,
+                forceMergeExplicitlyEnabledCounter,
+                forceMergeExplicitlyDisabledCounter,
+                forceMergeDefaultCounter,
+                downsampledForceMergeNeededCounter
+            );
         }
     }
 }