Use the sampling method in the data stream lifecycle

Author: gmarouli

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())
@@ -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,22 @@ private Set<Index> waitForInProgressOrTriggerDownsampling(
      */
     private void downsampleIndexOnce(
         DataStreamLifecycle.DownsamplingRound round,
+        DownsampleConfig.SamplingMethod requestedDownsamplingMethod,
         ProjectId projectId,
-        String sourceIndex,
+        IndexMetadata sourceIndexMetadata,
         String downsampleIndexName
     ) {
+        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,
-            new DownsampleConfig(round.fixedInterval(), null)
+            new DownsampleConfig(
+                round.fixedInterval(),
+                sourceIndexSamplingMethod == null ? requestedDownsamplingMethod : sourceIndexSamplingMethod
+            )
         );
         transportActionsDeduplicator.executeOnce(
             Tuple.tuple(projectId, request),
@@ -632,11 +643,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 +695,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;