Adding DynamoDB high availability features

.gitignore

@@ -206,3 +206,5 @@ bin/
 build
 site
 
+# ignore asdf files for jdk manager
+.tool-versions

mantis-control-plane/mantis-control-plane-client/src/main/java/io/mantisrx/server/master/client/HighAvailabilityServicesUtil.java

@@ -19,10 +19,14 @@
 import io.mantisrx.common.metrics.Metrics;
 import io.mantisrx.common.metrics.MetricsRegistry;
 import io.mantisrx.server.core.CoreConfiguration;
+import io.mantisrx.server.core.ILeaderMonitorFactory;
+import io.mantisrx.server.core.master.LocalLeaderFactory;
 import io.mantisrx.server.core.master.LocalMasterMonitor;
 import io.mantisrx.server.core.master.MasterDescription;
 import io.mantisrx.server.core.master.MasterMonitor;
-import io.mantisrx.server.core.zookeeper.CuratorService;
+import io.mantisrx.server.core.master.ZookeeperLeaderMonitorFactory;
+import io.mantisrx.server.core.master.ZookeeperMasterMonitor;
+import io.mantisrx.server.core.utils.ConfigUtils;
 import io.mantisrx.server.master.resourcecluster.ClusterID;
 import io.mantisrx.server.master.resourcecluster.ResourceClusterGateway;
 import io.mantisrx.server.master.resourcecluster.ResourceClusterGatewayClient;
@@ -71,7 +75,7 @@ public static HighAvailabilityServices createHAServices(CoreConfiguration config
     }
     else {
       if (HAServiceInstanceRef.get() == null) {
-          HAServiceInstanceRef.compareAndSet(null, new ZkHighAvailabilityServices(configuration));
+          HAServiceInstanceRef.compareAndSet(null, new HighAvailabilityServicesImpl(configuration));
       }
     }
 
@@ -121,99 +125,104 @@ protected void startUp() throws Exception {
     protected void shutDown() throws Exception {
     }
   }
+    private static class HighAvailabilityServicesImpl extends AbstractIdleService implements
+        HighAvailabilityServices {
+
+        private final MasterMonitor masterMonitor;
+        private final Counter resourceLeaderChangeCounter;
+        private final Counter resourceLeaderAlreadyRegisteredCounter;
+        private final AtomicInteger rmConnections = new AtomicInteger(0);
+        private final CoreConfiguration configuration;
+
+        public HighAvailabilityServicesImpl(CoreConfiguration configuration) {
+            this.configuration = configuration;
+            final ILeaderMonitorFactory factory = ConfigUtils.createInstance(configuration.getLeaderMonitorFactory(), ILeaderMonitorFactory.class);
+            if(factory instanceof LocalLeaderFactory) {
+                log.warn("using default non-local Zookeeper leader monitoring you should set: "+
+                    "mantis.leader.monitor.factory=io.mantisrx.server.core.master.ZookeeperLeaderMonitorFactory");
+                masterMonitor = new ZookeeperLeaderMonitorFactory().createLeaderMonitor(configuration);
+            } else {
+                masterMonitor = factory.createLeaderMonitor(configuration);
+            }
+            // this for backward compatibility, but should be modified to "HighAvailabilityServices" in the future
+            final String metricsGroup = masterMonitor instanceof ZookeeperMasterMonitor ? "ZkHighAvailabilityServices" :
+                "HighAvailabilityServices";
+
+            final Metrics metrics = MetricsRegistry.getInstance().registerAndGet(new Metrics.Builder()
+                .name(metricsGroup)
+                .addCounter("resourceLeaderChangeCounter")
+                .addCounter("resourceLeaderAlreadyRegisteredCounter")
+                .build());
+            resourceLeaderChangeCounter = metrics.getCounter("resourceLeaderChangeCounter");
+            resourceLeaderAlreadyRegisteredCounter = metrics.getCounter("resourceLeaderAlreadyRegisteredCounter");
 
-  /**
-   * Zookeeper based implementation of HighAvailabilityServices that finds the various leader instances
-   * through metadata stored on zookeeper.
-   */
-  private static class ZkHighAvailabilityServices extends AbstractIdleService implements
-      HighAvailabilityServices {
-
-    private final CuratorService curatorService;
-    private final Counter resourceLeaderChangeCounter;
-    private final Counter resourceLeaderAlreadyRegisteredCounter;
-    private final AtomicInteger rmConnections = new AtomicInteger(0);
-    private final CoreConfiguration configuration;
-
-    public ZkHighAvailabilityServices(CoreConfiguration configuration) {
-      curatorService = new CuratorService(configuration);
-      final Metrics metrics = MetricsRegistry.getInstance().registerAndGet(new Metrics.Builder()
-        .name("ZkHighAvailabilityServices")
-        .addCounter("resourceLeaderChangeCounter")
-        .addCounter("resourceLeaderAlreadyRegisteredCounter")
-        .build());
-      resourceLeaderChangeCounter = metrics.getCounter("resourceLeaderChangeCounter");
-      resourceLeaderAlreadyRegisteredCounter = metrics.getCounter("resourceLeaderAlreadyRegisteredCounter");
-      this.configuration = configuration;
-    }
-
-    @Override
-    protected void startUp() throws Exception {
-      curatorService.start();
-    }
-
-    @Override
-    protected void shutDown() throws Exception {
-      curatorService.shutdown();
-    }
-
-    @Override
-    public MantisMasterGateway getMasterClientApi() {
-      return new MantisMasterClientApi(curatorService.getMasterMonitor());
-    }
-
-    @Override
-    public MasterMonitor getMasterMonitor() {
-      return curatorService.getMasterMonitor();
-    }
-
-    @Override
-    public ResourceLeaderConnection<ResourceClusterGateway> connectWithResourceManager(
-        ClusterID clusterID) {
-      return new ResourceLeaderConnection<ResourceClusterGateway>() {
-        final MasterMonitor masterMonitor = curatorService.getMasterMonitor();
+        }
 
-        ResourceClusterGateway currentResourceClusterGateway =
-            new ResourceClusterGatewayClient(clusterID, masterMonitor.getLatestMaster(), configuration);
+        @Override
+        protected void startUp() throws Exception {
+            masterMonitor.start();
+        }
 
-        final String nameFormat =
-            "ResourceClusterGatewayCxn (" + rmConnections.getAndIncrement() + ")-%d";
-        final Scheduler scheduler =
-            Schedulers
-                .from(
-                    Executors
-                        .newSingleThreadExecutor(
-                            new ThreadFactoryBuilder().setNameFormat(nameFormat).build()));
+        @Override
+        protected void shutDown() throws Exception {
+            masterMonitor.shutdown();
+        }
 
-        final List<Subscription> subscriptions = new ArrayList<>();
+        @Override
+        public MantisMasterGateway getMasterClientApi() {
+            return new MantisMasterClientApi(masterMonitor);
+        }
 
         @Override
-        public ResourceClusterGateway getCurrent() {
-          return currentResourceClusterGateway;
+        public MasterMonitor getMasterMonitor() {
+            return masterMonitor;
         }
 
         @Override
-        public void register(ResourceLeaderChangeListener<ResourceClusterGateway> changeListener) {
-          Subscription subscription = masterMonitor
-              .getMasterObservable()
-              .observeOn(scheduler)
-              .subscribe(nextDescription -> {
-                log.info("nextDescription={}", nextDescription);
-
-                if (nextDescription.equals(((ResourceClusterGatewayClient)currentResourceClusterGateway).getMasterDescription())) {
-                    resourceLeaderAlreadyRegisteredCounter.increment();
-                    return;
+        public ResourceLeaderConnection<ResourceClusterGateway> connectWithResourceManager(
+            ClusterID clusterID) {
+            return new ResourceLeaderConnection<ResourceClusterGateway>() {
+                ResourceClusterGateway currentResourceClusterGateway =
+                    new ResourceClusterGatewayClient(clusterID, masterMonitor.getLatestMaster(), configuration);
+
+                final String nameFormat =
+                    "ResourceClusterGatewayCxn (" + rmConnections.getAndIncrement() + ")-%d";
+                final Scheduler scheduler =
+                    Schedulers
+                        .from(
+                            Executors
+                                .newSingleThreadExecutor(
+                                    new ThreadFactoryBuilder().setNameFormat(nameFormat).build()));
+
+                final List<Subscription> subscriptions = new ArrayList<>();
+
+                @Override
+                public ResourceClusterGateway getCurrent() {
+                    return currentResourceClusterGateway;
                 }
-                ResourceClusterGateway previous = currentResourceClusterGateway;
-                currentResourceClusterGateway = new ResourceClusterGatewayClient(clusterID, nextDescription, configuration);
-
-                resourceLeaderChangeCounter.increment();
-                changeListener.onResourceLeaderChanged(previous, currentResourceClusterGateway);
-              });
 
-          subscriptions.add(subscription);
+                @Override
+                public void register(ResourceLeaderChangeListener<ResourceClusterGateway> changeListener) {
+                    Subscription subscription = masterMonitor
+                        .getMasterObservable()
+                        .observeOn(scheduler)
+                        .subscribe(nextDescription -> {
+                            log.info("nextDescription={}", nextDescription);
+
+                            if (nextDescription.equals(((ResourceClusterGatewayClient)currentResourceClusterGateway).getMasterDescription())) {
+                                resourceLeaderAlreadyRegisteredCounter.increment();
+                                return;
+                            }
+                            ResourceClusterGateway previous = currentResourceClusterGateway;
+                            currentResourceClusterGateway = new ResourceClusterGatewayClient(clusterID, nextDescription, configuration);
+
+                            resourceLeaderChangeCounter.increment();
+                            changeListener.onResourceLeaderChanged(previous, currentResourceClusterGateway);
+                        });
+
+                    subscriptions.add(subscription);
+                }
+            };
         }
-      };
     }
-  }
 }

mantis-control-plane/mantis-control-plane-client/src/main/java/io/mantisrx/server/master/client/TestGetMasterMonitor.java

@@ -21,7 +21,7 @@
 import io.mantisrx.server.core.CoreConfiguration;
 import io.mantisrx.server.core.master.MasterDescription;
 import io.mantisrx.server.core.master.MasterMonitor;
-import io.mantisrx.server.core.zookeeper.CuratorService;
+import io.mantisrx.server.core.master.ZookeeperLeaderMonitorFactory;
 import io.mantisrx.server.master.client.config.StaticPropertiesConfigurationFactory;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -56,8 +56,7 @@ public static void main(String[] args) {
         final CountDownLatch latch = new CountDownLatch(5);
         StaticPropertiesConfigurationFactory configurationFactory = new StaticPropertiesConfigurationFactory(properties);
         CoreConfiguration config = configurationFactory.getConfig();
-        final CuratorService curatorService = new CuratorService(config);
-        MasterMonitor masterMonitor = curatorService.getMasterMonitor();
+        MasterMonitor masterMonitor = new ZookeeperLeaderMonitorFactory().createLeaderMonitor(config);
         masterMonitor.getMasterObservable()
                 .filter(new Func1<MasterDescription, Boolean>() {
                     @Override
@@ -73,7 +72,7 @@ public void call(MasterDescription masterDescription) {
                     }
                 })
                 .subscribe();
-        curatorService.start();
+        masterMonitor.start();
         try {
             latch.await();
         } catch (InterruptedException e) {

mantis-control-plane/mantis-control-plane-client/src/test/java/io/mantisrx/server/master/client/MasterClientWrapperTest.java

@@ -46,6 +46,8 @@ public class MasterClientWrapperTest {
         zkProps.put("mantis.zookeeper.leader.announcement.path", "/leader");
         zkProps.put("mantis.zookeeper.root", "/mantis/master");
         zkProps.put("mantis.localmode", "false");
+        zkProps.put("mantis.leader.monitor.factory","io.mantisrx.server.core.master.ZookeeperLeaderMonitorFactory");
+        zkProps.put("mantis.leader.elector.factory", "io.mantisrx.master.zk.ZookeeperLeadershipFactory");
     }
 
     MasterClientWrapper clientWrapper = null;

mantis-control-plane/mantis-control-plane-core/build.gradle

@@ -33,6 +33,7 @@ dependencies {
     api libraries.flinkRpcApi
     api libraries.flinkCore
     implementation libraries.commonsIo
+    implementation libraries.vavr
     compileOnly libraries.jsr305
     compileOnly libraries.flinkRpcImpl // Provided by copyLibs task
 

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/CoreConfiguration.java

@@ -77,4 +77,8 @@ public interface CoreConfiguration {
     @Config("mantis.asyncHttpClient.readTimeoutMs")
     @Default("10000")
     int getAsyncHttpClientReadTimeoutMs();
+
+    @Config("mantis.leader.monitor.factory")
+    @Default("io.mantisrx.server.core.master.LocalLeaderFactory")
+    String getLeaderMonitorFactory();
 }