Distributed ETCD with 3 pods. (#197)

* Deploy ETCD with 3 pods

* Expose etcd pod's number as a param named k8s_etcd_num_pods to client

coordinator/gscoordinator/cluster.py

@@ -161,6 +161,7 @@ def __init__(
         gie_graph_manager_image=None,
         coordinator_name=None,
         coordinator_service_name=None,
+        etcd_num_pods=None,
         etcd_cpu=None,
         etcd_mem=None,
         zookeeper_cpu=None,
@@ -231,6 +232,7 @@ def __init__(
 
         # etcd pod info
         self._etcd_image = etcd_image
+        self._etcd_num_pods = etcd_num_pods
         self._etcd_cpu = etcd_cpu
         self._etcd_mem = etcd_mem
 
@@ -390,32 +392,30 @@ def _create_etcd(self):
 
         time.sleep(1)
 
-        # create etcd deployment
+        # create etcd cluster
         etcd_builder = self._gs_etcd_builder_cls(
-            name=self._etcd_name,
-            labels=labels,
-            replicas=1,
-            image_pull_policy=self._image_pull_policy,
-        )
-
-        for name in self._image_pull_secrets:
-            etcd_builder.add_image_pull_secret(name)
-
-        etcd_builder.add_etcd_container(
-            name=self._etcd_container_name,
+            name_prefix=self._etcd_name,
+            container_name=self._etcd_container_name,
             service_name=self._etcd_service_name,
             image=self._etcd_image,
             cpu=self._etcd_cpu,
             mem=self._etcd_mem,
             preemptive=self._preemptive,
+            labels=labels,
+            image_pull_policy=self._image_pull_policy,
+            num_pods=self._etcd_num_pods,
+            restart_policy="Always",
+            image_pull_secrets=self._image_pull_secrets,
             listen_peer_service_port=self._random_etcd_listen_peer_service_port,
             listen_client_service_port=self._random_etcd_listen_client_service_port,
         )
-        self._resource_object.append(
-            self._app_api.create_namespaced_deployment(
-                self._namespace, etcd_builder.build()
+
+        for pod_builder in etcd_builder.build():
+            self._resource_object.append(
+                self._core_api.create_namespaced_pod(
+                    self._namespace, pod_builder.build()
+                )
             )
-        )
 
     def _create_vineyard_service(self):
         labels = {"name": self._engine_name}  # vineyard in engine pod

coordinator/gscoordinator/coordinator.py

@@ -831,6 +831,12 @@ def parse_sys_args():
         default="256Mi",
         help="Memory of engine container, suffix with ['Mi', 'Gi', 'Ti'].",
     )
+    parser.add_argument(
+        "--k8s_etcd_num_pods",
+        type=int,
+        default=3,
+        help="The number of etcd pods.",
+    )
     parser.add_argument(
         "--k8s_etcd_cpu",
         type=float,
@@ -918,6 +924,7 @@ def launch_graphscope():
             gie_graph_manager_image=args.k8s_gie_graph_manager_image,
             coordinator_name=args.k8s_coordinator_name,
             coordinator_service_name=args.k8s_coordinator_service_name,
+            etcd_num_pods=args.k8s_etcd_num_pods,
             etcd_cpu=args.k8s_etcd_cpu,
             etcd_mem=args.k8s_etcd_mem,
             zookeeper_cpu=args.k8s_zookeeper_cpu,

python/graphscope/client/session.py

@@ -160,6 +160,7 @@ def __init__(
         k8s_image_pull_secrets=gs_config.k8s_image_pull_secrets,
         k8s_coordinator_cpu=gs_config.k8s_coordinator_cpu,
         k8s_coordinator_mem=gs_config.k8s_coordinator_mem,
+        k8s_etcd_num_pods=gs_config.k8s_etcd_num_pods,
         k8s_etcd_cpu=gs_config.k8s_etcd_cpu,
         k8s_etcd_mem=gs_config.k8s_etcd_mem,
         k8s_zookeeper_cpu=gs_config.k8s_zookeeper_cpu,
@@ -230,6 +231,8 @@ def __init__(
 
             k8s_coordinator_mem (str, optional): Minimum number of memory request for coordinator pod. Defaults to '4Gi'.
 
+            k8s_etcd_num_pods (int, optional): The number of etcd pods. Defaults to 3.
+
             k8s_etcd_cpu (float, optional): Minimum number of CPU cores request for etcd pod. Defaults to 0.5.
 
             k8s_etcd_mem (str, optional): Minimum number of memory request for etcd pod. Defaults to '128Mi'.
@@ -343,6 +346,7 @@ def __init__(
             "k8s_zookeeper_image",
             "k8s_coordinator_cpu",
             "k8s_coordinator_mem",
+            "k8s_etcd_num_pods",
             "k8s_etcd_cpu",
             "k8s_etcd_mem",
             "k8s_zookeeper_cpu",
@@ -728,6 +732,7 @@ def _connect(self):
                 vineyard_cpu=self._config_params["k8s_vineyard_cpu"],
                 vineyard_mem=self._config_params["k8s_vineyard_mem"],
                 vineyard_shared_mem=self._config_params["k8s_vineyard_shared_mem"],
+                etcd_num_pods=self._config_params["k8s_etcd_num_pods"],
                 etcd_cpu=self._config_params["k8s_etcd_cpu"],
                 etcd_mem=self._config_params["k8s_etcd_mem"],
                 zookeeper_cpu=self._config_params["k8s_zookeeper_cpu"],

python/graphscope/config.py

@@ -47,6 +47,7 @@ class GSConfig(object):
     k8s_coordinator_mem = "4Gi"
 
     # etcd resource configuration
+    k8s_etcd_num_pods = 3
     k8s_etcd_cpu = 0.5
     k8s_etcd_mem = "128Mi"
 

python/graphscope/deploy/kubernetes/cluster.py

@@ -89,6 +89,9 @@ class KubernetesCluster(object):
         image_pull_secrets: list of str, optional
             A list of secret name used to pulling image. Defaults to None.
 
+        etcd_num_pods: int
+            The number of etcd pods.
+
         etcd_cpu: float
             Minimum number of CPU cores request for etcd pod.
 
@@ -178,6 +181,7 @@ def __init__(
         engine_mem=None,
         coordinator_cpu=None,
         coordinator_mem=None,
+        etcd_num_pods=None,
         etcd_cpu=None,
         etcd_mem=None,
         zookeeper_cpu=None,
@@ -211,6 +215,7 @@ def __init__(
         elif not isinstance(self._image_pull_secrets, list):
             self._image_pull_secrets = [self._image_pull_secrets]
 
+        self._etcd_num_pods = etcd_num_pods
         self._etcd_cpu = etcd_cpu
         self._etcd_mem = etcd_mem
         self._zookeeper_cpu = zookeeper_cpu
@@ -452,6 +457,7 @@ def _create_coordinator(self):
             coordinator_cpu=self._coordinator_cpu,
             coordinator_mem=self._coordinator_mem,
             coordinator_service_name=self._coordinator_service_name,
+            etcd_num_pods=self._etcd_num_pods,
             etcd_cpu=self._etcd_cpu,
             etcd_mem=self._etcd_mem,
             zookeeper_cpu=self._zookeeper_cpu,