Support choosing the downsampling method in data stream lifecycle (#137023)

Following #136813, we expose to data stream lifecycle the new sampling method config in the downsampling API. This will allow users to configure the sampling method directly in the lifecycle configuration. For example:
```
PUT _data_stream/my-ds/_lifecycle
{
  "data_retention": "10d",
  "downsampling_method": "last_value",
  "downsampling": [
     {
       "after": "1d",
       "fixed_interval": "5m
     }
  ]
}
```

Author: gmarouli

docs/changelog/137023.yaml

@@ -0,0 +1,5 @@
+pr: 137023
+summary: Support choosing the downsampling method in data stream lifecycle
+area: "Data streams"
+type: enhancement
+issues: []

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java

@@ -524,7 +524,10 @@ Set<Index> maybeExecuteDownsampling(ProjectState projectState, DataStream dataSt
                 // - has matching downsample rounds
                 // - is read-only
                 // So let's wait for an in-progress downsampling operation to succeed or trigger the last matching round
-                affectedIndices.addAll(waitForInProgressOrTriggerDownsampling(dataStream, backingIndexMeta, downsamplingRounds, project));
+                var downsamplingMethod = dataStream.getDataLifecycle().downsamplingMethod();
+                affectedIndices.addAll(
+                    waitForInProgressOrTriggerDownsampling(dataStream, backingIndexMeta, downsamplingRounds, downsamplingMethod, project)
+                );
             }
         }
 
@@ -541,6 +544,7 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
         DataStream dataStream,
         IndexMetadata backingIndex,
         List<DataStreamLifecycle.DownsamplingRound> downsamplingRounds,
+        DownsampleConfig.SamplingMethod downsamplingMethod,
         ProjectMetadata project
     ) {
         assert dataStream.getIndices().contains(backingIndex.getIndex())
@@ -556,7 +560,7 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
             String downsampleIndexName = DownsampleConfig.generateDownsampleIndexName(
                 DOWNSAMPLED_INDEX_PREFIX,
                 backingIndex,
-                round.config().getFixedInterval()
+                round.fixedInterval()
             );
             IndexMetadata targetDownsampleIndexMeta = project.index(downsampleIndexName);
             boolean targetDownsampleIndexExists = targetDownsampleIndexMeta != null;
@@ -568,7 +572,8 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
                     INDEX_DOWNSAMPLE_STATUS.get(targetDownsampleIndexMeta.getSettings()),
                     round,
                     lastRound,
-                    index,
+                    downsamplingMethod,
+                    backingIndex,
                     targetDownsampleIndexMeta.getIndex()
                 );
                 if (downsamplingNotComplete.isEmpty() == false) {
@@ -580,7 +585,7 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
                     // no maintenance needed for previously started downsampling actions and we are on the last matching round so it's time
                     // to kick off downsampling
                     affectedIndices.add(index);
-                    downsampleIndexOnce(round, project.id(), indexName, downsampleIndexName);
+                    downsampleIndexOnce(round, downsamplingMethod, project.id(), backingIndex, downsampleIndexName);
                 }
             }
         }
@@ -592,16 +597,30 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
      */
     private void downsampleIndexOnce(
         DataStreamLifecycle.DownsamplingRound round,
+        DownsampleConfig.SamplingMethod requestedDownsamplingMethod,
         ProjectId projectId,
-        String sourceIndex,
+        IndexMetadata sourceIndexMetadata,
         String downsampleIndexName
     ) {
+        // When an index is already downsampled with a method, we require all later downsampling rounds to use the same method.
+        // This is necessary to preserve the relation of the downsampled index to the raw data. For example, if an index is already
+        // downsampled and downsampled it again to 1 hour; we know that a document represents either the aggregated raw data of an hour
+        // or the last value of the raw data within this hour. If we mix the methods, we cannot derive any meaning from them.
+        // Furthermore, data stream lifecycle is configured on the data stream level and not on the individual index level, meaning that
+        // when a user changes downsampling method, some indices would not be able to be downsampled anymore.
+        // For this reason, when we encounter an already downsampled index, we use the source downsampling method which might be different
+        // from the requested one.
+        var sourceIndexSamplingMethod = DownsampleConfig.SamplingMethod.fromIndexMetadata(sourceIndexMetadata);
+        String sourceIndex = sourceIndexMetadata.getIndex().getName();
         DownsampleAction.Request request = new DownsampleAction.Request(
             TimeValue.THIRTY_SECONDS /* TODO should this be longer/configurable? */,
             sourceIndex,
             downsampleIndexName,
             null,
-            round.config()
+            new DownsampleConfig(
+                round.fixedInterval(),
+                sourceIndexSamplingMethod == null ? requestedDownsamplingMethod : sourceIndexSamplingMethod
+            )
         );
         transportActionsDeduplicator.executeOnce(
             Tuple.tuple(projectId, request),
@@ -632,11 +651,12 @@ private Set<Index> evaluateDownsampleStatus(
         IndexMetadata.DownsampleTaskStatus downsampleStatus,
         DataStreamLifecycle.DownsamplingRound currentRound,
         DataStreamLifecycle.DownsamplingRound lastRound,
-        Index backingIndex,
+        DownsampleConfig.SamplingMethod downsamplingMethod,
+        IndexMetadata backingIndex,
         Index downsampleIndex
     ) {
         Set<Index> affectedIndices = new HashSet<>();
-        String indexName = backingIndex.getName();
+        String indexName = backingIndex.getIndex().getName();
         String downsampleIndexName = downsampleIndex.getName();
         return switch (downsampleStatus) {
             case UNKNOWN -> {
@@ -683,15 +703,15 @@ private Set<Index> evaluateDownsampleStatus(
                 // NOTE that the downsample request is made through the deduplicator so it will only really be executed if
                 // there isn't one already in-flight. This can happen if a previous request timed-out, failed, or there was a
                 // master failover and data stream lifecycle needed to restart
-                downsampleIndexOnce(currentRound, projectId, indexName, downsampleIndexName);
-                affectedIndices.add(backingIndex);
+                downsampleIndexOnce(currentRound, downsamplingMethod, projectId, backingIndex, downsampleIndexName);
+                affectedIndices.add(backingIndex.getIndex());
                 yield affectedIndices;
             }
             case SUCCESS -> {
                 if (dataStream.getIndices().contains(downsampleIndex) == false) {
                     // at this point the source index is part of the data stream and the downsample index is complete but not
                     // part of the data stream. we need to replace the source index with the downsample index in the data stream
-                    affectedIndices.add(backingIndex);
+                    affectedIndices.add(backingIndex.getIndex());
                     replaceBackingIndexWithDownsampleIndexOnce(projectId, dataStream, indexName, downsampleIndexName);
                 }
                 yield affectedIndices;

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/rest/RestPutDataStreamLifecycleAction.java

@@ -21,6 +21,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Set;
 
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 import static org.elasticsearch.rest.RestUtils.getAckTimeout;
@@ -29,6 +30,9 @@
 @ServerlessScope(Scope.PUBLIC)
 public class RestPutDataStreamLifecycleAction extends BaseRestHandler {
 
+    private static final String SUPPORTS_DOWNSAMPLING_METHOD = "dlm.downsampling_method";
+    private static final Set<String> CAPABILITIES = Set.of(SUPPORTS_DOWNSAMPLING_METHOD);
+
     @Override
     public String getName() {
         return "put_data_lifecycles_action";
@@ -44,13 +48,14 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         try (XContentParser parser = request.contentParser()) {
             PutDataStreamLifecycleAction.Request putLifecycleRequest = PutDataStreamLifecycleAction.Request.parseRequest(
                 parser,
-                (dataRetention, enabled, downsampling) -> new PutDataStreamLifecycleAction.Request(
+                (dataRetention, enabled, downsamplingRounds, downsamplingMethod) -> new PutDataStreamLifecycleAction.Request(
                     getMasterNodeTimeout(request),
                     getAckTimeout(request),
                     Strings.splitStringByCommaToArray(request.param("name")),
                     dataRetention,
                     enabled,
-                    downsampling
+                    downsamplingRounds,
+                    downsamplingMethod
                 )
             );
             putLifecycleRequest.indicesOptions(IndicesOptions.fromRequest(request, putLifecycleRequest.indicesOptions()));
@@ -61,4 +66,9 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
             );
         }
     }
+
+    @Override
+    public Set<String> supportedCapabilities() {
+        return CAPABILITIES;
+    }
 }

modules/data-streams/src/test/java/org/elasticsearch/datastreams/MetadataIndexTemplateServiceTests.java

@@ -43,6 +43,8 @@
 import static org.elasticsearch.cluster.metadata.MetadataIndexTemplateService.composeDataLifecycles;
 import static org.elasticsearch.common.settings.Settings.builder;
 import static org.elasticsearch.datastreams.MetadataDataStreamRolloverServiceTests.createSettingsProvider;
+import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleFixtures.randomResettable;
+import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleFixtures.randomSamplingMethod;
 import static org.elasticsearch.indices.ShardLimitValidator.SETTING_CLUSTER_MAX_SHARDS_PER_NODE;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -146,51 +148,63 @@ public void testLifecycleComposition() {
             List<DataStreamLifecycle.Template> lifecycles = List.of();
             assertThat(composeDataLifecycles(lifecycles), nullValue());
         }
-        // One lifecycle results to this lifecycle as the final
+        // One lifecycle results in this lifecycle as the final
         {
+            ResettableValue<List<DataStreamLifecycle.DownsamplingRound>> downsamplingRounds = randomDownsampling();
             DataStreamLifecycle.Template lifecycle = DataStreamLifecycle.createDataLifecycleTemplate(
                 true,
                 randomRetention(),
-                randomDownsampling()
+                downsamplingRounds,
+                randomResettable(() -> randomSamplingMethod(downsamplingRounds.get()))
             );
             List<DataStreamLifecycle.Template> lifecycles = List.of(lifecycle);
             DataStreamLifecycle result = composeDataLifecycles(lifecycles).build();
             // Defaults to true
             assertThat(result.enabled(), equalTo(true));
             assertThat(result.dataRetention(), equalTo(lifecycle.dataRetention().get()));
-            assertThat(result.downsampling(), equalTo(lifecycle.downsampling().get()));
+            assertThat(result.downsamplingRounds(), equalTo(lifecycle.downsamplingRounds().get()));
+            assertThat(result.downsamplingMethod(), equalTo(lifecycle.downsamplingMethod().get()));
         }
         // If the last lifecycle is missing a property (apart from enabled) we keep the latest from the previous ones
         // Enabled is always true unless it's explicitly set to false
         {
+            List<DataStreamLifecycle.DownsamplingRound> downsamplingRounds = randomRounds();
             DataStreamLifecycle.Template lifecycle = DataStreamLifecycle.createDataLifecycleTemplate(
                 false,
                 randomPositiveTimeValue(),
-                randomRounds()
+                downsamplingRounds,
+                randomSamplingMethod(downsamplingRounds)
             );
             List<DataStreamLifecycle.Template> lifecycles = List.of(lifecycle, DataStreamLifecycle.Template.DATA_DEFAULT);
             DataStreamLifecycle result = composeDataLifecycles(lifecycles).build();
             assertThat(result.enabled(), equalTo(true));
             assertThat(result.dataRetention(), equalTo(lifecycle.dataRetention().get()));
-            assertThat(result.downsampling(), equalTo(lifecycle.downsampling().get()));
+            assertThat(result.downsamplingRounds(), equalTo(lifecycle.downsamplingRounds().get()));
+            assertThat(result.downsamplingMethod(), equalTo(lifecycle.downsamplingMethod().get()));
         }
-        // If both lifecycle have all properties, then the latest one overwrites all the others
+        // If both lifecycles have all properties, then the latest one overwrites all the others
         {
+            DownsampleConfig.SamplingMethod downsamplingMethod1 = randomFrom(DownsampleConfig.SamplingMethod.LAST_VALUE);
             DataStreamLifecycle.Template lifecycle1 = DataStreamLifecycle.createDataLifecycleTemplate(
                 false,
                 randomPositiveTimeValue(),
-                randomRounds()
+                randomRounds(),
+                downsamplingMethod1
             );
             DataStreamLifecycle.Template lifecycle2 = DataStreamLifecycle.createDataLifecycleTemplate(
                 true,
                 randomPositiveTimeValue(),
-                randomRounds()
+                randomRounds(),
+                downsamplingMethod1 == DownsampleConfig.SamplingMethod.LAST_VALUE
+                    ? DownsampleConfig.SamplingMethod.AGGREGATE
+                    : DownsampleConfig.SamplingMethod.LAST_VALUE
             );
             List<DataStreamLifecycle.Template> lifecycles = List.of(lifecycle1, lifecycle2);
             DataStreamLifecycle result = composeDataLifecycles(lifecycles).build();
             assertThat(result.enabled(), equalTo(lifecycle2.enabled()));
             assertThat(result.dataRetention(), equalTo(lifecycle2.dataRetention().get()));
-            assertThat(result.downsampling(), equalTo(lifecycle2.downsampling().get()));
+            assertThat(result.downsamplingRounds(), equalTo(lifecycle2.downsamplingRounds().get()));
+            assertThat(result.downsamplingMethod(), equalTo(lifecycle2.downsamplingMethod().get()));
         }
     }
 
@@ -255,7 +269,7 @@ private static List<DataStreamLifecycle.DownsamplingRound> randomRounds() {
         List<DataStreamLifecycle.DownsamplingRound> rounds = new ArrayList<>();
         var previous = new DataStreamLifecycle.DownsamplingRound(
             TimeValue.timeValueDays(randomIntBetween(1, 365)),
-            new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h"))
+            new DateHistogramInterval(randomIntBetween(1, 24) + "h")
         );
         rounds.add(previous);
         for (int i = 0; i < count; i++) {
@@ -268,9 +282,7 @@ private static List<DataStreamLifecycle.DownsamplingRound> randomRounds() {
 
     private static DataStreamLifecycle.DownsamplingRound nextRound(DataStreamLifecycle.DownsamplingRound previous) {
         var after = TimeValue.timeValueDays(previous.after().days() + randomIntBetween(1, 10));
-        var fixedInterval = new DownsampleConfig(
-            new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms")
-        );
+        var fixedInterval = new DateHistogramInterval((previous.fixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms");
         return new DataStreamLifecycle.DownsamplingRound(after, fixedInterval);
     }
 

modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java

@@ -154,7 +154,7 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
                 .setGeneration(3)
                 .setAllowCustomRouting(true)
                 .setIndexMode(IndexMode.STANDARD)
-                .setLifecycle(DataStreamLifecycle.createDataLifecycle(false, null, null))
+                .setLifecycle(DataStreamLifecycle.createDataLifecycle(false, null, null, null))
                 .setDataStreamOptions(DataStreamOptions.FAILURE_STORE_ENABLED)
                 .setFailureIndices(DataStream.DataStreamIndices.failureIndicesBuilder(failureStores).build())
                 .build();

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java

@@ -38,7 +38,9 @@
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
 import static org.elasticsearch.test.ESIntegTestCase.client;
+import static org.elasticsearch.test.ESTestCase.between;
 import static org.elasticsearch.test.ESTestCase.frequently;
+import static org.elasticsearch.test.ESTestCase.randomFrom;
 import static org.elasticsearch.test.ESTestCase.randomIntBetween;
 import static org.junit.Assert.assertTrue;
 
@@ -143,14 +145,18 @@ static void putComposableIndexTemplate(
     }
 
     static DataStreamLifecycle.Template randomDataLifecycleTemplate() {
+        ResettableValue<List<DataStreamLifecycle.DownsamplingRound>> downsampling = randomResettable(
+            DataStreamLifecycleFixtures::randomDownsamplingRounds
+        );
         return DataStreamLifecycle.createDataLifecycleTemplate(
             frequently(),
             randomResettable(ESTestCase::randomTimeValue),
-            randomResettable(DataStreamLifecycleFixtures::randomDownsamplingRounds)
+            downsampling,
+            randomResettable(() -> randomSamplingMethod(downsampling.get()))
         );
     }
 
-    private static <T> ResettableValue<T> randomResettable(Supplier<T> supplier) {
+    public static <T> ResettableValue<T> randomResettable(Supplier<T> supplier) {
         return switch (randomIntBetween(0, 2)) {
             case 0 -> ResettableValue.undefined();
             case 1 -> ResettableValue.reset();
@@ -164,7 +170,7 @@ private static List<DataStreamLifecycle.DownsamplingRound> randomDownsamplingRou
         List<DataStreamLifecycle.DownsamplingRound> rounds = new ArrayList<>();
         var previous = new DataStreamLifecycle.DownsamplingRound(
             TimeValue.timeValueDays(randomIntBetween(1, 365)),
-            new DownsampleConfig(new DateHistogramInterval(randomIntBetween(1, 24) + "h"))
+            new DateHistogramInterval(randomIntBetween(1, 24) + "h")
         );
         rounds.add(previous);
         for (int i = 0; i < count; i++) {
@@ -177,9 +183,19 @@ private static List<DataStreamLifecycle.DownsamplingRound> randomDownsamplingRou
 
     private static DataStreamLifecycle.DownsamplingRound nextRound(DataStreamLifecycle.DownsamplingRound previous) {
         var after = TimeValue.timeValueDays(previous.after().days() + randomIntBetween(1, 10));
-        var fixedInterval = new DownsampleConfig(
-            new DateHistogramInterval((previous.config().getFixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms")
-        );
+        var fixedInterval = new DateHistogramInterval((previous.fixedInterval().estimateMillis() * randomIntBetween(2, 5)) + "ms");
         return new DataStreamLifecycle.DownsamplingRound(after, fixedInterval);
     }
+
+    /**
+     * In order to produce valid data stream lifecycle configurations, the sampling method can be defined only when
+     * the downsampling rounds are also defined.
+     */
+    public static DownsampleConfig.SamplingMethod randomSamplingMethod(List<DataStreamLifecycle.DownsamplingRound> downsamplingRounds) {
+        if (downsamplingRounds == null || between(0, DownsampleConfig.SamplingMethod.values().length) == 0) {
+            return null;
+        } else {
+            return randomFrom(DownsampleConfig.SamplingMethod.values());
+        }
+    }
 }