Add new sampling method to the Downsample API (#136813)

When downsampling gauge metrics, we create an aggregate object that contains the min, max, sum and value_count values that allow us to respond to the min, max, sum, count and average aggregations without losing any accuracy.

However, we recognise that for certain cases keeping just the last value might be good enough. For this reason, we add one more configuration option in the downsample API that allows a user to choose the sampling method between:

aggregate: when supported, this will downsample a metric by summarising its values in a aggregate document. This remains the default
last_value: this will keep the last value and discard the rest.
```
POST /my-time-series-index/_downsample/my-downsampled-time-series-index
{
  "fixed_interval": "1d",
  "sampling_method": "last_value"
}
```
This PR introduces the sampling method only in the downsampling API. There will be follow ups that will introduce it in the lifecycle management features.

Relates: #128357.

Author: gmarouli

docs/changelog/136813.yaml

@@ -0,0 +1,5 @@
+pr: 136813
+summary: Add new sampling method to the Downsample API
+area: Downsampling
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/downsample/DownsampleConfig.java

@@ -9,12 +9,15 @@
 
 package org.elasticsearch.action.downsample;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Rounding;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
 import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
@@ -28,17 +31,21 @@
 
 import java.io.IOException;
 import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.Locale;
 import java.util.Objects;
 
 import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
+import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
 
 /**
  * This class holds the configuration details of a DownsampleAction that downsamples time series
  * (TSDB) indices. We have made great effort to simplify the rollup configuration and currently
- * only requires a fixed time interval. So, it has the following format:
+ * only requires a fixed time interval and optionally the sampling method. So, it has the following format:
  *
  *  {
- *    "fixed_interval" : "1d",
+ *    "fixed_interval": "1d",
+ *    "sampling_method": "aggregate"
  *  }
  *
  * fixed_interval is one or multiples of SI units and has no calendar-awareness (e.g. doesn't account
@@ -53,23 +60,28 @@
  * future extensions.
  */
 public class DownsampleConfig implements NamedWriteable, ToXContentObject {
+    public static final TransportVersion ADD_LAST_VALUE_DOWNSAMPLE_API = TransportVersion.fromName("add_last_value_downsample_api");
 
     private static final String NAME = "downsample/action/config";
     public static final String FIXED_INTERVAL = "fixed_interval";
+    public static final String SAMPLING_METHOD = "sampling_method";
     public static final String TIME_ZONE = "time_zone";
     public static final String DEFAULT_TIMEZONE = ZoneId.of("UTC").getId();
 
     private static final String timestampField = DataStreamTimestampFieldMapper.DEFAULT_PATH;
     private final DateHistogramInterval fixedInterval;
     private final String timeZone = DEFAULT_TIMEZONE;
     private final String intervalType = FIXED_INTERVAL;
+    @Nullable
+    private final SamplingMethod samplingMethod;
 
     private static final ConstructingObjectParser<DownsampleConfig, Void> PARSER;
+
     static {
         PARSER = new ConstructingObjectParser<>(NAME, a -> {
             DateHistogramInterval fixedInterval = (DateHistogramInterval) a[0];
             if (fixedInterval != null) {
-                return new DownsampleConfig(fixedInterval);
+                return new DownsampleConfig(fixedInterval, (SamplingMethod) a[1]);
             } else {
                 throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL + "] is required.");
             }
@@ -81,24 +93,48 @@ public class DownsampleConfig implements NamedWriteable, ToXContentObject {
             new ParseField(FIXED_INTERVAL),
             ObjectParser.ValueType.STRING
         );
+        PARSER.declareField(
+            optionalConstructorArg(),
+            p -> SamplingMethod.fromString(p.text()),
+            new ParseField(SAMPLING_METHOD),
+            ObjectParser.ValueType.STRING
+        );
     }
 
     /**
      * Create a new {@link DownsampleConfig} using the given configuration parameters.
      * @param fixedInterval the fixed interval to use for computing the date histogram for the rolled up documents (required).
+     * @deprecated please use {@link DownsampleConfig#DownsampleConfig(DateHistogramInterval, SamplingMethod)}, this method is being kept
+     * until the sampling method is completely integrated with ILM and DLM.
      */
+    @Deprecated
     public DownsampleConfig(final DateHistogramInterval fixedInterval) {
+        this(fixedInterval, null);
+    }
+
+    /**
+     * Create a new {@link DownsampleConfig} using the given configuration parameters.
+     * @param fixedInterval the fixed interval to use for computing the date histogram for the rolled up documents (required).
+     * @param samplingMethod the method used to downsample metrics, when null it default to {@link SamplingMethod#AGGREGATE}.
+     */
+    public DownsampleConfig(final DateHistogramInterval fixedInterval, @Nullable SamplingMethod samplingMethod) {
         if (fixedInterval == null) {
             throw new IllegalArgumentException("Parameter [" + FIXED_INTERVAL + "] is required.");
         }
         this.fixedInterval = fixedInterval;
+        this.samplingMethod = samplingMethod;
 
         // validate interval
         createRounding(this.fixedInterval.toString(), this.timeZone);
     }
 
     public DownsampleConfig(final StreamInput in) throws IOException {
         fixedInterval = new DateHistogramInterval(in);
+        if (in.getTransportVersion().supports(ADD_LAST_VALUE_DOWNSAMPLE_API)) {
+            samplingMethod = in.readOptionalWriteable(SamplingMethod::read);
+        } else {
+            samplingMethod = null;
+        }
     }
 
     /**
@@ -135,6 +171,9 @@ public static void validateSourceAndTargetIntervals(DownsampleConfig source, Dow
     @Override
     public void writeTo(final StreamOutput out) throws IOException {
         fixedInterval.writeTo(out);
+        if (out.getTransportVersion().supports(ADD_LAST_VALUE_DOWNSAMPLE_API)) {
+            out.writeOptionalWriteable(samplingMethod);
+        }
     }
 
     /**
@@ -180,6 +219,21 @@ public Rounding.Prepared createRounding() {
         return createRounding(fixedInterval.toString(), timeZone);
     }
 
+    /**
+     * @return the user configured sampling method
+     */
+    @Nullable
+    public SamplingMethod getSamplingMethod() {
+        return samplingMethod;
+    }
+
+    /**
+     * @return the sampling method that will be used based on this configuration.
+     */
+    public SamplingMethod getSamplingMethodOrDefault() {
+        return SamplingMethod.getOrDefault(samplingMethod);
+    }
+
     @Override
     public String getWriteableName() {
         return NAME;
@@ -195,7 +249,11 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
     }
 
     public XContentBuilder toXContentFragment(final XContentBuilder builder) throws IOException {
-        return builder.field(FIXED_INTERVAL, fixedInterval.toString());
+        builder.field(FIXED_INTERVAL, fixedInterval.toString());
+        if (samplingMethod != null) {
+            builder.field(SAMPLING_METHOD, samplingMethod.label);
+        }
+        return builder;
     }
 
     public static DownsampleConfig fromXContent(final XContentParser parser) throws IOException {
@@ -213,12 +271,13 @@ public boolean equals(final Object other) {
         final DownsampleConfig that = (DownsampleConfig) other;
         return Objects.equals(fixedInterval, that.fixedInterval)
             && Objects.equals(intervalType, that.intervalType)
-            && ZoneId.of(timeZone, ZoneId.SHORT_IDS).getRules().equals(ZoneId.of(that.timeZone, ZoneId.SHORT_IDS).getRules());
+            && ZoneId.of(timeZone, ZoneId.SHORT_IDS).getRules().equals(ZoneId.of(that.timeZone, ZoneId.SHORT_IDS).getRules())
+            && Objects.equals(samplingMethod, that.samplingMethod);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(fixedInterval, intervalType, ZoneId.of(timeZone));
+        return Objects.hash(fixedInterval, intervalType, ZoneId.of(timeZone), samplingMethod);
     }
 
     @Override
@@ -264,4 +323,95 @@ public static String generateDownsampleIndexName(
         }
         return prefix + fixedInterval + "-" + sourceIndexName;
     }
+
+    public enum SamplingMethod implements Writeable {
+        AGGREGATE((byte) 0, "aggregate"),
+        LAST_VALUE((byte) 1, "last_value");
+
+        private final byte id;
+        private final String label;
+
+        SamplingMethod(byte id, String label) {
+            this.id = id;
+            this.label = label;
+        }
+
+        byte id() {
+            return id;
+        }
+
+        public static SamplingMethod read(StreamInput in) throws IOException {
+            var id = in.readByte();
+            return switch (id) {
+                case 0 -> AGGREGATE;
+                case 1 -> LAST_VALUE;
+                default -> throw new IllegalArgumentException(
+                    "Sampling method id ["
+                        + id
+                        + "] is not one of the accepted ids "
+                        + Arrays.stream(values()).map(SamplingMethod::id).toList()
+                        + "."
+                );
+            };
+        }
+
+        /**
+         * Parses the configured sampling method from string (case-insensitive).
+         * @return the used sampling method, or null when the label is null.
+         */
+        @Nullable
+        public static SamplingMethod fromString(@Nullable String label) {
+            if (label == null) {
+                return null;
+            }
+            return switch (label.toLowerCase(Locale.ROOT)) {
+                case "aggregate" -> AGGREGATE;
+                case "last_value" -> LAST_VALUE;
+                default -> throw new IllegalArgumentException(
+                    "Sampling method ["
+                        + label
+                        + "] is not one of the accepted methods "
+                        + Arrays.stream(values()).map(SamplingMethod::toString).toList()
+                        + "."
+                );
+            };
+        }
+
+        /**
+         * Retrieves the configured sampling method from the index metadata. In case that it is null,
+         * it checks if the index is downsampled and returns the `aggregate` that was the only sampling
+         * method before we introduced last value.
+         * @return the used sampling method, or null if the index is not downsampled.
+         */
+        @Nullable
+        public static SamplingMethod fromIndexMetadata(IndexMetadata indexMetadata) {
+            SamplingMethod method = fromString(indexMetadata.getSettings().get(IndexMetadata.INDEX_DOWNSAMPLE_METHOD_KEY));
+            if (method != null) {
+                return method;
+            }
+            // Indices downsampled before the sampling method was introduced, will not have the sampling method in their metadata.
+            // For this reason, we first verify that they are indeed downsampled, and then we return `aggregate` because this was
+            // the only option available at the time.
+            boolean isIndexDownsampled = indexMetadata.getSettings().get(IndexMetadata.INDEX_DOWNSAMPLE_INTERVAL_KEY) != null;
+            return isIndexDownsampled ? AGGREGATE : null;
+        }
+
+        /**
+         * @return the sampling method that will be used based on this configuration. Default to {@link SamplingMethod#AGGREGATE}
+         * when the provided sampling method is null.
+         */
+        public static SamplingMethod getOrDefault(@Nullable SamplingMethod samplingMethod) {
+            return samplingMethod == null ? SamplingMethod.AGGREGATE : samplingMethod;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeByte(id);
+        }
+
+        @Override
+        public String toString() {
+            return label;
+        }
+    }
 }

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -1363,6 +1363,7 @@ public Index getResizeSourceIndex() {
 
     public static final String INDEX_DOWNSAMPLE_STATUS_KEY = "index.downsample.status";
     public static final String INDEX_DOWNSAMPLE_INTERVAL_KEY = "index.downsample.interval";
+    public static final String INDEX_DOWNSAMPLE_METHOD_KEY = "index.downsample.sampling_method";
     public static final Setting<String> INDEX_DOWNSAMPLE_SOURCE_UUID = Setting.simpleString(
         INDEX_DOWNSAMPLE_SOURCE_UUID_KEY,
         Property.IndexScope,
@@ -1411,6 +1412,12 @@ public String toString() {
         Property.InternalIndex
     );
 
+    public static final Setting<String> INDEX_DOWNSAMPLE_METHOD = Setting.simpleString(
+        INDEX_DOWNSAMPLE_METHOD_KEY,
+        Property.IndexScope,
+        Property.InternalIndex
+    );
+
     // LIFECYCLE_NAME is here an as optimization, see LifecycleSettings.LIFECYCLE_NAME and
     // LifecycleSettings.LIFECYCLE_NAME_SETTING for the 'real' version
     public static final String LIFECYCLE_NAME = "index.lifecycle.name";

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -89,6 +89,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexMetadata.INDEX_DOWNSAMPLE_ORIGIN_UUID,
                 IndexMetadata.INDEX_DOWNSAMPLE_STATUS,
                 IndexMetadata.INDEX_DOWNSAMPLE_INTERVAL,
+                IndexMetadata.INDEX_DOWNSAMPLE_METHOD,
                 SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG_SETTING,
                 SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN_SETTING,
                 SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO_SETTING,

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

@@ -0,0 +1 @@
+9199000

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

@@ -1 +1 @@
-shard_search_request_reshard_shard_count_summary,9198000
+add_last_value_downsample_api,9199000