[Zen2] Support rolling upgrades from Zen1

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

@@ -32,6 +32,8 @@
 import java.util.Map;
 import java.util.Optional;
 
+import static org.elasticsearch.cluster.coordination.Coordinator.ZEN1_BWC_TERM;
+
 /**
  * The core class of the cluster state coordination algorithm, directly implementing the
  * <a href="https://github.com/elastic/elasticsearch-formal-models/blob/master/ZenWithTerms/tla/ZenWithTerms.tla">formal model</a>
@@ -321,10 +323,15 @@ public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
                 getCurrentTerm());
         }
         if (clusterState.term() == getLastAcceptedTerm() && clusterState.version() <= getLastAcceptedVersion()) {
-            logger.debug("handlePublishRequest: ignored publish request due to version mismatch (expected: >[{}], actual: [{}])",
-                getLastAcceptedVersion(), clusterState.version());
-            throw new CoordinationStateRejectedException("incoming version " + clusterState.version() +
-                " lower or equal to current version " + getLastAcceptedVersion());
+            if (clusterState.term() == ZEN1_BWC_TERM) {
+                logger.debug("handling publish request in compatibility mode despite version mismatch (expected: >[{}], actual: [{}])",
+                    getLastAcceptedVersion(), clusterState.version());
+            } else {
+                logger.debug("handlePublishRequest: ignored publish request due to version mismatch (expected: >[{}], actual: [{}])",
+                    getLastAcceptedVersion(), clusterState.version());
+                throw new CoordinationStateRejectedException("incoming version " + clusterState.version() +
+                    " lower or equal to current version " + getLastAcceptedVersion());
+            }
         }
 
         logger.trace("handlePublishRequest: accepting publish request for version [{}] and term [{}]",

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

@@ -44,6 +44,7 @@
 import org.elasticsearch.cluster.service.ClusterApplier;
 import org.elasticsearch.cluster.service.ClusterApplier.ClusterApplyListener;
 import org.elasticsearch.cluster.service.MasterService;
+import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
@@ -86,6 +87,9 @@
 import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 
 public class Coordinator extends AbstractLifecycleComponent implements Discovery {
+
+    public static final long ZEN1_BWC_TERM = 0;
+
     private static final Logger logger = LogManager.getLogger(Coordinator.class);
 
     // the timeout for the publication of each value
@@ -121,6 +125,7 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery
     private long maxTermSeen;
     private final Reconfigurator reconfigurator;
     private final ClusterBootstrapService clusterBootstrapService;
+    private final DiscoveryUpgradeService discoveryUpgradeService;
     private final LagDetector lagDetector;
     private final ClusterFormationFailureHelper clusterFormationFailureHelper;
 
@@ -161,6 +166,8 @@ public Coordinator(String nodeName, Settings settings, ClusterSettings clusterSe
         masterService.setClusterStateSupplier(this::getStateForMasterService);
         this.reconfigurator = new Reconfigurator(settings, clusterSettings);
         this.clusterBootstrapService = new ClusterBootstrapService(settings, transportService);
+        this.discoveryUpgradeService = new DiscoveryUpgradeService(settings, clusterSettings, transportService, this::isBootstrapped,
+            joinHelper, peerFinder::getFoundPeers, this::unsafelySetConfigurationForUpgrade);
         this.lagDetector = new LagDetector(settings, transportService.getThreadPool(), n -> removeNode(n, "lagging"),
             transportService::getLocalNode);
         this.clusterFormationFailureHelper = new ClusterFormationFailureHelper(settings, this::getClusterFormationState,
@@ -256,6 +263,14 @@ PublishWithJoinResponse handlePublishRequest(PublishRequest publishRequest) {
                 throw new CoordinationStateRejectedException("no longer leading this publication's term: " + publishRequest);
             }
 
+            if (publishRequest.getAcceptedState().term() == ZEN1_BWC_TERM && getCurrentTerm() == ZEN1_BWC_TERM
+                && mode == Mode.FOLLOWER && Optional.of(sourceNode).equals(lastKnownLeader) == false) {
+
+                logger.debug("received cluster state from {} but currently following {}, rejecting", sourceNode, lastKnownLeader);
+                throw new CoordinationStateRejectedException("received cluster state from " + sourceNode + " but currently following "
+                    + lastKnownLeader + ", rejecting");
+            }
+
             ensureTermAtLeast(sourceNode, publishRequest.getAcceptedState().term());
             final PublishResponse publishResponse = coordinationState.get().handlePublishRequest(publishRequest);
 
@@ -323,7 +338,11 @@ private void startElection() {
                 final StartJoinRequest startJoinRequest
                     = new StartJoinRequest(getLocalNode(), Math.max(getCurrentTerm(), maxTermSeen) + 1);
                 logger.debug("starting election with {}", startJoinRequest);
-                getDiscoveredNodes().forEach(node -> joinHelper.sendStartJoinRequest(startJoinRequest, node));
+                getDiscoveredNodes().forEach(node -> {
+                    if (isZen1Node(node) == false) {
+                        joinHelper.sendStartJoinRequest(startJoinRequest, node);
+                    }
+                });
             }
         }
     }
@@ -384,6 +403,11 @@ void becomeCandidate(String method) {
 
             peerFinder.activate(coordinationState.get().getLastAcceptedState().nodes());
             clusterFormationFailureHelper.start();
+
+            if (getCurrentTerm() == ZEN1_BWC_TERM) {
+                discoveryUpgradeService.activate(lastKnownLeader);
+            }
+
             leaderChecker.setCurrentNodes(DiscoveryNodes.EMPTY_NODES);
             leaderChecker.updateLeader(null);
 
@@ -414,6 +438,7 @@ void becomeLeader(String method) {
 
         lastKnownLeader = Optional.of(getLocalNode());
         peerFinder.deactivate(getLocalNode());
+        discoveryUpgradeService.deactivate();
         clusterFormationFailureHelper.stop();
         closePrevotingAndElectionScheduler();
         preVoteCollector.update(getPreVoteResponse(), getLocalNode());
@@ -439,6 +464,7 @@ void becomeFollower(String method, DiscoveryNode leaderNode) {
 
         lastKnownLeader = Optional.of(leaderNode);
         peerFinder.deactivate(leaderNode);
+        discoveryUpgradeService.deactivate();
         clusterFormationFailureHelper.stop();
         closePrevotingAndElectionScheduler();
         cancelActivePublication();
@@ -647,9 +673,6 @@ public boolean setInitialConfiguration(final VotingConfiguration votingConfigura
                 return false;
             }
 
-            assert currentState.term() == 0 : currentState;
-            assert currentState.version() == 0 : currentState;
-
             if (mode != Mode.CANDIDATE) {
                 throw new CoordinationStateRejectedException("Cannot set initial configuration in mode " + mode);
             }
@@ -681,12 +704,60 @@ public boolean setInitialConfiguration(final VotingConfiguration votingConfigura
         }
     }
 
+    private boolean isBootstrapped() {
+        return getLastAcceptedState().getLastAcceptedConfiguration().isEmpty() == false;
+    }
+
+    private void unsafelySetConfigurationForUpgrade(VotingConfiguration votingConfiguration) {
+        assert Version.CURRENT.major == Version.V_6_6_0.major + 1 : "remove this method once unsafe upgrades are no longer needed";
+        synchronized (mutex) {
+            if (mode != Mode.CANDIDATE) {
+                throw new IllegalStateException("Cannot overwrite configuration in mode " + mode);
+            }
+
+            if (isBootstrapped()) {
+                throw new IllegalStateException("Cannot overwrite configuration: configuration is already set to "
+                    + getLastAcceptedState().getLastAcceptedConfiguration());
+            }
+
+            if (lastKnownLeader.map(Coordinator::isZen1Node).orElse(false) == false) {
+                throw new IllegalStateException("Cannot upgrade from last-known leader: " + lastKnownLeader);
+            }
+
+            if (getCurrentTerm() != ZEN1_BWC_TERM) {
+                throw new IllegalStateException("Cannot upgrade, term is " + getCurrentTerm());
+            }
+
+            logger.info("automatically bootstrapping during rolling upgrade, using initial configuration {}", votingConfiguration);
+
+            final long newTerm = 1;
+            final ClusterState currentState = getStateForMasterService();
+            final Builder builder = masterService.incrementVersion(currentState);
+            builder.metaData(MetaData.builder(currentState.metaData()).coordinationMetaData(
+                CoordinationMetaData.builder(currentState.metaData().coordinationMetaData())
+                    .term(newTerm)
+                    .lastAcceptedConfiguration(votingConfiguration)
+                    .lastCommittedConfiguration(votingConfiguration)
+                    .build()));
+            final ClusterState newClusterState = builder.build();
+
+            coordinationState.get().handleStartJoin(new StartJoinRequest(getLocalNode(), newClusterState.term()));
+            coordinationState.get().handlePublishRequest(new PublishRequest(newClusterState));
+
+            followersChecker.clearCurrentNodes();
+            followersChecker.updateFastResponseState(getCurrentTerm(), mode);
+
+            peerFinder.deactivate(getLocalNode());
+            peerFinder.activate(newClusterState.nodes());
+        }
+    }
+
     // Package-private for testing
     ClusterState improveConfiguration(ClusterState clusterState) {
         assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
 
         final Set<DiscoveryNode> liveNodes = StreamSupport.stream(clusterState.nodes().spliterator(), false)
-            .filter(this::hasJoinVoteFrom).collect(Collectors.toSet());
+            .filter(this::hasJoinVoteFrom).filter(discoveryNode -> isZen1Node(discoveryNode) == false).collect(Collectors.toSet());
         final VotingConfiguration newConfig = reconfigurator.reconfigure(liveNodes,
             clusterState.getVotingConfigExclusions().stream().map(VotingConfigExclusion::getNodeId).collect(Collectors.toSet()),
             clusterState.getLastAcceptedConfiguration());
@@ -967,7 +1038,9 @@ public void run() {
                             prevotingRound.close();
                         }
                         final ClusterState lastAcceptedState = coordinationState.get().getLastAcceptedState();
-                        prevotingRound = preVoteCollector.start(lastAcceptedState, getDiscoveredNodes());
+                        final List<DiscoveryNode> discoveredNodes
+                            = getDiscoveredNodes().stream().filter(n -> isZen1Node(n) == false).collect(Collectors.toList());
+                        prevotingRound = preVoteCollector.start(lastAcceptedState, discoveredNodes);
                     }
                 }
             }
@@ -1176,13 +1249,13 @@ protected void sendApplyCommit(DiscoveryNode destination, ApplyCommitRequest app
     }
 
     // TODO: only here temporarily for BWC development, remove once complete
-    public static Settings.Builder addZen1Attribute(Settings.Builder builder) {
-        return builder.put("node.attr.zen1", true);
+    public static Settings.Builder addZen1Attribute(boolean isZen1Node, Settings.Builder builder) {
+        return builder.put("node.attr.zen1", isZen1Node);
     }
 
     // TODO: only here temporarily for BWC development, remove once complete
     public static boolean isZen1Node(DiscoveryNode discoveryNode) {
         return discoveryNode.getVersion().before(Version.V_7_0_0) ||
-            discoveryNode.getAttributes().containsKey("zen1");
+            (Booleans.isTrue(discoveryNode.getAttributes().getOrDefault("zen1", "false")));
     }
 }