Supports Grakn Cluster failover functionality (#232)

## What is the goal of this PR?

We have added support for Grakn Cluster failover, allowing the client to switch between talking with one server to another. With this functionality, the users can now benefit from Cluster's high-availability and horizontal scalability.

Instantiating a client for Grakn Cluster is similar to instantiating a client for Grakn Core. When it connects to the server, it will retrieve complete information about the whole cluster and other servers that belong there.

```
GraknClient.Cluster client = new GraknClient.Cluster("172.0.0.1:1729");
```

When connected to Grakn Cluster, the client and session instances are bound to the whole cluster rather than a particular server. That means that if one server in the cluster crashes, the client and session instances are still fully operational and will simply failover to other servers that are available.

Transaction instances however, are bound to the server they connect to and cannot failover to other servers midway. Therefore if a server crashes, the transactions will simply fail.

Apart from the failover ability, the client behaviour when connecting to Grakn Cluster is otherwise identical to how it behaves when connecting to Grakn Core.

## What are the changes implemented in this PR?

Grakn Cluster specific implementations of the following component:

- `GraknClient.Cluster`: a client implementation that is designed to work with a cluster instance. It contains an array of `GraknClient.Core`, one for each server in the cluster
- `RPCSession.Cluster`: a session implementation that is designed to work with a cluster instance. It contains an array of `RPCSession.Core`, one for each server in the cluster.
- `RPCDatabaseManager.Cluster`: a database manager implementation for managing databases clusters

Author: Ganeshwara Hananda

.grabl/automation.yml

@@ -52,7 +52,7 @@ build:
         bazel run @graknlabs_dependencies//distribution/artifact:create-netrc
         bazel build //...
         bazel run @graknlabs_dependencies//tool/checkstyle:test-coverage
-        bazel test $(bazel query 'kind(checkstyle_test, //...)')
+        bazel test $(bazel query 'kind(checkstyle_test, //...)') --test_output=errors
     build-dependency:
       image: graknlabs-ubuntu-20.04
       command: |

BUILD

@@ -51,6 +51,7 @@ java_library(
         "@graknlabs_graql//java/pattern",
         "@graknlabs_graql//java/query",
         "@graknlabs_protocol//grpc/java:protocol",
+        "@graknlabs_protocol//grpc/java:protocol-cluster",
 
         "@graknlabs_grabl_tracing//client",
 

Grakn.java

@@ -114,6 +114,7 @@ interface Transaction extends AutoCloseable {
 
         enum Type {
             READ(0),
+            READ_REPLICA(2),
             WRITE(1);
 
             private final int id;

GraknClient.java

@@ -19,63 +19,171 @@
 
 package grakn.client;
 
-import grakn.client.Grakn.Client;
-import grakn.client.Grakn.DatabaseManager;
-import grakn.client.Grakn.Session;
+import grakn.client.common.exception.GraknClientException;
+import grakn.client.rpc.Address;
 import grakn.client.rpc.RPCDatabaseManager;
 import grakn.client.rpc.RPCSession;
+import grakn.common.collection.Pair;
+import grakn.protocol.cluster.ClusterProto;
+import grakn.protocol.cluster.GraknClusterGrpc;
 import io.grpc.Channel;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.StatusRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
-public class GraknClient implements Client {
+import static grakn.client.common.exception.ErrorMessage.Client.CLUSTER_NOT_AVAILABLE;
+import static grakn.common.collection.Collections.pair;
 
-    public static final String DEFAULT_URI = "localhost:1729";
+public class GraknClient {
+    public static final String DEFAULT_ADDRESS = "localhost:1729";
 
-    private final ManagedChannel channel;
-    private final DatabaseManager databases;
-
-    public GraknClient() {
-        this(DEFAULT_URI);
+    public static GraknClient.Core core() {
+        return core(DEFAULT_ADDRESS);
     }
 
-    public GraknClient(String address) {
-        channel = ManagedChannelBuilder.forTarget(address).usePlaintext().build();
-        databases = new RPCDatabaseManager(channel);
+    public static GraknClient.Core core(String address) {
+        return new GraknClient.Core(address);
     }
 
-    @Override
-    public Session session(String database, Session.Type type) {
-        return session(database, type, new GraknOptions());
+    public static GraknClient.Cluster cluster() {
+        return new GraknClient.Cluster(DEFAULT_ADDRESS);
     }
 
-    @Override
-    public Session session(String database, Session.Type type, GraknOptions options) {
-        return new RPCSession(this, database, type, options);
+    public static GraknClient.Cluster cluster(String address) {
+        return new GraknClient.Cluster(address);
     }
 
-    @Override
-    public DatabaseManager databases() {
-        return databases;
-    }
+    public static class Core implements Grakn.Client {
+        private final ManagedChannel channel;
+        private final RPCDatabaseManager.Core databases;
 
-    @Override
-    public boolean isOpen() {
-        return !channel.isShutdown();
-    }
+        private Core(String address) {
+            channel = ManagedChannelBuilder.forTarget(address).usePlaintext().build();
+            databases = new RPCDatabaseManager.Core(channel);
+        }
 
-    @Override
-    public void close() {
-        try {
-            channel.shutdown().awaitTermination(10, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
+        @Override
+        public RPCSession.Core session(String database, Grakn.Session.Type type) {
+            return session(database, type, new GraknOptions());
+        }
+
+        @Override
+        public RPCSession.Core session(String database, Grakn.Session.Type type, GraknOptions options) {
+            return new RPCSession.Core(this, database, type, options);
+        }
+
+        @Override
+        public RPCDatabaseManager.Core databases() {
+            return databases;
+        }
+
+        @Override
+        public boolean isOpen() {
+            return !channel.isShutdown();
+        }
+
+        @Override
+        public void close() {
+            try {
+                channel.shutdown().awaitTermination(10, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+
+        public Channel channel() {
+            return channel;
         }
     }
 
-    public Channel channel() {
-        return channel;
+    public static class Cluster implements Grakn.Client {
+        private static final Logger LOG = LoggerFactory.getLogger(Cluster.class);
+        private final Map<Address.Cluster.Server, Core> coreClientMap;
+        private final Core[] coreClientArray;
+        private final AtomicInteger selectedCoreClientIndex;
+        private GraknClusterGrpc.GraknClusterBlockingStub clusterDiscoveryRPC;
+        private final RPCDatabaseManager.Cluster databases;
+        private boolean isOpen;
+
+        private Cluster(String address) {
+            Pair<GraknClusterGrpc.GraknClusterBlockingStub, Set<Address.Cluster.Server>> discovery = discoverCluster(address);
+            clusterDiscoveryRPC = discovery.first();
+            coreClientMap = discovery.second().stream()
+                    .map(addr -> pair(addr, new Core(addr.client())))
+                    .collect(Collectors.toMap(Pair::first, Pair::second));
+            coreClientArray = coreClientMap.values().toArray(new Core[] {});
+            selectedCoreClientIndex = new AtomicInteger();
+            databases = new RPCDatabaseManager.Cluster(
+                    coreClientMap.entrySet().stream()
+                            .map(client -> pair(client.getKey(), client.getValue().databases()))
+                            .collect(Collectors.toMap(Pair::first, Pair::second))
+            );
+            isOpen = true;
+        }
+
+        @Override
+        public RPCSession.Cluster session(String database, Grakn.Session.Type type) {
+            return session(database, type, new GraknOptions());
+        }
+
+        @Override
+        public RPCSession.Cluster session(String database, Grakn.Session.Type type, GraknOptions options) {
+            return new RPCSession.Cluster(this, database, type, options, clusterDiscoveryRPC);
+        }
+
+        public GraknClusterGrpc.GraknClusterBlockingStub selectNextClusterDiscoveryRPC() {
+            Core selected = selectNextCoreClient();
+            clusterDiscoveryRPC = GraknClusterGrpc.newBlockingStub(selected.channel());
+            return clusterDiscoveryRPC;
+        }
+
+        @Override
+        public RPCDatabaseManager.Cluster databases() {
+            return databases;
+        }
+
+        @Override
+        public boolean isOpen() {
+            return isOpen;
+        }
+
+        @Override
+        public void close() {
+            coreClientMap.values().forEach(GraknClient.Core::close);
+            isOpen = false;
+        }
+
+        public Map<Address.Cluster.Server, Core> coreClients() {
+            return coreClientMap;
+        }
+
+        private Pair<GraknClusterGrpc.GraknClusterBlockingStub, Set<Address.Cluster.Server>> discoverCluster(String... addresses) {
+            for (String address: addresses) {
+                try (GraknClient.Core client = new Core(address)) {
+                    LOG.info("Performing server discovery to {}...", address);
+                    GraknClusterGrpc.GraknClusterBlockingStub clusterDiscoveryRPC = GraknClusterGrpc.newBlockingStub(client.channel());
+                    ClusterProto.Cluster.Discover.Res res =
+                            clusterDiscoveryRPC.clusterDiscover(ClusterProto.Cluster.Discover.Req.newBuilder().build());
+                    Set<Address.Cluster.Server> servers = res.getServersList().stream().map(Address.Cluster.Server::parse).collect(Collectors.toSet());
+                    LOG.info("Discovered {}", servers);
+                    return pair(clusterDiscoveryRPC, servers);
+                } catch (StatusRuntimeException e) {
+                    LOG.error("Server discovery to {} failed.", address);
+                }
+            }
+            throw new GraknClientException(CLUSTER_NOT_AVAILABLE.message((Object) addresses)); // remove ambiguity by casting to Object
+        }
+
+        private Core selectNextCoreClient() {
+            return coreClientArray[selectedCoreClientIndex.getAndIncrement() % coreClientMap.size()];
+        }
     }
 }

common/exception/ErrorMessage.java

@@ -36,6 +36,13 @@ public static class Client extends ErrorMessage {
                 new Client(4, "The required field 'res' of type '%s' was not set.");
         public static final Client UNKNOWN_REQUEST_ID =
                 new Client(5, "Received a response with unknown request id '%s'.");
+        public static final Client ILLEGAL_ARGUMENT = new Client(6, "Illegal argument passed into the method: '%s'.");
+
+        public static final Client CLUSTER_LEADER_NOT_YET_ELECTED =
+                new Client(7, "No leader has been elected for latest known term '%s'.");
+
+        public static final Client CLUSTER_NOT_AVAILABLE =
+                new Client(8, "Attempted connecting to these servers, but none are available: '%s'.");
 
         private static final String codePrefix = "CLI";
         private static final String messagePrefix = "Illegal Client State";

dependencies/graknlabs/repositories.bzl

@@ -50,8 +50,8 @@ def graknlabs_dependencies():
 def graknlabs_protocol():
     git_repository(
         name = "graknlabs_protocol",
-        remote = "https://github.com/graknlabs/protocol",
-        tag = "2.0.0-alpha-5", # sync-marker: do not remove this comment, this is used for sync-dependencies by @graknlabs_protocol
+        remote = "https://github.com/lolski/protocol",
+        commit = "b2cdf7d49617886a5643e1c1701e975a8418596c", # sync-marker: do not remove this comment, this is used for sync-dependencies by @graknlabs_protocol
     )
 
 def graknlabs_behaviour():

rpc/Address.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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 grakn.client.rpc;
+
+import java.util.Objects;
+
+import static java.lang.Integer.parseInt;
+
+public class Address {
+    public static class Cluster {
+        public static class Server {
+            private final String host;
+            private final int clientPort;
+            private final int serverPort;
+
+            public static Server parse(String address) {
+                String[] split = address.split(":");
+                return new Server(split[0], parseInt(split[1]), parseInt(split[2]));
+            }
+
+            public Server(String host, int clientPort, int serverPort) {
+                this.host = host;
+                this.clientPort = clientPort;
+                this.serverPort = serverPort;
+            }
+
+            public String host() {
+                return host;
+            }
+
+            public int clientPort() {
+                return clientPort;
+            }
+
+            public int serverPort() {
+                return serverPort;
+            }
+
+            public String server() {
+                return host + ":" + serverPort;
+            }
+
+            public String client() {
+                return host + ":" + clientPort;
+            }
+
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+                Server server = (Server) o;
+                return clientPort == server.clientPort &&
+                        serverPort == server.serverPort &&
+                        Objects.equals(host, server.host);
+            }
+
+            @Override
+            public int hashCode() {
+                return Objects.hash(host, clientPort, serverPort);
+            }
+
+            @Override
+            public String toString() {
+                return host + ":" + clientPort + ":" + serverPort;
+            }
+        }
+    }
+
+}