feat(coordinator): Adding a builtin rule for checking pod status (#4553)

As titled.

Author: zhanglei1949

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

@@ -107,6 +107,14 @@ spec:
               value: {{ .Values.frontend.service.gremlinPort | quote }}
             - name: GROOT_CYPHER_PORT
               value: {{ .Values.frontend.service.cypherPort | quote }}
+            - name: GROOT_FRONTEND_POD_SUFFIX
+              value: {{ printf "%s-frontend" (include "graphscope-store.name" .) }}
+            - name: GROOT_COORDINATOR_POD_SUFFIX
+              value: {{ printf "%s-coordinator" (include "graphscope-store.name" .) }}
+            - name: GROOT_PORTAAL_POD_SUFFIX
+              value: {{ printf "%s-portal" (include "graphscope-store.name" .) }}
+            - name: GROOT_STORE_POD_SUFFIX
+              value: {{ printf "%s-store" (include "graphscope-store.name" .) }}
             - name: INSTANCE_NAME
               value: {{ .Release.Name | quote }}
             - name: NAMESPACE

coordinator/gscoordinator/flex/core/alert/builtin_rules.py

@@ -119,6 +119,43 @@ def run_alert(self):
             if not available:
                 alert_message = self.generate_alert_message("-", message)
                 self.alert(alert_message)
+                
+class PodAvailableAlert(AlertRule):
+    def __init__(
+        self,
+        name,
+        severity,
+        metric_type,
+        conditions_description,
+        frequency,
+        message_collector,
+        enable=True,
+    ):
+        super().__init__(
+            name,
+            severity,
+            metric_type,
+            conditions_description,
+            frequency,
+            message_collector,
+            enable,
+        )
+
+    def run_alert(self):
+        """This function needs to handle exception by itself"""
+        try:
+            available = get_client_wrapper().pod_available()
+            if not available:
+                message = f"Pod unavailable: unknown reason"
+        except Exception as e:
+            available = False
+            message = "Pod unavailable: {0}".format(str(e))
+        finally:
+            # unable to distinguish whether frontend or executor is unavailable,
+            # so we set the target "-"
+            if not available:
+                alert_message = self.generate_alert_message("-", message)
+                self.alert(alert_message)
 
 
 def init_builtin_alert_rules(message_collector: AlertMessageCollector):
@@ -144,7 +181,17 @@ def init_builtin_alert_rules(message_collector: AlertMessageCollector):
             conditions_description="g.V().limit(1) failed",
             frequency=5,
             message_collector=message_collector,
-            enable=True,
+            enable=False, # GremlinServiceAvailableAlert is disabled by default
         )
         alert_rules[gremlin_service_available.id] = gremlin_service_available
+        pod_available = PodAvailableAlert(
+            name="PodAvailable",
+            severity="emergency",
+            metric_type="service",
+            conditions_description="pod not available",
+            frequency=3,
+            message_collector=message_collector,
+            enable=True,
+        )
+        alert_rules[pod_available.id] = pod_available
     return alert_rules

coordinator/gscoordinator/flex/core/client_wrapper.py

@@ -368,6 +368,8 @@ def get_storage_usage(self) -> GetStorageUsageResponse:
     def gremlin_service_available(self) -> bool:
         return self._client.gremlin_service_available()
 
+    def pod_available(self) -> bool:
+        return self._client.pod_available()
 
 client_wrapper = None
 

coordinator/gscoordinator/flex/core/config.py

@@ -109,6 +109,11 @@ def str_to_bool(s):
 except Exception as e:
     logger.warn("Invalid base64-encoded string found, use original value: %s", str(e))
 GROOT_FRONTEND_POD_SUFFIX = os.environ.get("GROOT_FRONTEND_POD_SUFFIX", "graphscope-store-frontend")
+GROOT_STORE_POD_SUFFIX = os.environ.get("GROOT_STORE_POD_SUFFIX", "graphscope-store-store")
+GROOT_COORDINATOR_POD_SUFFIX = os.environ.get("GROOT_COORDINATOR_POD_SUFFIX", "graphscope-store-coordinator")
+GROOT_PORTAL_POD_SUFFIX = os.environ.get("GROOT_PORTAL_POD_SUFFIX", "graphscope-store-portal")
+
+
 
 # dataloading service for groot
 STUDIO_WRAPPER_ENDPOINT = os.environ.get("STUDIO_WRAPPER_ENDPOINT", None)

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

@@ -29,11 +29,14 @@
 
 from gscoordinator.flex.core.config import CLUSTER_TYPE
 from gscoordinator.flex.core.config import CREATION_TIME
+from gscoordinator.flex.core.config import GROOT_COORDINATOR_POD_SUFFIX
 from gscoordinator.flex.core.config import GROOT_CYPHER_PORT
 from gscoordinator.flex.core.config import GROOT_FRONTEND_POD_SUFFIX
 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
+from gscoordinator.flex.core.config import GROOT_PORTAL_POD_SUFFIX
+from gscoordinator.flex.core.config import GROOT_STORE_POD_SUFFIX
 from gscoordinator.flex.core.config import GROOT_USERNAME
 from gscoordinator.flex.core.config import INSTANCE_NAME
 from gscoordinator.flex.core.config import NAMESPACE
@@ -126,6 +129,28 @@ def _fetch_endpoints_impl(self):
                     self._endpoints["cypher_endpoint"] = cypher_endpoint
                 logger.info(f"Update frontend endpoints: {str(endpoints)}")
 
+    def pod_available(self):
+        if CLUSTER_TYPE != "KUBERNETES":
+            return True
+        expected_prefixes = [
+            "{0}-{1}-".format(INSTANCE_NAME, GROOT_FRONTEND_POD_SUFFIX),
+            "{0}-{1}-".format(INSTANCE_NAME, GROOT_COORDINATOR_POD_SUFFIX),
+            "{0}-{1}-".format(INSTANCE_NAME, GROOT_STORE_POD_SUFFIX),
+            "{0}-{1}-".format(INSTANCE_NAME, GROOT_PORTAL_POD_SUFFIX),
+        ]
+        all_pod = self._core_api.list_namespaced_pod(NAMESPACE)
+        if len(all_pod.items) == 0:
+            raise RuntimeError("No pod found in namespace {0}".format(NAMESPACE))
+        for pod in all_pod.items:
+            for prefix in expected_prefixes:
+                if pod.metadata.name.startswith(prefix):
+                    if pod.status.phase != "Running":
+                        raise RuntimeError(
+                            "Pod {0} is not running, {1}".format(pod.metadata.name, pod.status.phase)
+                        )
+        return True
+
+
     def __del__(self):
         self._conn.close()
 

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

@@ -309,6 +309,10 @@ def gremlin_service_available(self) -> bool:
             except:  # noqa: E722
                 pass
             return True
+    
+    def pod_available(self) -> bool:
+        return self._graph.pod_available()
+        
 
 
 def init_groot_client(config: Config):