Generalize k8s_volumes and support mount pvc (#156)

* Generalize k8s_volumes and support mount PVC

* Close session in GIE function test

coordinator/gscoordinator/cluster.py

@@ -43,12 +43,11 @@
     K8SApiException = None
     K8SConfigException = None
 
-from graphscope.deploy.kubernetes.resource_builder import EmptyDirVolumeBuilder
 from graphscope.deploy.kubernetes.resource_builder import GSEngineBuilder
 from graphscope.deploy.kubernetes.resource_builder import GSEtcdBuilder
 from graphscope.deploy.kubernetes.resource_builder import GSGraphManagerBuilder
-from graphscope.deploy.kubernetes.resource_builder import HostPathVolumeBuilder
 from graphscope.deploy.kubernetes.resource_builder import ServiceBuilder
+from graphscope.deploy.kubernetes.resource_builder import VolumeBuilder
 from graphscope.deploy.kubernetes.resource_builder import resolve_volume_builder
 from graphscope.deploy.kubernetes.utils import delete_kubernetes_object
 from graphscope.deploy.kubernetes.utils import get_kubernetes_object_info
@@ -251,8 +250,9 @@ def _create_engine_replicaset(self):
         # volume1 is for vineyard ipc socket
         # MaxGraph: /home/maxgraph/data/vineyard
         engine_builder.add_volume(
-            EmptyDirVolumeBuilder(
+            VolumeBuilder(
                 name="vineyard-ipc-volume",
+                type="emptyDir",
                 field={},
                 mounts_list=[
                     {"mountPath": "/tmp/vineyard_workspace"},
@@ -262,8 +262,9 @@ def _create_engine_replicaset(self):
         )
         # volume2 is for shared memory
         engine_builder.add_volume(
-            EmptyDirVolumeBuilder(
+            VolumeBuilder(
                 name="host-shm",
+                type="emptyDir",
                 field={"medium": "Memory"},
                 mounts_list=[{"mountPath": "/dev/shm"}],
             )

interactive_engine/tests/function_test.sh

@@ -51,6 +51,7 @@ function _start {
 }
 
 function _stop {
+    curl -XPOST http://localhost:${_port} -d 'session.close()'
     _port=`cat $tmp_result | awk -F":" '{print $3}'`
     echo "stop port is ${_port}"
     kill -INT `lsof -i:${_port} -t`

python/graphscope/client/session.py

@@ -231,7 +231,9 @@ def __init__(
                 Minimum number of memory request for graphmanager container. Defaults to '4Gi'.
 
             k8s_volumes (dict, optional): A dict of k8s volume which represents a directory containing data, accessible to the
-                containers in a pod. Defaults to {}. Only 'hostPath' supported yet. For example, we can mount host path with:
+                containers in a pod. Defaults to {}.
+
+                For example, you can mount host path with:
 
                 k8s_volumes = {
                     "my-data": {
@@ -251,6 +253,22 @@ def __init__(
                     }
                 }
 
+                Or you can mount PVC with:
+
+                k8s_volumes = {
+                    "my-data": {
+                        "type": "persistentVolumeClaim",
+                        "field": {
+                            "claimName": "your-pvc-name"
+                        },
+                        "mounts": [
+                            {
+                                "mountPath": "<path1>"
+                            }
+                        ]
+                    }
+                }
+
                 Also, you can mount a single volume with:
 
                 k8s_volumes = {

python/graphscope/deploy/kubernetes/resource_builder.py

@@ -33,19 +33,12 @@ def _remove_nones(o):
 
 def resolve_volume_builder(name, value):
     """Resolve specified volume with value dict."""
-    supported_type = ["hostPath"]
     if "type" not in value or "field" not in value or "mounts" not in value:
         logger.warning("Volume %s must contains 'type' 'field' and 'mounts'", name)
         return None
-    type = value["type"]
-    if type not in supported_type:
-        logger.warning("%s of volume %s is unsupported yet", type, name)
-        return None
-    field = value["field"]
-    mounts_list = value["mounts"]
-    if type == "hostPath":
-        return HostPathVolumeBuilder(name=name, field=field, mounts_list=mounts_list)
-    return None
+    return VolumeBuilder(
+        name=name, type=value["type"], field=value["field"], mounts_list=value["mounts"]
+    )
 
 
 class NamespaceBuilder(object):
@@ -294,8 +287,10 @@ def build(self):
 class VolumeBuilder(object):
     """Builder for k8s volumes."""
 
-    def __init__(self, name, mounts_list):
+    def __init__(self, name, type, field, mounts_list):
         self._name = name
+        self._type = type
+        self._field = field
         self._mounts_list = mounts_list
         if not isinstance(self._mounts_list, list):
             self._mounts_list = [self._mounts_list]
@@ -304,34 +299,12 @@ def __init__(self, name, mounts_list):
             mount["name"] = self._name
 
     def build(self):
-        raise NotImplementedError
+        return {"name": self._name, self._type: self._field}
 
     def build_mount(self):
         return self._mounts_list
 
 
-class HostPathVolumeBuilder(VolumeBuilder):
-    """Builder for k8s host volume."""
-
-    def __init__(self, name, field, mounts_list):
-        super().__init__(name, mounts_list)
-        self._field = field
-
-    def build(self):
-        return {"name": self._name, "hostPath": self._field}
-
-
-class EmptyDirVolumeBuilder(VolumeBuilder):
-    """Builder for k8s empty-dir volumes."""
-
-    def __init__(self, name, field, mounts_list):
-        super().__init__(name, mounts_list)
-        self._field = field
-
-    def build(self):
-        return {"name": self._name, "emptyDir": self._field}
-
-
 class HttpHeaderBuilder(object):
     """Builder for k8s http header."""