Adding dry_run mode for setting data stream settings (#128269)

Author: masseyke

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

@@ -16,7 +16,6 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.CountDownActionListener;
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -102,53 +101,46 @@ protected void masterOperation(
             request.indices()
         );
         List<UpdateDataStreamSettingsAction.DataStreamSettingsResponse> dataStreamSettingsResponse = new ArrayList<>();
-        CountDownActionListener countDownListener = new CountDownActionListener(dataStreamNames.size() + 1, new ActionListener<>() {
-            @Override
-            public void onResponse(Void unused) {
-                listener.onResponse(new UpdateDataStreamSettingsAction.Response(dataStreamSettingsResponse));
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                listener.onFailure(e);
-            }
-        });
+        CountDownActionListener countDownListener = new CountDownActionListener(
+            dataStreamNames.size() + 1,
+            listener.delegateFailure(
+                (responseActionListener, unused) -> responseActionListener.onResponse(
+                    new UpdateDataStreamSettingsAction.Response(dataStreamSettingsResponse)
+                )
+            )
+        );
         countDownListener.onResponse(null);
         for (String dataStreamName : dataStreamNames) {
             updateSingleDataStream(
                 dataStreamName,
                 request.getSettings(),
+                request.isDryRun(),
                 request.masterNodeTimeout(),
                 request.ackTimeout(),
-                new ActionListener<>() {
-                    @Override
-                    public void onResponse(UpdateDataStreamSettingsAction.DataStreamSettingsResponse dataStreamResponse) {
-                        dataStreamSettingsResponse.add(dataStreamResponse);
-                        countDownListener.onResponse(null);
-                    }
-
-                    @Override
-                    public void onFailure(Exception e) {
-                        dataStreamSettingsResponse.add(
-                            new UpdateDataStreamSettingsAction.DataStreamSettingsResponse(
-                                dataStreamName,
-                                false,
-                                e.getMessage(),
-                                EMPTY,
-                                EMPTY,
-                                UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndicesSettingsResult.EMPTY
-                            )
-                        );
-                        countDownListener.onResponse(null);
-                    }
-                }
+                ActionListener.wrap(dataStreamResponse -> {
+                    dataStreamSettingsResponse.add(dataStreamResponse);
+                    countDownListener.onResponse(null);
+                }, e -> {
+                    dataStreamSettingsResponse.add(
+                        new UpdateDataStreamSettingsAction.DataStreamSettingsResponse(
+                            dataStreamName,
+                            false,
+                            e.getMessage(),
+                            EMPTY,
+                            EMPTY,
+                            UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndicesSettingsResult.EMPTY
+                        )
+                    );
+                    countDownListener.onResponse(null);
+                })
             );
         }
     }
 
     private void updateSingleDataStream(
         String dataStreamName,
         Settings settingsOverrides,
+        boolean dryRun,
         TimeValue masterNodeTimeout,
         TimeValue ackTimeout,
         ActionListener<UpdateDataStreamSettingsAction.DataStreamSettingsResponse> listener
@@ -187,13 +179,17 @@ private void updateSingleDataStream(
             );
             return;
         }
-        metadataDataStreamsService.updateSettings(masterNodeTimeout, ackTimeout, dataStreamName, settingsOverrides, new ActionListener<>() {
-            @Override
-            public void onResponse(AcknowledgedResponse acknowledgedResponse) {
-                if (acknowledgedResponse.isAcknowledged()) {
-                    updateSettingsOnIndices(dataStreamName, settingsOverrides, masterNodeTimeout, ackTimeout, listener);
+        metadataDataStreamsService.updateSettings(
+            masterNodeTimeout,
+            ackTimeout,
+            dataStreamName,
+            settingsOverrides,
+            dryRun,
+            listener.delegateFailure((dataStreamSettingsResponseActionListener, dataStream) -> {
+                if (dataStream != null) {
+                    updateSettingsOnIndices(dataStream, settingsOverrides, dryRun, masterNodeTimeout, ackTimeout, listener);
                 } else {
-                    listener.onResponse(
+                    dataStreamSettingsResponseActionListener.onResponse(
                         new UpdateDataStreamSettingsAction.DataStreamSettingsResponse(
                             dataStreamName,
                             false,
@@ -204,18 +200,14 @@ public void onResponse(AcknowledgedResponse acknowledgedResponse) {
                         )
                     );
                 }
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                listener.onFailure(e);
-            }
-        });
+            })
+        );
     }
 
     private void updateSettingsOnIndices(
-        String dataStreamName,
+        DataStream dataStream,
         Settings requestSettings,
+        boolean dryRun,
         TimeValue masterNodeTimeout,
         TimeValue ackTimeout,
         ActionListener<UpdateDataStreamSettingsAction.DataStreamSettingsResponse> listener
@@ -231,17 +223,15 @@ private void updateSettingsOnIndices(
                 appliedToDataStreamOnly.add(settingName);
             }
         }
-        final List<Index> concreteIndices = clusterService.state().metadata().dataStreams().get(dataStreamName).getIndices();
+        final List<Index> concreteIndices = dataStream.getIndices();
         final List<UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError> indexSettingErrors = new ArrayList<>();
 
-        CountDownActionListener indexCountDownListener = new CountDownActionListener(concreteIndices.size() + 1, new ActionListener<>() {
-            // Called when all indices for all settings are complete
-            @Override
-            public void onResponse(Void unused) {
-                DataStream dataStream = clusterService.state().metadata().dataStreams().get(dataStreamName);
-                listener.onResponse(
+        CountDownActionListener indexCountDownListener = new CountDownActionListener(
+            concreteIndices.size() + 1,
+            listener.delegateFailure(
+                (dataStreamSettingsResponseActionListener, unused) -> dataStreamSettingsResponseActionListener.onResponse(
                     new UpdateDataStreamSettingsAction.DataStreamSettingsResponse(
-                        dataStreamName,
+                        dataStream.getName(),
                         true,
                         null,
                         settingsFilter.filter(dataStream.getSettings()),
@@ -252,37 +242,33 @@ public void onResponse(Void unused) {
                             indexSettingErrors
                         )
                     )
-                );
-            }
+                )
+            )
+        );
 
-            @Override
-            public void onFailure(Exception e) {
-                listener.onFailure(e);
-            }
-        });
         indexCountDownListener.onResponse(null); // handles the case where there were zero indices
         Settings applyToIndexSettings = builder().loadFromMap(settingsToApply).build();
         for (Index index : concreteIndices) {
-            updateSettingsOnSingleIndex(index, applyToIndexSettings, masterNodeTimeout, ackTimeout, new ActionListener<>() {
-                @Override
-                public void onResponse(UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError indexSettingError) {
+            updateSettingsOnSingleIndex(
+                index,
+                applyToIndexSettings,
+                dryRun,
+                masterNodeTimeout,
+                ackTimeout,
+                indexCountDownListener.delegateFailure((listener1, indexSettingError) -> {
                     if (indexSettingError != null) {
                         indexSettingErrors.add(indexSettingError);
                     }
-                    indexCountDownListener.onResponse(null);
-                }
-
-                @Override
-                public void onFailure(Exception e) {
-                    indexCountDownListener.onFailure(e);
-                }
-            });
+                    listener1.onResponse(null);
+                })
+            );
         }
     }
 
     private void updateSettingsOnSingleIndex(
         Index index,
         Settings requestSettings,
+        boolean dryRun,
         TimeValue masterNodeTimeout,
         TimeValue ackTimeout,
         ActionListener<UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError> listener
@@ -302,18 +288,23 @@ private void updateSettingsOnSingleIndex(
                 );
                 return;
             }
-            updateSettingsService.updateSettings(
-                new UpdateSettingsClusterStateUpdateRequest(
-                    masterNodeTimeout,
-                    ackTimeout,
-                    requestSettings,
-                    UpdateSettingsClusterStateUpdateRequest.OnExisting.OVERWRITE,
-                    UpdateSettingsClusterStateUpdateRequest.OnStaticSetting.REOPEN_INDICES,
-                    index
-                ),
-                new ActionListener<>() {
-                    @Override
-                    public void onResponse(AcknowledgedResponse response) {
+            if (dryRun) {
+                /*
+                 * This is as far as we go with dry run mode. We get the benefit of having checked that all the indices that will be touced
+                 * are not blocked, but there is no value in going beyond this. So just respond to the listener and move on.
+                 */
+                listener.onResponse(null);
+            } else {
+                updateSettingsService.updateSettings(
+                    new UpdateSettingsClusterStateUpdateRequest(
+                        masterNodeTimeout,
+                        ackTimeout,
+                        requestSettings,
+                        UpdateSettingsClusterStateUpdateRequest.OnExisting.OVERWRITE,
+                        UpdateSettingsClusterStateUpdateRequest.OnStaticSetting.REOPEN_INDICES,
+                        index
+                    ),
+                    ActionListener.wrap(response -> {
                         UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError error;
                         if (response.isAcknowledged() == false) {
                             error = new UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError(
@@ -324,16 +315,13 @@ public void onResponse(AcknowledgedResponse response) {
                             error = null;
                         }
                         listener.onResponse(error);
-                    }
-
-                    @Override
-                    public void onFailure(Exception e) {
-                        listener.onResponse(
+                    },
+                        e -> listener.onResponse(
                             new UpdateDataStreamSettingsAction.DataStreamSettingsResponse.IndexSettingError(index.getName(), e.getMessage())
-                        );
-                    }
-                }
-            );
+                        )
+                    )
+                );
+            }
         }
 
     }

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

@@ -45,8 +45,10 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         try (XContentParser parser = request.contentParser()) {
             settings = Settings.fromXContent(parser);
         }
+        boolean dryRun = request.paramAsBoolean("dry_run", false);
         UpdateDataStreamSettingsAction.Request putDataStreamRequest = new UpdateDataStreamSettingsAction.Request(
             settings,
+            dryRun,
             RestUtils.getMasterNodeTimeout(request),
             RestUtils.getAckTimeout(request)
         ).indices(Strings.splitStringByCommaToArray(request.param("name")));

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

@@ -192,3 +192,103 @@ setup:
   - 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]" }
+
+---
+"Test dry run":
+  - 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
+        dry_run: true
+        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: {} }
+  - 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.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-policy" }
+  - match: { .$idx1name.settings.index.number_of_shards: "1" }
+  - match: { .$idx1name.settings.index.lifecycle.name: "my-policy" }

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

@@ -27,6 +27,11 @@
       ]
     },
     "params":{
+      "dry_run":{
+        "type":"boolean",
+        "description":"Perform a dry run but do not actually change any settings",
+        "default":false
+      },
       "timeout":{
         "type":"time",
         "description":"Specify timeout for acknowledging the cluster state update"