feat(coordinator): Add cypher endpoint info when list_service_status for groot

charts/graphscope-store/templates/portal/statefulset.yaml

@@ -105,6 +105,8 @@ spec:
               value: {{ .Values.frontend.service.servicePort | quote }}
             - name: GROOT_GREMLIN_PORT
               value: {{ .Values.frontend.service.gremlinPort | quote }}
+            - name: GROOT_CYPHER_PORT
+              value: {{ .Values.frontend.service.cypherPort | quote }}
             - name: INSTANCE_NAME
               value: {{ .Release.Name | quote }}
             - name: NAMESPACE

charts/graphscope-store/values.yaml

@@ -137,6 +137,8 @@ frontend:
 
     gremlinPort: 12312
 
+    cypherPort: 7687
+
     ## Internal port for communication between components.
     ##
     port: 55555

coordinator/gscoordinator/flex/core/config.py

@@ -101,6 +101,7 @@ def str_to_bool(s):
 # groot
 GROOT_GRPC_PORT = os.environ.get("GROOT_GRPC_PORT", 55556)
 GROOT_GREMLIN_PORT = os.environ.get("GROOT_GREMLIN_PORT", 12312)
+GROOT_CYPHER_PORT = os.environ.get("GROOT_CYPHER_PORT", 7687)
 GROOT_USERNAME = os.environ.get("GROOT_USERNAME", "")
 GROOT_PASSWORD = os.environ.get("GROOT_PASSWORD", "")
 try:

coordinator/gscoordinator/flex/core/insight/graph.py

@@ -25,9 +25,12 @@
 from gremlin_python.driver.client import Client
 from kubernetes import client as kube_client
 from kubernetes import config as kube_config
+from neo4j import GraphDatabase
+from neo4j import Session as Neo4jSession
 
 from gscoordinator.flex.core.config import CLUSTER_TYPE
 from gscoordinator.flex.core.config import CREATION_TIME
+from gscoordinator.flex.core.config import GROOT_CYPHER_PORT
 from gscoordinator.flex.core.config import GROOT_GREMLIN_PORT
 from gscoordinator.flex.core.config import GROOT_GRPC_PORT
 from gscoordinator.flex.core.config import GROOT_PASSWORD
@@ -46,7 +49,7 @@
 class GrootGraph(object):
     """Graph class for GraphScope store"""
 
-    def __init__(self, name, creation_time, gremlin_endpoint, grpc_endpoint):
+    def __init__(self, name, creation_time, gremlin_endpoint, grpc_endpoint, cypher_endpoint = None):
         self._id = "1"
         self._name = name
 
@@ -60,12 +63,15 @@ def __init__(self, name, creation_time, gremlin_endpoint, grpc_endpoint):
         )
         self._g = self._conn.g()
         self._schema = self._g.schema().to_dict()
-        self._gremlin_interface = {
+        self._endpoints = {
             "gremlin_endpoint": gremlin_endpoint,
             "grpc_endpoint": grpc_endpoint,
             "username": GROOT_USERNAME,
             "password": GROOT_PASSWORD,
         }
+        if cypher_endpoint:
+            self._endpoints["cypher_endpoint"] = cypher_endpoint
+
         # kubernetes
         if CLUSTER_TYPE == "KUBERNETES":
             self._api_client = resolve_api_client()
@@ -97,22 +103,41 @@ def _fetch_endpoints_impl(self):
                 grpc_endpoint, gremlin_endpoint, GROOT_USERNAME, GROOT_PASSWORD
             )
             g = conn.g()
+
+            cypher_raw_endpoint = "{0}:{1}".format(pod.status.pod_ip, GROOT_CYPHER_PORT)
+            cypher_endpoint = "neo4j://{0}".format(cypher_raw_endpoint)
+            try:
+                driver = GraphDatabase.driver(cypher_endpoint, auth=None)
+                sess = driver.session()
+                sess.run("MATCH (n) RETURN n LIMIT 1")
+            except Exception as e:
+                logger.warn(f"Cypher endpoint is not available: {str(e)}")
+                cypher_endpoint = None
+            else:
+                endpoints.append(cypher_raw_endpoint)
+                logger.info(f"Cypher endpoint is available: {cypher_endpoint}")
+                sess.close()
+                driver.close()
+
         except Exception as e:
             logger.warn(f"Failed to fetch frontend endpoints: {str(e)}")
         else:
             if (
-                gremlin_endpoint != self._gremlin_interface["gremlin_endpoint"]
-                or grpc_endpoint != self._gremlin_interface["grpc_endpoint"]
+                gremlin_endpoint != self._endpoints["gremlin_endpoint"]
+                or grpc_endpoint != self._endpoints["grpc_endpoint"]
+                or cypher_endpoint != self._endpoints.get("cypher_endpoint")
             ):
                 self._conn = conn
                 self._g = g
                 self._schema = self._g.schema().to_dict()
-                self._gremlin_interface = {
+                self._endpoints = {
                     "gremlin_endpoint": gremlin_endpoint,
                     "grpc_endpoint": grpc_endpoint,
                     "username": GROOT_USERNAME,
                     "password": GROOT_PASSWORD,
                 }
+                if cypher_endpoint:
+                    self._endpoints["cypher_endpoint"] = cypher_endpoint
                 logger.info(f"Update frontend endpoints: {str(endpoints)}")
 
     def __del__(self):
@@ -131,8 +156,8 @@ def name(self):
         return self._name
 
     @property
-    def gremlin_interface(self):
-        return self._gremlin_interface
+    def groot_endpoints(self):
+        return self._endpoints
 
     @property
     def schema(self):
@@ -284,12 +309,27 @@ def get_groot_graph_from_local():
             client.close()
             break
         time.sleep(5)
+    # test whether cypher endpoint is ready
+    try:
+        cypher_endpoint = f"neo4j://{host}:${GROOT_CYPHER_PORT}"
+        driver = GraphDatabase.driver(cypher_endpoint, auth=None)
+        sess = driver.session()
+        sess.run("MATCH (n) RETURN n LIMIT 1")
+    except Exception as e:
+        logger.warn(f"Cypher endpoint is not available: {str(e)}")
+        cypher_endpoint = None
+    else:
+        logger.info(f"Cypher endpoint is available: {cypher_endpoint}")
+        sess.close()
+        driver.close()    
+
     # groot graph
     return GrootGraph(
         name=INSTANCE_NAME,
         creation_time=CREATION_TIME,
         gremlin_endpoint=gremlin_endpoint,
         grpc_endpoint=grpc_endpoint,
+        cypher_endpoint=cypher_endpoint,
     )
 
 

coordinator/gscoordinator/flex/core/insight/groot.py

@@ -99,18 +99,21 @@ def check_graph_exists(self, graph_id: str):
             raise RuntimeError(f"Graph {graph_id} not exist.")
 
     def list_service_status(self) -> List[dict]:
-        gremlin_interface = self._graph.gremlin_interface
-        return [
+        groot_endpoints = self._graph.groot_endpoints
+        res = [
             {
                 "graph_id": self._graph.id,
                 "status": "Running",
                 "start_time": CREATION_TIME,
                 "sdk_endpoints": {
-                    "gremlin": gremlin_interface["gremlin_endpoint"],
-                    "grpc": gremlin_interface["grpc_endpoint"],
+                    "gremlin": groot_endpoints["gremlin_endpoint"],
+                    "grpc": groot_endpoints["grpc_endpoint"],
                 },
             }
         ]
+        if "cypher_endpoint" in res[0]:
+            res[0]["sdk_endpoints"]["cypher"] = groot_endpoints["cypher_endpoint"]
+        return res
 
     def create_graph(self, graph: dict) -> dict:
         raise RuntimeError("Create graph is not supported yet.")
@@ -284,12 +287,12 @@ def get_storage_usage(self) -> dict:
 
     def gremlin_service_available(self) -> bool:
         try:
-            gremlin_interface = self._graph.gremlin_interface
+            groot_endpoints = self._graph.groot_endpoints
             client = Client(
-                gremlin_interface["gremlin_endpoint"],
+                groot_endpoints["gremlin_endpoint"],
                 "g",
-                username=gremlin_interface["username"],
-                password=gremlin_interface["password"],
+                username=groot_endpoints["username"],
+                password=groot_endpoints["password"],
             )
             client.submit(
                 "g.with('evaluationTimeout', 5000).V().limit(1)"

coordinator/gscoordinator/flex/core/insight/utils.py

@@ -25,7 +25,7 @@
 
 def convert_to_configini(graph, ds_manager, config):
     # for bulk loader to connect to groot
-    gremlin_interface = graph.gremlin_interface
+    groot_endpoints = graph.groot_endpoints
     # column mapping config
     column_mapping_config = {}
     # project
@@ -112,12 +112,12 @@ def convert_to_configini(graph, ds_manager, config):
     # custom_config
     custom_config = {
         "separatr": "\\\\|",  # fixed
-        "graphEndpoint": gremlin_interface["grpc_endpoint"],
+        "graphEndpoint": groot_endpoints["grpc_endpoint"],
         "project": project,
         "outputTable": output_table,
         "columnMappingConfig": json.dumps(column_mapping_config),
-        "authUsername": gremlin_interface["username"],
-        "authPassword": gremlin_interface["password"],
+        "authUsername": groot_endpoints["username"],
+        "authPassword": groot_endpoints["password"],
         "dataSinkType": "volume",
         "odpsVolumeProject": project,
         # "-" is not allowed