[Zen2] Implement Tombstone REST APIs (#36007)

* [Zen2] Implement Tombstone REST APIs

* Adds REST API for withdrawing votes and clearing vote withdrawls
* Tests added to Netty4 module since we need a real Network impl. for Http endpoints

modules/transport-netty4/src/test/java/org/elasticsearch/rest/discovery/Zen2RestApiIT.java

@@ -0,0 +1,166 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.rest.discovery;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.ESNetty4IntegTestCase;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.Node;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.cluster.coordination.ClusterBootstrapService;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.routing.UnassignedInfo;
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.discovery.zen.ElectMasterService;
+import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.test.discovery.TestZenDiscovery;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.core.Is.is;
+
+// These tests are here today so they have access to a proper REST client. They cannot be in :server:integTest since the REST client needs a
+// proper transport implementation, and they cannot be REST tests today since they need to restart nodes. When #35599 and friends land we
+// should be able to move these tests to run against a proper cluster instead. TODO do this.
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, transportClientRatio = 0, autoMinMasterNodes = false)
+public class Zen2RestApiIT extends ESNetty4IntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal))
+            .put(TestZenDiscovery.USE_ZEN2.getKey(), true)
+            .put(ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.getKey(), Integer.MAX_VALUE)
+            .put(ClusterBootstrapService.INITIAL_MASTER_NODE_COUNT_SETTING.getKey(), 2)
+            .build();
+    }
+
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false; // enable http
+    }
+
+    public void testRollingRestartOfTwoNodeCluster() throws Exception {
+        final List<String> nodes = internalCluster().startNodes(2);
+        createIndex("test",
+            Settings.builder()
+                .put(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), TimeValue.ZERO) // assign shards
+                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 2) // causes rebalancing
+                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 1)
+                .build());
+        ensureGreen("test");
+
+        RestClient restClient = getRestClient();
+
+        internalCluster().rollingRestart(new InternalTestCluster.RestartCallback() {
+            @Override
+            public void doAfterNodes(int n, Client client) throws IOException {
+                ensureGreen("test");
+                Response response =
+                    restClient.performRequest(new Request("POST", "/_cluster/withdrawn_votes/" + internalCluster().getNodeNames()[n]));
+                assertThat(response.getStatusLine().getStatusCode(), is(200));
+            }
+
+            @Override
+            public Settings onNodeStopped(String nodeName) throws IOException {
+                String viaNode = randomValueOtherThan(nodeName, () -> randomFrom(nodes));
+
+                List<Node> allNodes = restClient.getNodes();
+                try {
+                    restClient.setNodes(
+                        Collections.singletonList(
+                            new Node(
+                                HttpHost.create(
+                                    internalCluster().getInstance(HttpServerTransport.class, viaNode)
+                                        .boundAddress().publishAddress().toString()
+                                )
+                            )
+                        )
+                    );
+                    Response deleteResponse = restClient.performRequest(new Request("DELETE", "/_cluster/withdrawn_votes"));
+                    assertThat(deleteResponse.getStatusLine().getStatusCode(), is(200));
+
+                    ClusterHealthResponse clusterHealthResponse = client(viaNode).admin().cluster().prepareHealth()
+                        .setWaitForEvents(Priority.LANGUID)
+                        .setWaitForNodes(Integer.toString(1))
+                        .setTimeout(TimeValue.timeValueSeconds(30L))
+                        .setWaitForYellowStatus()
+                        .get();
+                    assertFalse(nodeName, clusterHealthResponse.isTimedOut());
+                    return Settings.EMPTY;
+                } finally {
+                    restClient.setNodes(allNodes);
+                }
+            }
+        });
+        ensureStableCluster(2);
+        ensureGreen("test");
+        assertThat(internalCluster().size(), is(2));
+    }
+
+    public void testClearVotingTombstonesNotWaitingForRemoval() throws Exception {
+        List<String> nodes = internalCluster().startNodes(3);
+        RestClient restClient = getRestClient();
+        Response response = restClient.performRequest(new Request("POST", "/_cluster/withdrawn_votes/" + nodes.get(2)));
+        assertThat(response.getStatusLine().getStatusCode(), is(200));
+        assertThat(response.getEntity().getContentLength(), is(0L));
+        Response deleteResponse = restClient.performRequest(new Request("DELETE", "/_cluster/withdrawn_votes/?wait_for_removal=false"));
+        assertThat(deleteResponse.getStatusLine().getStatusCode(), is(200));
+        assertThat(deleteResponse.getEntity().getContentLength(), is(0L));
+    }
+
+    public void testClearVotingTombstonesWaitingForRemoval() throws Exception {
+        List<String> nodes = internalCluster().startNodes(3);
+        RestClient restClient = getRestClient();
+        String nodeToWithdraw = nodes.get(randomIntBetween(0, 2));
+        Response response = restClient.performRequest(new Request("POST", "/_cluster/withdrawn_votes/" + nodeToWithdraw));
+        assertThat(response.getStatusLine().getStatusCode(), is(200));
+        assertThat(response.getEntity().getContentLength(), is(0L));
+        internalCluster().stopRandomNode(InternalTestCluster.nameFilter(nodeToWithdraw));
+        Response deleteResponse = restClient.performRequest(new Request("DELETE", "/_cluster/withdrawn_votes"));
+        assertThat(deleteResponse.getStatusLine().getStatusCode(), is(200));
+        assertThat(deleteResponse.getEntity().getContentLength(), is(0L));
+    }
+
+    public void testFailsOnUnknownNode() throws Exception {
+        internalCluster().startNodes(3);
+        RestClient restClient = getRestClient();
+        try {
+            restClient.performRequest(new Request("POST", "/_cluster/withdrawn_votes/invalid"));
+            fail("Invalid node name should throw.");
+        } catch (ResponseException e) {
+            assertThat(e.getResponse().getStatusLine().getStatusCode(), is(400));
+            assertThat(
+                e.getCause().getMessage(),
+                Matchers.containsString("add voting tombstones request for [invalid] matched no master-eligible nodes")
+            );
+        }
+    }
+}

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -225,6 +225,7 @@
 import org.elasticsearch.rest.action.RestFieldCapabilitiesAction;
 import org.elasticsearch.rest.action.RestMainAction;
 import org.elasticsearch.rest.action.admin.cluster.RestCancelTasksAction;
+import org.elasticsearch.rest.action.admin.cluster.RestClearVotingTombstonesAction;
 import org.elasticsearch.rest.action.admin.cluster.RestClusterAllocationExplainAction;
 import org.elasticsearch.rest.action.admin.cluster.RestClusterGetSettingsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestClusterHealthAction;
@@ -254,6 +255,7 @@
 import org.elasticsearch.rest.action.admin.cluster.RestRestoreSnapshotAction;
 import org.elasticsearch.rest.action.admin.cluster.RestSnapshotsStatusAction;
 import org.elasticsearch.rest.action.admin.cluster.RestVerifyRepositoryAction;
+import org.elasticsearch.rest.action.admin.cluster.RestAddVotingTombstonesAction;
 import org.elasticsearch.rest.action.admin.indices.RestAnalyzeAction;
 import org.elasticsearch.rest.action.admin.indices.RestClearIndicesCacheAction;
 import org.elasticsearch.rest.action.admin.indices.RestCloseIndexAction;
@@ -543,6 +545,8 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
                 catActions.add((AbstractCatAction) a);
             }
         };
+        registerHandler.accept(new RestAddVotingTombstonesAction(settings, restController));
+        registerHandler.accept(new RestClearVotingTombstonesAction(settings, restController));
         registerHandler.accept(new RestMainAction(settings, restController));
         registerHandler.accept(new RestNodesInfoAction(settings, restController, settingsFilter));
         registerHandler.accept(new RestRemoteClusterInfoAction(settings, restController));

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/AddVotingTombstonesResponse.java

@@ -21,14 +21,16 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import java.io.IOException;
 
 /**
  * A response to {@link AddVotingTombstonesRequest} indicating that voting tombstones have been added for the requested nodes and these
  * nodes have been removed from the voting configuration.
  */
-public class AddVotingTombstonesResponse extends ActionResponse {
+public class AddVotingTombstonesResponse extends ActionResponse implements ToXContentObject {
 
     public AddVotingTombstonesResponse() {
     }
@@ -46,4 +48,9 @@ public void readFrom(StreamInput in) throws IOException {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        return builder;
+    }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/configuration/ClearVotingTombstonesResponse.java

@@ -21,13 +21,15 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import java.io.IOException;
 
 /**
  * A response to {@link ClearVotingTombstonesRequest} indicating that voting tombstones have been cleared from the cluster state.
  */
-public class ClearVotingTombstonesResponse extends ActionResponse {
+public class ClearVotingTombstonesResponse extends ActionResponse implements ToXContentObject {
     public ClearVotingTombstonesResponse() {
     }
 
@@ -44,4 +46,9 @@ public void readFrom(StreamInput in) throws IOException {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        return builder;
+    }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestAddVotingTombstonesAction.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.rest.action.admin.cluster;
+
+import org.elasticsearch.action.admin.cluster.configuration.AddVotingTombstonesAction;
+import org.elasticsearch.action.admin.cluster.configuration.AddVotingTombstonesRequest;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+
+public class RestAddVotingTombstonesAction extends BaseRestHandler {
+
+    private static final TimeValue DEFAULT_TIMEOUT = TimeValue.timeValueSeconds(30L);
+
+    public RestAddVotingTombstonesAction(Settings settings, RestController controller) {
+        super(settings);
+        controller.registerHandler(RestRequest.Method.POST, "/_cluster/withdrawn_votes/{node_name}", this);
+    }
+
+    @Override
+    public String getName() {
+        return "add_voting_tombstones_action";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        String nodeName = request.param("node_name");
+        AddVotingTombstonesRequest addVotingTombstonesRequest = new AddVotingTombstonesRequest(
+            new String[]{nodeName},
+            TimeValue.parseTimeValue(request.param("timeout"), DEFAULT_TIMEOUT, getClass().getSimpleName() + ".timeout")
+        );
+        return channel -> client.execute(
+            AddVotingTombstonesAction.INSTANCE,
+            addVotingTombstonesRequest,
+            new RestToXContentListener<>(channel)
+        );
+    }
+}

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClearVotingTombstonesAction.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.rest.action.admin.cluster;
+
+import org.elasticsearch.action.admin.cluster.configuration.ClearVotingTombstonesAction;
+import org.elasticsearch.action.admin.cluster.configuration.ClearVotingTombstonesRequest;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+
+public class RestClearVotingTombstonesAction extends BaseRestHandler {
+
+    public RestClearVotingTombstonesAction(Settings settings, RestController controller) {
+        super(settings);
+        controller.registerHandler(RestRequest.Method.DELETE, "/_cluster/withdrawn_votes", this);
+    }
+
+    @Override
+    public String getName() {
+        return "clear_voting_tombstones_action";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
+        ClearVotingTombstonesRequest req = new ClearVotingTombstonesRequest();
+        if (request.hasParam("wait_for_removal")) {
+            req.setWaitForRemoval(request.paramAsBoolean("wait_for_removal", true));
+        }
+        return channel -> client.execute(ClearVotingTombstonesAction.INSTANCE, req, new RestToXContentListener<>(channel));
+    }
+}