Expose minimum_master_nodes in cluster state

To safely support rolling upgrades from 6.x to 7.x we need the 7.x nodes to
have access to the `minimum_master_nodes` setting, but this setting is
otherwise unnecessary in 7.x and we would like to remove it.  Since a rolling
upgrade from 6.x to 7.x involves the 7.x nodes joining a 6.x master, we can
avoid the need for setting `minimum_master_nodes` on the 7.x nodes by copying
the value set on the 6.x master.

This change exposes the master's node-level value for `minimum_master_nodes`
via a field in the cluster state.

Author: DaveCTurner

server/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateAction.java

@@ -127,6 +127,7 @@ private void buildResponse(final ClusterStateRequest request,
         ClusterState.Builder builder = ClusterState.builder(currentState.getClusterName());
         builder.version(currentState.version());
         builder.stateUUID(currentState.stateUUID());
+        builder.minimumMasterNodesOnPublishingMaster(currentState.getMinimumMasterNodesOnPublishingMaster());
 
         if (request.nodes()) {
             builder.nodes(currentState.nodes());

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -23,6 +23,7 @@
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlocks;
@@ -178,17 +179,19 @@ default boolean isPrivate() {
 
     private final boolean wasReadFromDiff;
 
+    private final int minimumMasterNodesOnPublishingMaster;
+
     // built on demand
     private volatile RoutingNodes routingNodes;
 
     public ClusterState(long version, String stateUUID, ClusterState state) {
         this(state.clusterName, version, stateUUID, state.metaData(), state.routingTable(), state.nodes(), state.blocks(),
-                state.customs(), false);
+                state.customs(), -1, false);
     }
 
     public ClusterState(ClusterName clusterName, long version, String stateUUID, MetaData metaData, RoutingTable routingTable,
                         DiscoveryNodes nodes, ClusterBlocks blocks, ImmutableOpenMap<String, Custom> customs,
-                        boolean wasReadFromDiff) {
+                        int minimumMasterNodesOnPublishingMaster, boolean wasReadFromDiff) {
         this.version = version;
         this.stateUUID = stateUUID;
         this.clusterName = clusterName;
@@ -197,6 +200,7 @@ public ClusterState(ClusterName clusterName, long version, String stateUUID, Met
         this.nodes = nodes;
         this.blocks = blocks;
         this.customs = customs;
+        this.minimumMasterNodesOnPublishingMaster = minimumMasterNodesOnPublishingMaster;
         this.wasReadFromDiff = wasReadFromDiff;
     }
 
@@ -290,6 +294,8 @@ public Set<VotingConfigExclusion> getVotingConfigExclusions() {
         return coordinationMetaData().getVotingConfigExclusions();
     }
 
+    public int getMinimumMasterNodesOnPublishingMaster() { return minimumMasterNodesOnPublishingMaster; }
+
     // Used for testing and logging to determine how this cluster state was send over the wire
     public boolean wasReadFromDiff() {
         return wasReadFromDiff;
@@ -436,6 +442,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
         if (metrics.contains(Metric.MASTER_NODE)) {
             builder.field("master_node", nodes().getMasterNodeId());
+            builder.field("minimum_master_nodes", minimumMasterNodesOnPublishingMaster);
         }
 
         if (metrics.contains(Metric.BLOCKS)) {
@@ -644,7 +651,7 @@ public static class Builder {
         private ClusterBlocks blocks = ClusterBlocks.EMPTY_CLUSTER_BLOCK;
         private final ImmutableOpenMap.Builder<String, Custom> customs;
         private boolean fromDiff;
-
+        private int minimumMasterNodesOnPublishingMaster = -1;
 
         public Builder(ClusterState state) {
             this.clusterName = state.clusterName;
@@ -655,6 +662,7 @@ public Builder(ClusterState state) {
             this.metaData = state.metaData();
             this.blocks = state.blocks();
             this.customs = ImmutableOpenMap.builder(state.customs());
+            this.minimumMasterNodesOnPublishingMaster = state.minimumMasterNodesOnPublishingMaster;
             this.fromDiff = false;
         }
 
@@ -715,6 +723,11 @@ public Builder stateUUID(String uuid) {
             return this;
         }
 
+        public Builder minimumMasterNodesOnPublishingMaster(int minimumMasterNodesOnPublishingMaster) {
+            this.minimumMasterNodesOnPublishingMaster = minimumMasterNodesOnPublishingMaster;
+            return this;
+        }
+
         public Builder putCustom(String type, Custom custom) {
             customs.put(type, custom);
             return this;
@@ -739,7 +752,8 @@ public ClusterState build() {
             if (UNKNOWN_UUID.equals(uuid)) {
                 uuid = UUIDs.randomBase64UUID();
             }
-            return new ClusterState(clusterName, version, uuid, metaData, routingTable, nodes, blocks, customs.build(), fromDiff);
+            return new ClusterState(clusterName, version, uuid, metaData, routingTable, nodes, blocks, customs.build(),
+                minimumMasterNodesOnPublishingMaster, fromDiff);
         }
 
         public static byte[] toBytes(ClusterState state) throws IOException {
@@ -782,6 +796,7 @@ public static ClusterState readFrom(StreamInput in, DiscoveryNode localNode) thr
             Custom customIndexMetaData = in.readNamedWriteable(Custom.class);
             builder.putCustom(customIndexMetaData.getWriteableName(), customIndexMetaData);
         }
+        builder.minimumMasterNodesOnPublishingMaster = in.getVersion().onOrAfter(Version.V_7_0_0) ? in.readVInt() : -1;
         return builder.build();
     }
 
@@ -807,6 +822,9 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeNamedWriteable(cursor.value);
             }
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeVInt(minimumMasterNodesOnPublishingMaster);
+        }
     }
 
     private static class ClusterStateDiff implements Diff<ClusterState> {
@@ -829,6 +847,8 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
 
         private final Diff<ImmutableOpenMap<String, Custom>> customs;
 
+        private final int minimumMasterNodesOnPublishingMaster;
+
         ClusterStateDiff(ClusterState before, ClusterState after) {
             fromUuid = before.stateUUID;
             toUuid = after.stateUUID;
@@ -839,6 +859,7 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
             metaData = after.metaData.diff(before.metaData);
             blocks = after.blocks.diff(before.blocks);
             customs = DiffableUtils.diff(before.customs, after.customs, DiffableUtils.getStringKeySerializer(), CUSTOM_VALUE_SERIALIZER);
+            minimumMasterNodesOnPublishingMaster = after.minimumMasterNodesOnPublishingMaster;
         }
 
         ClusterStateDiff(StreamInput in, DiscoveryNode localNode) throws IOException {
@@ -851,6 +872,7 @@ private static class ClusterStateDiff implements Diff<ClusterState> {
             metaData = MetaData.readDiffFrom(in);
             blocks = ClusterBlocks.readDiffFrom(in);
             customs = DiffableUtils.readImmutableOpenMapDiff(in, DiffableUtils.getStringKeySerializer(), CUSTOM_VALUE_SERIALIZER);
+            minimumMasterNodesOnPublishingMaster = in.getVersion().onOrAfter(Version.V_7_0_0) ? in.readVInt() : -1;
         }
 
         @Override
@@ -864,6 +886,9 @@ public void writeTo(StreamOutput out) throws IOException {
             metaData.writeTo(out);
             blocks.writeTo(out);
             customs.writeTo(out);
+            if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+                out.writeVInt(minimumMasterNodesOnPublishingMaster);
+            }
         }
 
         @Override
@@ -883,9 +908,9 @@ public ClusterState apply(ClusterState state) {
             builder.metaData(metaData.apply(state.metaData));
             builder.blocks(blocks.apply(state.blocks));
             builder.customs(customs.apply(state.customs));
+            builder.minimumMasterNodesOnPublishingMaster(minimumMasterNodesOnPublishingMaster);
             builder.fromDiff(true);
             return builder.build();
         }
-
     }
 }

server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java

@@ -90,7 +90,7 @@ public JoinHelper(Settings settings, AllocationService allocationService, Master
         this.masterService = masterService;
         this.transportService = transportService;
         this.joinTimeout = JOIN_TIMEOUT_SETTING.get(settings);
-        this.joinTaskExecutor = new JoinTaskExecutor(allocationService, logger) {
+        this.joinTaskExecutor = new JoinTaskExecutor(settings, allocationService, logger) {
 
             @Override
             public ClusterTasksResult<JoinTaskExecutor.Task> execute(ClusterState currentState, List<JoinTaskExecutor.Task> joiningTasks)

server/src/main/java/org/elasticsearch/cluster/coordination/JoinTaskExecutor.java

@@ -29,7 +29,9 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.discovery.DiscoverySettings;
+import org.elasticsearch.discovery.zen.ElectMasterService;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -45,6 +47,8 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
 
     private final Logger logger;
 
+    private final int minimumMasterNodesOnLocalNode;
+
     public static class Task {
 
         private final DiscoveryNode node;
@@ -80,9 +84,10 @@ public boolean isFinishElectionTask() {
         private static final String FINISH_ELECTION_TASK_REASON = "_FINISH_ELECTION_";
     }
 
-    public JoinTaskExecutor(AllocationService allocationService, Logger logger) {
+    public JoinTaskExecutor(Settings settings, AllocationService allocationService, Logger logger) {
         this.allocationService = allocationService;
         this.logger = logger;
+        minimumMasterNodesOnLocalNode = ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.get(settings);
     }
 
     @Override
@@ -185,7 +190,9 @@ protected ClusterState.Builder becomeMasterAndTrimConflictingNodes(ClusterState
         // or removed by us above
         ClusterState tmpState = ClusterState.builder(currentState).nodes(nodesBuilder).blocks(ClusterBlocks.builder()
             .blocks(currentState.blocks())
-            .removeGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID)).build();
+            .removeGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID))
+            .minimumMasterNodesOnPublishingMaster(minimumMasterNodesOnLocalNode)
+            .build();
         logger.trace("becomeMasterAndTrimConflictingNodes: {}", tmpState.nodes());
         return ClusterState.builder(allocationService.deassociateDeadNodes(tmpState, false, "removed dead nodes on election"));
     }

server/src/main/java/org/elasticsearch/discovery/zen/NodeJoinController.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 
 import java.util.ArrayList;
@@ -59,9 +60,10 @@ public class NodeJoinController {
     private ElectionContext electionContext = null;
 
 
-    public NodeJoinController(MasterService masterService, AllocationService allocationService, ElectMasterService electMaster) {
+    public NodeJoinController(Settings settings, MasterService masterService, AllocationService allocationService,
+                              ElectMasterService electMaster) {
         this.masterService = masterService;
-        joinTaskExecutor = new JoinTaskExecutor(allocationService, logger) {
+        joinTaskExecutor = new JoinTaskExecutor(settings, allocationService, logger) {
             @Override
             public void clusterStatePublished(ClusterChangedEvent event) {
                 electMaster.logMinimumMasterNodesWarningIfNecessary(event.previousState(), event.state());

server/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java

@@ -220,7 +220,7 @@ public ZenDiscovery(Settings settings, ThreadPool threadPool, TransportService t
         this.membership = new MembershipAction(transportService, new MembershipListener(), onJoinValidators);
         this.joinThreadControl = new JoinThreadControl();
 
-        this.nodeJoinController = new NodeJoinController(masterService, allocationService, electMaster);
+        this.nodeJoinController = new NodeJoinController(settings, masterService, allocationService, electMaster);
         this.nodeRemovalExecutor = new ZenNodeRemovalClusterStateTaskExecutor(allocationService, electMaster, this::submitRejoin, logger);
 
         masterService.setClusterStateSupplier(this::clusterState);

server/src/test/java/org/elasticsearch/cluster/serialization/ClusterSerializationTests.java

@@ -67,7 +67,8 @@ public void testClusterStateSerialization() throws Exception {
             .add(newNode("node3")).localNodeId("node1").masterNodeId("node2").build();
 
         ClusterState clusterState = ClusterState.builder(new ClusterName("clusterName1"))
-            .nodes(nodes).metaData(metaData).routingTable(routingTable).build();
+            .nodes(nodes).metaData(metaData).routingTable(routingTable)
+            .minimumMasterNodesOnPublishingMaster(randomIntBetween(-1, 10)).build();
 
         AllocationService strategy = createAllocationService();
         clusterState = ClusterState.builder(clusterState).routingTable(strategy.reroute(clusterState, "reroute").routingTable()).build();
@@ -78,6 +79,9 @@ public void testClusterStateSerialization() throws Exception {
         assertThat(serializedClusterState.getClusterName().value(), equalTo(clusterState.getClusterName().value()));
 
         assertThat(serializedClusterState.routingTable().toString(), equalTo(clusterState.routingTable().toString()));
+
+        assertThat(serializedClusterState.getMinimumMasterNodesOnPublishingMaster(),
+            equalTo(clusterState.getMinimumMasterNodesOnPublishingMaster()));
     }
 
     public void testRoutingTableSerialization() throws Exception {

server/src/test/java/org/elasticsearch/discovery/zen/MinimumMasterNodesInClusterStateIT.java

@@ -0,0 +1,66 @@
+/*
+ * 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.discovery.zen;
+
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.elasticsearch.discovery.zen.ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING;
+import static org.elasticsearch.test.InternalTestCluster.nameFilter;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.isIn;
+
+@ESIntegTestCase.ClusterScope(numDataNodes = 0, numClientNodes = 0)
+public class MinimumMasterNodesInClusterStateIT extends ESIntegTestCase {
+
+    public void testMasterPublishes() throws Exception {
+        final String firstNode = internalCluster().startNode();
+
+        {
+            final ClusterState localState
+                = client(firstNode).admin().cluster().state(new ClusterStateRequest().local(true)).get().getState();
+            assertThat(localState.getMinimumMasterNodesOnPublishingMaster(), equalTo(1));
+            assertFalse(DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.exists(localState.metaData().settings()));
+        }
+
+        final List<String> secondThirdNodes = internalCluster().startNodes(2);
+        assertThat(internalCluster().getMasterName(), equalTo(firstNode));
+
+        final List<String> allNodes = Stream.concat(Stream.of(firstNode), secondThirdNodes.stream()).collect(Collectors.toList());
+        for (final String node : allNodes) {
+            final ClusterState localState = client(node).admin().cluster().state(new ClusterStateRequest().local(true)).get().getState();
+            assertThat(localState.getMinimumMasterNodesOnPublishingMaster(), equalTo(1));
+            assertThat(DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.get(localState.metaData().settings()), equalTo(2));
+        }
+
+        internalCluster().stopRandomNode(nameFilter(firstNode));
+        assertThat(internalCluster().getMasterName(), isIn(secondThirdNodes));
+
+        for (final String node : secondThirdNodes) {
+            final ClusterState localState = client(node).admin().cluster().state(new ClusterStateRequest().local(true)).get().getState();
+            assertThat(localState.getMinimumMasterNodesOnPublishingMaster(), equalTo(2));
+            assertThat(DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING.get(localState.metaData().settings()), equalTo(2));
+        }
+    }
+}

server/src/test/java/org/elasticsearch/discovery/zen/NodeJoinControllerTests.java

@@ -141,7 +141,7 @@ private void setupMasterServiceAndNodeJoinController(ClusterState initialState)
             throw new IllegalStateException("method setupMasterServiceAndNodeJoinController can only be called once");
         }
         masterService = ClusterServiceUtils.createMasterService(threadPool, initialState);
-        nodeJoinController = new NodeJoinController(masterService, createAllocationService(Settings.EMPTY),
+        nodeJoinController = new NodeJoinController(Settings.EMPTY, masterService, createAllocationService(Settings.EMPTY),
             new ElectMasterService(Settings.EMPTY));
     }
 

server/src/test/java/org/elasticsearch/indices/cluster/ClusterStateChanges.java

@@ -213,7 +213,7 @@ allocationService, new AliasValidator(), environment,
             transportService, clusterService, threadPool, createIndexService, actionFilters, indexNameExpressionResolver);
 
         nodeRemovalExecutor = new NodeRemovalClusterStateTaskExecutor(allocationService, logger);
-        joinTaskExecutor = new JoinTaskExecutor(allocationService, logger);
+        joinTaskExecutor = new JoinTaskExecutor(Settings.EMPTY, allocationService, logger);
     }
 
     public ClusterState createIndex(ClusterState state, CreateIndexRequest request) {