Expose replica info when getting Cluster databases (typedb#186)

## What is the goal of this PR?

We now expose additional information when fetching databases in Cluster. The database name, server address, primary/secondary state, and term number are all available.

This is a **breaking change**. `client.databases().all()` now returns `List[Database]` (previously it was `List[str]`). To get the old functionality (listing the names), use `[db.name() for db in client.databases().all()]`.

Additionally, `client.databases().delete()` has been deleted, and superseded by the new syntax: `client.databases().get(name).delete()`.

## What are the changes implemented in this PR?

- Delete `client.databases().delete`, and add `client.databases().get`, which returns a new `Database` object that has a `delete` method
- Add `GraknClientCluster` and `DatabaseManagerCluster`, which return `DatabaseCluster` objects when `all` and `get` are called. `DatabaseCluster` objects expose their database name, server address, primary/secondary state and term number.

Author: Alex Walker

dependencies/graknlabs/artifacts.bzl

@@ -27,15 +27,15 @@ def graknlabs_grakn_core_artifacts():
         artifact_name = "grakn-core-server-{platform}-{version}.{ext}",
         tag_source = deployment["artifact.release"],
         commit_source = deployment["artifact.snapshot"],
-        commit = "4d449aa198fd5cceca54cb3889114ab5aa1b8e5e",
+        commit = "5f96fb705f8f256e8cb1b918fe85893056c827d2",
     )
 
 def graknlabs_grakn_cluster_artifacts():
     native_artifact_files(
         name = "graknlabs_grakn_cluster_artifact",
         group_name = "graknlabs_grakn_cluster",
-        artifact_name = "grakn-cluster-server-{platform}-{version}.{ext}",
+        artifact_name = "grakn-cluster-all-{platform}-{version}.{ext}",
         tag_source = deployment_private["artifact.release"],
         commit_source = deployment_private["artifact.snapshot"],
-        commit = "55bb7a5bb99fecf6990b0e8ed3220b264f8de452",
+        commit = "cc3382fe6be239973f1e7d8437a71683e3dff809",
     )

grakn/client.py

@@ -28,10 +28,10 @@
 from grakn.common.exception import GraknClientException  # noqa # pylint: disable=unused-import
 from grakn.concept.type.value_type import ValueType  # noqa # pylint: disable=unused-import
 from grakn.options import GraknOptions, GraknClusterOptions
-from grakn.rpc.cluster.failsafe_task import FailsafeTask
-from grakn.rpc.cluster.replica_info import ReplicaInfo
+from grakn.rpc.cluster.failsafe_task import _FailsafeTask
+from grakn.rpc.cluster.database import DatabaseCluster, _DatabaseClusterRPC  # noqa # pylint: disable=unused-import
 from grakn.rpc.cluster.server_address import ServerAddress
-from grakn.rpc.cluster.database_manager import _DatabaseManagerClusterRPC
+from grakn.rpc.cluster.database_manager import DatabaseManagerCluster, _DatabaseManagerClusterRPC
 from grakn.rpc.cluster.session import SessionClusterRPC
 from grakn.rpc.database_manager import DatabaseManager, _DatabaseManagerRPC
 from grakn.rpc.session import Session, SessionType, _SessionRPC
@@ -46,7 +46,7 @@ def core(address=DEFAULT_ADDRESS) -> "GraknClient":
         return _ClientRPC(address)
 
     @staticmethod
-    def cluster(addresses: List[str]) -> "GraknClient":
+    def cluster(addresses: List[str]) -> "GraknClientCluster":
         return _ClientClusterRPC(addresses)
 
     @abstractmethod
@@ -65,6 +65,10 @@ def is_open(self) -> bool:
     def close(self) -> None:
         pass
 
+    @abstractmethod
+    def is_cluster(self) -> bool:
+        pass
+
     @abstractmethod
     def __enter__(self):
         pass
@@ -74,6 +78,13 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         pass
 
 
+class GraknClientCluster(GraknClient):
+
+    @abstractmethod
+    def databases(self) -> DatabaseManagerCluster:
+        pass
+
+
 class _ClientRPC(GraknClient):
 
     def __init__(self, address: str):
@@ -97,6 +108,9 @@ def close(self):
         self._channel.close()
         self._is_open = False
 
+    def is_cluster(self) -> bool:
+        return False
+
     def __enter__(self):
         return self
 
@@ -112,13 +126,13 @@ def channel(self):
 
 
 # _ClientClusterRPC must live in this package because of circular ref with GraknClient
-class _ClientClusterRPC(GraknClient):
+class _ClientClusterRPC(GraknClientCluster):
 
     def __init__(self, addresses: List[str]):
         self._core_clients: Dict[ServerAddress, _ClientRPC] = {addr: _ClientRPC(addr.client()) for addr in self._fetch_cluster_servers(addresses)}
         self._grakn_cluster_grpc_stubs = {addr: GraknClusterStub(client.channel()) for (addr, client) in self._core_clients.items()}
-        self._database_managers = _DatabaseManagerClusterRPC({addr: client.databases() for (addr, client) in self._core_clients.items()})
-        self._replica_info_map: Dict[str, ReplicaInfo] = {}
+        self._database_managers = _DatabaseManagerClusterRPC(self, {addr: client.databases() for (addr, client) in self._core_clients.items()})
+        self._cluster_databases: Dict[str, _DatabaseClusterRPC] = {}
         self._is_open = True
 
     def session(self, database: str, session_type: SessionType, options=None) -> Session:
@@ -132,7 +146,7 @@ def _session_primary_replica(self, database: str, session_type: SessionType, opt
     def _session_any_replica(self, database: str, session_type: SessionType, options=None) -> SessionClusterRPC:
         return _OpenSessionFailsafeTask(database, session_type, options, self).run_any_replica()
 
-    def databases(self) -> DatabaseManager:
+    def databases(self) -> DatabaseManagerCluster:
         return self._database_managers
 
     def is_open(self) -> bool:
@@ -143,14 +157,17 @@ def close(self) -> None:
             client.close()
         self._is_open = False
 
+    def is_cluster(self) -> bool:
+        return True
+
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
-    def replica_info_map(self) -> Dict[str, ReplicaInfo]:
-        return self._replica_info_map
+    def cluster_databases(self) -> Dict[str, _DatabaseClusterRPC]:
+        return self._cluster_databases
 
     def cluster_members(self) -> Set[ServerAddress]:
         return set(self._core_clients.keys())
@@ -176,12 +193,12 @@ def _fetch_cluster_servers(self, addresses: List[str]) -> Set[ServerAddress]:
         raise GraknClientException("Unable to connect to Grakn Cluster. Attempted connecting to the cluster members, but none are available: %s" % str(addresses))
 
 
-class _OpenSessionFailsafeTask(FailsafeTask):
+class _OpenSessionFailsafeTask(_FailsafeTask):
 
     def __init__(self, database: str, session_type: SessionType, options: GraknClusterOptions, client: "_ClientClusterRPC"):
         super().__init__(client, database)
         self.session_type = session_type
         self.options = options
 
-    def run(self, replica: ReplicaInfo.Replica):
+    def run(self, replica: _DatabaseClusterRPC.Replica):
         return SessionClusterRPC(self.client, replica.address(), self.database, self.session_type, self.options)

grakn/grakn_proto_builder.py

@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from typing import Type, Union
+from typing import Union
 
 import grakn_protocol.protobuf.options_pb2 as options_proto
 

grakn/rpc/cluster/database.py

@@ -0,0 +1,134 @@
+#
+# 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.
+#
+from typing import Dict, Optional, Set
+
+import grakn_protocol.protobuf.cluster.database_pb2 as database_proto
+
+from grakn.rpc.cluster.server_address import ServerAddress
+from grakn.rpc.database import DatabaseCluster, _DatabaseRPC
+
+
+class _DatabaseClusterRPC(DatabaseCluster):
+
+    def __init__(self, database_manager_cluster, database: str):
+        self._databases: Dict[ServerAddress, _DatabaseRPC] = {}
+        for address in database_manager_cluster.database_managers():
+            database_manager = database_manager_cluster.database_managers()[address]
+            self._databases[address] = _DatabaseRPC(database_manager, name=database)
+        self._name = database
+        self._database_manager_cluster = database_manager_cluster
+        self._replicas: Set["_DatabaseClusterRPC.Replica"] = set()
+
+    @staticmethod
+    def of(proto_db: database_proto.Database, database_manager_cluster) -> "_DatabaseClusterRPC":
+        assert proto_db.replicas
+        database: str = proto_db.name
+        database_cluster_rpc = _DatabaseClusterRPC(database_manager_cluster, database)
+        for proto_replica in proto_db.replicas:
+            database_cluster_rpc.replicas().add(_DatabaseClusterRPC.Replica.of(proto_replica, database_cluster_rpc))
+        print("Discovered database cluster: " + str(database_cluster_rpc))
+        return database_cluster_rpc
+
+    def primary_replica(self) -> Optional["_DatabaseClusterRPC.Replica"]:
+        primaries = [replica for replica in self._replicas if replica.is_primary()]
+        return max(primaries, key=lambda r: r.term) if primaries else None
+
+    def preferred_secondary_replica(self) -> "_DatabaseClusterRPC.Replica":
+        return next(iter([replica for replica in self._replicas if replica.is_preferred_secondary()]), next(iter(self._replicas)))
+
+    def name(self) -> str:
+        return self._name
+
+    def delete(self) -> None:
+        for address in self._databases:
+            if self._database_manager_cluster.database_managers()[address].contains(self._name):
+                self._databases[address].delete()
+
+    def replicas(self):
+        return self._replicas
+
+    def __str__(self):
+        return self._name
+
+    class Replica:
+
+        def __init__(self, database: "_DatabaseClusterRPC", address: ServerAddress, term: int, is_primary: bool, is_preferred_secondary: bool):
+            self._database = database
+            self._replica_id = _DatabaseClusterRPC.Replica.Id(address, database.name())
+            self._term = term
+            self._is_primary = is_primary
+            self._is_preferred_secondary = is_preferred_secondary
+
+        @staticmethod
+        def of(proto_replica: database_proto.Database.Replica, database: "_DatabaseClusterRPC") -> "_DatabaseClusterRPC.Replica":
+            return _DatabaseClusterRPC.Replica(database, ServerAddress.parse(proto_replica.address), proto_replica.term,
+                                               proto_replica.primary, proto_replica.preferred_secondary)
+
+        def replica_id(self) -> "_DatabaseClusterRPC.Replica.Id":
+            return self._replica_id
+
+        def term(self) -> int:
+            return self._term
+
+        def is_primary(self) -> bool:
+            return self._is_primary
+
+        def is_preferred_secondary(self) -> bool:
+            return self._is_preferred_secondary
+
+        def address(self) -> ServerAddress:
+            return self._replica_id.address()
+
+        def __eq__(self, other):
+            if self is other:
+                return True
+            if not other or type(self) != type(other):
+                return False
+            return self._replica_id == other.replica_id() and self._term == other.term() and self._is_primary == other.is_primary() and self._is_preferred_secondary == other.is_preferred_secondary()
+
+        def __hash__(self):
+            return hash((self._replica_id, self._is_primary, self._is_preferred_secondary, self._term))
+
+        def __str__(self):
+            return "%s:%s:%d" % (str(self._replica_id), "P" if self._is_primary else "S", self._term)
+
+        class Id:
+
+            def __init__(self, address: ServerAddress, database: str):
+                self._address = address
+                self._database = database
+
+            def address(self) -> ServerAddress:
+                return self._address
+
+            def database(self) -> str:
+                return self._database
+
+            def __eq__(self, other):
+                if self is other:
+                    return True
+                if not other or type(self) != type(other):
+                    return False
+                return self._address == other.address() and self._database == other.database()
+
+            def __hash__(self):
+                return hash((self._address, self._database))
+
+            def __str__(self):
+                return "%s/%s" % (str(self._address), self._database)

grakn/rpc/cluster/database_manager.py

@@ -16,18 +16,33 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-
+from abc import abstractmethod
 from typing import Dict, List
 
+import grakn_protocol.protobuf.cluster.database_pb2 as database_proto
+
 from grakn.common.exception import GraknClientException
+from grakn.rpc.cluster.database import _DatabaseClusterRPC
 from grakn.rpc.cluster.server_address import ServerAddress
+from grakn.rpc.database import DatabaseCluster
 from grakn.rpc.database_manager import DatabaseManager, _DatabaseManagerRPC
 
 
-class _DatabaseManagerClusterRPC(DatabaseManager):
+class DatabaseManagerCluster(DatabaseManager):
+
+    @abstractmethod
+    def get(self, name: str) -> DatabaseCluster:
+        pass
+
+    @abstractmethod
+    def all(self) -> List[DatabaseCluster]:
+        pass
 
-    def __init__(self, database_managers: Dict[ServerAddress, "_DatabaseManagerRPC"]):
-        assert database_managers
+
+class _DatabaseManagerClusterRPC(DatabaseManagerCluster):
+
+    def __init__(self, client, database_managers: Dict[ServerAddress, "_DatabaseManagerRPC"]):
+        self._client = client
         self._database_managers = database_managers
 
     def contains(self, name: str) -> bool:
@@ -44,16 +59,27 @@ def create(self, name: str) -> None:
             if not database_manager.contains(name):
                 database_manager.create(name)
 
-    def delete(self, name: str) -> None:
-        for database_manager in self._database_managers.values():
-            if database_manager.contains(name):
-                database_manager.delete(name)
+    def get(self, name: str) -> DatabaseCluster:
+        errors = []
+        for address in self._database_managers:
+            try:
+                database_get_req = database_proto.Database.Get.Req()
+                database_get_req.name = name
+                res = self._client.grakn_cluster_grpc_stub(address).database_get(database_get_req)
+                return _DatabaseClusterRPC.of(res.database, self)
+            except GraknClientException as e:
+                errors.append(e)
+        raise GraknClientException("Attempted connecting to all cluster members, but the following errors occurred: " + str([str(e) for e in errors]))
 
-    def all(self) -> List[str]:
+    def all(self) -> List[DatabaseCluster]:
         errors = []
-        for database_manager in self._database_managers.values():
+        for address in self._database_managers:
             try:
-                return database_manager.all()
+                res = self._client.grakn_cluster_grpc_stub(address).database_all(database_proto.Database.All.Req())
+                return [_DatabaseClusterRPC.of(db, self) for db in res.databases]
             except GraknClientException as e:
                 errors.append(e)
         raise GraknClientException("Attempted connecting to all cluster members, but the following errors occurred: " + str([str(e) for e in errors]))
+
+    def database_managers(self) -> Dict[ServerAddress, _DatabaseManagerRPC]:
+        return self._database_managers