GET _data_stream displays both ILM and DSL information (#99947)

This add support to the `GET _data_stream` API for displaying the value
of the `index.lifecycle.prefer_ilm` setting both at the backing index
level and at the top level (top level meaning, similarly to the existing
`ilm_policy` field, the value in the index template that's backing the
data stream), an `ilm_policy` field for each backing index displaying
the actual ILM policy configured for the index itself, a `managed_by`
field for each backing index indicating who manages this index (the
possible values are: `Index Lifecycle Management`, `Data stream
lifecycle`, and `Unmanaged`).

This also adds a top level field to indicate which system would manage
the next generation index for this data stream based on the current
configuration. This field is called `next_generation_managed_by` and the
same values as the indices level `managed_by` field has are available.

An example output for a data stream that has 2 backing indices managed
by ILM and the write index by DSL:

```
{
	"data_streams": [{
		"name": "datastream-psnyudmbitp",
		"timestamp_field": {
			"name": "@timestamp"
		},
		"indices": [{
			"index_name": ".ds-datastream-psnyudmbitp-2023.09.27-000001",
			"index_uuid": "kyw0WEXvS8-ahchYS10NRQ",
                        "prefer_ilm": true,
			"ilm_policy": "policy-uVBEI",
			"managed_by": "Index Lifecycle Management"
		}, {
			"index_name": ".ds-datastream-psnyudmbitp-2023.09.27-000002",
			"index_uuid": "pDLdc4DERwO54GRzDr4krw",
			"prefer_ilm": true,
			"ilm_policy": "policy-uVBEI",
			"managed_by": "Index Lifecycle Management"
		}, {
			"index_name": ".ds-datastream-psnyudmbitp-2023.09.27-000003",
			"index_uuid": "gYZirLKcS3mlc1c3oHRpYw",
			"prefer_ilm": false,
			"ilm_policy": "policy-uVBEI",
                        "managed_by": "Data stream lifecycle"
		}],
		"generation": 3,
		"status": "YELLOW",
		"template": "indextemplate-obcvkbjqand",
		"lifecycle": {
			"enabled": true,
			"data_retention": "90d"
		},
		"ilm_policy": "policy-uVBEI",
                "next_generation_managed_by": "Data stream lifecycle",
		"prefer_ilm": false,
		"hidden": false,
		"system": false,
		"allow_custom_routing": false,
		"replicated": false
	}]
}
```

docs/changelog/99947.yaml

@@ -0,0 +1,5 @@
+pr: 99947
+summary: GET `_data_stream` displays both ILM and DSL information
+area: Data streams
+type: feature
+issues: []

docs/reference/data-streams/change-mappings-and-settings.asciidoc

@@ -573,15 +573,21 @@ stream's oldest backing index.
       "indices": [
         {
           "index_name": ".ds-my-data-stream-2099.03.07-000001", <1>
-          "index_uuid": "Gpdiyq8sRuK9WuthvAdFbw"
+          "index_uuid": "Gpdiyq8sRuK9WuthvAdFbw",
+          "prefer_ilm": true,
+          "managed_by": "Unmanaged"
         },
         {
           "index_name": ".ds-my-data-stream-2099.03.08-000002",
-          "index_uuid": "_eEfRrFHS9OyhqWntkgHAQ"
+          "index_uuid": "_eEfRrFHS9OyhqWntkgHAQ",
+          "prefer_ilm": true,
+          "managed_by": "Unmanaged"
         }
       ],
       "generation": 2,
       "status": "GREEN",
+      "next_generation_managed_by": "Unmanaged",
+      "prefer_ilm": true,
       "template": "my-data-stream-template",
       "hidden": false,
       "system": false,

docs/reference/data-streams/downsampling-manual.asciidoc

@@ -358,11 +358,15 @@ This returns:
       "indices": [
         {
           "index_name": ".ds-my-data-stream-2023.07.26-000001", <1>
-          "index_uuid": "ltOJGmqgTVm4T-Buoe7Acg"
+          "index_uuid": "ltOJGmqgTVm4T-Buoe7Acg",
+          "prefer_ilm": true,
+          "managed_by": "Data stream lifecycle"
         }
       ],
       "generation": 1,
       "status": "GREEN",
+      "next_generation_managed_by": "Data stream lifecycle",
+      "prefer_ilm": true,
       "template": "my-data-stream-template",
       "hidden": false,
       "system": false,

docs/reference/indices/get-data-stream.asciidoc

@@ -225,7 +225,7 @@ cluster can not write into this data stream or change its mappings.
 
 `lifecycle`::
 (object)
-Functionality in preview:[]. Contains the configuration for the data stream lifecycle management of this data stream.
+Contains the configuration for the data stream lifecycle management of this data stream.
 +
 .Properties of `lifecycle`
 [%collapsible%open]
@@ -265,11 +265,17 @@ The API returns the following response:
       "indices": [
         {
           "index_name": ".ds-my-data-stream-2099.03.07-000001",
-          "index_uuid": "xCEhwsp8Tey0-FLNFYVwSg"
+          "index_uuid": "xCEhwsp8Tey0-FLNFYVwSg",
+          "prefer_ilm": true,
+          "ilm_policy": "my-lifecycle-policy",
+          "managed_by": "Index Lifecycle Management"
         },
         {
           "index_name": ".ds-my-data-stream-2099.03.08-000002",
-          "index_uuid": "PA_JquKGSiKcAKBA8DJ5gw"
+          "index_uuid": "PA_JquKGSiKcAKBA8DJ5gw",
+          "prefer_ilm": true,
+          "ilm_policy": "my-lifecycle-policy",
+          "managed_by": "Index Lifecycle Management"
         }
       ],
       "generation": 2,
@@ -277,6 +283,8 @@ The API returns the following response:
         "my-meta-field": "foo"
       },
       "status": "GREEN",
+      "next_generation_managed_by": "Index Lifecycle Management",
+      "prefer_ilm": true,
       "template": "my-index-template",
       "ilm_policy": "my-lifecycle-policy",
       "hidden": false,
@@ -292,7 +300,10 @@ The API returns the following response:
       "indices": [
         {
           "index_name": ".ds-my-data-stream-two-2099.03.08-000001",
-          "index_uuid": "3liBu2SYS5axasRt6fUIpA"
+          "index_uuid": "3liBu2SYS5axasRt6fUIpA",
+          "prefer_ilm": true,
+          "ilm_policy": "my-lifecycle-policy",
+          "managed_by": "Index Lifecycle Management"
         }
       ],
       "generation": 1,
@@ -300,6 +311,8 @@ The API returns the following response:
         "my-meta-field": "foo"
       },
       "status": "YELLOW",
+      "next_generation_managed_by": "Index Lifecycle Management",
+      "prefer_ilm": true,
       "template": "my-index-template",
       "ilm_policy": "my-lifecycle-policy",
       "hidden": false,

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java

@@ -1320,11 +1320,17 @@ public class DataStreamIT extends ESIntegTestCase {
         ).actionGet();
         assertThat(response.getDataStreams().size(), is(1));
         DataStreamInfo metricsFooDataStream = response.getDataStreams().get(0);
-        assertThat(metricsFooDataStream.getDataStream().getName(), is("metrics-foo"));
+        DataStream dataStream = metricsFooDataStream.getDataStream();
+        assertThat(dataStream.getName(), is("metrics-foo"));
         assertThat(metricsFooDataStream.getDataStreamStatus(), is(ClusterHealthStatus.YELLOW));
         assertThat(metricsFooDataStream.getIndexTemplate(), is("template_for_foo"));
         assertThat(metricsFooDataStream.getIlmPolicy(), is(nullValue()));
-        assertThat(metricsFooDataStream.getDataStream().getLifecycle(), is(lifecycle));
+        assertThat(dataStream.getLifecycle(), is(lifecycle));
+        assertThat(metricsFooDataStream.templatePreferIlmValue(), is(true));
+        GetDataStreamAction.Response.IndexProperties indexProperties = metricsFooDataStream.getIndexSettingsValues()
+            .get(dataStream.getWriteIndex());
+        assertThat(indexProperties.ilmPolicyName(), is(nullValue()));
+        assertThat(indexProperties.preferIlm(), is(true));
     }
 
     private static void assertBackingIndex(String backingIndex, String timestampFieldPathInMapping, Map<?, ?> expectedMapping) {

modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/GetDataStreamsTransportAction.java

@@ -11,6 +11,8 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
+import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.IndexProperties;
+import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.ManagedBy;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
 import org.elasticsearch.cluster.ClusterState;
@@ -21,6 +23,7 @@ import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
@@ -39,9 +42,12 @@ import org.elasticsearch.transport.TransportService;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.index.IndexSettings.PREFER_ILM_SETTING;
+
 public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction<
     GetDataStreamAction.Request,
     GetDataStreamAction.Response> {
@@ -95,6 +101,7 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
         List<GetDataStreamAction.Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
         for (DataStream dataStream : dataStreams) {
             final String indexTemplate;
+            boolean indexTemplatePreferIlmValue = true;
             String ilmPolicyName = null;
             if (dataStream.isSystem()) {
                 SystemDataStreamDescriptor dataStreamDescriptor = systemIndices.findMatchingDataStreamDescriptor(dataStream.getName());
@@ -104,13 +111,15 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
                         dataStreamDescriptor.getComposableIndexTemplate(),
                         dataStreamDescriptor.getComponentTemplates()
                     );
-                    ilmPolicyName = settings.get("index.lifecycle.name");
+                    ilmPolicyName = settings.get(IndexMetadata.LIFECYCLE_NAME);
+                    indexTemplatePreferIlmValue = PREFER_ILM_SETTING.get(settings);
                 }
             } else {
                 indexTemplate = MetadataIndexTemplateService.findV2Template(state.metadata(), dataStream.getName(), false);
                 if (indexTemplate != null) {
                     Settings settings = MetadataIndexTemplateService.resolveSettings(state.metadata(), indexTemplate);
-                    ilmPolicyName = settings.get("index.lifecycle.name");
+                    ilmPolicyName = settings.get(IndexMetadata.LIFECYCLE_NAME);
+                    indexTemplatePreferIlmValue = PREFER_ILM_SETTING.get(settings);
                 } else {
                     LOGGER.warn(
                         "couldn't find any matching template for data stream [{}]. has it been restored (and possibly renamed)"
@@ -125,18 +134,35 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
                 dataStream.getIndices().stream().map(Index::getName).toArray(String[]::new)
             );
 
+            Map<Index, IndexProperties> backingIndicesSettingsValues = new HashMap<>();
+            Metadata metadata = state.getMetadata();
+            for (Index index : dataStream.getIndices()) {
+                IndexMetadata indexMetadata = metadata.index(index);
+                Boolean preferIlm = PREFER_ILM_SETTING.get(indexMetadata.getSettings());
+                assert preferIlm != null : "must use the default prefer ilm setting value, if nothing else";
+                ManagedBy managedBy;
+                if (metadata.isIndexManagedByILM(indexMetadata)) {
+                    managedBy = ManagedBy.ILM;
+                } else if (dataStream.isIndexManagedByDataStreamLifecycle(index, metadata::index)) {
+                    managedBy = ManagedBy.LIFECYCLE;
+                } else {
+                    managedBy = ManagedBy.UNMANAGED;
+                }
+                backingIndicesSettingsValues.put(index, new IndexProperties(preferIlm, indexMetadata.getLifecyclePolicyName(), managedBy));
+            }
+
             GetDataStreamAction.Response.TimeSeries timeSeries = null;
             if (dataStream.getIndexMode() == IndexMode.TIME_SERIES) {
                 List<Tuple<Instant, Instant>> ranges = new ArrayList<>();
                 Tuple<Instant, Instant> current = null;
                 String previousIndexName = null;
                 for (Index index : dataStream.getIndices()) {
-                    IndexMetadata metadata = state.getMetadata().index(index);
-                    if (metadata.getIndexMode() != IndexMode.TIME_SERIES) {
+                    IndexMetadata indexMetadata = metadata.index(index);
+                    if (indexMetadata.getIndexMode() != IndexMode.TIME_SERIES) {
                         continue;
                     }
-                    Instant start = metadata.getTimeSeriesStart();
-                    Instant end = metadata.getTimeSeriesEnd();
+                    Instant start = indexMetadata.getTimeSeriesStart();
+                    Instant end = indexMetadata.getTimeSeriesEnd();
                     if (current == null) {
                         current = new Tuple<>(start, end);
                     } else if (current.v2().compareTo(start) == 0) {
@@ -175,7 +201,9 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
                     streamHealth.getStatus(),
                     indexTemplate,
                     ilmPolicyName,
-                    timeSeries
+                    timeSeries,
+                    backingIndicesSettingsValues,
+                    indexTemplatePreferIlmValue
                 )
             );
         }

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

@@ -8,15 +8,33 @@
 package org.elasticsearch.datastreams.action;
 
 import org.elasticsearch.action.datastreams.GetDataStreamAction.Response;
+import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.ManagedBy;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentFactory;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 
 public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase<Response> {
 
@@ -43,13 +61,198 @@ public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase
         return new Response(instance.getDataStreams().stream().map(this::mutateInstance).toList());
     }
 
+    @SuppressWarnings("unchecked")
+    public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Exception {
+        // we'll test a data stream with 3 backing indices - two managed by ILM (having the ILM policy configured for them)
+        // and one without any ILM policy configured
+        String dataStreamName = "logs";
+
+        Index firstGenerationIndex = new Index(getDefaultBackingIndexName(dataStreamName, 1), UUIDs.base64UUID());
+        Index secondGenerationIndex = new Index(getDefaultBackingIndexName(dataStreamName, 2), UUIDs.base64UUID());
+        Index writeIndex = new Index(getDefaultBackingIndexName(dataStreamName, 3), UUIDs.base64UUID());
+        List<Index> indices = List.of(firstGenerationIndex, secondGenerationIndex, writeIndex);
+        {
+            // data stream has an enabled lifecycle
+            DataStream logs = new DataStream(
+                "logs",
+                indices,
+                3,
+                null,
+                false,
+                false,
+                false,
+                true,
+                IndexMode.STANDARD,
+                new DataStreamLifecycle()
+            );
+
+            String ilmPolicyName = "rollover-30days";
+            Map<Index, Response.IndexProperties> indexSettingsValues = Map.of(
+                firstGenerationIndex,
+                new Response.IndexProperties(true, ilmPolicyName, ManagedBy.ILM),
+                secondGenerationIndex,
+                new Response.IndexProperties(false, ilmPolicyName, ManagedBy.LIFECYCLE),
+                writeIndex,
+                new Response.IndexProperties(false, null, ManagedBy.LIFECYCLE)
+            );
+
+            Response.DataStreamInfo dataStreamInfo = new Response.DataStreamInfo(
+                logs,
+                ClusterHealthStatus.GREEN,
+                "index-template",
+                null,
+                null,
+                indexSettingsValues,
+                false
+            );
+            Response response = new Response(List.of(dataStreamInfo));
+            XContentBuilder contentBuilder = XContentFactory.jsonBuilder();
+            response.toXContent(contentBuilder, ToXContent.EMPTY_PARAMS);
+
+            BytesReference bytes = BytesReference.bytes(contentBuilder);
+            try (XContentParser parser = createParser(JsonXContent.jsonXContent, bytes)) {
+                Map<String, Object> map = parser.map();
+                List<Object> dataStreams = (List<Object>) map.get(Response.DATA_STREAMS_FIELD.getPreferredName());
+                assertThat(dataStreams.size(), is(1));
+                Map<String, Object> dataStreamMap = (Map<String, Object>) dataStreams.get(0);
+                assertThat(dataStreamMap.get(DataStream.NAME_FIELD.getPreferredName()), is(dataStreamName));
+
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.LIFECYCLE_FIELD.getPreferredName()), is(Map.of("enabled", true)));
+                assertThat(
+                    dataStreamMap.get(Response.DataStreamInfo.NEXT_GENERATION_INDEX_MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.LIFECYCLE.displayValue)
+                );
+
+                List<Object> indicesRepresentation = (List<Object>) dataStreamMap.get(DataStream.INDICES_FIELD.getPreferredName());
+                Map<String, Object> firstGenIndexRepresentation = (Map<String, Object>) indicesRepresentation.get(0);
+                assertThat(firstGenIndexRepresentation.get("index_name"), is(firstGenerationIndex.getName()));
+                assertThat(firstGenIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(true));
+                assertThat(firstGenIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(ilmPolicyName));
+                assertThat(
+                    firstGenIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.ILM.displayValue)
+                );
+
+                Map<String, Object> secondGenIndexRepresentation = (Map<String, Object>) indicesRepresentation.get(1);
+                assertThat(secondGenIndexRepresentation.get("index_name"), is(secondGenerationIndex.getName()));
+                assertThat(secondGenIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
+                assertThat(
+                    secondGenIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()),
+                    is(ilmPolicyName)
+                );
+                assertThat(
+                    secondGenIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.LIFECYCLE.displayValue)
+                );
+
+                // the write index is managed by data stream lifecycle
+                Map<String, Object> writeIndexRepresentation = (Map<String, Object>) indicesRepresentation.get(2);
+                assertThat(writeIndexRepresentation.get("index_name"), is(writeIndex.getName()));
+                assertThat(writeIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
+                assertThat(writeIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(
+                    writeIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.LIFECYCLE.displayValue)
+                );
+            }
+        }
+
+        {
+            // data stream has a lifecycle that's not enabled
+            DataStream logs = new DataStream(
+                "logs",
+                indices,
+                3,
+                null,
+                false,
+                false,
+                false,
+                true,
+                IndexMode.STANDARD,
+                new DataStreamLifecycle(null, null, false)
+            );
+
+            String ilmPolicyName = "rollover-30days";
+            Map<Index, Response.IndexProperties> indexSettingsValues = Map.of(
+                firstGenerationIndex,
+                new Response.IndexProperties(true, ilmPolicyName, ManagedBy.ILM),
+                secondGenerationIndex,
+                new Response.IndexProperties(true, ilmPolicyName, ManagedBy.ILM),
+                writeIndex,
+                new Response.IndexProperties(false, null, ManagedBy.UNMANAGED)
+            );
+
+            Response.DataStreamInfo dataStreamInfo = new Response.DataStreamInfo(
+                logs,
+                ClusterHealthStatus.GREEN,
+                "index-template",
+                null,
+                null,
+                indexSettingsValues,
+                false
+            );
+            Response response = new Response(List.of(dataStreamInfo));
+            XContentBuilder contentBuilder = XContentFactory.jsonBuilder();
+            response.toXContent(contentBuilder, ToXContent.EMPTY_PARAMS);
+
+            BytesReference bytes = BytesReference.bytes(contentBuilder);
+            try (XContentParser parser = createParser(JsonXContent.jsonXContent, bytes)) {
+                Map<String, Object> map = parser.map();
+                List<Object> dataStreams = (List<Object>) map.get(Response.DATA_STREAMS_FIELD.getPreferredName());
+                assertThat(dataStreams.size(), is(1));
+                Map<String, Object> dataStreamMap = (Map<String, Object>) dataStreams.get(0);
+                assertThat(dataStreamMap.get(DataStream.NAME_FIELD.getPreferredName()), is(dataStreamName));
+                // note that the prefer_ilm value is displayed at the top level even if the template backing the data stream doesn't have a
+                // policy specified anymore
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(dataStreamMap.get(Response.DataStreamInfo.LIFECYCLE_FIELD.getPreferredName()), is(Map.of("enabled", false)));
+                assertThat(
+                    dataStreamMap.get(Response.DataStreamInfo.NEXT_GENERATION_INDEX_MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.UNMANAGED.displayValue)
+                );
+
+                List<Object> indicesRepresentation = (List<Object>) dataStreamMap.get(DataStream.INDICES_FIELD.getPreferredName());
+                Map<String, Object> firstGenIndexRepresentation = (Map<String, Object>) indicesRepresentation.get(0);
+                assertThat(firstGenIndexRepresentation.get("index_name"), is(firstGenerationIndex.getName()));
+                assertThat(firstGenIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(true));
+                assertThat(firstGenIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(ilmPolicyName));
+                assertThat(
+                    firstGenIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.ILM.displayValue)
+                );
+
+                // the write index is managed by data stream lifecycle
+                Map<String, Object> writeIndexRepresentation = (Map<String, Object>) indicesRepresentation.get(2);
+                assertThat(writeIndexRepresentation.get("index_name"), is(writeIndex.getName()));
+                assertThat(writeIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
+                assertThat(writeIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(
+                    writeIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.UNMANAGED.displayValue)
+                );
+            }
+        }
+    }
+
+    public void testManagedByDisplayValuesDontAccidentalyChange() {
+        // UI might derive logic based on the display values so any changes should be coordinated with the UI team
+        assertThat(ManagedBy.ILM.displayValue, is("Index Lifecycle Management"));
+        assertThat(ManagedBy.LIFECYCLE.displayValue, is("Data stream lifecycle"));
+        assertThat(ManagedBy.UNMANAGED.displayValue, is("Unmanaged"));
+    }
+
     private Response.DataStreamInfo mutateInstance(Response.DataStreamInfo instance) {
         var dataStream = instance.getDataStream();
         var status = instance.getDataStreamStatus();
         var indexTemplate = instance.getIndexTemplate();
         var ilmPolicyName = instance.getIlmPolicy();
         var timeSeries = instance.getTimeSeries();
-        switch (randomIntBetween(0, 4)) {
+        var indexSettings = instance.getIndexSettingsValues();
+        var templatePreferIlm = instance.templatePreferIlmValue();
+        switch (randomIntBetween(0, 6)) {
             case 0 -> dataStream = randomValueOtherThan(dataStream, DataStreamTestHelper::randomInstance);
             case 1 -> status = randomValueOtherThan(status, () -> randomFrom(ClusterHealthStatus.values()));
             case 2 -> indexTemplate = randomBoolean() && indexTemplate != null ? null : randomAlphaOfLengthBetween(2, 10);
@@ -57,8 +260,22 @@ public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase
             case 4 -> timeSeries = randomBoolean() && timeSeries != null
                 ? null
                 : randomValueOtherThan(timeSeries, () -> new Response.TimeSeries(generateRandomTimeSeries()));
+            case 5 -> indexSettings = randomValueOtherThan(
+                indexSettings,
+                () -> randomBoolean()
+                    ? Map.of()
+                    : Map.of(
+                        new Index(randomAlphaOfLengthBetween(50, 100), UUIDs.base64UUID()),
+                        new Response.IndexProperties(
+                            randomBoolean(),
+                            randomAlphaOfLengthBetween(50, 100),
+                            randomBoolean() ? ManagedBy.ILM : ManagedBy.LIFECYCLE
+                        )
+                    )
+            );
+            case 6 -> templatePreferIlm = templatePreferIlm ? false : true;
         }
-        return new Response.DataStreamInfo(dataStream, status, indexTemplate, ilmPolicyName, timeSeries);
+        return new Response.DataStreamInfo(dataStream, status, indexTemplate, ilmPolicyName, timeSeries, indexSettings, templatePreferIlm);
     }
 
     private List<Tuple<Instant, Instant>> generateRandomTimeSeries() {
@@ -70,6 +287,21 @@ public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase
         return timeSeries;
     }
 
+    private Map<Index, Response.IndexProperties> generateRandomIndexSettingsValues() {
+        Map<Index, Response.IndexProperties> values = new HashMap<>();
+        for (int i = 0; i < randomIntBetween(0, 3); i++) {
+            values.put(
+                new Index(randomAlphaOfLengthBetween(50, 100), UUIDs.base64UUID()),
+                new Response.IndexProperties(
+                    randomBoolean(),
+                    randomAlphaOfLengthBetween(50, 100),
+                    randomBoolean() ? ManagedBy.ILM : ManagedBy.LIFECYCLE
+                )
+            );
+        }
+        return values;
+    }
+
     private Response.DataStreamInfo generateRandomDataStreamInfo() {
         List<Tuple<Instant, Instant>> timeSeries = randomBoolean() ? generateRandomTimeSeries() : null;
         return new Response.DataStreamInfo(
@@ -77,7 +309,9 @@ public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase
             ClusterHealthStatus.GREEN,
             randomAlphaOfLengthBetween(2, 10),
             randomAlphaOfLengthBetween(2, 10),
-            timeSeries != null ? new Response.TimeSeries(timeSeries) : null
+            timeSeries != null ? new Response.TimeSeries(timeSeries) : null,
+            generateRandomIndexSettingsValues(),
+            randomBoolean()
         );
     }
 }

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/10_basic.yml

@@ -311,6 +311,77 @@ setup:
         name: simple-data-stream2
   - is_true: acknowledged
 
+---
+"Get data stream and check DSL and ILM information":
+  - skip:
+      version: " - 8.10.99"
+      reason: "data streams DSL and ILM mixing information available in 8.11+"
+
+  - do:
+      allowed_warnings:
+        - "index template [mixing-dsl-template] has index patterns [mixing-dsl-stream] matching patterns from existing older templates
+        [global] with patterns (global => [*]); this template [mixing-dsl-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: mixing-dsl-template
+        body:
+          index_patterns: [mixing-dsl-stream]
+          template:
+            mappings:
+              properties:
+                '@timestamp':
+                  type: date_nanos
+            lifecycle:
+              data_retention: "30d"
+              enabled: false
+            settings:
+              index.lifecycle.prefer_ilm: false
+              index.lifecycle.name: "missing_ilm_policy"
+          data_stream: {}
+
+  - do:
+      indices.create_data_stream:
+        name: mixing-dsl-stream
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_stream:
+        name: mixing-dsl-stream
+  - match: { data_streams.0.name: mixing-dsl-stream }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - match: { data_streams.0.ilm_policy: "missing_ilm_policy" }
+  - match: { data_streams.0.prefer_ilm: false }
+  - match: { data_streams.0.next_generation_managed_by: "Index Lifecycle Management" }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.prefer_ilm: false }
+  - match: { data_streams.0.indices.0.ilm_policy: "missing_ilm_policy" }
+  - match: { data_streams.0.indices.0.managed_by: "Index Lifecycle Management" }
+
+  - do:
+      indices.put_data_lifecycle:
+        name: "*"
+        body: >
+          {
+            "data_retention": "30d",
+            "enabled": true
+          }
+
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_stream:
+        name: mixing-dsl-stream
+  - match: { data_streams.0.name: mixing-dsl-stream }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - match: { data_streams.0.ilm_policy: "missing_ilm_policy" }
+  - match: { data_streams.0.prefer_ilm: false }
+  - match: { data_streams.0.next_generation_managed_by: "Data stream lifecycle" }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.prefer_ilm: false }
+  - match: { data_streams.0.indices.0.ilm_policy: "missing_ilm_policy" }
+  - match: { data_streams.0.indices.0.managed_by: "Data stream lifecycle" }
+
 ---
 "Delete data stream with backing indices":
   - skip:

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -145,8 +145,9 @@ public class TransportVersions {
     public static final TransportVersion WAIT_FOR_CLUSTER_STATE_IN_RECOVERY_ADDED = def(8_502_00_0);
     public static final TransportVersion RECOVERY_COMMIT_TOO_NEW_EXCEPTION_ADDED = def(8_503_00_0);
     public static final TransportVersion NODE_INFO_COMPONENT_VERSIONS_ADDED = def(8_504_00_0);
-
     public static final TransportVersion COMPACT_FIELD_CAPS_ADDED = def(8_505_00_0);
+    public static final TransportVersion DATA_STREAM_RESPONSE_INDEX_PROPERTIES = def(8_506_00_0);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/action/datastreams/GetDataStreamAction.java

@@ -23,6 +23,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
@@ -32,8 +33,11 @@ import java.io.IOException;
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.TransportVersions.DATA_STREAM_RESPONSE_INDEX_PROPERTIES;
+
 public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response> {
 
     public static final GetDataStreamAction INSTANCE = new GetDataStreamAction();
@@ -142,12 +146,28 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
     }
 
     public static class Response extends ActionResponse implements ToXContentObject {
+
+        public enum ManagedBy {
+            ILM("Index Lifecycle Management"),
+            LIFECYCLE("Data stream lifecycle"),
+            UNMANAGED("Unmanaged");
+
+            public final String displayValue;
+
+            ManagedBy(String displayValue) {
+                this.displayValue = displayValue;
+            }
+        }
+
         public static final ParseField DATA_STREAMS_FIELD = new ParseField("data_streams");
 
         public static class DataStreamInfo implements SimpleDiffable<DataStreamInfo>, ToXContentObject {
 
             public static final ParseField STATUS_FIELD = new ParseField("status");
             public static final ParseField INDEX_TEMPLATE_FIELD = new ParseField("template");
+            public static final ParseField PREFER_ILM = new ParseField("prefer_ilm");
+            public static final ParseField MANAGED_BY = new ParseField("managed_by");
+            public static final ParseField NEXT_GENERATION_INDEX_MANAGED_BY = new ParseField("next_generation_managed_by");
             public static final ParseField ILM_POLICY_FIELD = new ParseField("ilm_policy");
             public static final ParseField LIFECYCLE_FIELD = new ParseField("lifecycle");
             public static final ParseField HIDDEN_FIELD = new ParseField("hidden");
@@ -167,28 +187,39 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             private final String ilmPolicyName;
             @Nullable
             private final TimeSeries timeSeries;
+            private final Map<Index, IndexProperties> indexSettingsValues;
+            private final boolean templatePreferIlmValue;
 
             public DataStreamInfo(
                 DataStream dataStream,
                 ClusterHealthStatus dataStreamStatus,
                 @Nullable String indexTemplate,
                 @Nullable String ilmPolicyName,
-                @Nullable TimeSeries timeSeries
+                @Nullable TimeSeries timeSeries,
+                Map<Index, IndexProperties> indexSettingsValues,
+                boolean templatePreferIlmValue
             ) {
                 this.dataStream = dataStream;
                 this.dataStreamStatus = dataStreamStatus;
                 this.indexTemplate = indexTemplate;
                 this.ilmPolicyName = ilmPolicyName;
                 this.timeSeries = timeSeries;
+                this.indexSettingsValues = indexSettingsValues;
+                this.templatePreferIlmValue = templatePreferIlmValue;
             }
 
+            @SuppressWarnings("unchecked")
             DataStreamInfo(StreamInput in) throws IOException {
                 this(
                     new DataStream(in),
                     ClusterHealthStatus.readFrom(in),
                     in.readOptionalString(),
                     in.readOptionalString(),
-                    in.getTransportVersion().onOrAfter(TransportVersions.V_8_3_0) ? in.readOptionalWriteable(TimeSeries::new) : null
+                    in.getTransportVersion().onOrAfter(TransportVersions.V_8_3_0) ? in.readOptionalWriteable(TimeSeries::new) : null,
+                    in.getTransportVersion().onOrAfter(DATA_STREAM_RESPONSE_INDEX_PROPERTIES)
+                        ? in.readMap(Index::new, IndexProperties::new)
+                        : Map.of(),
+                    in.getTransportVersion().onOrAfter(DATA_STREAM_RESPONSE_INDEX_PROPERTIES) ? in.readBoolean() : true
                 );
             }
 
@@ -215,6 +246,14 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                 return timeSeries;
             }
 
+            public Map<Index, IndexProperties> getIndexSettingsValues() {
+                return indexSettingsValues;
+            }
+
+            public boolean templatePreferIlmValue() {
+                return templatePreferIlmValue;
+            }
+
             @Override
             public void writeTo(StreamOutput out) throws IOException {
                 dataStream.writeTo(out);
@@ -224,6 +263,10 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                 if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_3_0)) {
                     out.writeOptionalWriteable(timeSeries);
                 }
+                if (out.getTransportVersion().onOrAfter(DATA_STREAM_RESPONSE_INDEX_PROPERTIES)) {
+                    out.writeMap(indexSettingsValues);
+                    out.writeBoolean(templatePreferIlmValue);
+                }
             }
 
             @Override
@@ -242,7 +285,27 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                     .startObject()
                     .field(DataStream.NAME_FIELD.getPreferredName(), DataStream.TIMESTAMP_FIELD_NAME)
                     .endObject();
-                builder.xContentList(DataStream.INDICES_FIELD.getPreferredName(), dataStream.getIndices());
+
+                builder.field(DataStream.INDICES_FIELD.getPreferredName());
+                if (dataStream.getIndices() == null) {
+                    builder.nullValue();
+                } else {
+                    builder.startArray();
+                    for (Index index : dataStream.getIndices()) {
+                        builder.startObject();
+                        index.toXContentFragment(builder);
+                        IndexProperties indexProperties = indexSettingsValues.get(index);
+                        if (indexProperties != null) {
+                            builder.field(PREFER_ILM.getPreferredName(), indexProperties.preferIlm());
+                            if (indexProperties.ilmPolicyName() != null) {
+                                builder.field(ILM_POLICY_FIELD.getPreferredName(), indexProperties.ilmPolicyName());
+                            }
+                            builder.field(MANAGED_BY.getPreferredName(), indexProperties.managedBy.displayValue);
+                        }
+                        builder.endObject();
+                    }
+                    builder.endArray();
+                }
                 builder.field(DataStream.GENERATION_FIELD.getPreferredName(), dataStream.getGeneration());
                 if (dataStream.getMetadata() != null) {
                     builder.field(DataStream.METADATA_FIELD.getPreferredName(), dataStream.getMetadata());
@@ -258,6 +321,8 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                 if (ilmPolicyName != null) {
                     builder.field(ILM_POLICY_FIELD.getPreferredName(), ilmPolicyName);
                 }
+                builder.field(NEXT_GENERATION_INDEX_MANAGED_BY.getPreferredName(), getNextGenerationManagedBy().displayValue);
+                builder.field(PREFER_ILM.getPreferredName(), templatePreferIlmValue);
                 builder.field(HIDDEN_FIELD.getPreferredName(), dataStream.isHidden());
                 builder.field(SYSTEM_FIELD.getPreferredName(), dataStream.isSystem());
                 builder.field(ALLOW_CUSTOM_ROUTING.getPreferredName(), dataStream.isAllowCustomRouting());
@@ -280,21 +345,55 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                 return builder;
             }
 
+            /**
+             * Computes and returns which system will manage the next generation for this data stream.
+             */
+            public ManagedBy getNextGenerationManagedBy() {
+                // both ILM and DSL are configured so let's check the prefer_ilm setting to see which system takes precedence
+                if (ilmPolicyName != null && dataStream.getLifecycle() != null && dataStream.getLifecycle().isEnabled()) {
+                    return templatePreferIlmValue ? ManagedBy.ILM : ManagedBy.LIFECYCLE;
+                }
+
+                if (ilmPolicyName != null) {
+                    return ManagedBy.ILM;
+                }
+
+                if (dataStream.getLifecycle() != null && dataStream.getLifecycle().isEnabled()) {
+                    return ManagedBy.LIFECYCLE;
+                }
+
+                return ManagedBy.UNMANAGED;
+            }
+
             @Override
             public boolean equals(Object o) {
-                if (this == o) return true;
-                if (o == null || getClass() != o.getClass()) return false;
+                if (this == o) {
+                    return true;
+                }
+                if (o == null || getClass() != o.getClass()) {
+                    return false;
+                }
                 DataStreamInfo that = (DataStreamInfo) o;
-                return dataStream.equals(that.dataStream)
+                return templatePreferIlmValue == that.templatePreferIlmValue
+                    && Objects.equals(dataStream, that.dataStream)
                     && dataStreamStatus == that.dataStreamStatus
                     && Objects.equals(indexTemplate, that.indexTemplate)
                     && Objects.equals(ilmPolicyName, that.ilmPolicyName)
-                    && Objects.equals(timeSeries, that.timeSeries);
+                    && Objects.equals(timeSeries, that.timeSeries)
+                    && Objects.equals(indexSettingsValues, that.indexSettingsValues);
             }
 
             @Override
             public int hashCode() {
-                return Objects.hash(dataStream, dataStreamStatus, indexTemplate, ilmPolicyName, timeSeries);
+                return Objects.hash(
+                    dataStream,
+                    dataStreamStatus,
+                    indexTemplate,
+                    ilmPolicyName,
+                    timeSeries,
+                    indexSettingsValues,
+                    templatePreferIlmValue
+                );
             }
         }
 
@@ -326,6 +425,23 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             }
         }
 
+        /**
+         * Encapsulates the configured properties we want to display for each backing index.
+         * They'll usually be settings values, but could also be additional properties derived from settings.
+         */
+        public record IndexProperties(boolean preferIlm, @Nullable String ilmPolicyName, ManagedBy managedBy) implements Writeable {
+            public IndexProperties(StreamInput in) throws IOException {
+                this(in.readBoolean(), in.readOptionalString(), in.readEnum(ManagedBy.class));
+            }
+
+            @Override
+            public void writeTo(StreamOutput out) throws IOException {
+                out.writeBoolean(preferIlm);
+                out.writeOptionalString(ilmPolicyName);
+                out.writeEnum(managedBy);
+            }
+        }
+
         private final List<DataStreamInfo> dataStreams;
         @Nullable
         private final RolloverConfiguration rolloverConfiguration;

server/src/main/java/org/elasticsearch/index/Index.java

@@ -103,9 +103,14 @@ public class Index implements Writeable, ToXContentObject {
     @Override
     public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
         builder.startObject();
+        toXContentFragment(builder);
+        return builder.endObject();
+    }
+
+    public XContentBuilder toXContentFragment(final XContentBuilder builder) throws IOException {
         builder.field(INDEX_NAME_KEY, name);
         builder.field(INDEX_UUID_KEY, uuid);
-        return builder.endObject();
+        return builder;
     }
 
     public static Index fromXContent(final XContentParser parser) throws IOException {

x-pack/plugin/ilm/src/internalClusterTest/java/org/elasticsearch/xpack/ilm/DataStreamAndIndexLifecycleMixingTests.java

@@ -15,6 +15,7 @@ import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.datastreams.CreateDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
+import org.elasticsearch.action.datastreams.GetDataStreamAction.Response.ManagedBy;
 import org.elasticsearch.action.datastreams.lifecycle.ExplainIndexDataStreamLifecycle;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
@@ -56,10 +57,13 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
 
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.backingIndexEqualTo;
 import static org.elasticsearch.cluster.metadata.MetadataIndexTemplateService.DEFAULT_TIMESTAMP_FIELD;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -798,6 +802,129 @@ public class DataStreamAndIndexLifecycleMixingTests extends ESIntegTestCase {
         });
     }
 
+    public void testGetDataStreamResponse() throws Exception {
+        // ILM rolls over every 2 documents
+        RolloverAction rolloverIlmAction = new RolloverAction(RolloverConditions.newBuilder().addMaxIndexDocsCondition(2L).build());
+        Phase hotPhase = new Phase("hot", TimeValue.ZERO, Map.of(rolloverIlmAction.getWriteableName(), rolloverIlmAction));
+        LifecyclePolicy lifecyclePolicy = new LifecyclePolicy(policy, Map.of("hot", hotPhase));
+        PutLifecycleAction.Request putLifecycleRequest = new PutLifecycleAction.Request(lifecyclePolicy);
+        assertAcked(client().execute(PutLifecycleAction.INSTANCE, putLifecycleRequest).get());
+
+        putComposableIndexTemplate(
+            indexTemplateName,
+            null,
+            List.of(dataStreamName + "*"),
+            Settings.builder().put(LifecycleSettings.LIFECYCLE_NAME, policy).build(),
+            null,
+            null
+        );
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
+
+        indexDocs(dataStreamName, 2);
+
+        // wait to rollover
+        assertBusy(() -> {
+            GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[] { dataStreamName });
+            GetDataStreamAction.Response getDataStreamResponse = client().execute(GetDataStreamAction.INSTANCE, getDataStreamRequest)
+                .actionGet();
+            assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
+            assertThat(getDataStreamResponse.getDataStreams().get(0).getDataStream().getIndices().size(), is(2));
+        });
+
+        // prefer_ilm false in the index template
+        putComposableIndexTemplate(
+            indexTemplateName,
+            null,
+            List.of(dataStreamName + "*"),
+            Settings.builder().put(LifecycleSettings.LIFECYCLE_NAME, policy).put(IndexSettings.PREFER_ILM, false).build(),
+            null,
+            null
+        );
+
+        client().execute(
+            PutDataStreamLifecycleAction.INSTANCE,
+            new PutDataStreamLifecycleAction.Request(new String[] { dataStreamName }, TimeValue.timeValueDays(90))
+        ).actionGet();
+
+        // rollover again - at this point this data stream should have 2 backing indices managed by ILM and the write index managed by
+        // data stream lifecycle
+        indexDocs(dataStreamName, 2);
+
+        assertBusy(() -> {
+            GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[] { dataStreamName });
+            GetDataStreamAction.Response getDataStreamResponse = client().execute(GetDataStreamAction.INSTANCE, getDataStreamRequest)
+                .actionGet();
+            assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
+            GetDataStreamAction.Response.DataStreamInfo dataStreamInfo = getDataStreamResponse.getDataStreams().get(0);
+            List<Index> indices = dataStreamInfo.getDataStream().getIndices();
+            assertThat(indices.size(), is(3));
+
+            // the prefer_ilm value from the template should be reflected in the response at the top level
+            assertThat(dataStreamInfo.templatePreferIlmValue(), is(false));
+            // the template ILM policy should still be reflected at the top level
+            assertThat(dataStreamInfo.getIlmPolicy(), is(policy));
+
+            List<String> backingIndices = getBackingIndices(dataStreamName);
+            String firstGenerationIndex = backingIndices.get(0);
+            String secondGenerationIndex = backingIndices.get(1);
+            String writeIndex = backingIndices.get(2);
+            assertThat(
+                indices.stream().map(i -> i.getName()).toList(),
+                containsInAnyOrder(firstGenerationIndex, secondGenerationIndex, writeIndex)
+            );
+
+            Function<String, Optional<Index>> backingIndexSupplier = indexName -> indices.stream()
+                .filter(index -> index.getName().equals(indexName))
+                .findFirst();
+
+            // let's assert the policy is reported for all indices (as it's present in the index template) and the value of the
+            // prefer_ilm setting remains true for the first 2 generations and is false for the write index (the generation after rollover)
+            Optional<Index> firstGenSettings = backingIndexSupplier.apply(firstGenerationIndex);
+            assertThat(firstGenSettings.isPresent(), is(true));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(firstGenSettings.get()).preferIlm(), is(true));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(firstGenSettings.get()).ilmPolicyName(), is(policy));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(firstGenSettings.get()).managedBy(), is(ManagedBy.ILM));
+            Optional<Index> secondGenSettings = backingIndexSupplier.apply(secondGenerationIndex);
+            assertThat(secondGenSettings.isPresent(), is(true));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(secondGenSettings.get()).preferIlm(), is(true));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(secondGenSettings.get()).ilmPolicyName(), is(policy));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(secondGenSettings.get()).managedBy(), is(ManagedBy.ILM));
+            Optional<Index> writeIndexSettings = backingIndexSupplier.apply(writeIndex);
+            assertThat(writeIndexSettings.isPresent(), is(true));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(writeIndexSettings.get()).preferIlm(), is(false));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(writeIndexSettings.get()).ilmPolicyName(), is(policy));
+            assertThat(dataStreamInfo.getIndexSettingsValues().get(writeIndexSettings.get()).managedBy(), is(ManagedBy.LIFECYCLE));
+
+            // with the current configuratino, the next generation index will be managed by DSL
+            assertThat(dataStreamInfo.getNextGenerationManagedBy(), is(ManagedBy.LIFECYCLE));
+        });
+
+        // remove ILM policy and prefer_ilm from template
+        putComposableIndexTemplate(indexTemplateName, null, List.of(dataStreamName + "*"), Settings.builder().build(), null, null);
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[] { dataStreamName });
+        GetDataStreamAction.Response getDataStreamResponse = client().execute(GetDataStreamAction.INSTANCE, getDataStreamRequest)
+            .actionGet();
+        assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
+        GetDataStreamAction.Response.DataStreamInfo dataStreamInfo = getDataStreamResponse.getDataStreams().get(0);
+        // since the ILM related settings are gone from the index template, this data stream should now be managed by lifecycle
+        assertThat(dataStreamInfo.getNextGenerationManagedBy(), is(ManagedBy.LIFECYCLE));
+
+        // disable data stream lifecycle on the data stream. the future generations will be UNMANAGED
+        client().execute(
+            PutDataStreamLifecycleAction.INSTANCE,
+            new PutDataStreamLifecycleAction.Request(new String[] { dataStreamName }, TimeValue.timeValueDays(90), false)
+        ).actionGet();
+
+        getDataStreamRequest = new GetDataStreamAction.Request(new String[] { dataStreamName });
+        getDataStreamResponse = client().execute(GetDataStreamAction.INSTANCE, getDataStreamRequest).actionGet();
+        assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
+        dataStreamInfo = getDataStreamResponse.getDataStreams().get(0);
+        // since the ILM related settings are gone from the index template and the lifeclcye is disabled, this data stream should now be
+        // managed unmanaged
+        assertThat(dataStreamInfo.getNextGenerationManagedBy(), is(ManagedBy.UNMANAGED));
+    }
+
     static void indexDocs(String dataStream, int numDocs) {
         BulkRequest bulkRequest = new BulkRequest();
         for (int i = 0; i < numDocs; i++) {