Add class that supports reconfiguring zokeeper

4 files changed, 201 insertions, 1 deletions

diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ReconfigurableVespaZooKeeperServer.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ReconfigurableVespaZooKeeperServer.java
new file mode 100644
index 00000000000..0a025d53fc4
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ReconfigurableVespaZooKeeperServer.java
@@ -0,0 +1,20 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.zookeeper;
+
+import com.google.inject.Inject;
+import com.yahoo.cloud.config.ZookeeperServerConfig;
+import com.yahoo.component.AbstractComponent;
+
+/**
+ * Starts or reconfigures zookeeper cluster
+ *
+ * @author Harald Musum
+ */
+public class ReconfigurableVespaZooKeeperServer extends AbstractComponent implements VespaZooKeeperServer {
+
+    @Inject
+    public ReconfigurableVespaZooKeeperServer(Reconfigurer reconfigurer, ZookeeperServerConfig zookeeperServerConfig) {
+        reconfigurer.startOrReconfigure(zookeeperServerConfig);
+    }
+
+}
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
new file mode 100644
index 00000000000..52c6fda94be
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/Reconfigurer.java
@@ -0,0 +1,103 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.zookeeper;
+
+import com.google.inject.Inject;
+import com.yahoo.cloud.config.ZookeeperServerConfig;
+import com.yahoo.component.AbstractComponent;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.admin.ZooKeeperAdmin;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.stream.Collectors;
+
+/**
+ * Starts zookeeper server and supports reconfiguring zookeeper cluster. Created as a component
+ * without any config injected, to make sure that it is not recreated when config changes.
+ *
+ * @author hmusum
+ */
+public class Reconfigurer extends AbstractComponent {
+    private static final java.util.logging.Logger log = java.util.logging.Logger.getLogger(Reconfigurer.class.getName());
+
+    private ZooKeeperRunner zooKeeperRunner;
+
+    @Inject
+    public Reconfigurer() {
+        log.log(Level.FINE, "Created ZooKeeperReconfigurer");
+    }
+
+    // For testing only
+    void startOrReconfigure(ZookeeperServerConfig newConfig) {
+        if (zooKeeperRunner == null)
+            zooKeeperRunner = startServer(newConfig);
+
+        if (shouldReconfigure(newConfig))
+            reconfigure(newConfig);
+    }
+
+    boolean shouldReconfigure(ZookeeperServerConfig newConfig) {
+        ZookeeperServerConfig existingConfig = zooKeeperRunner.zookeeperServerConfig();
+        if (!newConfig.dynamicReconfiguration() || existingConfig == null) return false;
+
+        return !newConfig.equals(existingConfig);
+    }
+
+    private ZooKeeperRunner startServer(ZookeeperServerConfig zookeeperServerConfig) {
+        //System.out.println("Starting server with config " + zookeeperServerConfig);
+        return new ZooKeeperRunner(zookeeperServerConfig);
+    }
+
+    void reconfigure(ZookeeperServerConfig newConfig) {
+        ZookeeperServerConfig existingConfig = zooKeeperRunner.zookeeperServerConfig();
+
+        List<String> originalServers = List.copyOf(servers(existingConfig));
+        log.log(Level.INFO, "Original servers: " + originalServers);
+
+        List<String> joiningServers = servers(newConfig);
+        List<String> leavingServers = setDifference(originalServers, joiningServers);
+        List<String> addedServers = setDifference(joiningServers, leavingServers);
+
+        log.log(Level.INFO, "Will reconfigure zookeeper cluster. Joining servers: " + joiningServers +
+                            ", leaving servers: " + leavingServers +
+                            ", new members" + addedServers);
+
+        int sessionTimeoutInSeconds = 30;
+        try {
+            ZooKeeperAdmin zooKeeperAdmin = new ZooKeeperAdmin(connectionSpec(existingConfig), sessionTimeoutInSeconds, null);
+
+            long fromConfig = -1;
+            zooKeeperAdmin.reconfigure(joiningServers, originalServers, addedServers, fromConfig, null);
+        } catch (IOException | KeeperException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    List<String> currentServers() {
+        if (zooKeeperRunner == null) return List.of();
+
+        return servers(zooKeeperRunner.zookeeperServerConfig());
+    }
+
+    // Returns items in set a that are not in set b
+    List<String> setDifference(List<String> a, List<String> b) {
+        Set<String> ret = new HashSet<>(a);
+        ret.removeAll(b);
+        return new ArrayList<>(ret);
+    }
+
+    private String connectionSpec(ZookeeperServerConfig config) {
+        return String.join(",", servers(config));
+    }
+
+    private List<String> servers(ZookeeperServerConfig config) {
+        return config.server().stream()
+                .map(server -> server.hostname() + ":" + server.quorumPort() + ":" + server.electionPort())
+                .collect(Collectors.toList());
+    }
+
+}
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZooKeeperRunner.java b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZooKeeperRunner.java
index 492417cef96..9b6624857ac 100644
--- a/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZooKeeperRunner.java
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/main/java/com/yahoo/vespa/zookeeper/ZooKeeperRunner.java
@@ -5,6 +5,7 @@ import com.yahoo.cloud.config.ZookeeperServerConfig;
 import com.yahoo.security.tls.TransportSecurityUtils;
 
 import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import static com.yahoo.vespa.defaults.Defaults.getDefaults;
 import static com.yahoo.vespa.zookeeper.Configurator.zookeeperServerHostnames;
@@ -15,7 +16,8 @@ import static com.yahoo.vespa.zookeeper.Configurator.zookeeperServerHostnames;
  * @author Harald Musum
  */
 public class ZooKeeperRunner implements Runnable {
-    private static final java.util.logging.Logger log = java.util.logging.Logger.getLogger(ZooKeeperRunner.class.getName());
+    private static final Logger log = java.util.logging.Logger.getLogger(ZooKeeperRunner.class.getName());
+
     private final Thread zkServerThread;
     private final ZookeeperServerConfig zookeeperServerConfig;
 
diff --git a/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java b/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
new file mode 100644
index 00000000000..cce9b4f6ff8
--- /dev/null
+++ b/zookeeper-server/zookeeper-server-3.5.6/src/test/java/com/yahoo/vespa/zookeeper/ReconfigurerTest.java
@@ -0,0 +1,75 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.zookeeper;
+
+import com.yahoo.cloud.config.ZookeeperServerConfig;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests dynamic reconfiguration of zookeeper cluster.
+ */
+public class ReconfigurerTest {
+
+    private File cfgFile;
+    private File idFile;
+
+    @Rule
+    public TemporaryFolder folder = new TemporaryFolder();
+
+    @Before
+    public void setup() throws IOException {
+        cfgFile = folder.newFile();
+        idFile = folder.newFile("myid");
+    }
+
+    @Test
+    public void testStartupAndReconfigure() {
+        Reconfigurer reconfigurer = new Reconfigurer();
+        reconfigurer.startOrReconfigure(createConfig(1));
+
+        // Created config has dynamicReconfig set to false
+        assertFalse(reconfigurer.shouldReconfigure(createConfig(2)));
+
+        // Created config has dynamicReconfig set to true
+        assertTrue(reconfigurer.shouldReconfigure(createConfigAllowReconfiguring(2)));
+    }
+
+    private ZookeeperServerConfig createConfigAllowReconfiguring(int numberOfServers) {
+        return createConfig(numberOfServers, true);
+    }
+
+    private ZookeeperServerConfig createConfig(int numberOfServers) {
+        return createConfig(numberOfServers, false);
+    }
+
+    private ZookeeperServerConfig createConfig(int numberOfServers, boolean dynamicReconfiguration) {
+        ZookeeperServerConfig.Builder builder = new ZookeeperServerConfig.Builder();
+        builder.zooKeeperConfigFile(cfgFile.getAbsolutePath());
+        builder.myidFile(idFile.getAbsolutePath());
+        IntStream.range(0, numberOfServers).forEach(i -> {
+            builder.server(newServer(i, "localhost", i, i + 1));
+        });
+        builder.myid(0);
+        builder.dynamicReconfiguration(dynamicReconfiguration);
+        return builder.build();
+    }
+
+    private ZookeeperServerConfig.Server.Builder newServer(int id, String hostName, int electionPort, int quorumPort) {
+        ZookeeperServerConfig.Server.Builder builder = new ZookeeperServerConfig.Server.Builder();
+        builder.id(id);
+        builder.hostname(hostName);
+        builder.electionPort(electionPort);
+        builder.quorumPort(quorumPort);
+        return builder;
+    }
+
+}