Use bulk reconfig mode

5 files changed, 40 insertions, 54 deletions

diff --git a/zookeeper-server/zookeeper-server-3.7.0/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdminImpl.java b/zookeeper-server/zookeeper-server-3.7.0/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdminImpl.java
index acd5cd07a28..157306b5664 100644
--- a/zookeeper-server/zookeeper-server-3.7.0/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdminImpl.java
+++ b/zookeeper-server/zookeeper-server-3.7.0/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdminImpl.java
@@ -19,11 +19,11 @@ public class VespaZooKeeperAdminImpl implements VespaZooKeeperAdmin {
     private static final Logger log = java.util.logging.Logger.getLogger(VespaZooKeeperAdminImpl.class.getName());
 
     @Override
-    public void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException {
+    public void reconfigure(String connectionSpec, String servers) throws ReconfigException {
         try (ZooKeeperAdmin zooKeeperAdmin = createAdmin(connectionSpec)) {
             long fromConfig = -1;
             // Using string parameters because the List variant of reconfigure fails to join empty lists (observed on 3.5.6, fixed in 3.7.0)
-            byte[] appliedConfig = zooKeeperAdmin.reconfigure(joiningServers, leavingServers, null, fromConfig, null);
+            byte[] appliedConfig = zooKeeperAdmin.reconfigure(null, null, servers, fromConfig, null);
             log.log(Level.INFO, "Applied ZooKeeper config: " + new String(appliedConfig, StandardCharsets.UTF_8));
         }
         catch (   KeeperException.ReconfigInProgress
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Configurator.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Configurator.java
index f95704f8b58..c65035106f1 100644
--- a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Configurator.java
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Configurator.java
@@ -85,7 +85,7 @@ public class Configurator {
         sb.append("reconfigEnabled=true").append("\n");
         sb.append("skipACL=yes").append("\n");
         ensureThisServerIsRepresented(config.myid(), config.server());
-        config.server().forEach(server -> addServerToCfg(sb, server, config.clientPort()));
+        config.server().forEach(server -> sb.append(serverSpec(server, config.clientPort(), server.joining())).append("\n"));
         sb.append(new TlsQuorumConfig().createConfig(vespaTlsConfig));
         sb.append(new TlsClientServerConfig().createConfig(vespaTlsConfig));
         return sb.toString();
@@ -110,7 +110,8 @@ public class Configurator {
         }
     }
 
-    private void addServerToCfg(StringBuilder sb, ZookeeperServerConfig.Server server, int clientPort) {
+    static String serverSpec(ZookeeperServerConfig.Server server, int clientPort, boolean joining) {
+        StringBuilder sb = new StringBuilder();
         sb.append("server.")
           .append(server.id())
           .append("=")
@@ -119,7 +120,7 @@ public class Configurator {
           .append(server.quorumPort())
           .append(":")
           .append(server.electionPort());
-        if (server.joining()) {
+        if (joining) {
             // Servers that are joining an existing cluster must be marked as observers. Note that this will NOT
             // actually make the server an observer, but prevent it from forming an ensemble independently of the
             // existing cluster.
@@ -129,8 +130,8 @@ public class Configurator {
               .append("observer");
         }
         sb.append(";")
-          .append(clientPort)
-          .append("\n");
+          .append(clientPort);
+        return sb.toString();
     }
 
     static List<String> zookeeperServerHostnames(ZookeeperServerConfig zookeeperServerConfig) {
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
index 4452e74bb10..514d45c7e77 100644
--- a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
@@ -10,13 +10,14 @@ import com.yahoo.yolean.Exceptions;
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import java.util.stream.Collectors;
+
+import static com.yahoo.vespa.zookeeper.Configurator.serverSpec;
+import static java.util.stream.Collectors.toList;
 
 /**
  * Starts zookeeper server and supports reconfiguring zookeeper cluster. Keep this as a component
@@ -84,30 +85,34 @@ public class Reconfigurer extends AbstractComponent {
         return runner;
     }
 
+    // TODO jonmv: unify server config writing here and in Configurator
+    // TODO jonmv: read dynamic file, discard if old quorum impossible (config file + .dynamic.<id>)
+    // TODO jonmv: if dynamic file, all unlisted servers are observers; otherwise joiners are observers
+    // TODO jonmv: use bulk mode, i.e., supply only new servers
+    // TODO jonmv: always reconfigure when dynamic config is used, and to only non-retired nodes
+    // TODO jonmv: verify reconfig by issuing a dummy write
+    // TODO jonmv: wrap Curator in Provider, for Curator shutdown
+    // TODO jonmv: scale down to 1 server as well
+    // TODO jonmv: unit test this
     private void reconfigure(ZookeeperServerConfig newConfig) {
         Instant reconfigTriggered = Instant.now();
         // No point in trying to reconfigure if there is only one server in the new ensemble,
         // the others will be shutdown or are about to be shutdown
         if (newConfig.server().size() == 1) shutdownAndDie(Duration.ZERO);
 
-        List<String> newServers = difference(servers(newConfig), servers(activeConfig));
-        String leavingServerIds = String.join(",", serverIdsDifference(activeConfig, newConfig));
-        String joiningServersSpec = String.join(",", newServers);
-        leavingServerIds = leavingServerIds.isEmpty() ? null : leavingServerIds;
-        joiningServersSpec = joiningServersSpec.isEmpty() ? null : joiningServersSpec;
-        log.log(Level.INFO, "Will reconfigure ZooKeeper cluster. \nJoining servers: " + joiningServersSpec +
-                            "\nleaving servers: " + leavingServerIds +
+        String newServers = String.join(",", servers(newConfig));
+        log.log(Level.INFO, "Will reconfigure ZooKeeper cluster." +
                             "\nServers in active config:" + servers(activeConfig) +
                             "\nServers in new config:" + servers(newConfig));
         String connectionSpec = localConnectionSpec(activeConfig);
         Instant now = Instant.now();
-        Duration reconfigTimeout = reconfigTimeout(newServers.size());
+        Duration reconfigTimeout = reconfigTimeout(newConfig.server().size());
         Instant end = now.plus(reconfigTimeout);
         // Loop reconfiguring since we might need to wait until another reconfiguration is finished before we can succeed
         for (int attempt = 1; now.isBefore(end); attempt++) {
             try {
                 Instant reconfigStarted = Instant.now();
-                vespaZooKeeperAdmin.reconfigure(connectionSpec, joiningServersSpec, leavingServerIds);
+                vespaZooKeeperAdmin.reconfigure(connectionSpec, newServers);
                 Instant reconfigEnded = Instant.now();
                 log.log(Level.INFO, "Reconfiguration completed in " +
                                     Duration.between(reconfigTriggered, reconfigEnded) +
@@ -146,24 +151,10 @@ public class Reconfigurer extends AbstractComponent {
         return HostName.getLocalhost() + ":" + config.clientPort();
     }
 
-    private static List<String> serverIdsDifference(ZookeeperServerConfig oldConfig, ZookeeperServerConfig newConfig) {
-        return difference(servers(oldConfig), servers(newConfig)).stream()
-                                                                 .map(server -> server.substring(0, server.indexOf('=')))
-                                                                 .collect(Collectors.toList());
-    }
-
     private static List<String> servers(ZookeeperServerConfig config) {
-        // See https://zookeeper.apache.org/doc/r3.6.3/zookeeperReconfig.html#sc_reconfig_clientport for format
         return config.server().stream()
-                     .map(server -> server.id() + "=" + server.hostname() + ":" + server.quorumPort() + ":" +
-                                    server.electionPort() + ";" + config.clientPort())
-                     .collect(Collectors.toList());
-    }
-
-    private static <T> List<T> difference(List<T> list1, List<T> list2) {
-        List<T> copy = new ArrayList<>(list1);
-        copy.removeAll(list2);
-        return copy;
+                     .map(server -> serverSpec(server, config.clientPort(), false))
+                     .collect(toList());
     }
 
 }
diff --git a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdmin.java b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdmin.java
index 8809dca0def..59c9628bcab 100644
--- a/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdmin.java
+++ b/zookeeper-server/zookeeper-server-common/src/main/java/com/yahoo/vespa/zookeeper/VespaZooKeeperAdmin.java
@@ -10,7 +10,7 @@ import java.time.Duration;
  */
 public interface VespaZooKeeperAdmin {
 
-    void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException;
+    void reconfigure(String connectionSpec, String servers) throws ReconfigException;
 
     /* Timeout for connecting to ZooKeeper */
     default Duration sessionTimeout() { return Duration.ofSeconds(30); }
diff --git a/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java b/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
index 1f551dc1e27..b868fd65de3 100644
--- a/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
+++ b/zookeeper-server/zookeeper-server-common/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
@@ -51,8 +51,8 @@ public class ReconfigurerTest {
         ZookeeperServerConfig nextConfig = createConfig(5, true);
         reconfigurer.startOrReconfigure(nextConfig);
         assertEquals("node1:2181", reconfigurer.connectionSpec());
-        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers());
-        assertNull("No servers are leaving", reconfigurer.leavingServers());
+        assertEquals("server.0=node0:2182:2183;2181,server.1=node1:2182:2183;2181,server.2=node2:2182:2183;2181,server.3=node3:2182:2183;2181,server.4=node4:2182:2183;2181",
+                     reconfigurer.servers());
         assertEquals(1, reconfigurer.reconfigurations());
         assertSame(nextConfig, reconfigurer.activeConfig());
 
@@ -66,16 +66,16 @@ public class ReconfigurerTest {
         reconfigurer.startOrReconfigure(nextConfig);
         assertEquals(2, reconfigurer.reconfigurations());
         assertEquals("node1:2181", reconfigurer.connectionSpec());
-        assertNull("No servers are joining", reconfigurer.joiningServers());
-        assertEquals("3,4", reconfigurer.leavingServers());
+        assertEquals("server.0=node0:2182:2183;2181,server.1=node1:2182:2183;2181,server.2=node2:2182:2183;2181",
+                     reconfigurer.servers());
         assertSame(nextConfig, reconfigurer.activeConfig());
 
         // Cluster loses node1, but node3 joins. Indices are shuffled.
         nextConfig = createConfig(3, true, 1);
         reconfigurer.startOrReconfigure(nextConfig);
         assertEquals(3, reconfigurer.reconfigurations());
-        assertEquals("1=node2:2182:2183;2181,2=node3:2182:2183;2181", reconfigurer.joiningServers());
-        assertEquals("1,2", reconfigurer.leavingServers());
+        assertEquals("server.0=node0:2182:2183;2181,server.1=node2:2182:2183;2181,server.2=node3:2182:2183;2181",
+                     reconfigurer.servers());
         assertSame(nextConfig, reconfigurer.activeConfig());
     }
 
@@ -89,8 +89,8 @@ public class ReconfigurerTest {
         ZookeeperServerConfig nextConfig = createConfig(5, true);
         reconfigurer.startOrReconfigure(nextConfig);
         assertEquals("node1:2181", reconfigurer.connectionSpec());
-        assertEquals("3=node3:2182:2183;2181,4=node4:2182:2183;2181", reconfigurer.joiningServers());
-        assertNull("No servers are leaving", reconfigurer.leavingServers());
+        assertEquals("server.0=node0:2182:2183;2181,server.1=node1:2182:2183;2181,server.2=node2:2182:2183;2181,server.3=node3:2182:2183;2181,server.4=node4:2182:2183;2181",
+                     reconfigurer.servers());
         assertEquals(1, reconfigurer.reconfigurations());
         assertSame(nextConfig, reconfigurer.activeConfig());
     }
@@ -163,12 +163,8 @@ public class ReconfigurerTest {
             return zooKeeperAdmin.connectionSpec;
         }
 
-        String joiningServers() {
-            return zooKeeperAdmin.joiningServers;
-        }
-
-        String leavingServers() {
-            return zooKeeperAdmin.leavingServers;
+        String servers() {
+            return zooKeeperAdmin.servers;
         }
 
         int reconfigurations() {
@@ -192,8 +188,7 @@ public class ReconfigurerTest {
     private static class TestableVespaZooKeeperAdmin implements VespaZooKeeperAdmin {
 
         String connectionSpec;
-        String joiningServers;
-        String leavingServers;
+        String servers;
         int reconfigurations = 0;
 
         private int failures = 0;
@@ -205,12 +200,11 @@ public class ReconfigurerTest {
         }
 
         @Override
-        public void reconfigure(String connectionSpec, String joiningServers, String leavingServers) throws ReconfigException {
+        public void reconfigure(String connectionSpec, String servers) throws ReconfigException {
             if (++attempts < failures)
                 throw new ReconfigException("Reconfig failed");
             this.connectionSpec = connectionSpec;
-            this.joiningServers = joiningServers;
-            this.leavingServers = leavingServers;
+            this.servers = servers;
             this.reconfigurations++;
         }