Make DatabaseManager more fault tolerant in Cluster (typedb#178)

## What is the goal of this PR?

We've introduced a couple of fault tolerance improvements into the Grakn Cluster version of DatabaseManager:

- `contains` and `all` will now try to retrieve information from every node, instead of failing if the first node is down
- `create` and `delete` will now be applied to every working node, even if some nodes are down.

For now, this means `create` will no longer throw if the database already exists, and `delete` will no longer throw if the database doesn't exist.

## What are the changes implemented in this PR?

- `contains` and `all` will now try to retrieve information from every node, instead of failing if the first node is down
- `create` and `delete` will now be applied to every working node, even if some nodes are down.

Author: Alex Walker

dependencies/graknlabs/artifacts.bzl

@@ -27,7 +27,7 @@ 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 = "3227b9e07c9c2317c0e7eab29259204f92433d76",
     )
 
 def graknlabs_grakn_cluster_artifacts():
@@ -37,5 +37,5 @@ def graknlabs_grakn_cluster_artifacts():
         artifact_name = "grakn-cluster-server-{platform}-{version}.{ext}",
         tag_source = deployment_private["artifact.release"],
         commit_source = deployment_private["artifact.snapshot"],
-        commit = "93cbc149d7b9ce588e52d8132c8a0b2265658a3c",
+        commit = "892073639b024904a413154011fb42d9dae090f3",
     )

dependencies/graknlabs/repositories.bzl

@@ -37,5 +37,5 @@ def graknlabs_behaviour():
     git_repository(
         name = "graknlabs_behaviour",
         remote = "https://github.com/graknlabs/behaviour",
-        commit = "5a3b731b3ef154b5b1bd95b788dc374bd8873746" # sync-marker: do not remove this comment, this is used for sync-dependencies by @graknlabs_behaviour
+        commit = "9f1cf29952dddaaee96a9ce3b982a8e4d6d45c48" # sync-marker: do not remove this comment, this is used for sync-dependencies by @graknlabs_behaviour
     )

grakn/client.py

@@ -44,8 +44,8 @@ def core(address=DEFAULT_ADDRESS) -> "GraknClient":
         return _RPCGraknClient(address)
 
     @staticmethod
-    def cluster(address=DEFAULT_ADDRESS) -> "GraknClient":
-        return _RPCGraknClientCluster(address)
+    def cluster(addresses: List[str]) -> "GraknClient":
+        return _RPCGraknClientCluster(addresses)
 
     @abstractmethod
     def session(self, database: str, session_type: SessionType, options: GraknOptions = None) -> Session:
@@ -112,8 +112,8 @@ def channel(self):
 # _RPCGraknClientCluster must live in this package because of circular ref with GraknClient
 class _RPCGraknClientCluster(GraknClient):
 
-    def __init__(self, address: str):
-        self._core_clients: Dict[Address.Server, _RPCGraknClient] = {addr: _RPCGraknClient(addr.client()) for addr in self._discover_cluster([address])}
+    def __init__(self, addresses: List[str]):
+        self._core_clients: Dict[Address.Server, _RPCGraknClient] = {addr: _RPCGraknClient(addr.client()) for addr in self._discover_cluster(addresses)}
         self._grakn_cluster_grpc_stubs = {addr: GraknClusterStub(client.channel()) for (addr, client) in self._core_clients.items()}
         self._databases = _RPCDatabaseManagerCluster({addr: client.databases() for (addr, client) in self._core_clients.items()})
         self._is_open = True

grakn/rpc/cluster/database_manager.py

@@ -19,6 +19,7 @@
 
 from typing import Dict, List
 
+from grakn.common.exception import GraknClientException
 from grakn.rpc.cluster.address import Address
 from grakn.rpc.database_manager import DatabaseManager, _RPCDatabaseManager
 
@@ -30,15 +31,29 @@ def __init__(self, database_managers: Dict[Address.Server, "_RPCDatabaseManager"
         self._database_managers = database_managers
 
     def contains(self, name: str) -> bool:
-        return next(iter(self._database_managers.values())).contains(name)
+        errors = []
+        for database_manager in self._database_managers.values():
+            try:
+                return database_manager.contains(name)
+            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 create(self, name: str) -> None:
         for database_manager in self._database_managers.values():
-            database_manager.create(name)
+            if not database_manager.contains(name):
+                database_manager.create(name)
 
     def delete(self, name: str) -> None:
         for database_manager in self._database_managers.values():
-            database_manager.delete(name)
+            if database_manager.contains(name):
+                database_manager.delete(name)
 
     def all(self) -> List[str]:
-        return next(iter(self._database_managers.values())).all()
+        errors = []
+        for database_manager in self._database_managers.values():
+            try:
+                return database_manager.all()
+            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]))

tests/behaviour/background/cluster/environment.py

@@ -22,12 +22,19 @@
 from tests.behaviour.context import Context
 
 
+IGNORE_TAGS = ["ignore", "ignore-client-python", "ignore-cluster"]
+
+
 def before_all(context: Context):
     environment_base.before_all(context)
-    context.client = GraknClient.cluster()
+    context.client = GraknClient.cluster([GraknClient.DEFAULT_ADDRESS])
 
 
 def before_scenario(context: Context, scenario):
+    for tag in IGNORE_TAGS:
+        if tag in scenario.effective_tags:
+            scenario.skip("tagged with @" + tag)
+            return
     environment_base.before_scenario(context, scenario)
 
 

tests/behaviour/background/core/environment.py

@@ -22,12 +22,19 @@
 from tests.behaviour.context import Context
 
 
+IGNORE_TAGS = ["ignore", "ignore-client-python", "ignore-core"]
+
+
 def before_all(context: Context):
     environment_base.before_all(context)
     context.client = GraknClient.core()
 
 
 def before_scenario(context: Context, scenario):
+    for tag in IGNORE_TAGS:
+        if tag in scenario.effective_tags:
+            scenario.skip("tagged with @" + tag)
+            return
     environment_base.before_scenario(context, scenario)
 
 

tests/behaviour/background/environment_base.py

@@ -25,20 +25,13 @@
 
 import time
 
-IGNORE_TAGS = ["ignore", "ignore-client-python"]
-
 
 def before_all(context: Context):
     context.THREAD_POOL_SIZE = 32
     context.client = GraknClient.core()
 
 
 def before_scenario(context: Context, scenario):
-    for tag in IGNORE_TAGS:
-        if tag in scenario.effective_tags:
-            scenario.skip("tagged with @" + tag)
-            return
-
     for database in context.client.databases().all():
         context.client.databases().delete(database)
     context.sessions = []

tests/integration/test_cluster_failover.py

@@ -32,7 +32,7 @@
 class TestClusterFailover(TestCase):
 
     def setUp(self):
-        with GraknClient.cluster("localhost:11729") as client:
+        with GraknClient.cluster(["localhost:11729", "localhost:21729", "localhost:31729"]) as client:
             if "grakn" in client.databases().all():
                 client.databases().delete("grakn")
             client.databases().create("grakn")
@@ -56,7 +56,7 @@ def get_primary_replica(self):
                 return self.get_primary_replica()
 
     def test_put_entity_type_to_crashed_primary_replica(self):
-        with GraknClient.cluster("localhost:11729") as client:
+        with GraknClient.cluster(["localhost:11729", "localhost:21729", "localhost:31729"]) as client:
             assert client.databases().contains("grakn")
             primary_replica = self.get_primary_replica()
             print("Performing operations against the primary replica " + str(primary_replica))

tools/behave_rule.bzl

@@ -64,7 +64,7 @@ def _rule_implementation(ctx):
            echo Starting Grakn Server
            mkdir ./grakn_distribution/"$DIRECTORY"/grakn_test
            ./grakn_distribution/"$DIRECTORY"/grakn server --data grakn_test &
-           sleep 8
+           sleep 9
 
            """
     # TODO: If two step files have the same name, we should rename the second one to prevent conflict