Adding rest actions to get and set data stream settings

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

@@ -26,13 +26,15 @@ public class DataStreamFeatures implements FeatureSpecification {
         "data_stream.downsample.default_aggregate_metric_fix"
     );
 
+    public static final NodeFeature LOGS_STREAM_FEATURE = new NodeFeature("logs_stream");
+
     @Override
     public Set<NodeFeature> getFeatures() {
         return Set.of(DataStream.DATA_STREAM_FAILURE_STORE_FEATURE);
     }
 
     @Override
     public Set<NodeFeature> getTestFeatures() {
-        return Set.of(DATA_STREAM_FAILURE_STORE_TSDB_FIX, DOWNSAMPLE_AGGREGATE_DEFAULT_METRIC_FIX);
+        return Set.of(DATA_STREAM_FAILURE_STORE_TSDB_FIX, DOWNSAMPLE_AGGREGATE_DEFAULT_METRIC_FIX, LOGS_STREAM_FEATURE);
     }
 }

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.action.datastreams.lifecycle.GetDataStreamLifecycleAction;
 import org.elasticsearch.action.datastreams.lifecycle.PutDataStreamLifecycleAction;
 import org.elasticsearch.client.internal.OriginSettingClient;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -70,10 +71,12 @@
 import org.elasticsearch.datastreams.rest.RestCreateDataStreamAction;
 import org.elasticsearch.datastreams.rest.RestDataStreamsStatsAction;
 import org.elasticsearch.datastreams.rest.RestDeleteDataStreamAction;
+import org.elasticsearch.datastreams.rest.RestGetDataStreamSettingsAction;
 import org.elasticsearch.datastreams.rest.RestGetDataStreamsAction;
 import org.elasticsearch.datastreams.rest.RestMigrateToDataStreamAction;
 import org.elasticsearch.datastreams.rest.RestModifyDataStreamsAction;
 import org.elasticsearch.datastreams.rest.RestPromoteDataStreamAction;
+import org.elasticsearch.datastreams.rest.RestUpdateDataStreamSettingsAction;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.health.HealthIndicatorService;
 import org.elasticsearch.index.IndexSettingProvider;
@@ -280,6 +283,10 @@ public List<RestHandler> getRestHandlers(
         handlers.add(new RestGetDataStreamOptionsAction());
         handlers.add(new RestPutDataStreamOptionsAction());
         handlers.add(new RestDeleteDataStreamOptionsAction());
+        if (DataStream.LOGS_STREAM_FEATURE_FLAG) {
+            handlers.add(new RestGetDataStreamSettingsAction());
+            handlers.add(new RestUpdateDataStreamSettingsAction());
+        }
         return handlers;
     }
 

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

@@ -0,0 +1,51 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.datastreams.rest;
+
+import org.elasticsearch.action.datastreams.GetDataStreamSettingsAction;
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestUtils;
+import org.elasticsearch.rest.Scope;
+import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.elasticsearch.rest.RestRequest.Method.GET;
+
+@ServerlessScope(Scope.PUBLIC)
+public class RestGetDataStreamSettingsAction extends BaseRestHandler {
+    @Override
+    public String getName() {
+        return "gett_data_stream_settings_action";
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(GET, "/_data_stream/{name}/_settings"));
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        GetDataStreamSettingsAction.Request getDataStreamRequest = new GetDataStreamSettingsAction.Request(
+            RestUtils.getMasterNodeTimeout(request)
+        ).indices(Strings.splitStringByCommaToArray(request.param("name")));
+        return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).execute(
+            GetDataStreamSettingsAction.INSTANCE,
+            getDataStreamRequest,
+            new RestRefCountedChunkedToXContentListener<>(channel)
+        );
+    }
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.datastreams.rest;
+
+import org.elasticsearch.action.datastreams.UpdateDataStreamSettingsAction;
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestUtils;
+import org.elasticsearch.rest.Scope;
+import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestCancellableNodeClient;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.elasticsearch.rest.RestRequest.Method.PUT;
+
+@ServerlessScope(Scope.PUBLIC)
+public class RestUpdateDataStreamSettingsAction extends BaseRestHandler {
+
+    @Override
+    public String getName() {
+        return "update_data_stream_settings_action";
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(PUT, "/_data_stream/{name}/_settings"));
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        Settings settings;
+        try (XContentParser parser = request.contentParser()) {
+            settings = Settings.fromXContent(parser);
+        }
+        UpdateDataStreamSettingsAction.Request putDataStreamRequest = new UpdateDataStreamSettingsAction.Request(
+            settings,
+            RestUtils.getMasterNodeTimeout(request),
+            RestUtils.getAckTimeout(request)
+        ).indices(Strings.splitStringByCommaToArray(request.param("name")));
+        return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).execute(
+            UpdateDataStreamSettingsAction.INSTANCE,
+            putDataStreamRequest,
+            new RestRefCountedChunkedToXContentListener<>(channel)
+        );
+    }
+}

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/240_data_stream_settings.yml

@@ -0,0 +1,194 @@
+setup:
+  - skip:
+      features: allowed_warnings
+
+---
+"Test single data stream":
+  - requires:
+      cluster_features: [ "logs_stream" ]
+      reason: requires setting 'logs_stream' to get or set data stream settings
+  - do:
+      allowed_warnings:
+        - "index template [my-template] has index patterns [my-data-stream-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template
+        body:
+          index_patterns: [ my-data-stream-* ]
+          data_stream: { }
+          template:
+            settings:
+              number_of_replicas: 0
+              lifecycle.name: my-policy
+
+  - do:
+      indices.create_data_stream:
+        name: my-data-stream-1
+
+  - do:
+      cluster.health:
+        index: "my-data-stream-1"
+        wait_for_status: green
+
+  - do:
+      indices.get_data_stream_settings:
+        name: my-data-stream-1
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.settings: {} }
+  - match: { data_streams.0.effective_settings.index.number_of_shards: null }
+  - match: { data_streams.0.effective_settings.index.number_of_replicas: "0" }
+  - match: { data_streams.0.effective_settings.index.lifecycle.name: "my-policy" }
+
+  - do:
+      indices.get_data_stream:
+        name: my-data-stream-1
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.settings: {} }
+  - match: { data_streams.0.effective_settings: null }
+
+  - do:
+      indices.put_data_stream_settings:
+        name: my-data-stream-1
+        body:
+          index:
+            number_of_shards: 2
+            lifecycle.name: my-new-policy
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.applied_to_data_stream: true }
+  - match: { data_streams.0.index_settings_results.applied_to_data_stream_only: [index.number_of_shards]}
+  - match: { data_streams.0.index_settings_results.applied_to_data_stream_and_backing_indices: [index.lifecycle.name] }
+  - match: { data_streams.0.settings.index.number_of_shards: "2" }
+  - match: { data_streams.0.settings.index.lifecycle.name: "my-new-policy" }
+  - match: { data_streams.0.effective_settings.index.number_of_shards: "2" }
+  - match: { data_streams.0.effective_settings.index.number_of_replicas: "0" }
+  - match: { data_streams.0.effective_settings.index.lifecycle.name: "my-new-policy" }
+
+  - do:
+      indices.rollover:
+        alias: "my-data-stream-1"
+
+  - do:
+      cluster.health:
+        index: "my-data-stream-1"
+        wait_for_status: green
+
+  - do:
+      indices.get_data_stream_settings:
+        name: my-data-stream-1
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.settings.index.number_of_shards: "2" }
+  - match: { data_streams.0.effective_settings.index.number_of_shards: "2" }
+  - match: { data_streams.0.effective_settings.index.number_of_replicas: "0" }
+  - match: { data_streams.0.effective_settings.index.lifecycle.name: "my-new-policy" }
+
+  - do:
+      indices.get_data_stream:
+        name: my-data-stream-1
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.settings.index.number_of_shards: "2" }
+  - match: { data_streams.0.settings.index.lifecycle.name: "my-new-policy" }
+  - match: { data_streams.0.effective_settings: null }
+
+  - do:
+      indices.get_data_stream:
+        name: my-data-stream-1
+  - set: { data_streams.0.indices.0.index_name: idx0name }
+  - set: { data_streams.0.indices.1.index_name: idx1name }
+
+  - do:
+      indices.get_settings:
+        index: my-data-stream-1
+  - match: { .$idx0name.settings.index.number_of_shards: "1" }
+  - match: { .$idx0name.settings.index.lifecycle.name: "my-new-policy" }
+  - match: { .$idx1name.settings.index.number_of_shards: "2" }
+  - match: { .$idx1name.settings.index.lifecycle.name: "my-new-policy" }
+
+---
+"Test multiple data streams":
+  - requires:
+      cluster_features: [ "logs_stream" ]
+      reason: requires setting 'logs_stream' to get or set data stream settings
+  - do:
+      allowed_warnings:
+        - "index template [my-template] has index patterns [my-data-stream-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template
+        body:
+          index_patterns: [ my-data-stream-* ]
+          data_stream: { }
+          template:
+            settings:
+              number_of_replicas: 0
+              lifecycle.name: my-policy
+
+  - do:
+      indices.create_data_stream:
+        name: my-data-stream-1
+
+  - do:
+      indices.create_data_stream:
+        name: my-data-stream-2
+
+  - do:
+      indices.create_data_stream:
+        name: my-data-stream-3
+
+  - do:
+      cluster.health:
+        index: "my-data-stream-1,my-data-stream-2,my-data-stream-3"
+        wait_for_status: green
+
+  - do:
+      indices.get_data_stream_settings:
+        name: "*"
+  - length: { data_streams: 3 }
+
+  - do:
+      indices.get_data_stream_settings:
+        name: "my-data-stream-1,my-data-stream-2"
+  - length: { data_streams: 2 }
+
+  - do:
+      indices.put_data_stream_settings:
+        name: my-data-stream-1,my-data-stream-2
+        body:
+          index:
+            number_of_shards: 2
+            lifecycle.name: my-new-policy
+  - length: { data_streams: 2 }
+
+---
+"Test invalid settings":
+  - requires:
+      cluster_features: [ "logs_stream" ]
+      reason: requires setting 'logs_stream' to get or set data stream settings
+  - do:
+      allowed_warnings:
+        - "index template [my-template] has index patterns [my-data-stream-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template
+        body:
+          index_patterns: [ my-data-stream-* ]
+          data_stream: { }
+          template:
+            settings:
+              number_of_replicas: 0
+              lifecycle.name: my-policy
+
+  - do:
+      indices.create_data_stream:
+        name: my-data-stream-1
+
+  - do:
+      cluster.health:
+        index: "my-data-stream-1"
+        wait_for_status: green
+
+  - do:
+      indices.put_data_stream_settings:
+        name: my-data-stream-1
+        body:
+          index:
+            fake_setting: 1234
+  - match: { data_streams.0.name: my-data-stream-1 }
+  - match: { data_streams.0.applied_to_data_stream: false }
+  - match: { data_streams.0.error: "Cannot set the following settings on a data stream: [index.fake_setting]" }

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

@@ -0,0 +1,40 @@
+{
+  "indices.get_data_stream_settings":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/data-streams.html",
+      "description":"Gets a data stream's settings"
+    },
+    "stability":"stable",
+    "visibility": "feature_flag",
+    "feature_flag": "logs_stream",
+    "headers":{
+      "accept": [ "application/json"]
+    },
+    "url":{
+      "paths":[
+        {
+          "path":"/_data_stream/{name}/_settings",
+          "methods":[
+            "GET"
+          ],
+          "parts":{
+            "name":{
+              "type":"string",
+              "description":"The name of the data stream or data stream pattern"
+            }
+          }
+        }
+      ]
+    },
+    "params":{
+      "timeout":{
+        "type":"time",
+        "description":"Specify timeout for acknowledging the cluster state update"
+      },
+      "master_timeout":{
+        "type":"time",
+        "description":"Specify timeout for connection to master"
+      }
+    }
+  }
+}