Generalize the Kubernetes deployment interface to make it extensible (#…

…269)

* Some General modification for GraphScope kubernetes deployment

* Bump up vineyard version to v0.2.1.

Co-authored-by: Tao He <sighingnow@gmail.com>

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.15
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.2.1
       options:
         --shm-size 4096m
     strategy:
@@ -135,7 +135,7 @@ jobs:
     defaults:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
-    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.15
+    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.2.1
     steps:
     - name: Install Dependencies
       run: |
@@ -175,7 +175,7 @@ jobs:
     defaults:
       run:
         shell: scl enable devtoolset-7 -- bash --noprofile --norc -eo pipefail {0}
-    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.1.15
+    container: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.2.1
     steps:
     - name: Install Dependencies
       run: |
@@ -305,6 +305,10 @@ jobs:
         pushd python && python3 setup.py build_proto && popd
         pushd coordinator && python3 setup.py build_builtin && popd
 
+    - name: Setup tmate session debug
+      if: false
+      uses: mxschmitt/action-tmate@v2
+
     - name: App Test with Expr Off
       if: matrix.experimental == 'OFF'
       run: |

.github/workflows/local.yml

@@ -83,7 +83,7 @@ jobs:
 
     - name: Install libvineyard
       run: |
-        git clone -b v0.1.15 --single-branch --depth=1 https://github.com/alibaba/libvineyard.git
+        git clone -b v0.2.1 --single-branch --depth=1 https://github.com/alibaba/libvineyard.git
         cd libvineyard
         git submodule update --init
         mkdir build && pushd build
@@ -102,6 +102,10 @@ jobs:
         # Default to /usr/local
         make install WITH_LEARNING_ENGINE=ON
 
+    - name: DEBUG
+      if: false
+      uses: mxschmitt/action-tmate@v2
+
     - name: Run Local Test
       run: |
         export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64

analytical_engine/CMakeLists.txt

@@ -109,7 +109,7 @@ endif ()
 find_package(libgrapelite REQUIRED)
 include_directories(${LIBGRAPELITE_INCLUDE_DIRS})
 
-find_package(vineyard 0.1.15 REQUIRED)
+find_package(vineyard 0.2.1 REQUIRED)
 include_directories(${VINEYARD_INCLUDE_DIRS})
 add_compile_options(-DENABLE_SELECTOR)
 

analytical_engine/core/grape_instance.cc

@@ -130,8 +130,10 @@ bl::result<void> GrapeInstance::unloadGraph(const rpc::GSParams& params) {
       VY_OK_OR_RAISE(client_->DelData(frag_id, false, true));
     }
     MPI_Barrier(comm_spec_.comm());
-    if (comm_spec_.worker_id() == 0) {
-      VINEYARD_SUPPRESS(client_->DelData(frag_group_id, false, true));
+    if (exists) {
+      if (comm_spec_.worker_id() == 0) {
+        VINEYARD_SUPPRESS(client_->DelData(frag_group_id, false, true));
+      }
     }
   }
   return object_manager_.RemoveObject(graph_name);

analytical_engine/core/launcher.cc

@@ -43,8 +43,13 @@ void VineyardServer::Start() {
   auto ts = std::chrono::duration_cast<std::chrono::milliseconds>(
                 std::chrono::system_clock::now().time_since_epoch())
                 .count();
-  // random vineyard socket
-  vineyard_socket_ = "/tmp/vineyard.sock." + std::to_string(ts);
+  if (getenv("VINEYARD_IPC_SOCKET")) {
+    vineyard_socket_ =
+        vineyard::ExpandEnvironmentVariables("$VINEYARD_IPC_SOCKET");
+  } else {
+    // random vineyard socket
+    vineyard_socket_ = "/tmp/vineyard.sock." + std::to_string(ts);
+  }
 
   std::string vineyardd_path;
   if (getenv("VINEYARD_HOME")) {

analytical_engine/core/loader/arrow_fragment_appender.h

@@ -306,10 +306,17 @@ class ArrowFragmentAppender {
                                                         buffer->size());
 
     std::shared_ptr<arrow::csv::TableReader> reader;
+#if defined(ARROW_VERSION) && ARROW_VERSION >= 4000000
+    ARROW_OK_ASSIGN_OR_RAISE(
+        reader, arrow::csv::TableReader::Make(
+                    arrow::io::IOContext(pool), input, read_options,
+                    parse_options, arrow::csv::ConvertOptions::Defaults()));
+#else
     ARROW_OK_ASSIGN_OR_RAISE(
         reader,
         arrow::csv::TableReader::Make(pool, input, read_options, parse_options,
                                       arrow::csv::ConvertOptions::Defaults()));
+#endif
 
     std::shared_ptr<arrow::Table> table;
     ARROW_OK_ASSIGN_OR_RAISE(table, reader->Read());

coordinator/gscoordinator/cluster.py

@@ -159,6 +159,8 @@ class KubernetesClusterLauncher(Launcher):
     _mars_scheduler_port = 7103  # fixed
     _mars_worker_port = 7104  # fixed
 
+    _MAXGRAPH_MANAGER_HOST = "http://%s.%s.svc.cluster.local:8080"
+
     def __init__(
         self,
         namespace=None,
@@ -335,6 +337,9 @@ def get_namespace(self):
     def get_gie_graph_manager_service_name(self):
         return self._gie_graph_manager_service_name
 
+    def get_manager_host(self):
+        return "http://{0}".format(self._get_graph_manager_service_endpoint())
+
     @property
     def preemptive(self):
         return self._preemptive
@@ -483,6 +488,7 @@ def _create_engine_replicaset(self):
 
         # add engine container
         engine_builder.add_engine_container(
+            cmd=["tail", "-f", "/dev/null"],
             name=self._engine_container_name,
             image=self._gs_image,
             cpu=self._engine_cpu,
@@ -702,12 +708,14 @@ def _create_interactive_engine_service(self):
 
         # add manager container
         graph_manager_builder.add_manager_container(
+            cmd=["/bin/bash", "-c", "--"],
+            args=["/home/maxgraph/manager-entrypoint.sh"],
             name=self._gie_manager_container_name,
             image=self._gie_graph_manager_image,
             cpu=self._gie_graph_manager_cpu,
             mem=self._gie_graph_manager_mem,
             preemptive=self._preemptive,
-            port=self._interactive_engine_manager_port,
+            ports=self._interactive_engine_manager_port,
         )
 
         # add zookeeper container
@@ -717,7 +725,7 @@ def _create_interactive_engine_service(self):
             cpu=self._zookeeper_cpu,
             mem=self._zookeeper_mem,
             preemptive=self._preemptive,
-            port=self._zookeeper_port,
+            ports=self._zookeeper_port,
         )
 
         self._resource_object.append(
@@ -897,6 +905,16 @@ def _get_etcd_service_endpoint(self):
         )
         return endpoints[0]
 
+    def _get_graph_manager_service_endpoint(self):
+        # Always len(endpoints) >= 1
+        endpoints = get_service_endpoints(
+            api_client=self._api_client,
+            namespace=self._namespace,
+            name=self._gie_graph_manager_service_name,
+            type="ClusterIP",
+        )
+        return endpoints[0]
+
     def _launch_analytical_engine_locally(self):
         logger.info(
             "Starting GAE rpc service on {} ...".format(

coordinator/gscoordinator/coordinator.py

@@ -84,7 +84,6 @@
 
 ENGINE_CONTAINER = "engine"
 VINEYARD_CONTAINER = "vineyard"
-MAXGRAPH_MANAGER_HOST = "http://%s.%s.svc.cluster.local:8080"
 
 logger = logging.getLogger("graphscope")
 
@@ -129,9 +128,6 @@ def __init__(self, launcher, dangling_timeout_seconds, log_level="INFO"):
         if self._launcher_type == types_pb2.K8S:
             self._pods_list = self._launcher.get_pods_list()
             self._k8s_namespace = self._launcher.get_namespace()
-            self._gie_graph_manager_service_name = (
-                self._launcher.get_gie_graph_manager_service_name()
-            )
         else:
             self._pods_list = []  # locally launched
             self._k8s_namespace = ""
@@ -478,10 +474,7 @@ def CreateInteractiveInstance(self, request, context):
         }
 
         if self._launcher_type == types_pb2.K8S:
-            manager_host = MAXGRAPH_MANAGER_HOST % (
-                self._gie_graph_manager_service_name,
-                self._k8s_namespace,
-            )
+            post_url = "{0}/instance/create".format(self._launcher.get_manager_host())
             params.update(
                 {
                     "schemaJson": schema_json,
@@ -492,7 +485,6 @@ def CreateInteractiveInstance(self, request, context):
                     "gremlinServerMem": gremlin_server_mem,
                 }
             )
-            post_url = "%s/instance/create" % manager_host
             engine_params = [
                 "{}:{}".format(key, value)
                 for key, value in request.engine_params.items()
@@ -547,10 +539,7 @@ def CreateInteractiveInstance(self, request, context):
     def CloseInteractiveInstance(self, request, context):
         object_id = request.object_id
         if self._launcher_type == types_pb2.K8S:
-            manager_host = MAXGRAPH_MANAGER_HOST % (
-                self._gie_graph_manager_service_name,
-                self._k8s_namespace,
-            )
+            manager_host = self._launcher.get_manager_host()
             pod_name_list = ",".join(self._pods_list)
             close_url = "%s/instance/close?graphName=%ld&podNameList=%s&containerName=%s&waitingForDelete=%s" % (
                 manager_host,

coordinator/gscoordinator/template/CMakeLists.template

@@ -43,7 +43,7 @@ set(CMAKE_THREAD_PREFER_PTHREAD ON)
 find_package(Threads REQUIRED)
 
 # find vineyard-----------------------------------------------------------------
-find_package(vineyard 0.1.15 REQUIRED)
+find_package(vineyard 0.2.1 REQUIRED)
 include_directories(${VINEYARD_INCLUDE_DIRS})
 add_compile_options(-DENABLE_SELECTOR)
 

interactive_engine/deploy/docker/dockerfile/coordinator.application.properties

@@ -6,7 +6,7 @@ zookeeper.connect=zookeeper:2181
 
 platform.kind:YARN
 
-coordinator.port=8183
+coordinator.port=0
 resource.frontend.count=1
 resource.ingestnode.count=1
 resource.executor.count=resource_executor_count

interactive_engine/pom.xml

@@ -18,7 +18,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <log4j2.version>[2.13.2,)</log4j2.version>
+        <log4j2.version>[2.6.1,)</log4j2.version>
         <hadoop.version>2.8.4</hadoop.version>
         <grpc.version>1.14.0</grpc.version>
         <protobuf.version>3.5.1</protobuf.version>

interactive_engine/src/coordinator/pom.xml

@@ -14,7 +14,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <log4j2.version>2.11.0</log4j2.version>
+        <log4j2.version>2.6.1</log4j2.version>
     </properties>
 
     <dependencies>

interactive_engine/src/executor/runtime/native/CMakeLists.txt

@@ -44,7 +44,7 @@ find_package(Threads REQUIRED)
 #  we need edge src/dst ids in etable.
 add_definitions(-DENDPOINT_LISTS)
 
-find_package(vineyard 0.1.15 REQUIRED)
+find_package(vineyard 0.2.1 REQUIRED)
 add_library(native_store global_store_ffi.cc
                          htap_ds_impl.cc
                          graph_builder_ffi.cc

interactive_engine/src/instance-manager/src/main/java/com/alibaba/maxgraph/admin/controller/InstanceManagerController.java

@@ -56,7 +56,7 @@
 @RequestMapping("/instance")
 public class InstanceManagerController {
     private static final Logger logger = LoggerFactory.getLogger(InstanceManagerController.class);
-    private static final long LAUNCH_MAX_TIME_LILL = 5 * 60 * 1000;
+    private static final long LAUNCH_MAX_TIME_LILL = 10 * 60 * 1000;
     private static final String NAME_PATTERN = "[a-z0-9]+[a-z0-9\\-.]*[a-z0-9]+";
 
     @Autowired

k8s/gsvineyard.Dockerfile

@@ -14,7 +14,7 @@ RUN cd /tmp && \
     make -j`nproc` && \
     make install && \
     cd /tmp && \
-    git clone -b v0.1.15 https://github.com/alibaba/libvineyard.git --depth=1 && \
+    git clone -b v0.2.1 https://github.com/alibaba/libvineyard.git --depth=1 && \
     cd libvineyard && \
     git submodule update --init && \
     mkdir -p /tmp/libvineyard/build && \

python/graphscope/client/session.py

@@ -34,8 +34,10 @@
 from queue import Empty as EmptyQueue
 
 try:
+    from kubernetes import client as kube_client
     from kubernetes import config as kube_config
 except ImportError:
+    kube_client = None
     kube_config = None
 
 import graphscope
@@ -753,9 +755,15 @@ def _connect(self):
                 or self._config_params["k8s_gs_image"] is None
             ):
                 raise K8sError("None image found.")
-            api_client = kube_config.new_client_from_config(
-                **self._config_params["k8s_client_config"]
-            )
+            if isinstance(
+                self._config_params["k8s_client_config"],
+                kube_client.api_client.ApiClient,
+            ):
+                api_client = self._config_params["k8s_client_config"]
+            else:
+                api_client = kube_config.new_client_from_config(
+                    **self._config_params["k8s_client_config"]
+                )
             self._launcher = KubernetesClusterLauncher(
                 api_client=api_client,
                 namespace=self._config_params["k8s_namespace"],