Support different downsampling methods through ILM (#136951)

Following #136813, we expose to ILM the new sampling method config in the downsampling API. This will allow users to configure the sampling method in their downsample action of their ILM policies. For example:
```
PUT _ilm/policy/datastream_policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_docs": 1
          },
          "downsample": {
  	          "fixed_interval": "1h",
  	          "force_merge_index": false,
                  "sampling_method": "aggregate"
  	  }
        }
      }
    }
  }
}
```

Author: gmarouli

docs/changelog/136951.yaml

@@ -0,0 +1,5 @@
+pr: 136951
+summary: Support different downsampling methods through ILM
+area: "ILM+SLM"
+type: enhancement
+issues: []

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

@@ -19,9 +19,13 @@ 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`.
 
+`sampling_method` {applies_to}`stack: ga 9.3`
+:   (Optional, string) The sampling method that will be used to sample metrics; there are two methods available `aggregate` and
+the `last_value`. Defaults to `aggregate`.
 
 ## Example [ilm-downsample-ex]
 

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

@@ -0,0 +1 @@
+9210000

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

@@ -1 +1 @@
-ilm_searchable_snapshot_opt_out_clone,9209000
+add_sample_method_downsample_ilm,9210000

server/src/test/java/org/elasticsearch/action/downsample/DownsampleConfigTests.java

@@ -50,12 +50,11 @@ public static DownsampleConfig randomConfig() {
     }
 
     private static DownsampleConfig.SamplingMethod randomSamplingMethod() {
-        return switch (between(0, 2)) {
-            case 0 -> null;
-            case 1 -> DownsampleConfig.SamplingMethod.AGGREGATE;
-            case 2 -> DownsampleConfig.SamplingMethod.LAST_VALUE;
-            default -> throw new AssertionError("Illegal randomisation branch");
-        };
+        if (between(0, DownsampleConfig.SamplingMethod.values().length) == 0) {
+            return null;
+        } else {
+            return randomFrom(DownsampleConfig.SamplingMethod.values());
+        }
     }
 
     @Override

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

@@ -495,7 +495,7 @@ private IlmForceMergeInPolicies addIlmPolicies(Metadata.Builder metadataBuilder)
                     new Phase(
                         "hot",
                         TimeValue.ZERO,
-                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.MINUTE, null, hotForceMergeEnabled))
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.MINUTE, null, hotForceMergeEnabled, null))
                     )
                 )
             ),
@@ -506,13 +506,13 @@ private IlmForceMergeInPolicies addIlmPolicies(Metadata.Builder metadataBuilder)
                     new Phase(
                         "warm",
                         TimeValue.ZERO,
-                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.HOUR, null, warmForceMergeEnabled))
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.HOUR, null, warmForceMergeEnabled, null))
                     ),
                     "cold",
                     new Phase(
                         "cold",
                         TimeValue.timeValueDays(3),
-                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.DAY, null, coldForceMergeEnabled))
+                        Map.of("downsample", new DownsampleAction(DateHistogramInterval.DAY, null, coldForceMergeEnabled, null))
                     )
                 )
             ),

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
@@ -49,6 +50,7 @@ public class DownsampleAction implements LifecycleAction {
 
     private static final Logger logger = LogManager.getLogger(DownsampleAction.class);
     public static final TransportVersion ILM_FORCE_MERGE_IN_DOWNSAMPLING = TransportVersion.fromName("ilm_downsample_force_merge");
+    public static final TransportVersion ADD_SAMPLE_METHOD_DOWNSAMPLE_ILM = TransportVersion.fromName("add_sample_method_downsample_ilm");
 
     public static final String NAME = "downsample";
     public static final String DOWNSAMPLED_INDEX_PREFIX = "downsample-";
@@ -57,6 +59,7 @@ public class DownsampleAction implements LifecycleAction {
     public static final TimeValue DEFAULT_WAIT_TIMEOUT = new TimeValue(1, TimeUnit.DAYS);
     private static final ParseField FIXED_INTERVAL_FIELD = new ParseField(DownsampleConfig.FIXED_INTERVAL);
     private static final ParseField FORCE_MERGE_INDEX = new ParseField("force_merge_index");
+    private static final ParseField SAMPLING_METHOD_FIELD = new ParseField("sampling_method");
     private static final ParseField WAIT_TIMEOUT_FIELD = new ParseField("wait_timeout");
     static final String BWC_CLEANUP_TARGET_INDEX_NAME = "cleanup-target-index";
     private static final BiFunction<String, LifecycleExecutionState, String> DOWNSAMPLED_INDEX_NAME_SUPPLIER = (
@@ -65,7 +68,7 @@ public class DownsampleAction implements LifecycleAction {
 
     private static final ConstructingObjectParser<DownsampleAction, Void> PARSER = new ConstructingObjectParser<>(
         NAME,
-        a -> new DownsampleAction((DateHistogramInterval) a[0], (TimeValue) a[1], (Boolean) a[2])
+        a -> new DownsampleAction((DateHistogramInterval) a[0], (TimeValue) a[1], (Boolean) a[2], (DownsampleConfig.SamplingMethod) a[3])
     );
 
     static {
@@ -82,23 +85,37 @@ public class DownsampleAction implements LifecycleAction {
             ObjectParser.ValueType.STRING
         );
         PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), FORCE_MERGE_INDEX);
+        PARSER.declareField(
+            optionalConstructorArg(),
+            p -> DownsampleConfig.SamplingMethod.fromString(p.text()),
+            SAMPLING_METHOD_FIELD,
+            ObjectParser.ValueType.STRING
+        );
     }
 
     private final DateHistogramInterval fixedInterval;
+    @Nullable
+    private final DownsampleConfig.SamplingMethod samplingMethod;
     private final TimeValue waitTimeout;
     private final Boolean forceMergeIndex;
 
     public static DownsampleAction parse(XContentParser parser) {
         return PARSER.apply(parser, null);
     }
 
-    public DownsampleAction(final DateHistogramInterval fixedInterval, final TimeValue waitTimeout, Boolean forceMergeIndex) {
+    public DownsampleAction(
+        final DateHistogramInterval fixedInterval,
+        final TimeValue waitTimeout,
+        final Boolean forceMergeIndex,
+        final DownsampleConfig.SamplingMethod samplingMethod
+    ) {
         if (fixedInterval == null) {
             throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL_FIELD.getPreferredName() + "] is required.");
         }
         this.fixedInterval = fixedInterval;
         this.waitTimeout = waitTimeout == null ? DEFAULT_WAIT_TIMEOUT : waitTimeout;
         this.forceMergeIndex = forceMergeIndex;
+        this.samplingMethod = samplingMethod;
     }
 
     public DownsampleAction(StreamInput in) throws IOException {
@@ -107,7 +124,10 @@ public DownsampleAction(StreamInput in) throws IOException {
             in.getTransportVersion().onOrAfter(TransportVersions.V_8_10_X)
                 ? TimeValue.parseTimeValue(in.readString(), WAIT_TIMEOUT_FIELD.getPreferredName())
                 : DEFAULT_WAIT_TIMEOUT,
-            in.getTransportVersion().supports(ILM_FORCE_MERGE_IN_DOWNSAMPLING) ? in.readOptionalBoolean() : null
+            in.getTransportVersion().supports(ILM_FORCE_MERGE_IN_DOWNSAMPLING) ? in.readOptionalBoolean() : null,
+            in.getTransportVersion().supports(ADD_SAMPLE_METHOD_DOWNSAMPLE_ILM)
+                ? in.readOptionalWriteable(DownsampleConfig.SamplingMethod::read)
+                : null
         );
     }
 
@@ -122,6 +142,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().supports(ILM_FORCE_MERGE_IN_DOWNSAMPLING)) {
             out.writeOptionalBoolean(forceMergeIndex);
         }
+        if (out.getTransportVersion().supports(ADD_SAMPLE_METHOD_DOWNSAMPLE_ILM)) {
+            out.writeOptionalWriteable(samplingMethod);
+        }
     }
 
     @Override
@@ -132,6 +155,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (forceMergeIndex != null) {
             builder.field(FORCE_MERGE_INDEX.getPreferredName(), forceMergeIndex);
         }
+        if (samplingMethod != null) {
+            builder.field(SAMPLING_METHOD_FIELD.getPreferredName(), samplingMethod.toString());
+        }
         builder.endObject();
         return builder;
     }
@@ -153,6 +179,21 @@ public Boolean forceMergeIndex() {
         return forceMergeIndex;
     }
 
+    /**
+     * @return the sampling method configured in the ILM policy or null
+     */
+    @Nullable
+    public DownsampleConfig.SamplingMethod samplingMethod() {
+        return samplingMethod;
+    }
+
+    /**
+     * @return the sampling method that will be applied when the downsample occurs.
+     */
+    public DownsampleConfig.SamplingMethod samplingMethodOrDefault() {
+        return DownsampleConfig.SamplingMethod.getOrDefault(samplingMethod);
+    }
+
     @Override
     public boolean isSafeAction() {
         return false;
@@ -246,7 +287,14 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         );
 
         // Here is where the actual downsample action takes place
-        DownsampleStep downsampleStep = new DownsampleStep(downsampleKey, waitForDownsampleIndexKey, client, fixedInterval, waitTimeout);
+        DownsampleStep downsampleStep = new DownsampleStep(
+            downsampleKey,
+            waitForDownsampleIndexKey,
+            fixedInterval,
+            waitTimeout,
+            samplingMethod,
+            client
+        );
 
         // Wait until the downsampled index is recovered. We again wait until the configured threshold is breached and
         // if the downsampled index has not successfully recovered until then, we rewind to the "cleanup-downsample-index"
@@ -345,12 +393,13 @@ public boolean equals(Object o) {
         DownsampleAction that = (DownsampleAction) o;
         return Objects.equals(this.fixedInterval, that.fixedInterval)
             && Objects.equals(this.waitTimeout, that.waitTimeout)
-            && Objects.equals(this.forceMergeIndex, that.forceMergeIndex);
+            && Objects.equals(this.forceMergeIndex, that.forceMergeIndex)
+            && Objects.equals(this.samplingMethod, that.samplingMethod);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(fixedInterval, waitTimeout, forceMergeIndex);
+        return Objects.hash(fixedInterval, waitTimeout, forceMergeIndex, samplingMethod);
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java

@@ -34,17 +34,20 @@ public class DownsampleStep extends AsyncActionStep {
 
     private final DateHistogramInterval fixedInterval;
     private final TimeValue waitTimeout;
+    private final DownsampleConfig.SamplingMethod samplingMethod;
 
     public DownsampleStep(
         final StepKey key,
         final StepKey nextStepKey,
-        final Client client,
         final DateHistogramInterval fixedInterval,
-        final TimeValue waitTimeout
+        final TimeValue waitTimeout,
+        final DownsampleConfig.SamplingMethod samplingMethod,
+        final Client client
     ) {
         super(key, nextStepKey, client);
         this.fixedInterval = fixedInterval;
         this.waitTimeout = waitTimeout;
+        this.samplingMethod = samplingMethod;
     }
 
     @Override
@@ -95,7 +98,7 @@ public void performAction(
     }
 
     void performDownsampleIndex(ProjectId projectId, String indexName, String downsampleIndexName, ActionListener<Void> listener) {
-        DownsampleConfig config = new DownsampleConfig(fixedInterval);
+        DownsampleConfig config = new DownsampleConfig(fixedInterval, samplingMethod);
         DownsampleAction.Request request = new DownsampleAction.Request(
             TimeValue.MAX_VALUE,
             indexName,
@@ -119,9 +122,13 @@ public TimeValue getWaitTimeout() {
         return waitTimeout;
     }
 
+    public DownsampleConfig.SamplingMethod getSamplingMethod() {
+        return samplingMethod;
+    }
+
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), fixedInterval, waitTimeout);
+        return Objects.hash(super.hashCode(), fixedInterval, waitTimeout, samplingMethod);
     }
 
     @Override
@@ -136,7 +143,10 @@ public boolean equals(Object obj) {
             return false;
         }
         DownsampleStep other = (DownsampleStep) obj;
-        return super.equals(obj) && Objects.equals(fixedInterval, other.fixedInterval) && Objects.equals(waitTimeout, other.waitTimeout);
+        return super.equals(obj)
+            && Objects.equals(fixedInterval, other.fixedInterval)
+            && Objects.equals(waitTimeout, other.waitTimeout)
+            && Objects.equals(samplingMethod, other.samplingMethod);
     }
 
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleType.java

@@ -471,6 +471,22 @@ static void validateDownsamplingIntervals(Collection<Phase> phases) {
                         + "]"
                 );
             }
+            var firstSamplingMethod = firstDownsample.v2().samplingMethodOrDefault();
+            var secondSamplingMethod = secondDownsample.v2().samplingMethodOrDefault();
+            if (Objects.equals(firstSamplingMethod, secondSamplingMethod) == false) {
+                // All phases need to use the same downsampling method
+                throw new IllegalArgumentException(
+                    "Downsampling method ["
+                        + secondSamplingMethod
+                        + "] for phase ["
+                        + secondDownsample.v1()
+                        + "] must be compatible with the method ["
+                        + firstSamplingMethod
+                        + "] for phase ["
+                        + firstDownsample.v1()
+                        + "]"
+                );
+            }
             firstDownsample = secondDownsample;
         }
     }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/action/TimeSeriesUsageTransportActionTests.java

@@ -94,7 +94,7 @@ private LifecyclePolicy randomPolicy(Map<String, PhaseConfig> phases) {
             if (phaseConfig.hasDownsampling) {
                 actions.put(
                     DownsampleAction.NAME,
-                    new DownsampleAction(new DateHistogramInterval("1m"), null, phaseConfig.hasDownsamplingForceMerge)
+                    new DownsampleAction(new DateHistogramInterval("1m"), null, phaseConfig.hasDownsamplingForceMerge, null)
                 );
             }
             if (phaseConfig.hasForceMerge) {