* Deploy the engines when the session calls the function of creating …

…engine instances.

* Deploy the vineyard deployment if not exist.
* Add a new option for KubernetesLauncher to deploy the engines in the eager mode or lazy mode.
* Split the create_engine_instances into two steps.
- Allocate the engine instances.
- Distribute the relevant process to the engine instances.
* Add a new pytest on kubernetes to test the lazy mode.
* Add a new label for engine pods to indicate the specific engine.
* Set the engine_selector of all engine kubernetes resources.
* Create different engine statefulset based on object_id for gae and gle.
* Delete all engine kubernetes resources based on engine_selector for gae and gie.

Signed-off-by: Ye Cao <caoye.cao@alibaba-inc.com>

coordinator/gscoordinator/cluster_builder.py

@@ -54,6 +54,7 @@ def __init__(
         engine_cpu,
         engine_mem,
         engine_pod_node_selector,
+        engine_pod_prefix,
         glog_level,
         image_pull_policy,
         image_pull_secrets,
@@ -80,7 +81,7 @@ def __init__(
         with_mars,
         dataset_proxy,
     ):
-        self._gs_prefix = "gs-engine-"
+        self._gs_prefix = engine_pod_prefix
         self._analytical_prefix = "gs-analytical-"
         self._interactive_frontend_prefix = "gs-interactive-frontend-"
 
@@ -101,7 +102,9 @@ def __init__(
             "app.kubernetes.io/instance": self._instance_id,
             "app.kubernetes.io/version": __version__,
             "app.kubernetes.io/component": "engine",
+            "app.kubernetes.io/engine_selector": self.engine_stateful_set_name,
         }
+
         self._frontend_labels = self._engine_labels.copy()
         self._frontend_labels["app.kubernetes.io/component"] = "frontend"
 
@@ -216,25 +219,6 @@ def get_base_machine_env(self):
         ]
         return env
 
-    def get_vineyard_socket_volume(self):
-        name = "vineyard-ipc-socket"
-        volume = kube_client.V1Volume(name=name)
-        if self._vineyard_deployment is None:
-            empty_dir = kube_client.V1EmptyDirVolumeSource()
-            volume.empty_dir = empty_dir
-        else:
-            path = f"/var/run/vineyard-kubernetes/{self._namespace}/{self._vineyard_deployment}"
-            host_path = kube_client.V1HostPathVolumeSource(path=path)
-            host_path.type = "Directory"
-            volume.host_path = host_path
-
-        source_volume_mount = kube_client.V1VolumeMount(
-            name=name, mount_path="/tmp/vineyard_workspace"
-        )
-        destination_volume_mount = source_volume_mount
-
-        return volume, source_volume_mount, destination_volume_mount
-
     def get_shm_volume(self):
         name = "host-shm"
         volume = kube_client.V1Volume(name=name)
@@ -358,6 +342,20 @@ def get_dataset_container(self, volume_mounts):
         container.security_context = kube_client.V1SecurityContext(privileged=True)
         return container
 
+    def get_vineyard_socket_volume_from_vineyard_deployment(self):
+        name = "vineyard-ipc-socket"
+
+        # Notice, the path must be same as the one in vineyardd_types.go
+        # https://github.com/v6d-io/v6d/blob/main/k8s/apis/k8s/v1alpha1/vineyardd_types.go#L125
+        path = f"/var/run/vineyard-kubernetes/{self._namespace}/{self._vineyard_deployment}"
+        host_path = kube_client.V1HostPathVolumeSource(path=path)
+        host_path.type = "Directory"
+        volume = kube_client.V1Volume(name=name, host_path=host_path)
+        volume_mount = kube_client.V1VolumeMount(
+            name=name, mount_path="/tmp/vineyard_workspace"
+        )
+        return volume, volume_mount
+
     def get_engine_pod_spec(self):
         containers = []
         volumes = []
@@ -366,6 +364,14 @@ def get_engine_pod_spec(self):
         volumes = [shm_volume[0]]
         engine_volume_mounts = [shm_volume[2]]
 
+        if self.vineyard_deployment_exists():
+            (
+                volume,
+                volume_mount,
+            ) = self.get_vineyard_socket_volume_from_vineyard_deployment()
+            volumes.append(volume)
+            engine_volume_mounts.append(volume_mount)
+
         if self._volumes and self._volumes is not None:
             udf_volumes = ResourceBuilder.get_user_defined_volumes(self._volumes)
             volumes.extend(udf_volumes[0])

coordinator/gscoordinator/coordinator.py

@@ -865,6 +865,12 @@ def parse_sys_args():
         default=False,
         help="Mount the aliyun dataset bucket as a volume by ossfs.",
     )
+    parser.add_argument(
+        "--k8s_deploy_mode",
+        type=str,
+        default="eager",
+        help="The deploying mode of graphscope, eager or lazy.",
+    )
     parser.add_argument(
         "--monitor",
         type=str2bool,
@@ -933,6 +939,7 @@ def get_launcher(args):
             with_mars=args.k8s_with_mars,
             enabled_engines=args.k8s_enabled_engines,
             dataset_proxy=args.dataset_proxy,
+            deploy_mode=args.k8s_deploy_mode,
         )
     elif args.cluster_type == "hosts":
         launcher = LocalLauncher(