[DLM] Introduce default rollover cluster setting & expose it via APIs (#94240)

For managing data streams with DLM we chose to have one cluster setting that will determine the rollover conditions for all data streams. This PR introduces this cluster setting, it exposes it via the 3 existing APIs under the flag `include_defaults` and adjusts DLM to use it. The feature remains behind a feature flag.

docs/changelog/94240.yaml

@@ -0,0 +1,5 @@
+pr: 94240
+summary: "[DLM] Introduce default rollover cluster setting & expose it via APIs"
+area: DLM
+type: feature
+issues: []

docs/reference/indices/get-component-template.asciidoc

@@ -1,4 +1,4 @@
-[[getting-component-templates]] 
+[[getting-component-templates]]
 === Get component template API
 ++++
 <titleabbrev>Get component template</titleabbrev>
@@ -10,7 +10,7 @@ Retrieves information about one or more component templates.
 
 [source,console]
 --------------------------------------------------
-PUT /_component_template/template_1 
+PUT /_component_template/template_1
 {
   "template": {
     "settings": {
@@ -71,6 +71,9 @@ include::{docdir}/rest-api/common-parms.asciidoc[tag=local]
 
 include::{docdir}/rest-api/common-parms.asciidoc[tag=master-timeout]
 
+`include_defaults`::
+(Optional, Boolean) Functionality in experimental:[]. If `true`, return all default settings in the response.
+Defaults to `false`.
 
 [[get-component-template-api-example]]
 ==== {api-examples-title}

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

@@ -99,6 +99,10 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=ds-expand-wildcards]
 +
 Defaults to `open`.
 
+`include_defaults`::
+(Optional, Boolean) Functionality in experimental:[]. If `true`, return all default settings in the response.
+Defaults to `false`.
+
 [role="child_attributes"]
 [[get-data-stream-api-response-body]]
 ==== {api-response-body-title}
@@ -216,6 +220,25 @@ If `true`, the data stream this data stream allows custom routing on write reque
 (Boolean)
 If `true`, the data stream is created and managed by {ccr} and the local
 cluster can not write into this data stream or change its mappings.
+
+`lifecycle`::
+(object)
+Functionality in experimental:[]. Contains the configuration for the data lifecycle management of this data stream.
++
+.Properties of `lifecycle`
+[%collapsible%open]
+=====
+`data_retention`::
+(string)
+If defined, every document added to this data stream will be stored at least for this time frame. Any time after this
+duration the document could be deleted. When empty, every document in this data stream will be stored indefinitely.
+
+`rollover`::
+(object)
+The conditions which will trigger the rollover of a backing index as configured by the cluster setting
+`cluster.dlm.default.rollover`. This property is an implementation detail and it will only be retrieved when the query
+param `include_defaults` is set to `true`. The contents of this field are subject to change.
+=====
 ====
 
 [[get-data-stream-api-example]]

docs/reference/indices/get-index-template.asciidoc

@@ -63,6 +63,9 @@ include::{docdir}/rest-api/common-parms.asciidoc[tag=local]
 
 include::{docdir}/rest-api/common-parms.asciidoc[tag=master-timeout]
 
+`include_defaults`::
+(Optional, Boolean) Functionality in experimental:[]. If `true`, return all default settings in the response.
+Defaults to `false`.
 
 [[get-template-api-example]]
 ==== {api-examples-title}

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

@@ -17,12 +17,14 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.health.ClusterStateHealth;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
@@ -46,6 +48,7 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
 
     private static final Logger LOGGER = LogManager.getLogger(GetDataStreamsTransportAction.class);
     private final SystemIndices systemIndices;
+    private final ClusterSettings clusterSettings;
 
     @Inject
     public GetDataStreamsTransportAction(
@@ -68,6 +71,7 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
             ThreadPool.Names.SAME
         );
         this.systemIndices = systemIndices;
+        clusterSettings = clusterService.getClusterSettings();
     }
 
     @Override
@@ -77,14 +81,15 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
         ClusterState state,
         ActionListener<GetDataStreamAction.Response> listener
     ) throws Exception {
-        listener.onResponse(innerOperation(state, request, indexNameExpressionResolver, systemIndices));
+        listener.onResponse(innerOperation(state, request, indexNameExpressionResolver, systemIndices, clusterSettings));
     }
 
     static GetDataStreamAction.Response innerOperation(
         ClusterState state,
         GetDataStreamAction.Request request,
         IndexNameExpressionResolver indexNameExpressionResolver,
-        SystemIndices systemIndices
+        SystemIndices systemIndices,
+        ClusterSettings clusterSettings
     ) {
         List<DataStream> dataStreams = getDataStreams(state, indexNameExpressionResolver, request);
         List<GetDataStreamAction.Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
@@ -165,7 +170,12 @@ public class GetDataStreamsTransportAction extends TransportMasterNodeReadAction
                 )
             );
         }
-        return new GetDataStreamAction.Response(dataStreamInfos);
+        return new GetDataStreamAction.Response(
+            dataStreamInfos,
+            request.includeDefaults() && DataLifecycle.isEnabled()
+                ? clusterSettings.get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING)
+                : null
+        );
     }
 
     static List<DataStream> getDataStreams(

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

@@ -10,6 +10,7 @@ package org.elasticsearch.datastreams.rest;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
@@ -39,6 +40,9 @@ public class RestGetDataStreamsAction extends BaseRestHandler {
         GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(
             Strings.splitStringByCommaToArray(request.param("name"))
         );
+        if (DataLifecycle.isEnabled()) {
+            getDataStreamsRequest.includeDefaults(request.paramAsBoolean("include_defaults", false));
+        }
         getDataStreamsRequest.indicesOptions(IndicesOptions.fromRequest(request, getDataStreamsRequest.indicesOptions()));
         return channel -> client.execute(GetDataStreamAction.INSTANCE, getDataStreamsRequest, new RestToXContentListener<>(channel));
     }

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

@@ -14,6 +14,7 @@ import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.indices.SystemIndices;
@@ -149,7 +150,13 @@ public class GetDataStreamsTransportActionTests extends ESTestCase {
         }
 
         var req = new GetDataStreamAction.Request(new String[] {});
-        var response = GetDataStreamsTransportAction.innerOperation(state, req, resolver, systemIndices);
+        var response = GetDataStreamsTransportAction.innerOperation(
+            state,
+            req,
+            resolver,
+            systemIndices,
+            ClusterSettings.createBuiltInClusterSettings()
+        );
         assertThat(response.getDataStreams(), hasSize(2));
         assertThat(response.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStream1));
         assertThat(response.getDataStreams().get(0).getTimeSeries().temporalRanges(), contains(new Tuple<>(sixHoursAgo, twoHoursAhead)));
@@ -164,7 +171,13 @@ public class GetDataStreamsTransportActionTests extends ESTestCase {
             mBuilder.remove(dataStream.getIndices().get(1).getName());
             state = ClusterState.builder(state).metadata(mBuilder).build();
         }
-        response = GetDataStreamsTransportAction.innerOperation(state, req, resolver, systemIndices);
+        response = GetDataStreamsTransportAction.innerOperation(
+            state,
+            req,
+            resolver,
+            systemIndices,
+            ClusterSettings.createBuiltInClusterSettings()
+        );
         assertThat(response.getDataStreams(), hasSize(2));
         assertThat(response.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStream1));
         assertThat(

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/190_managed_by_data_lifecycle.yml

@@ -25,6 +25,7 @@
   - do:
       indices.get_data_stream:
         name: "dlm-managed-data-stream"
+        include_defaults: true
   - match: { data_streams.0.name: dlm-managed-data-stream }
   - match: { data_streams.0.timestamp_field.name: '@timestamp' }
   - match: { data_streams.0.generation: 1 }
@@ -34,3 +35,4 @@
   - match: { data_streams.0.template: 'template-with-lifecycle' }
   - match: { data_streams.0.hidden: false }
   - match: { data_streams.0.lifecycle.data_retention: '30d' }
+  - is_true: data_streams.0.lifecycle.rollover

modules/dlm/src/internalClusterTest/java/org/elasticsearch/dlm/DataLifecycleServiceIT.java

@@ -9,8 +9,6 @@ package org.elasticsearch.dlm;
 
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
-import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
-import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
 import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -62,6 +60,7 @@ public class DataLifecycleServiceIT extends ESIntegTestCase {
     protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
         Settings.Builder settings = Settings.builder().put(super.nodeSettings(nodeOrdinal, otherSettings));
         settings.put(DataLifecycleService.DLM_POLL_INTERVAL, "1s");
+        settings.put(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING.getKey(), "min_docs=1,max_docs=1");
         return settings.build();
     }
 
@@ -70,15 +69,6 @@ public class DataLifecycleServiceIT extends ESIntegTestCase {
         DataLifecycle lifecycle = new DataLifecycle();
 
         putComposableIndexTemplate("id1", null, List.of("metrics-foo*"), null, null, lifecycle);
-        Iterable<DataLifecycleService> dataLifecycleServices = internalCluster().getInstances(DataLifecycleService.class);
-
-        for (DataLifecycleService dataLifecycleService : dataLifecycleServices) {
-            dataLifecycleService.setDefaultRolloverRequestSupplier((target) -> {
-                RolloverRequest rolloverRequest = new RolloverRequest(target, null);
-                rolloverRequest.setConditions(RolloverConditions.newBuilder().addMaxIndexDocsCondition(1L).build());
-                return rolloverRequest;
-            });
-        }
         String dataStreamName = "metrics-foo";
         CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
         client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
@@ -104,15 +94,7 @@ public class DataLifecycleServiceIT extends ESIntegTestCase {
         DataLifecycle lifecycle = new DataLifecycle(TimeValue.timeValueMillis(0));
 
         putComposableIndexTemplate("id1", null, List.of("metrics-foo*"), null, null, lifecycle);
-        Iterable<DataLifecycleService> dataLifecycleServices = internalCluster().getInstances(DataLifecycleService.class);
-
-        for (DataLifecycleService dataLifecycleService : dataLifecycleServices) {
-            dataLifecycleService.setDefaultRolloverRequestSupplier((target) -> {
-                RolloverRequest rolloverRequest = new RolloverRequest(target, null);
-                rolloverRequest.setConditions(RolloverConditions.newBuilder().addMaxIndexDocsCondition(1L).build());
-                return rolloverRequest;
-            });
-        }
+
         String dataStreamName = "metrics-foo";
         CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
         client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();

modules/dlm/src/main/java/org/elasticsearch/dlm/DataLifecycleService.java

@@ -23,6 +23,7 @@ import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -32,7 +33,6 @@ import org.elasticsearch.common.scheduler.SchedulerEngine;
 import org.elasticsearch.common.scheduler.TimeValueSchedule;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
@@ -46,7 +46,6 @@ import java.time.Clock;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.function.Function;
 import java.util.function.LongSupplier;
 
 /**
@@ -77,11 +76,9 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
     private final Clock clock;
     private volatile boolean isMaster = false;
     private volatile TimeValue pollInterval;
+    private volatile RolloverConditions rolloverConditions;
     private SchedulerEngine.Job scheduledJob;
     private final SetOnce<SchedulerEngine> scheduler = new SetOnce<>();
-    // we use this rollover supplier to facilitate testing until we'll be able to read the
-    // rollover configuration from a cluster setting
-    private Function<String, RolloverRequest> defaultRolloverRequestSupplier;
 
     public DataLifecycleService(
         Settings settings,
@@ -99,7 +96,7 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
         this.nowSupplier = nowSupplier;
         this.scheduledJob = null;
         this.pollInterval = DLM_POLL_INTERVAL_SETTING.get(settings);
-        this.defaultRolloverRequestSupplier = this::getDefaultRolloverRequest;
+        this.rolloverConditions = clusterService.getClusterSettings().get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING);
     }
 
     /**
@@ -108,6 +105,8 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
     public void init() {
         clusterService.addListener(this);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(DLM_POLL_INTERVAL_SETTING, this::updatePollInterval);
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING, this::updateRolloverConditions);
     }
 
     @Override
@@ -178,7 +177,7 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
 
     private void maybeExecuteRollover(ClusterState state, DataStream dataStream) {
         if (dataStream.isIndexManagedByDLM(dataStream.getWriteIndex(), state.metadata()::index)) {
-            RolloverRequest rolloverRequest = defaultRolloverRequestSupplier.apply(dataStream.getName());
+            RolloverRequest rolloverRequest = getDefaultRolloverRequest(dataStream.getName());
             transportActionsDeduplicator.executeOnce(
                 rolloverRequest,
                 ActionListener.noop(),
@@ -290,16 +289,7 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
 
     private RolloverRequest getDefaultRolloverRequest(String dataStream) {
         RolloverRequest rolloverRequest = new RolloverRequest(dataStream, null).masterNodeTimeout(TimeValue.MAX_VALUE);
-        rolloverRequest.setConditions(
-            RolloverConditions.newBuilder()
-                // TODO get rollover from cluster setting once we have it
-                .addMaxIndexAgeCondition(TimeValue.timeValueDays(7))
-                .addMaxPrimaryShardSizeCondition(ByteSizeValue.ofGb(50))
-                .addMaxPrimaryShardDocsCondition(200_000_000L)
-                // don't rollover an empty index
-                .addMinIndexDocsCondition(1L)
-                .build()
-        );
+        rolloverRequest.setConditions(rolloverConditions);
         return rolloverRequest;
     }
 
@@ -308,6 +298,10 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
         maybeScheduleJob();
     }
 
+    private void updateRolloverConditions(RolloverConditions newRolloverConditions) {
+        this.rolloverConditions = newRolloverConditions;
+    }
+
     private void cancelJob() {
         if (scheduler.get() != null) {
             scheduler.get().remove(DATA_LIFECYCLE_JOB_NAME);
@@ -340,9 +334,4 @@ public class DataLifecycleService implements ClusterStateListener, Closeable, Sc
         scheduledJob = new SchedulerEngine.Job(DATA_LIFECYCLE_JOB_NAME, new TimeValueSchedule(pollInterval));
         scheduler.get().add(scheduledJob);
     }
-
-    // package visibility for testing
-    void setDefaultRolloverRequestSupplier(Function<String, RolloverRequest> defaultRolloverRequestSupplier) {
-        this.defaultRolloverRequestSupplier = defaultRolloverRequestSupplier;
-    }
 }

rest-api-spec/src/main/resources/rest-api-spec/api/cluster.get_component_template.json

@@ -39,6 +39,10 @@
       "local":{
         "type":"boolean",
         "description":"Return local information, do not retrieve the state from master node (default: false)"
+      },
+      "include_defaults":{
+        "type":"boolean",
+        "description":"Return all default configurations for the component template (default: false)"
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_data_stream.json

@@ -43,6 +43,10 @@
         ],
         "default":"open",
         "description":"Whether wildcard expressions should get expanded to open or closed indices (default: open)"
+      },
+      "include_defaults":{
+        "type":"boolean",
+        "description":"Return all relevant default configurations for the data stream (default: false)"
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_index_template.json

@@ -43,6 +43,10 @@
       "local":{
         "type":"boolean",
         "description":"Return local information, do not retrieve the state from master node (default: false)"
+      },
+      "include_defaults":{
+        "type":"boolean",
+        "description":"Return all relevant default configurations for the index template (default: false)"
       }
     }
   }

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/cluster.component_template/10_basic.yml

@@ -140,3 +140,32 @@
   - match: {component_templates.0.name: test-lifecycle}
   - match: {component_templates.0.component_template.version: 1}
   - match: {component_templates.0.component_template.template.lifecycle: {data_retention: "10d"}}
+
+---
+"Get data lifecycle with default rollover":
+  - skip:
+      version: " - 8.7.99"
+      reason: "Data Lifecycle template was added after 8.7"
+
+  - do:
+      cluster.put_component_template:
+        name: test-lifecycle
+        body:
+          template:
+            lifecycle:
+              data_retention: "10d"
+          version: 1
+          _meta:
+            foo: bar
+            baz:
+              eggplant: true
+
+  - do:
+      cluster.get_component_template:
+        name: test-lifecycle
+        include_defaults: true
+
+  - match: {component_templates.0.name: test-lifecycle}
+  - match: {component_templates.0.component_template.version: 1}
+  - match: {component_templates.0.component_template.template.lifecycle.data_retention: "10d"}
+  - is_true: component_templates.0.component_template.template.lifecycle.rollover

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.get_index_template/10_basic.yml

@@ -136,6 +136,35 @@ setup:
   - match: {index_templates.0.index_template.index_patterns: ["data-stream-with-lifecycle-*"]}
   - match: {index_templates.0.index_template.template.settings: {index: {number_of_shards: '1'}}}
   - match: {index_templates.0.index_template.template.mappings: {properties: {field: {type: keyword}}}}
+  - match: {index_templates.0.index_template.template.lifecycle.data_retention: "30d"}
+
+---
+"Get data lifecycle with default rollover":
+  - skip:
+      version: " - 8.7.99"
+      reason: "Data lifecycle in index templates was added after 8.7"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [test-lifecycle] has index patterns [data-stream-with-lifecycle-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [test-lifecycle] will take precedence during new index creation"
+      indices.put_index_template:
+        name: test-lifecycle
+        body:
+          index_patterns: data-stream-with-lifecycle-*
+          template:
+            lifecycle:
+              data_retention: "30d"
+          data_stream: {}
+  - do:
+      indices.get_index_template:
+        name: test-lifecycle
+        include_defaults: true
+
+  - match: {index_templates.0.name: test-lifecycle}
+  - match: {index_templates.0.index_template.index_patterns: ["data-stream-with-lifecycle-*"]}
+  - match: {index_templates.0.index_template.template.lifecycle.data_retention: "30d"}
+  - is_true: index_templates.0.index_template.template.lifecycle.rollover
 
 ---
 "Reject data lifecycle without data stream configuration":

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverConditions.java

@@ -11,6 +11,7 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.core.Nullable;
@@ -256,6 +257,64 @@ public class RolloverConditions implements Writeable, ToXContentObject {
         return builder;
     }
 
+    /**
+     * Parses a cluster setting configuration, it expects it to have the following format: "condition1=value1,condition2=value2"
+     * @throws SettingsException if the input is invalid, if there are unknown conditions or invalid format values.
+     * @throws IllegalArgumentException if the input is null or blank.
+     */
+    public static RolloverConditions parseSetting(String input, String setting) {
+        if (Strings.isNullOrBlank(input)) {
+            throw new IllegalArgumentException("The rollover conditions cannot be null or blank");
+        }
+        String[] sConditions = input.split(",");
+        RolloverConditions.Builder builder = newBuilder();
+        for (String sCondition : sConditions) {
+            String[] keyValue = sCondition.split("=");
+            if (keyValue.length != 2) {
+                throw new SettingsException("Invalid condition: '{}', format must be 'condition=value'", sCondition);
+            }
+            var condition = keyValue[0];
+            var value = keyValue[1];
+            if (MAX_SIZE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMaxIndexSizeCondition(ByteSizeValue.parseBytesSizeValue(value, setting));
+            } else if (MAX_PRIMARY_SHARD_SIZE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMaxPrimaryShardSizeCondition(ByteSizeValue.parseBytesSizeValue(value, setting));
+            } else if (MAX_AGE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMaxIndexAgeCondition(TimeValue.parseTimeValue(value, setting));
+            } else if (MAX_DOCS_FIELD.getPreferredName().equals(condition)) {
+                builder.addMaxIndexDocsCondition(parseLong(value, setting));
+            } else if (MAX_PRIMARY_SHARD_DOCS_FIELD.getPreferredName().equals(condition)) {
+                builder.addMaxPrimaryShardDocsCondition(parseLong(value, setting));
+            } else if (MIN_SIZE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMinIndexSizeCondition(ByteSizeValue.parseBytesSizeValue(value, setting));
+            } else if (MIN_PRIMARY_SHARD_SIZE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMinPrimaryShardSizeCondition(ByteSizeValue.parseBytesSizeValue(value, setting));
+            } else if (MIN_AGE_FIELD.getPreferredName().equals(condition)) {
+                builder.addMinIndexAgeCondition(TimeValue.parseTimeValue(value, setting));
+            } else if (MIN_DOCS_FIELD.getPreferredName().equals(condition)) {
+                builder.addMinIndexDocsCondition(parseLong(value, setting));
+            } else if (MIN_PRIMARY_SHARD_DOCS_FIELD.getPreferredName().equals(condition)) {
+                builder.addMinPrimaryShardDocsCondition(parseLong(value, condition));
+            } else {
+                throw new SettingsException("Unknown condition: '{}'", condition);
+            }
+        }
+        return builder.build();
+    }
+
+    private static Long parseLong(String sValue, String settingName) {
+        try {
+            return Long.parseLong(sValue);
+        } catch (NumberFormatException e) {
+            throw new SettingsException(
+                "Invalid value '{}' in setting '{}', the value is expected to be of type long",
+                sValue,
+                settingName,
+                e.getMessage()
+            );
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

server/src/main/java/org/elasticsearch/action/admin/indices/template/get/GetComponentTemplateAction.java

@@ -8,11 +8,14 @@
 
 package org.elasticsearch.action.admin.indices.template.get;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.action.support.master.MasterNodeReadRequest;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Nullable;
@@ -43,22 +46,32 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
 
         @Nullable
         private String name;
+        private boolean includeDefaults;
 
         public Request() {}
 
         public Request(String name) {
             this.name = name;
+            this.includeDefaults = false;
         }
 
         public Request(StreamInput in) throws IOException {
             super(in);
             name = in.readOptionalString();
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                includeDefaults = in.readBoolean();
+            } else {
+                includeDefaults = false;
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeOptionalString(name);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeBoolean(includeDefaults);
+            }
         }
 
         @Override
@@ -74,12 +87,27 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
             return this;
         }
 
+        /**
+         * Sets the flag to signal that in the response the default values will also be displayed.
+         */
+        public Request includeDefaults(boolean includeDefaults) {
+            this.includeDefaults = includeDefaults;
+            return this;
+        }
+
         /**
          * The name of the component templates.
          */
         public String name() {
             return this.name;
         }
+
+        /**
+         * True if in the response the default values will be displayed.
+         */
+        public boolean includeDefaults() {
+            return includeDefaults;
+        }
     }
 
     public static class Response extends ActionResponse implements ToXContentObject {
@@ -88,14 +116,27 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
         public static final ParseField COMPONENT_TEMPLATE = new ParseField("component_template");
 
         private final Map<String, ComponentTemplate> componentTemplates;
+        @Nullable
+        private final RolloverConditions rolloverConditions;
 
         public Response(StreamInput in) throws IOException {
             super(in);
             componentTemplates = in.readMap(StreamInput::readString, ComponentTemplate::new);
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                rolloverConditions = in.readOptionalWriteable(RolloverConditions::new);
+            } else {
+                rolloverConditions = null;
+            }
         }
 
         public Response(Map<String, ComponentTemplate> componentTemplates) {
             this.componentTemplates = componentTemplates;
+            this.rolloverConditions = null;
+        }
+
+        public Response(Map<String, ComponentTemplate> componentTemplates, @Nullable RolloverConditions rolloverConditions) {
+            this.componentTemplates = componentTemplates;
+            this.rolloverConditions = rolloverConditions;
         }
 
         public Map<String, ComponentTemplate> getComponentTemplates() {
@@ -105,6 +146,9 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeMap(componentTemplates, StreamOutput::writeString, (o, v) -> v.writeTo(o));
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeOptionalWriteable(rolloverConditions);
+            }
         }
 
         @Override
@@ -112,12 +156,13 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Response that = (Response) o;
-            return Objects.equals(componentTemplates, that.componentTemplates);
+            return Objects.equals(componentTemplates, that.componentTemplates)
+                && Objects.equals(rolloverConditions, that.rolloverConditions);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(componentTemplates);
+            return Objects.hash(componentTemplates, rolloverConditions);
         }
 
         @Override
@@ -127,7 +172,8 @@ public class GetComponentTemplateAction extends ActionType<GetComponentTemplateA
             for (Map.Entry<String, ComponentTemplate> componentTemplate : this.componentTemplates.entrySet()) {
                 builder.startObject();
                 builder.field(NAME.getPreferredName(), componentTemplate.getKey());
-                builder.field(COMPONENT_TEMPLATE.getPreferredName(), componentTemplate.getValue(), params);
+                builder.field(COMPONENT_TEMPLATE.getPreferredName());
+                componentTemplate.getValue().toXContent(builder, params, rolloverConditions);
                 builder.endObject();
             }
             builder.endArray();

server/src/main/java/org/elasticsearch/action/admin/indices/template/get/GetComposableIndexTemplateAction.java

@@ -8,11 +8,14 @@
 
 package org.elasticsearch.action.admin.indices.template.get;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.action.support.master.MasterNodeReadRequest;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Nullable;
@@ -40,6 +43,7 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
 
         @Nullable
         private final String name;
+        private boolean includeDefaults;
 
         /**
          * @param name A template name or pattern, or {@code null} to retrieve all templates.
@@ -49,17 +53,34 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
                 throw new IllegalArgumentException("template name may not contain ','");
             }
             this.name = name;
+            this.includeDefaults = false;
         }
 
         public Request(StreamInput in) throws IOException {
             super(in);
             name = in.readOptionalString();
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                includeDefaults = in.readBoolean();
+            } else {
+                includeDefaults = false;
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeOptionalString(name);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeBoolean(includeDefaults);
+            }
+        }
+
+        public void includeDefaults(boolean includeDefaults) {
+            this.includeDefaults = includeDefaults;
+        }
+
+        public boolean includeDefaults() {
+            return includeDefaults;
         }
 
         @Override
@@ -98,14 +119,26 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
         public static final ParseField INDEX_TEMPLATE = new ParseField("index_template");
 
         private final Map<String, ComposableIndexTemplate> indexTemplates;
+        private final RolloverConditions rolloverConditions;
 
         public Response(StreamInput in) throws IOException {
             super(in);
             indexTemplates = in.readMap(StreamInput::readString, ComposableIndexTemplate::new);
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                rolloverConditions = in.readOptionalWriteable(RolloverConditions::new);
+            } else {
+                rolloverConditions = null;
+            }
         }
 
         public Response(Map<String, ComposableIndexTemplate> indexTemplates) {
             this.indexTemplates = indexTemplates;
+            this.rolloverConditions = null;
+        }
+
+        public Response(Map<String, ComposableIndexTemplate> indexTemplates, RolloverConditions rolloverConditions) {
+            this.indexTemplates = indexTemplates;
+            this.rolloverConditions = rolloverConditions;
         }
 
         public Map<String, ComposableIndexTemplate> indexTemplates() {
@@ -115,6 +148,9 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeMap(indexTemplates, StreamOutput::writeString, (o, v) -> v.writeTo(o));
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeOptionalWriteable(rolloverConditions);
+            }
         }
 
         @Override
@@ -122,12 +158,12 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             GetComposableIndexTemplateAction.Response that = (GetComposableIndexTemplateAction.Response) o;
-            return Objects.equals(indexTemplates, that.indexTemplates);
+            return Objects.equals(indexTemplates, that.indexTemplates) && Objects.equals(rolloverConditions, that.rolloverConditions);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(indexTemplates);
+            return Objects.hash(indexTemplates, rolloverConditions);
         }
 
         @Override
@@ -137,7 +173,8 @@ public class GetComposableIndexTemplateAction extends ActionType<GetComposableIn
             for (Map.Entry<String, ComposableIndexTemplate> indexTemplate : this.indexTemplates.entrySet()) {
                 builder.startObject();
                 builder.field(NAME.getPreferredName(), indexTemplate.getKey());
-                builder.field(INDEX_TEMPLATE.getPreferredName(), indexTemplate.getValue(), params);
+                builder.field(INDEX_TEMPLATE.getPreferredName());
+                indexTemplate.getValue().toXContent(builder, params, rolloverConditions);
                 builder.endObject();
             }
             builder.endArray();

server/src/main/java/org/elasticsearch/action/admin/indices/template/get/TransportGetComponentTemplateAction.java

@@ -16,10 +16,12 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -31,6 +33,8 @@ public class TransportGetComponentTemplateAction extends TransportMasterNodeRead
     GetComponentTemplateAction.Request,
     GetComponentTemplateAction.Response> {
 
+    private final ClusterSettings clusterSettings;
+
     @Inject
     public TransportGetComponentTemplateAction(
         TransportService transportService,
@@ -50,6 +54,7 @@ public class TransportGetComponentTemplateAction extends TransportMasterNodeRead
             GetComponentTemplateAction.Response::new,
             ThreadPool.Names.SAME
         );
+        clusterSettings = clusterService.getClusterSettings();
     }
 
     @Override
@@ -65,27 +70,33 @@ public class TransportGetComponentTemplateAction extends TransportMasterNodeRead
         ActionListener<GetComponentTemplateAction.Response> listener
     ) {
         Map<String, ComponentTemplate> allTemplates = state.metadata().componentTemplates();
+        Map<String, ComponentTemplate> results;
 
         // If we did not ask for a specific name, then we return all templates
         if (request.name() == null) {
-            listener.onResponse(new GetComponentTemplateAction.Response(allTemplates));
-            return;
-        }
-
-        final Map<String, ComponentTemplate> results = new HashMap<>();
-        String name = request.name();
-        if (Regex.isSimpleMatchPattern(name)) {
-            for (Map.Entry<String, ComponentTemplate> entry : allTemplates.entrySet()) {
-                if (Regex.simpleMatch(name, entry.getKey())) {
-                    results.put(entry.getKey(), entry.getValue());
+            results = allTemplates;
+        } else {
+            results = new HashMap<>();
+            String name = request.name();
+            if (Regex.isSimpleMatchPattern(name)) {
+                for (Map.Entry<String, ComponentTemplate> entry : allTemplates.entrySet()) {
+                    if (Regex.simpleMatch(name, entry.getKey())) {
+                        results.put(entry.getKey(), entry.getValue());
+                    }
                 }
+            } else if (allTemplates.containsKey(name)) {
+                results.put(name, allTemplates.get(name));
+            } else {
+                throw new ResourceNotFoundException("component template matching [" + request.name() + "] not found");
+
             }
-        } else if (allTemplates.containsKey(name)) {
-            results.put(name, allTemplates.get(name));
+        }
+        if (request.includeDefaults() && DataLifecycle.isEnabled()) {
+            listener.onResponse(
+                new GetComponentTemplateAction.Response(results, clusterSettings.get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING))
+            );
         } else {
-            throw new ResourceNotFoundException("component template matching [" + request.name() + "] not found");
+            listener.onResponse(new GetComponentTemplateAction.Response(results));
         }
-
-        listener.onResponse(new GetComponentTemplateAction.Response(results));
     }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/template/get/TransportGetComposableIndexTemplateAction.java

@@ -16,10 +16,12 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -31,6 +33,8 @@ public class TransportGetComposableIndexTemplateAction extends TransportMasterNo
     GetComposableIndexTemplateAction.Request,
     GetComposableIndexTemplateAction.Response> {
 
+    private final ClusterSettings clusterSettings;
+
     @Inject
     public TransportGetComposableIndexTemplateAction(
         TransportService transportService,
@@ -50,6 +54,7 @@ public class TransportGetComposableIndexTemplateAction extends TransportMasterNo
             GetComposableIndexTemplateAction.Response::new,
             ThreadPool.Names.SAME
         );
+        clusterSettings = clusterService.getClusterSettings();
     }
 
     @Override
@@ -65,27 +70,34 @@ public class TransportGetComposableIndexTemplateAction extends TransportMasterNo
         ActionListener<GetComposableIndexTemplateAction.Response> listener
     ) {
         Map<String, ComposableIndexTemplate> allTemplates = state.metadata().templatesV2();
-
+        Map<String, ComposableIndexTemplate> results;
         // If we did not ask for a specific name, then we return all templates
         if (request.name() == null) {
-            listener.onResponse(new GetComposableIndexTemplateAction.Response(allTemplates));
-            return;
-        }
-
-        final Map<String, ComposableIndexTemplate> results = new HashMap<>();
-        String name = request.name();
-        if (Regex.isSimpleMatchPattern(name)) {
-            for (Map.Entry<String, ComposableIndexTemplate> entry : allTemplates.entrySet()) {
-                if (Regex.simpleMatch(name, entry.getKey())) {
-                    results.put(entry.getKey(), entry.getValue());
+            results = allTemplates;
+        } else {
+            results = new HashMap<>();
+            String name = request.name();
+            if (Regex.isSimpleMatchPattern(name)) {
+                for (Map.Entry<String, ComposableIndexTemplate> entry : allTemplates.entrySet()) {
+                    if (Regex.simpleMatch(name, entry.getKey())) {
+                        results.put(entry.getKey(), entry.getValue());
+                    }
                 }
+            } else if (allTemplates.containsKey(name)) {
+                results.put(name, allTemplates.get(name));
+            } else {
+                throw new ResourceNotFoundException("index template matching [" + request.name() + "] not found");
             }
-        } else if (allTemplates.containsKey(name)) {
-            results.put(name, allTemplates.get(name));
+        }
+        if (request.includeDefaults() && DataLifecycle.isEnabled()) {
+            listener.onResponse(
+                new GetComposableIndexTemplateAction.Response(
+                    results,
+                    clusterSettings.get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING)
+                )
+            );
         } else {
-            throw new ResourceNotFoundException("index template matching [" + request.name() + "] not found");
+            listener.onResponse(new GetComposableIndexTemplateAction.Response(results));
         }
-
-        listener.onResponse(new GetComposableIndexTemplateAction.Response(results));
     }
 }

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

@@ -12,10 +12,12 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.MasterNodeReadRequest;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -46,11 +48,17 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
 
         private String[] names;
         private IndicesOptions indicesOptions = IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
+        private boolean includeDefaults = false;
 
         public Request(String[] names) {
             this.names = names;
         }
 
+        public Request(String[] names, boolean includeDefaults) {
+            this.names = names;
+            this.includeDefaults = includeDefaults;
+        }
+
         public String[] getNames() {
             return names;
         }
@@ -64,6 +72,11 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             super(in);
             this.names = in.readOptionalStringArray();
             this.indicesOptions = IndicesOptions.readIndicesOptions(in);
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                this.includeDefaults = in.readBoolean();
+            } else {
+                this.includeDefaults = false;
+            }
         }
 
         @Override
@@ -71,6 +84,9 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             super.writeTo(out);
             out.writeOptionalStringArray(names);
             indicesOptions.writeIndicesOptions(out);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeBoolean(includeDefaults);
+            }
         }
 
         @Override
@@ -78,12 +94,14 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return Arrays.equals(names, request.names) && indicesOptions.equals(request.indicesOptions);
+            return Arrays.equals(names, request.names)
+                && indicesOptions.equals(request.indicesOptions)
+                && includeDefaults == request.includeDefaults;
         }
 
         @Override
         public int hashCode() {
-            int result = Objects.hash(indicesOptions);
+            int result = Objects.hash(indicesOptions, includeDefaults);
             result = 31 * result + Arrays.hashCode(names);
             return result;
         }
@@ -98,6 +116,10 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             return indicesOptions;
         }
 
+        public boolean includeDefaults() {
+            return includeDefaults;
+        }
+
         public Request indicesOptions(IndicesOptions indicesOptions) {
             this.indicesOptions = indicesOptions;
             return this;
@@ -113,6 +135,11 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             this.names = indices;
             return this;
         }
+
+        public Request includeDefaults(boolean includeDefaults) {
+            this.includeDefaults = includeDefaults;
+            return this;
+        }
     }
 
     public static class Response extends ActionResponse implements ToXContentObject {
@@ -202,6 +229,14 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
 
             @Override
             public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                return toXContent(builder, params, null);
+            }
+
+            /**
+             * Converts the response to XContent and passes the RolloverConditions, when provided, to the data stream.
+             */
+            public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nullable RolloverConditions rolloverConditions)
+                throws IOException {
                 builder.startObject();
                 builder.field(DataStream.NAME_FIELD.getPreferredName(), dataStream.getName());
                 builder.field(DataStream.TIMESTAMP_FIELD_FIELD.getPreferredName(), dataStream.getTimeStampField());
@@ -215,7 +250,8 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
                     builder.field(INDEX_TEMPLATE_FIELD.getPreferredName(), indexTemplate);
                 }
                 if (dataStream.getLifecycle() != null) {
-                    builder.field(LIFECYCLE_FIELD.getPreferredName(), dataStream.getLifecycle());
+                    builder.field(LIFECYCLE_FIELD.getPreferredName());
+                    dataStream.getLifecycle().toXContent(builder, params, rolloverConditions);
                 }
                 if (ilmPolicyName != null) {
                     builder.field(ILM_POLICY_FIELD.getPreferredName(), ilmPolicyName);
@@ -289,22 +325,42 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
         }
 
         private final List<DataStreamInfo> dataStreams;
+        @Nullable
+        private final RolloverConditions rolloverConditions;
 
         public Response(List<DataStreamInfo> dataStreams) {
+            this(dataStreams, null);
+        }
+
+        public Response(List<DataStreamInfo> dataStreams, @Nullable RolloverConditions rolloverConditions) {
             this.dataStreams = dataStreams;
+            this.rolloverConditions = rolloverConditions;
         }
 
         public Response(StreamInput in) throws IOException {
-            this(in.readList(DataStreamInfo::new));
+            this(
+                in.readList(DataStreamInfo::new),
+                in.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()
+                    ? in.readOptionalWriteable(RolloverConditions::new)
+                    : null
+            );
         }
 
         public List<DataStreamInfo> getDataStreams() {
             return dataStreams;
         }
 
+        @Nullable
+        public RolloverConditions getRolloverConditions() {
+            return rolloverConditions;
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeList(dataStreams);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_8_0) && DataLifecycle.isEnabled()) {
+                out.writeOptionalWriteable(rolloverConditions);
+            }
         }
 
         @Override
@@ -312,7 +368,7 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             builder.startObject();
             builder.startArray(DATA_STREAMS_FIELD.getPreferredName());
             for (DataStreamInfo dataStream : dataStreams) {
-                dataStream.toXContent(builder, params);
+                dataStream.toXContent(builder, params, rolloverConditions);
             }
             builder.endArray();
             builder.endObject();
@@ -324,12 +380,12 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Response response = (Response) o;
-            return dataStreams.equals(response.dataStreams);
+            return dataStreams.equals(response.dataStreams) && Objects.equals(rolloverConditions, response.rolloverConditions);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(dataStreams);
+            return Objects.hash(dataStreams, rolloverConditions);
         }
     }
 

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

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.Strings;
@@ -130,8 +131,17 @@ public class ComponentTemplate implements SimpleDiffable<ComponentTemplate>, ToX
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params, null);
+    }
+
+    /**
+     * Converts the component template to XContent and passes the RolloverConditions, when provided, to the template.
+     */
+    public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nullable RolloverConditions rolloverConditions)
+        throws IOException {
         builder.startObject();
-        builder.field(TEMPLATE.getPreferredName(), this.template, params);
+        builder.field(TEMPLATE.getPreferredName());
+        this.template.toXContent(builder, params, rolloverConditions);
         if (this.version != null) {
             builder.field(VERSION.getPreferredName(), this.version);
         }

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.TransportVersion;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.Strings;
@@ -256,10 +257,19 @@ public class ComposableIndexTemplate implements SimpleDiffable<ComposableIndexTe
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params, null);
+    }
+
+    /**
+     * Converts the composable index template to XContent and passes the RolloverConditions, when provided, to the template.
+     */
+    public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nullable RolloverConditions rolloverConditions)
+        throws IOException {
         builder.startObject();
         builder.stringListField(INDEX_PATTERNS.getPreferredName(), this.indexPatterns);
         if (this.template != null) {
-            builder.field(TEMPLATE.getPreferredName(), this.template, params);
+            builder.field(TEMPLATE.getPreferredName());
+            this.template.toXContent(builder, params, rolloverConditions);
         }
         if (this.componentTemplates != null) {
             builder.stringListField(COMPOSED_OF.getPreferredName(), this.componentTemplates);

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

@@ -9,11 +9,13 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.Build;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.core.Booleans;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
@@ -25,6 +27,8 @@ import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 /**
@@ -32,12 +36,43 @@ import java.util.Objects;
  */
 public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentObject {
 
+    public static final Setting<RolloverConditions> CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING = new Setting<>(
+        "cluster.dlm.default.rollover",
+        "max_age=7d,max_primary_shard_size=50gb,min_docs=1,max_primary_shard_docs=200000000",
+        (s) -> RolloverConditions.parseSetting(s, "cluster.dlm.default.rollover"),
+        new RolloverConditionsValidator(),
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+
+    /**
+     * We require the default rollover conditions to have min_docs set to a non-negative number to avoid empty indices
+     * and to have at least one MAX condition set to ensure that the rollover will be triggered.
+     */
+    static class RolloverConditionsValidator implements Setting.Validator<RolloverConditions> {
+
+        @Override
+        public void validate(RolloverConditions value) {
+            List<String> errors = new ArrayList<>(2);
+            if (value.getMinDocs() == null && value.getMinPrimaryShardDocs() == null) {
+                errors.add("Either min_docs or min_primary_shard_docs rollover conditions should be set and greater than 0.");
+            }
+            if (value.hasMaxConditions() == false) {
+                errors.add("At least one max_* rollover condition must be set.");
+            }
+            if (errors.isEmpty() == false) {
+                throw new IllegalArgumentException(String.join(" ", errors));
+            }
+        }
+    }
+
     private static final boolean FEATURE_FLAG_ENABLED;
 
     public static final DataLifecycle EMPTY = new DataLifecycle();
     public static final String DLM_ORIGIN = "data_lifecycle";
 
     private static final ParseField DATA_RETENTION_FIELD = new ParseField("data_retention");
+    private static final ParseField ROLLOVER_FIELD = new ParseField("rollover");
 
     private static final ConstructingObjectParser<DataLifecycle, Void> PARSER = new ConstructingObjectParser<>(
         "lifecycle",
@@ -115,11 +150,23 @@ public class DataLifecycle implements SimpleDiffable<DataLifecycle>, ToXContentO
         return Strings.toString(this, true, true);
     }
 
+    @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params, null);
+    }
+
+    /**
+     * Converts the data lifecycle to XContent and injects the RolloverConditions if they exist.
+     */
+    public XContentBuilder toXContent(XContentBuilder builder, Params ignored, @Nullable RolloverConditions rolloverConditions)
+        throws IOException {
         builder.startObject();
         if (dataRetention != null) {
             builder.field(DATA_RETENTION_FIELD.getPreferredName(), dataRetention.getStringRep());
         }
+        if (rolloverConditions != null) {
+            builder.field(ROLLOVER_FIELD.getPreferredName(), rolloverConditions);
+        }
         builder.endObject();
         return builder;
     }

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

@@ -13,6 +13,7 @@ import org.apache.lucene.index.PointValues;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.action.admin.indices.rollover.RolloverInfo;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.cluster.Diff;
@@ -830,6 +831,14 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params, null);
+    }
+
+    /**
+     * Converts the data stream to XContent and passes the RolloverConditions, when provided, to the lifecycle.
+     */
+    public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nullable RolloverConditions rolloverConditions)
+        throws IOException {
         builder.startObject();
         builder.field(NAME_FIELD.getPreferredName(), name);
         builder.field(TIMESTAMP_FIELD_FIELD.getPreferredName(), TIMESTAMP_FIELD);
@@ -846,7 +855,8 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
             builder.field(INDEX_MODE.getPreferredName(), indexMode);
         }
         if (lifecycle != null) {
-            builder.field(LIFECYCLE.getPreferredName(), lifecycle);
+            builder.field(LIFECYCLE.getPreferredName());
+            lifecycle.toXContent(builder, params, rolloverConditions);
         }
         builder.endObject();
         return builder;

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.TransportVersion;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.compress.CompressedXContent;
@@ -210,6 +211,14 @@ public class Template implements SimpleDiffable<Template>, ToXContentObject {
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return toXContent(builder, params, null);
+    }
+
+    /**
+     * Converts the template to XContent and passes the RolloverConditions, when provided, to the lifecycle.
+     */
+    public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nullable RolloverConditions rolloverConditions)
+        throws IOException {
         builder.startObject();
         if (this.settings != null) {
             builder.startObject(SETTINGS.getPreferredName());
@@ -238,7 +247,8 @@ public class Template implements SimpleDiffable<Template>, ToXContentObject {
             builder.endObject();
         }
         if (this.lifecycle != null) {
-            builder.field(LIFECYCLE.getPreferredName(), lifecycle);
+            builder.field(LIFECYCLE.getPreferredName());
+            lifecycle.toXContent(builder, params, rolloverConditions);
         }
         builder.endObject();
         return builder;

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

@@ -35,6 +35,7 @@ import org.elasticsearch.cluster.coordination.LeaderChecker;
 import org.elasticsearch.cluster.coordination.MasterHistory;
 import org.elasticsearch.cluster.coordination.NoMasterBlockService;
 import org.elasticsearch.cluster.coordination.Reconfigurator;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.cluster.metadata.IndexGraveyard;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.routing.OperationRouting;
@@ -565,7 +566,8 @@ public final class ClusterSettings extends AbstractScopedSettings {
         TcpTransport.isUntrustedRemoteClusterEnabled() ? RemoteClusterPortSettings.TCP_NO_DELAY : null,
         TcpTransport.isUntrustedRemoteClusterEnabled() ? RemoteClusterPortSettings.TCP_REUSE_ADDRESS : null,
         TcpTransport.isUntrustedRemoteClusterEnabled() ? RemoteClusterPortSettings.TCP_SEND_BUFFER_SIZE : null,
-        StatelessSecureSettings.STATELESS_SECURE_SETTINGS
+        StatelessSecureSettings.STATELESS_SECURE_SETTINGS,
+        DataLifecycle.isEnabled() ? DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING : null
     ).filter(Objects::nonNull).collect(Collectors.toSet());
 
     static List<SettingUpgrader<?>> BUILT_IN_SETTING_UPGRADERS = Collections.emptyList();

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetComponentTemplateAction.java

@@ -10,6 +10,7 @@ package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.template.get.GetComponentTemplateAction;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
@@ -48,7 +49,9 @@ public class RestGetComponentTemplateAction extends BaseRestHandler {
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
 
         final GetComponentTemplateAction.Request getRequest = new GetComponentTemplateAction.Request(request.param("name"));
-
+        if (DataLifecycle.isEnabled()) {
+            getRequest.includeDefaults(request.paramAsBoolean("include_defaults", false));
+        }
         getRequest.local(request.paramAsBoolean("local", getRequest.local()));
         getRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getRequest.masterNodeTimeout()));
 

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetComposableIndexTemplateAction.java

@@ -10,6 +10,7 @@ package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.template.get.GetComposableIndexTemplateAction;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.DataLifecycle;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
@@ -47,7 +48,9 @@ public class RestGetComposableIndexTemplateAction extends BaseRestHandler {
 
         getRequest.local(request.paramAsBoolean("local", getRequest.local()));
         getRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getRequest.masterNodeTimeout()));
-
+        if (DataLifecycle.isEnabled()) {
+            getRequest.includeDefaults(request.paramAsBoolean("include_defaults", false));
+        }
         final boolean implicitAll = getRequest.name() == null;
 
         return channel -> client.execute(GetComposableIndexTemplateAction.INSTANCE, getRequest, new RestToXContentListener<>(channel) {

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/RolloverConditionsTests.java

@@ -10,6 +10,7 @@ package org.elasticsearch.action.admin.indices.rollover;
 
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.TimeValue;
@@ -24,6 +25,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+
 public class RolloverConditionsTests extends AbstractXContentSerializingTestCase<RolloverConditions> {
 
     @Override
@@ -38,6 +42,10 @@ public class RolloverConditionsTests extends AbstractXContentSerializingTestCase
 
     @Override
     protected RolloverConditions createTestInstance() {
+        return randomRolloverConditions();
+    }
+
+    public static RolloverConditions randomRolloverConditions() {
         ByteSizeValue maxSize = randomBoolean() ? randomByteSizeValue() : null;
         ByteSizeValue maxPrimaryShardSize = randomBoolean() ? randomByteSizeValue() : null;
         Long maxDocs = randomBoolean() ? randomNonNegativeLong() : null;
@@ -63,7 +71,7 @@ public class RolloverConditionsTests extends AbstractXContentSerializingTestCase
             .build();
     }
 
-    private ByteSizeValue randomByteSizeValue() {
+    private static ByteSizeValue randomByteSizeValue() {
         ByteSizeUnit unit = randomFrom(ByteSizeUnit.values());
         return new ByteSizeValue(randomNonNegativeLong() / unit.toBytes(1), unit);
     }
@@ -170,6 +178,88 @@ public class RolloverConditionsTests extends AbstractXContentSerializingTestCase
         assertTrue(rolloverConditions.areConditionsMet(Map.of(maxAgeCondition, true, minDocsCondition, true, minAgeCondition, true)));
     }
 
+    public void testClusterSettingParsing() {
+        RolloverConditions defaultSetting = RolloverConditions.parseSetting(
+            "max_age=7d,max_primary_shard_size=50gb,min_docs=1,max_primary_shard_docs=200000000",
+            "test-setting"
+        );
+
+        assertThat(defaultSetting.getMaxSize(), nullValue());
+        assertThat(defaultSetting.getMaxPrimaryShardSize(), equalTo(ByteSizeValue.ofGb(50)));
+        assertThat(defaultSetting.getMaxAge(), equalTo(TimeValue.timeValueDays(7)));
+        assertThat(defaultSetting.getMaxDocs(), nullValue());
+        assertThat(defaultSetting.getMaxPrimaryShardDocs(), equalTo(200_000_000L));
+
+        assertThat(defaultSetting.getMinSize(), nullValue());
+        assertThat(defaultSetting.getMinPrimaryShardSize(), nullValue());
+        assertThat(defaultSetting.getMinAge(), nullValue());
+        assertThat(defaultSetting.getMinDocs(), equalTo(1L));
+        assertThat(defaultSetting.getMinPrimaryShardDocs(), nullValue());
+
+        var maxSize = ByteSizeValue.ofGb(randomIntBetween(1, 100));
+        var maxPrimaryShardSize = ByteSizeValue.ofGb(randomIntBetween(1, 50));
+        var maxAge = TimeValue.timeValueMillis(randomMillisUpToYear9999());
+        var maxDocs = randomLongBetween(1_000_000, 1_000_000_000);
+        var maxPrimaryShardDocs = randomLongBetween(1_000_000, 1_000_000_000);
+
+        var minSize = ByteSizeValue.ofGb(randomIntBetween(1, 100));
+        var minPrimaryShardSize = ByteSizeValue.ofGb(randomIntBetween(1, 50));
+        var minAge = TimeValue.timeValueMillis(randomMillisUpToYear9999());
+        var minDocs = randomLongBetween(1_000_000, 1_000_000_000);
+        var minPrimaryShardDocs = randomLongBetween(1_000_000, 1_000_000_000);
+        String setting = "max_size="
+            + maxSize.getStringRep()
+            + ",max_primary_shard_size="
+            + maxPrimaryShardSize.getStringRep()
+            + ",max_age="
+            + maxAge.getStringRep()
+            + ",max_docs="
+            + maxDocs
+            + ",max_primary_shard_docs="
+            + maxPrimaryShardDocs
+            + ",min_size="
+            + minSize.getStringRep()
+            + ",min_primary_shard_size="
+            + minPrimaryShardSize.getStringRep()
+            + ",min_age="
+            + minAge.getStringRep()
+            + ",min_docs="
+            + minDocs
+            + ",min_primary_shard_docs="
+            + minPrimaryShardDocs;
+        RolloverConditions randomSetting = RolloverConditions.parseSetting(setting, "test2");
+        assertThat(randomSetting.getMaxAge(), equalTo(maxAge));
+        assertThat(randomSetting.getMaxPrimaryShardSize(), equalTo(maxPrimaryShardSize));
+        assertThat(randomSetting.getMaxDocs(), equalTo(maxDocs));
+        assertThat(randomSetting.getMaxPrimaryShardDocs(), equalTo(maxPrimaryShardDocs));
+        assertThat(randomSetting.getMaxSize(), equalTo(maxSize));
+
+        assertThat(randomSetting.getMinAge(), equalTo(minAge));
+        assertThat(randomSetting.getMinPrimaryShardSize(), equalTo(minPrimaryShardSize));
+        assertThat(randomSetting.getMinPrimaryShardDocs(), equalTo(minPrimaryShardDocs));
+        assertThat(randomSetting.getMinDocs(), equalTo(minDocs));
+        assertThat(randomSetting.getMinSize(), equalTo(minSize));
+
+        IllegalArgumentException invalid = expectThrows(
+            IllegalArgumentException.class,
+            () -> RolloverConditions.parseSetting("", "empty-setting")
+        );
+        assertEquals("The rollover conditions cannot be null or blank", invalid.getMessage());
+        SettingsException unknown = expectThrows(
+            SettingsException.class,
+            () -> RolloverConditions.parseSetting("unknown_condition=?", "unknown-setting")
+        );
+        assertEquals("Unknown condition: 'unknown_condition'", unknown.getMessage());
+        SettingsException numberFormat = expectThrows(
+            SettingsException.class,
+            () -> RolloverConditions.parseSetting("max_docs=one", "invalid-number-setting")
+        );
+        assertEquals(
+            "Invalid value 'one' in setting 'invalid-number-setting', the value is expected to be of type long",
+            numberFormat.getMessage()
+        );
+    }
+
     private static final List<Consumer<RolloverConditions.Builder>> conditionsGenerator = Arrays.asList(
         (builder) -> builder.addMaxIndexDocsCondition(randomNonNegativeLong()),
         (builder) -> builder.addMaxIndexSizeCondition(ByteSizeValue.ofBytes(randomNonNegativeLong())),

server/src/test/java/org/elasticsearch/cluster/metadata/ComponentTemplateTests.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditionsTests;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
@@ -18,6 +20,8 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
@@ -26,6 +30,7 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.Map;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class ComponentTemplateTests extends SimpleDiffableSerializationTestCase<ComponentTemplate> {
@@ -244,4 +249,36 @@ public class ComponentTemplateTests extends SimpleDiffableSerializationTestCase<
             assertThat(Template.mappingsEquals(m1, m2), equalTo(true));
         }
     }
+
+    public void testXContentSerializationWithRollover() throws IOException {
+        Settings settings = null;
+        CompressedXContent mappings = null;
+        Map<String, AliasMetadata> aliases = null;
+        if (randomBoolean()) {
+            settings = randomSettings();
+        }
+        if (randomBoolean()) {
+            mappings = randomMappings();
+        }
+        if (randomBoolean()) {
+            aliases = randomAliases();
+        }
+        DataLifecycle lifecycle = randomLifecycle();
+        ComponentTemplate template = new ComponentTemplate(
+            new Template(settings, mappings, aliases, lifecycle),
+            randomNonNegativeLong(),
+            null
+        );
+
+        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            builder.humanReadable(true);
+            RolloverConditions rolloverConditions = RolloverConditionsTests.randomRolloverConditions();
+            template.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConditions);
+            String serialized = Strings.toString(builder);
+            assertThat(serialized, containsString("rollover"));
+            for (String label : rolloverConditions.getConditions().keySet()) {
+                assertThat(serialized, containsString(label));
+            }
+        }
+    }
 }

server/src/test/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplateTests.java

@@ -8,19 +8,26 @@
 
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditionsTests;
 import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.SimpleDiffableSerializationTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTestCase<ComposableIndexTemplate> {
@@ -100,6 +107,10 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
         return Collections.singletonMap(aliasName, aliasMeta);
     }
 
+    private static DataLifecycle randomLifecycle() {
+        return new DataLifecycle(randomMillisUpToYear9999());
+    }
+
     private static CompressedXContent randomMappings(ComposableIndexTemplate.DataStreamTemplate dataStreamTemplate) {
         try {
             if (dataStreamTemplate != null) {
@@ -272,4 +283,42 @@ public class ComposableIndexTemplateTests extends SimpleDiffableSerializationTes
         assertThat(ComposableIndexTemplate.componentTemplatesEquals(List.of(randomAlphaOfLength(5)), List.of()), equalTo(false));
         assertThat(ComposableIndexTemplate.componentTemplatesEquals(List.of(), List.of(randomAlphaOfLength(5))), equalTo(false));
     }
+
+    public void testXContentSerializationWithRollover() throws IOException {
+        Settings settings = null;
+        CompressedXContent mappings = null;
+        Map<String, AliasMetadata> aliases = null;
+        ComposableIndexTemplate.DataStreamTemplate dataStreamTemplate = randomDataStreamTemplate();
+        if (randomBoolean()) {
+            settings = randomSettings();
+        }
+        if (randomBoolean()) {
+            mappings = randomMappings(dataStreamTemplate);
+        }
+        if (randomBoolean()) {
+            aliases = randomAliases();
+        }
+        DataLifecycle lifecycle = randomLifecycle();
+        Template template = new Template(settings, mappings, aliases, lifecycle);
+        new ComposableIndexTemplate(
+            List.of(randomAlphaOfLength(4)),
+            template,
+            List.of(),
+            randomNonNegativeLong(),
+            randomNonNegativeLong(),
+            null,
+            dataStreamTemplate
+        );
+
+        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            builder.humanReadable(true);
+            RolloverConditions rolloverConditions = RolloverConditionsTests.randomRolloverConditions();
+            template.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConditions);
+            String serialized = Strings.toString(builder);
+            assertThat(serialized, containsString("rollover"));
+            for (String label : rolloverConditions.getConditions().keySet()) {
+                assertThat(serialized, containsString(label));
+            }
+        }
+    }
 }

server/src/test/java/org/elasticsearch/cluster/metadata/DataLifecycleSerializationTests.java

@@ -1,45 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License
- * 2.0 and the Server Side Public License, v 1; you may not use this file except
- * in compliance with, at your election, the Elastic License 2.0 or the Server
- * Side Public License, v 1.
- */
-
-package org.elasticsearch.cluster.metadata;
-
-import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.test.AbstractXContentSerializingTestCase;
-import org.elasticsearch.xcontent.XContentParser;
-
-import java.io.IOException;
-
-public class DataLifecycleSerializationTests extends AbstractXContentSerializingTestCase<DataLifecycle> {
-
-    @Override
-    protected Writeable.Reader<DataLifecycle> instanceReader() {
-        return DataLifecycle::new;
-    }
-
-    @Override
-    protected DataLifecycle createTestInstance() {
-        if (randomBoolean()) {
-            return new DataLifecycle();
-        } else {
-            return new DataLifecycle(randomMillisUpToYear9999());
-        }
-    }
-
-    @Override
-    protected DataLifecycle mutateInstance(DataLifecycle instance) throws IOException {
-        if (instance.getDataRetention() == null) {
-            return new DataLifecycle(randomMillisUpToYear9999());
-        }
-        return new DataLifecycle(instance.getDataRetention().millis() + randomMillisUpToYear9999());
-    }
-
-    @Override
-    protected DataLifecycle doParseInstance(XContentParser parser) throws IOException {
-        return DataLifecycle.fromXContent(parser);
-    }
-}

server/src/test/java/org/elasticsearch/cluster/metadata/DataLifecycleTests.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditionsTests;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentType;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+
+public class DataLifecycleTests extends AbstractXContentSerializingTestCase<DataLifecycle> {
+
+    @Override
+    protected Writeable.Reader<DataLifecycle> instanceReader() {
+        return DataLifecycle::new;
+    }
+
+    @Override
+    protected DataLifecycle createTestInstance() {
+        if (randomBoolean()) {
+            return new DataLifecycle();
+        } else {
+            return new DataLifecycle(randomMillisUpToYear9999());
+        }
+    }
+
+    @Override
+    protected DataLifecycle mutateInstance(DataLifecycle instance) throws IOException {
+        if (instance.getDataRetention() == null) {
+            return new DataLifecycle(randomMillisUpToYear9999());
+        }
+        return new DataLifecycle(instance.getDataRetention().millis() + randomMillisUpToYear9999());
+    }
+
+    @Override
+    protected DataLifecycle doParseInstance(XContentParser parser) throws IOException {
+        return DataLifecycle.fromXContent(parser);
+    }
+
+    public void testXContentSerializationWithRollover() throws IOException {
+        DataLifecycle dataLifecycle = createTestInstance();
+        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            builder.humanReadable(true);
+            RolloverConditions rolloverConditions = RolloverConditionsTests.randomRolloverConditions();
+            dataLifecycle.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConditions);
+            String serialized = Strings.toString(builder);
+            assertThat(serialized, containsString("rollover"));
+            for (String label : rolloverConditions.getConditions().keySet()) {
+                assertThat(serialized, containsString(label));
+            }
+        }
+    }
+
+    public void testDefaultClusterSetting() {
+        ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        RolloverConditions rolloverConditions = clusterSettings.get(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING);
+        assertThat(rolloverConditions.getMaxAge(), equalTo(TimeValue.timeValueDays(7)));
+        assertThat(rolloverConditions.getMaxPrimaryShardSize(), equalTo(ByteSizeValue.ofGb(50)));
+        assertThat(rolloverConditions.getMaxPrimaryShardDocs(), equalTo(200_000_000L));
+        assertThat(rolloverConditions.getMinDocs(), equalTo(1L));
+        assertThat(rolloverConditions.getMaxSize(), nullValue());
+        assertThat(rolloverConditions.getMaxDocs(), nullValue());
+        assertThat(rolloverConditions.getMinAge(), nullValue());
+        assertThat(rolloverConditions.getMinSize(), nullValue());
+        assertThat(rolloverConditions.getMinPrimaryShardSize(), nullValue());
+        assertThat(rolloverConditions.getMinPrimaryShardDocs(), nullValue());
+
+    }
+
+    public void testInvalidClusterSetting() {
+        {
+            IllegalArgumentException exception = expectThrows(
+                IllegalArgumentException.class,
+                () -> DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING.get(
+                    Settings.builder().put(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING.getKey(), "").build()
+                )
+            );
+            assertThat(exception.getMessage(), equalTo("The rollover conditions cannot be null or blank"));
+        }
+        {
+            IllegalArgumentException exception = expectThrows(
+                IllegalArgumentException.class,
+                () -> DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING.get(
+                    Settings.builder().put(DataLifecycle.CLUSTER_DLM_DEFAULT_ROLLOVER_SETTING.getKey(), "min_docs=1").build()
+                )
+            );
+            assertThat(exception.getMessage(), equalTo("At least one max_* rollover condition must be set."));
+        }
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java

@@ -7,8 +7,11 @@
  */
 package org.elasticsearch.cluster.metadata;
 
+import org.apache.lucene.tests.util.LuceneTestCase;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.rollover.MaxAgeCondition;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditionsTests;
 import org.elasticsearch.action.admin.indices.rollover.RolloverInfo;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
@@ -23,7 +26,11 @@ import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.test.AbstractXContentSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -37,7 +44,10 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.function.Predicate;
 
+import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
 import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.newInstance;
+import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.randomIndexInstances;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.hasItems;
@@ -545,7 +555,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
             // never remove them all
             indicesToRemove.remove(0);
         }
-        var indicesToAdd = DataStreamTestHelper.randomIndexInstances();
+        var indicesToAdd = randomIndexInstances();
         var postSnapshotIndices = new ArrayList<>(preSnapshotDataStream.getIndices());
         postSnapshotIndices.removeAll(indicesToRemove);
         postSnapshotIndices.addAll(indicesToAdd);
@@ -590,7 +600,7 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         var indicesToAdd = new ArrayList<Index>();
         while (indicesToAdd.isEmpty()) {
             // ensure at least one index
-            indicesToAdd.addAll(DataStreamTestHelper.randomIndexInstances());
+            indicesToAdd.addAll(randomIndexInstances());
         }
 
         var postSnapshotDataStream = new DataStream(
@@ -1125,4 +1135,40 @@ public class DataStreamTests extends AbstractXContentSerializingTestCase<DataStr
         }
         return newInstance(dataStreamName, backingIndices, backingIndicesCount, null, false, lifecycle);
     }
+
+    public void testXContentSerializationWithRollover() throws IOException {
+        String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        List<Index> indices = randomIndexInstances();
+        long generation = indices.size() + ESTestCase.randomLongBetween(1, 128);
+        indices.add(new Index(getDefaultBackingIndexName(dataStreamName, generation), UUIDs.randomBase64UUID(LuceneTestCase.random())));
+        Map<String, Object> metadata = null;
+        if (randomBoolean()) {
+            metadata = Map.of("key", "value");
+        }
+
+        DataStream dataStream = new DataStream(
+            dataStreamName,
+            indices,
+            generation,
+            metadata,
+            randomBoolean(),
+            randomBoolean(),
+            false, // Some tests don't work well with system data streams, since these data streams require special handling
+            System::currentTimeMillis,
+            randomBoolean(),
+            randomBoolean() ? IndexMode.STANDARD : null, // IndexMode.TIME_SERIES triggers validation that many unit tests doesn't pass
+            new DataLifecycle(randomMillisUpToYear9999())
+        );
+
+        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            builder.humanReadable(true);
+            RolloverConditions rolloverConditions = RolloverConditionsTests.randomRolloverConditions();
+            dataStream.toXContent(builder, ToXContent.EMPTY_PARAMS, rolloverConditions);
+            String serialized = Strings.toString(builder);
+            assertThat(serialized, containsString("rollover"));
+            for (String label : rolloverConditions.getConditions().keySet()) {
+                assertThat(serialized, containsString(label));
+            }
+        }
+    }
 }