ILM: Make UpdateSettingsStep retryable (#51235) (#51298)

This makes the UpdateSettingsStep retryable. This step updates settings needed
during the execution of ILM actions (mark indexes as read-only, change
allocation configurations, mark indexing complete, etc)

As the index updates are idempotent in nature (PUT requests and are applied only
if the values have changed) and the settings values are seldom user-configurable
(aside from the allocate action) the testing for this change goes along the
lines of artificially simulating a setting update failure on a particular value
update, which is followed by a successful step execution (a retry) in an
environment outside of ILM (the step executions are triggered manually).

(cherry picked from commit 8391b0aba469f39532bfc2796b76148167dc0289)
Signed-off-by: Andrei Dan <andrei.dan@elastic.co>

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/UpdateSettingsStep.java

@@ -28,6 +28,11 @@ public class UpdateSettingsStep extends AsyncActionStep {
         this.settings = settings;
     }
 
+    @Override
+    public boolean isRetryable() {
+        return true;
+    }
+
     @Override
     public void performAction(IndexMetaData indexMetaData, ClusterState currentState, ClusterStateObserver observer, Listener listener) {
         UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(indexMetaData.getIndex().getName()).settings(settings);

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/UpdateSettingsStepTests.java

@@ -0,0 +1,172 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ilm;
+
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Setting.Property;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.core.ilm.AsyncActionStep;
+import org.elasticsearch.xpack.core.ilm.Step.StepKey;
+import org.elasticsearch.xpack.core.ilm.UpdateSettingsStep;
+import org.junit.After;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.test.ESIntegTestCase.Scope.SUITE;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.ilm.UpdateSettingsStepTests.SettingsTestingService.INVALID_VALUE;
+import static org.hamcrest.Matchers.is;
+
+@ClusterScope(scope = SUITE, supportsDedicatedMasters = false, numDataNodes = 1, numClientNodes = 0)
+public class UpdateSettingsStepTests extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(SettingsListenerPlugin.class);
+    }
+
+    private static final SettingsTestingService service = new SettingsTestingService();
+
+    public static class SettingsListenerPlugin extends Plugin {
+
+        @Override
+        public List<Setting<?>> getSettings() {
+            return Arrays.asList(SettingsTestingService.VALUE);
+        }
+
+        @Override
+        public void onIndexModule(IndexModule module) {
+            module.addSettingsUpdateConsumer(SettingsTestingService.VALUE, service::setValue, service::validate);
+        }
+
+        @Override
+        public Collection<Object> createComponents(Client client, ClusterService clusterService, ThreadPool threadPool,
+                                                   ResourceWatcherService resourceWatcherService, ScriptService scriptService,
+                                                   NamedXContentRegistry xContentRegistry, Environment environment,
+                                                   NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry) {
+            return Collections.singletonList(service);
+        }
+    }
+
+    public static class SettingsListenerModule extends AbstractModule {
+        private final SettingsTestingService service;
+
+        SettingsListenerModule(SettingsTestingService service) {
+            this.service = service;
+        }
+
+        @Override
+        protected void configure() {
+            bind(SettingsTestingService.class).toInstance(service);
+        }
+    }
+
+    static class SettingsTestingService {
+        public static final String INVALID_VALUE = "INVALID";
+        static Setting<String> VALUE = Setting.simpleString("index.test.setting", Property.Dynamic, Property.IndexScope);
+
+        public volatile String value;
+
+        void setValue(String value) {
+            this.value = value;
+        }
+
+        void validate(String value) {
+            if (value.equals(INVALID_VALUE)) {
+                throw new IllegalArgumentException("[" + INVALID_VALUE + "] is not supported");
+            }
+        }
+
+        void resetValues() {
+            this.value = "";
+        }
+    }
+
+    @After
+    public void resetSettingValue() {
+        service.resetValues();
+    }
+
+    public void testUpdateSettingsStepRetriesOnError() throws InterruptedException {
+        assertAcked(client().admin().indices().prepareCreate("test").setSettings(Settings.builder()
+            .build()).get());
+
+        ClusterState state = clusterService().state();
+        IndexMetaData indexMetaData = state.metaData().index("test");
+        ThreadPool threadPool = internalCluster().getInstance(ThreadPool.class);
+        ClusterStateObserver observer = new ClusterStateObserver(clusterService(), null, logger, threadPool.getThreadContext());
+
+        CountDownLatch latch = new CountDownLatch(2);
+
+        // fail the first setting update by using an invalid valid
+        Settings invalidValueSetting = Settings.builder().put("index.test.setting", INVALID_VALUE).build();
+        UpdateSettingsStep step = new UpdateSettingsStep(
+            new StepKey("hot", "action", "updateSetting"), new StepKey("hot", "action", "validate"), client(),
+            invalidValueSetting);
+
+        step.performAction(indexMetaData, state, observer, new AsyncActionStep.Listener() {
+            @Override
+            public void onResponse(boolean complete) {
+                latch.countDown();
+                fail("expected the test to fail when updating the setting to an invalid value");
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                latch.countDown();
+
+                // use a valid setting value so the second update call is successful
+                Settings validIndexSetting = Settings.builder().put("index.test.setting", "valid").build();
+                UpdateSettingsStep step = new UpdateSettingsStep(
+                    new StepKey("hot", "action", "updateSetting"), new StepKey("hot", "action", "validate"), client(),
+                    validIndexSetting);
+
+                step.performAction(indexMetaData, state, observer, new AsyncActionStep.Listener() {
+                    @Override
+                    public void onResponse(boolean complete) {
+                        latch.countDown();
+                        assertThat(complete, is(true));
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        latch.countDown();
+                        fail("unexpected failure when trying to update setting to a valid value");
+                    }
+                });
+            }
+        });
+
+
+        latch.await(10, TimeUnit.SECONDS);
+
+        for (SettingsTestingService instance : internalCluster().getDataNodeInstances(SettingsTestingService.class)) {
+            assertThat(instance.value, is("valid"));
+        }
+    }
+}