#! /usr/bin/env python3 # -*- coding: utf-8 -*- # # Copyright 2020 Alibaba Group Holding Limited. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import base64 import copy import json import logging import os import random import shlex import subprocess import sys import time from graphscope.proto import message_pb2 from gscoordinator.cluster_builder import EngineCluster from gscoordinator.cluster_builder import MarsCluster try: from kubernetes import client as kube_client from kubernetes import config as kube_config from kubernetes import watch as kube_watch from kubernetes.client import AppsV1Api from kubernetes.client import CoreV1Api from kubernetes.client.rest import ApiException as K8SApiException from kubernetes.config import ConfigException as K8SConfigException except ImportError: kube_client = None kube_config = None kube_watch = None AppsV1Api = None CoreV1Api = None K8SApiException = None K8SConfigException = None from graphscope.config import Config from graphscope.deploy.kubernetes.utils import delete_kubernetes_object from graphscope.deploy.kubernetes.utils import get_kubernetes_object_info from graphscope.deploy.kubernetes.utils import resolve_api_client from graphscope.framework.utils import PipeWatcher from graphscope.framework.utils import get_tempdir from graphscope.proto import types_pb2 from gscoordinator.constants import ANALYTICAL_CONTAINER_NAME from gscoordinator.constants import GRAPHLEARN_CONTAINER_NAME from gscoordinator.constants import GRAPHLEARN_TORCH_CONTAINER_NAME from gscoordinator.constants import INTERACTIVE_EXECUTOR_CONTAINER_NAME from gscoordinator.launcher import AbstractLauncher from gscoordinator.utils import ANALYTICAL_ENGINE_PATH from gscoordinator.utils import GRAPHSCOPE_HOME from gscoordinator.utils import INTERACTIVE_ENGINE_SCRIPT from gscoordinator.utils import WORKSPACE from gscoordinator.utils import ResolveMPICmdPrefix from gscoordinator.utils import delegate_command_to_pod from gscoordinator.utils import parse_as_glog_level from gscoordinator.utils import replace_string_in_dict from gscoordinator.utils import run_kube_cp_command logger = logging.getLogger("graphscope") class FakeKubeResponse: def __init__(self, obj): self.data = json.dumps(obj) class KubernetesClusterLauncher(AbstractLauncher): def __init__(self, config: Config): super().__init__() self._serving = False self._api_client = resolve_api_client() self._core_api = kube_client.CoreV1Api(self._api_client) self._apps_api = kube_client.AppsV1Api(self._api_client) self._pytorchjobs_api = kube_client.CustomObjectsApi(self._api_client) self._resource_object = ResourceManager(self._api_client) self._config: Config = config self._config.kubernetes_launcher.engine.post_setup() launcher_config = config.kubernetes_launcher # glog level self._glog_level = parse_as_glog_level(config.log_level) # Session Config self._num_workers = config.session.num_workers self._instance_id = config.session.instance_id self._timeout_seconds = config.session.timeout_seconds self._retry_time_seconds = config.session.retry_time_seconds # Vineyard Config # self._vineyard_socket = config.vineyard.socket self._vineyard_rpc_port = config.vineyard.rpc_port self._vineyard_deployment = config.vineyard.deployment_name # Launcher Config self._namespace = launcher_config.namespace self._delete_namespace = launcher_config.delete_namespace # Coordinator Config self._coordinator_name = config.coordinator.deployment_name self._coordinator_service_name = self._coordinator_name self._image_registry = launcher_config.image.registry self._image_repository = launcher_config.image.repository self._image_tag = launcher_config.image.tag self._image_pull_policy = launcher_config.image.pull_policy self._image_pull_secrets = launcher_config.image.pull_secrets self._vineyard_resource = config.vineyard.resource self._volumes = launcher_config.volumes self._owner_references = self.get_coordinator_owner_references() self._engine_pod_prefix = "gs-engine-" self._vineyard_image = config.vineyard.image self._vineyard_mem = config.vineyard.resource.requests.memory self._vineyard_cpu = config.vineyard.resource.requests.cpu self._service_type = launcher_config.service_type self._waiting_for_delete = launcher_config.waiting_for_delete # check the validity of deploy mode self._deploy_mode = launcher_config.deployment_mode if self._deploy_mode not in ["eager", "lazy"]: logger.error( "Invalid mode %s, choose from 'eager' or 'lazy'. Proceeding with default mode: 'eager'", self._deploy_mode, ) self._deploy_mode = "eager" self._vineyard_pod_name_list = [] # set the kube config file self._k8s_config_file = launcher_config.config_file if self._k8s_config_file is None: self._k8s_config_file = os.environ.get("KUBECONFIG", "~/.kube/config") if self._vineyard_deployment is not None: self._deploy_vineyard_deployment_if_not_exist() # check the if the vineyard deployment is ready again if not self._check_if_vineyard_deployment_exist(): # if not ready, then set the vineyard deployment to None logger.error( "Vineyard deployment %s is not ready, please check the deployment status." "Proceeding with none vineyard deployment mode.", self._vineyard_deployment, ) self._vineyard_deployment = None # if the vineyard deployment is not set and use the eager mode, # which means deploy the engine as a single pod and there is no # external vineyard deployment. The vineyard objects are not # shared between the engine pods, so report an error here and set # the mode to eager. if self._deploy_mode == "lazy" and self._vineyard_deployment is None: logger.error( "Lazy mode is only possible with a vineyard deployment, " "please add a vineyard deployment name by k8s_vineyard_deployment='vineyardd-sample'. " "Proceeding with default mode: 'eager'" ) self._deploy_mode = "eager" self._pod_name_list = [] self._pod_ip_list = [] self._pod_host_ip_list = [] # analytical engine self._analytical_pod_name = [] self._analytical_pod_ip = [] self._analytical_pod_host_ip = [] # analytical java engine self._analytical_java_pod_name = [] self._analytical_java_pod_ip = [] self._analytical_java_pod_host_ip = [] # interactive engine self._interactive_resource_object = {} self._interactive_pod_name = {} self._interactive_pod_ip = {} self._interactive_pod_host_ip = {} # graphlearn engine self._graphlearn_resource_object = {} self._graphlearn_pod_name = {} self._graphlearn_pod_ip = {} self._graphlearn_pod_host_ip = {} # graphlearn_torch engine self._graphlearn_torch_resource_object = {} self._graphlearn_torch_pod_name = {} self._graphlearn_torch_pod_ip = {} self._graphlearn_torch_pod_host_ip = {} self._analytical_engine_endpoint = None self._mars_service_endpoint = None self._analytical_engine_process = None self._random_analytical_engine_rpc_port = random.randint(56001, 57000) # interactive engine # executor inter-processing port # executor rpc port # frontend port self._interactive_port = 8233 # 8000 ~ 9000 is exposed self._graphlearn_start_port = 8000 # 9001 ~ 10001 is exposed self._graphlearn_torch_start_port = 9001 self._graphlearn_services = {} self._graphlearn_instance_processes = {} self._graphlearn_torch_services = {} self._graphlearn_torch_instance_processes = {} # workspace self._instance_workspace = os.path.join(WORKSPACE, self._instance_id) os.makedirs(self._instance_workspace, exist_ok=True) self._session_workspace = None self._engine_cluster = self._build_engine_cluster() self._vineyard_socket = self._engine_cluster.vineyard_ipc_socket self._vineyard_service_endpoint = None self._vineyard_internal_service_endpoint = None self._mars_service_endpoint = None if self._config.kubernetes_launcher.mars.enable: self._mars_cluster = MarsCluster( self._instance_id, self._namespace, self._service_type ) def __del__(self): self.stop() def type(self): return types_pb2.K8S # the argument `with_analytical_` means whether to add the analytical engine # container to the engine statefulsets, and the other three arguments are similar. def _build_engine_cluster(self): return EngineCluster( config=self._config, engine_pod_prefix=self._engine_pod_prefix, graphlearn_start_port=self._graphlearn_start_port, graphlearn_torch_start_port=self._graphlearn_torch_start_port, ) def get_coordinator_owner_references(self): owner_references = [] if self._coordinator_name: try: deployment = self._apps_api.read_namespaced_deployment( self._coordinator_name, self._namespace ) owner_references.append( kube_client.V1OwnerReference( api_version="apps/v1", kind="Deployment", name=self._coordinator_name, uid=deployment.metadata.uid, ) ) except K8SApiException: logger.error("Coordinator %s not found", self._coordinator_name) return owner_references def waiting_for_delete(self): return self._waiting_for_delete def get_namespace(self): return self._namespace def get_vineyard_stream_info(self): if self._vineyard_deployment is not None: hosts = [ f"{self._namespace}:{host}" for host in self._vineyard_pod_name_list ] else: hosts = [f"{self._namespace}:{host}" for host in self._pod_name_list] return "kubernetes", hosts def set_session_workspace(self, session_id): self._session_workspace = os.path.join(self._instance_workspace, session_id) os.makedirs(self._session_workspace, exist_ok=True) def launch_etcd(self): pass def configure_etcd_endpoint(self): pass @property def hosts(self): """list of pod name""" return self._pod_name_list @property def hosts_list(self): return self._get_analytical_hosts() @property def vineyard_endpoint(self) -> str: if self._check_if_vineyard_deployment_exist(): return self._vineyard_service_endpoint else: return self._vineyard_internal_endpoint def distribute_file(self, path): pod_name_list, _, _ = self._allocate_analytical_engine() for pod in pod_name_list: container = ANALYTICAL_CONTAINER_NAME try: # The library may exist in the analytical pod. test_cmd = f"test -f {path}" logger.debug(delegate_command_to_pod(test_cmd, pod, container)) logger.info("Library exists, skip distribute") except RuntimeError: cmd = f"mkdir -p {os.path.dirname(path)}" logger.debug(delegate_command_to_pod(cmd, pod, container)) logger.debug(run_kube_cp_command(path, path, pod, container, True)) def close_analytical_instance(self): pass def launch_vineyard(self): """Launch vineyardd in k8s cluster.""" # vineyardd is auto launched in vineyardd container # args = f"vineyardd \ # -socket {self._engine_cluster._sock} -etcd_endpoint http://{self._pod_ip_list[0]}:2379" pass def close_etcd(self): # etcd is managed by vineyard pass def close_vineyard(self): # No need to close vineyardd # Use delete deployment instead pass def check_if_engine_exist(self, engine_type, object_id=None): """Checks if the engine with the given type exists. Args: engine_type: The type of engine to check for. object_id: The object id of the engine to check for. Returns: True if the engine exists, False otherwise. """ if object_id: engine_pod_name_dict = getattr(self, f"_{engine_type}_pod_name") engine_pod_name_list = engine_pod_name_dict.get(object_id, []) engine_pod_ip_dict = getattr(self, f"_{engine_type}_pod_ip") engine_pod_ip_list = engine_pod_ip_dict.get(object_id, []) engine_pod_host_ip_dict = getattr(self, f"_{engine_type}_pod_host_ip") engine_pod_host_ip_list = engine_pod_host_ip_dict.get(object_id, []) else: engine_pod_name_list = getattr(self, f"_{engine_type}_pod_name") engine_pod_ip_list = getattr(self, f"_{engine_type}_pod_ip") engine_pod_host_ip_list = getattr(self, f"_{engine_type}_pod_host_ip") return engine_pod_name_list and engine_pod_ip_list and engine_pod_host_ip_list def deploy_engine(self, engine_type, object_id=None): """Deploys the engine with the given type. Args: engine_type: The type of engine to deploy. object_id: The object ID to deploy the engine with. Returns: A tuple of the pod names, IP addresses, and host IP addresses of the deployed engine and the response of the engine and service. """ if not self.check_if_engine_exist(engine_type, object_id): self._engine_pod_prefix = f"gs-{engine_type}-" + ( f"{object_id}-" if object_id else "" ).replace("_", "-") self._config.kubernetes_launcher.engine.enable_gae = ( engine_type == "analytical" ) self._config.kubernetes_launcher.engine.enable_gae_java = ( engine_type == "analytical-java" ) self._config.kubernetes_launcher.engine.enable_gie = ( engine_type == "interactive" ) self._config.kubernetes_launcher.engine.enable_gle = ( engine_type == "graphlearn" ) self._config.kubernetes_launcher.engine.enable_glt = ( engine_type == "graphlearn-torch" ) self._engine_cluster = self._build_engine_cluster() response = self._create_engine_stateful_set() self._waiting_for_services_ready() if object_id: resource_object = getattr(self, f"_{engine_type}_resource_object") pod_name = getattr(self, f"_{engine_type}_pod_name") pod_ip = getattr(self, f"_{engine_type}_pod_ip") pod_host_ip = getattr(self, f"_{engine_type}_pod_host_ip") resource_object[object_id] = response pod_name[object_id] = self._pod_name_list pod_ip[object_id] = self._pod_ip_list pod_host_ip[object_id] = self._pod_host_ip_list else: # Set the engine pod info setattr(self, f"_{engine_type}_pod_name", self._pod_name_list) setattr(self, f"_{engine_type}_pod_ip", self._pod_ip_list) setattr(self, f"_{engine_type}_pod_host_ip", self._pod_host_ip_list) return ( ( getattr(self, f"_{engine_type}_pod_name") if object_id is None else getattr(self, f"_{engine_type}_pod_name")[object_id] ), ( getattr(self, f"_{engine_type}_pod_ip") if object_id is None else getattr(self, f"_{engine_type}_pod_ip")[object_id] ), ( getattr(self, f"_{engine_type}_pod_host_ip") if object_id is None else getattr(self, f"_{engine_type}_pod_host_ip")[object_id] ), ) def delete_engine_stateful_set_with_object_id(self, engine_type, object_id): """delete the engine stateful set with the given object id. Args: engine_type(str): the type of engine object_id (int): The object id of the engine to delete. """ resource_object = getattr(self, f"_{engine_type}_resource_object") obj = resource_object.get(object_id, {}) if obj: delete_kubernetes_object( api_client=self._api_client, target=obj, wait=self._waiting_for_delete, timeout_seconds=self._timeout_seconds, ) pod_name = getattr(self, f"_{engine_type}_pod_name") pod_ip = getattr(self, f"_{engine_type}_pod_ip") pod_host_ip = getattr(self, f"_{engine_type}_pod_host_ip") del resource_object[object_id] del pod_name[object_id] del pod_ip[object_id] del pod_host_ip[object_id] def deploy_analytical_engine(self): return self.deploy_engine("analytical") def deploy_analytical_java_engine(self): return self.deploy_engine("analytical-java") def deploy_interactive_engine(self, object_id): pod_name_list, pod_ip_list, pod_host_ip_list = self.deploy_engine( "interactive", object_id ) try: response = self._core_api.read_namespaced_pod( pod_name_list[0], self._namespace ) except K8SApiException: logger.exception( "Get pod %s error, please check if the pod is ready", pod_name_list[0], ) owner_references = [ kube_client.V1OwnerReference( api_version=response.metadata.owner_references[0].api_version, kind=response.metadata.owner_references[0].kind, name=response.metadata.owner_references[0].name, uid=response.metadata.owner_references[0].uid, ) ] name = f"gs-interactive-frontend-{object_id}-{self._instance_id}" self._create_frontend_deployment(name, owner_references) return pod_name_list, pod_ip_list, pod_host_ip_list def deploy_graphlearn_engine(self, object_id): return self.deploy_engine("graphlearn", object_id) def deploy_graphlearn_torch_engine(self, object_id): return self.deploy_engine("graphlearn-torch", object_id) def delete_interactive_engine(self, object_id): self.delete_engine_stateful_set_with_object_id("interactive", object_id) def delete_graphlearn_engine(self, object_id): self.delete_engine_stateful_set_with_object_id("graphlearn", object_id) def delete_graphlearn_torch_engine(self, object_id): self.delete_engine_stateful_set_with_object_id("graphlearn-torch", object_id) def _allocate_interactive_engine(self, object_id): # check the interactive engine flag if not self._config.kubernetes_launcher.engine.enable_gie: raise NotImplementedError("Interactive engine not enabled") # allocate analytical engine based on the mode if self._deploy_mode == "eager": return self._pod_name_list, self._pod_ip_list, self._pod_host_ip_list return self.deploy_interactive_engine(object_id) def _distribute_interactive_process( self, hosts, object_id: int, schema_path: str, params: dict, with_cypher: bool, engine_selector: str, ): """ Args: hosts (str): hosts of the graph. object_id (int): object id of the graph. schema_path (str): path of the schema file. engine_selector(str): the label selector of the engine. """ env = os.environ.copy() env["GRAPHSCOPE_HOME"] = GRAPHSCOPE_HOME container = INTERACTIVE_EXECUTOR_CONTAINER_NAME params = "\n".join([f"{k}={v}" for k, v in params.items()]) params = base64.b64encode(params.encode("utf-8")).decode("utf-8") neo4j_disabled = "true" if not with_cypher else "false" cmd = [ INTERACTIVE_ENGINE_SCRIPT, "create_gremlin_instance_on_k8s", self._session_workspace, str(object_id), schema_path, hosts, container, str(self._interactive_port), # executor port str(self._interactive_port + 1), # executor rpc port str(self._interactive_port + 2), # frontend gremlin port str(self._interactive_port + 3), # frontend cypher port self._coordinator_name, engine_selector, neo4j_disabled, params, ] self._interactive_port += 4 logger.info("Create GIE instance with command: %s", " ".join(cmd)) process = subprocess.Popen( cmd, start_new_session=True, cwd=os.getcwd(), env=env, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, universal_newlines=True, ) return process def create_interactive_instance( self, object_id: int, schema_path: str, params: dict, with_cypher: bool ): pod_name_list, _, _ = self._allocate_interactive_engine(object_id) if not pod_name_list: raise RuntimeError("Failed to allocate interactive engine") hosts = ",".join(pod_name_list) engine_selector = "gs-engine-" + self._instance_id if self._deploy_mode == "lazy": engine_selector = ( "gs-interactive-" + str(object_id) + "-" + self._instance_id ) return self._distribute_interactive_process( hosts, object_id, schema_path, params, with_cypher, engine_selector ) def close_interactive_instance(self, object_id): if self._deploy_mode == "lazy": logger.info("Close interactive instance with object id: %d", object_id) self.delete_interactive_engine(object_id) return None pod_name_list, _, _ = self._allocate_interactive_engine(object_id) hosts = ",".join(pod_name_list) env = os.environ.copy() env["GRAPHSCOPE_HOME"] = GRAPHSCOPE_HOME container = INTERACTIVE_EXECUTOR_CONTAINER_NAME cmd = [ INTERACTIVE_ENGINE_SCRIPT, "close_gremlin_instance_on_k8s", self._session_workspace, str(object_id), hosts, container, self._instance_id, ] logger.info("Close GIE instance with command: %s", " ".join(cmd)) process = subprocess.Popen( cmd, start_new_session=True, cwd=os.getcwd(), env=env, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, bufsize=1, ) return process def _create_mars_scheduler(self): logger.info("Launching mars scheduler pod for GraphScope ...") deployment = self._mars_cluster.get_mars_deployment() deployment.metadata.owner_references = self._owner_references response = self._apps_api.create_namespaced_deployment( self._namespace, deployment ) self._resource_object.append(response) # The function is used to inject vineyard as a sidecar container into the workload # and return the json string of new workload which is injected with vineyard sidecar # # Assume we have a workload json as below: # # { # "apiVersion": "apps/v1", # "kind": "Deployment", # "metadata": { # "name": "nginx-deployment", # "namespace": "vineyard-job" # }, # "spec": { # "selector": { # "matchLabels": { # "app": "nginx" # } # }, # "template": { # "metadata": { # "labels": { # "app": "nginx" # } # }, # "spec": { # "containers": [ # { # "name": "nginx", # "image": "nginx:1.14.2", # "ports": [ # { # "containerPort": 80 # } # ] # } # ] # } # } # } # } # # The function will return a new workload json as below: # # { # "apiVersion": "apps/v1", # "kind": "Deployment", # "metadata": { # "creationTimestamp": null, # "name": "nginx-deployment", # "namespace": "vineyard-job" # }, # "spec": { # "selector": { # "matchLabels": { # "app": "nginx" # } # } # }, # "template": { # "metadata": null, # "labels": { # "app": "nginx", # "app.vineyard.io/name": "vineyard-sidecar" # }, # "spec": { # "containers": [ # { # "command": null, # "image": "nginx:1.14.2", # "name": "nginx", # "ports": [ # { # "containerPort": 80 # } # ], # "volumeMounts": [ # { # "mountPath": "/var/run", # "name": "vineyard-socket" # } # ] # }, # { # "command": [ # "/bin/bash", # "-c", # "/usr/bin/wait-for-it.sh -t 60 vineyard-sidecar-etcd-service.vineyard-job.svc.cluster.local:2379; \\\n # sleep 1; /usr/local/bin/vineyardd --sync_crds true --socket /var/run/vineyard.sock --size 256Mi \\\n # --stream_threshold 80 --etcd_cmd etcd --etcd_prefix /vineyard \\\n # --etcd_endpoint http://vineyard-sidecar-etcd-service:2379\n" # ], # "env": [ # { # "name": "VINEYARDD_UID", # "value": null # }, # { # "name": "VINEYARDD_NAME", # "value": "vineyard-sidecar" # }, # { # "name": "VINEYARDD_NAMESPACE", # "value": "vineyard-job" # } # ], # "image": "vineyardcloudnative/vineyardd:latest", # "imagePullPolicy": "IfNotPresent", # "name": "vineyard-sidecar", # "ports": [ # { # "containerPort": 9600, # "name": "vineyard-rpc", # "protocol": "TCP" # } # ], # "volumeMounts": [ # { # "mountPath": "/var/run", # "name": "vineyard-socket" # } # ] # } # ], # "volumes": [ # { # "emptyDir": {}, # "name": "vineyard-socket" # } # ] # } # } # } def _inject_vineyard_as_sidecar(self, workload): import vineyard # create the annotations for the workload's template if not exists if workload.spec.template.metadata.annotations is None: workload.spec.template.metadata.annotations = {} # create the labels for the workload's template if not exists if workload.spec.template.metadata.labels is None: workload.spec.template.metadata.labels = {} workload_json = json.dumps( self._api_client.sanitize_for_serialization(workload) ) sts_name = ( f"{self._engine_cluster.engine_stateful_set_name}-{self._instance_id}" ) owner_reference_json = self._get_owner_reference_as_json() # inject vineyard sidecar into the workload # # the name is used to specify the name of the sidecar container, which is also the # labelSelector of the rpc service and the etcd service. # # the apply_resources is used to apply resources to the kubernetes cluster during # the injection. # # for more details about vineyardctl inject, please refer to the link below: # https://github.com/v6d-io/v6d/tree/main/k8s/cmd#vineyardctl-inject new_workload_json = vineyard.deploy.vineyardctl.inject( kubeconfig=self._k8s_config_file, resource=workload_json, sidecar_volume_mountpath="/tmp/vineyard_workspace", name=sts_name + "-vineyard", apply_resources=True, owner_references=owner_reference_json, sidecar_image=self._vineyard_image, sidecar_cpu=self._vineyard_cpu, sidecar_memory=self._vineyard_mem, sidecar_service_type=self._service_type, output="json", capture=True, ) normalized_workload_json = json.loads(new_workload_json) final_workload_json = json.loads(normalized_workload_json["workload"]) fake_kube_response = FakeKubeResponse(final_workload_json) new_workload = self._api_client.deserialize(fake_kube_response, type(workload)) return new_workload def _create_engine_stateful_set(self): logger.info("Creating engine pods...") stateful_set = self._engine_cluster.get_engine_stateful_set() if self._vineyard_deployment is not None: # schedule engine statefulset to the same node with vineyard deployment stateful_set = self._add_pod_affinity_for_vineyard_deployment( workload=stateful_set ) else: stateful_set = self._inject_vineyard_as_sidecar(stateful_set) response = self._apps_api.create_namespaced_stateful_set( self._namespace, stateful_set ) self._resource_object.append(response) return response def _create_frontend_deployment(self, name=None, owner_references=None): logger.info("Creating frontend pods...") deployment = self._engine_cluster.get_interactive_frontend_deployment() if name is not None: deployment.metadata.name = name deployment.metadata.owner_references = owner_references response = self._apps_api.create_namespaced_deployment( self._namespace, deployment ) self._resource_object.append(response) def _create_frontend_service(self): logger.info("Creating frontend service...") service = self._engine_cluster.get_interactive_frontend_service(8233, 7687) service.metadata.owner_references = self._owner_references response = self._core_api.create_namespaced_service(self._namespace, service) self._resource_object.append(response) def _create_graphlearn_service(self, object_id): logger.info("Creating graphlearn service...") service = self._engine_cluster.get_graphlearn_service( object_id, self._graphlearn_start_port ) service.metadata.owner_references = self._owner_references response = self._core_api.create_namespaced_service(self._namespace, service) self._graphlearn_services[object_id] = response self._resource_object.append(response) def _create_graphlearn_torch_service(self, object_id): logger.info("Creating graphlearn torch service...") service = self._engine_cluster.get_graphlearn_torch_service( object_id, self._graphlearn_torch_start_port ) service.metadata.owner_references = self._owner_references response = self._core_api.create_namespaced_service(self._namespace, service) self._graphlearn_torch_services[object_id] = response self._resource_object.append(response) def get_engine_config(self): config = { "vineyard_service_name": self._engine_cluster.vineyard_service_name, "vineyard_rpc_endpoint": self._vineyard_service_endpoint, } if self._config.kubernetes_launcher.mars.enable: config["mars_endpoint"] = self._mars_service_endpoint return config def _create_services(self): self._create_engine_stateful_set() if self._config.kubernetes_launcher.engine.enable_gie: self._create_frontend_deployment(owner_references=self._owner_references) # self._create_frontend_service() if self._config.kubernetes_launcher.mars.enable: # scheduler used by Mars self._create_mars_scheduler() def _waiting_for_services_ready(self): logger.info("Waiting for services ready...") selector = "" namespace = self._namespace start_time = time.time() event_messages = [] while True: # TODO: Add label selector to filter out deployments. statefulsets = self._apps_api.list_namespaced_stateful_set(namespace) service_available = False for rs in statefulsets.items: if rs.metadata.name == self._engine_cluster.engine_stateful_set_name: # logger.info( # "Engine pod: %s ready / %s total", # rs.status.ready_replicas, # self._num_workers, # ) if rs.status.ready_replicas == self._num_workers: # service is ready service_available = True break # check container status labels = rs.spec.selector.match_labels selector = ",".join(f"{k}={v}" for k, v in labels.items()) pods = self._core_api.list_namespaced_pod( namespace=namespace, label_selector=selector ) for pod in pods.items: pod_name = pod.metadata.name field_selector = "involvedObject.name=" + pod_name stream = kube_watch.Watch().stream( self._core_api.list_namespaced_event, namespace, field_selector=field_selector, timeout_seconds=1, ) for event in stream: msg = f"[{pod_name}]: {event['object'].message}" if msg not in event_messages: event_messages.append(msg) logger.info(msg) if event["object"].reason == "Failed": raise RuntimeError("Kubernetes event error: " + msg) if service_available: break if self._timeout_seconds + start_time < time.time(): raise TimeoutError("GraphScope Engines launching timeout.") time.sleep(self._retry_time_seconds) self._pod_name_list = [] self._pod_ip_list = [] self._pod_host_ip_list = [] pods = self._core_api.list_namespaced_pod( namespace=namespace, label_selector=selector ) for pod in pods.items: self._pod_name_list.append(pod.metadata.name) self._pod_ip_list.append(pod.status.pod_ip) self._pod_host_ip_list.append(pod.status.host_ip) assert len(self._pod_ip_list) > 0 self._vineyard_service_endpoint = ( self._engine_cluster.get_vineyard_service_endpoint(self._api_client) ) self._vineyard_internal_endpoint = ( f"{self._pod_ip_list[0]}:{self._engine_cluster._vineyard_service_port}" ) logger.info("GraphScope engines pod is ready.") logger.info("Engines pod name list: %s", self._pod_name_list) logger.info("Engines pod ip list: %s", self._pod_ip_list) logger.info("Engines pod host ip list: %s", self._pod_host_ip_list) logger.info("Vineyard service endpoint: %s", self._vineyard_service_endpoint) if self._config.kubernetes_launcher.mars.enable: self._mars_service_endpoint = self._mars_cluster.get_mars_service_endpoint( self._api_client ) logger.info("Mars service endpoint: %s", self._mars_service_endpoint) # the function will add the podAffinity to the engine workload so that the workload # will be scheduled to the same node with vineyard deployment. # e.g. the vineyard deployment is named "vineyard-deployment" and the namespace is "graphscope-system", # the podAffinity will be added to the engine workload as below: # spec: # affinity: # podAffinity: # requiredDuringSchedulingIgnoredDuringExecution: # - labelSelector: # matchExpressions: # - key: app.kubernetes.io/instance # operator: In # values: # - graphscope-system-vineyard-deployment # [vineyard deployment namespace]-[vineyard deployment name] # topologyKey: kubernetes.io/hostname def _add_pod_affinity_for_vineyard_deployment(self, workload): import vineyard workload_json = json.dumps( self._api_client.sanitize_for_serialization(workload) ) new_workload_json = vineyard.deploy.vineyardctl.schedule.workload( kubeconfig=self._k8s_config_file, resource=workload_json, vineyardd_name=self._vineyard_deployment, vineyardd_namespace=self._namespace, capture=True, ) normalized_workload_json = json.loads(new_workload_json) fake_kube_response = FakeKubeResponse(normalized_workload_json) new_workload = self._api_client.deserialize(fake_kube_response, type(workload)) return new_workload def _dump_resource_object(self): resource = {} if self._delete_namespace: resource[self._namespace] = "Namespace" else: # coordinator info resource[self._coordinator_name] = "Deployment" resource[self._coordinator_service_name] = "Service" self._resource_object.dump(extra_resource=resource) def _get_analytical_hosts(self): pod_name_list = self._pod_name_list if self._analytical_pod_name: pod_name_list = self._analytical_pod_name return pod_name_list def _allocate_analytical_engine(self): # allocate analytical engine based on the mode if self._deploy_mode == "eager": return self._pod_name_list, self._pod_ip_list, self._pod_host_ip_list else: if self._config.kubernetes_launcher.engine.enable_gae: return self.deploy_analytical_engine() elif self._config.kubernetes_launcher.engine.enable_gae_java: return self.deploy_analytical_java_engine() else: logger.warning("analytical is not enabled, skip allocating") def _distribute_analytical_process(self, pod_name_list, pod_ip_list): # generate and distribute hostfile hosts = os.path.join(get_tempdir(), "hosts_of_nodes") with open(hosts, "w") as f: for i, pod_ip in enumerate(pod_ip_list): f.write(f"{pod_ip} {pod_name_list[i]}\n") container = ANALYTICAL_CONTAINER_NAME for pod in pod_name_list: logger.debug( run_kube_cp_command(hosts, "/tmp/hosts_of_nodes", pod, container, True) ) # launch engine rmcp = ResolveMPICmdPrefix(rsh_agent=True) cmd, mpi_env = rmcp.resolve(self._num_workers, pod_name_list) cmd.append(ANALYTICAL_ENGINE_PATH) cmd.extend(["--host", "0.0.0.0"]) cmd.extend(["--port", str(self._random_analytical_engine_rpc_port)]) cmd.extend(["-v", str(self._glog_level)]) mpi_env["GLOG_v"] = str(self._glog_level) cmd.extend(["--vineyard_socket", self._engine_cluster.vineyard_ipc_socket]) logger.info("Analytical engine launching command: %s", " ".join(cmd)) env = os.environ.copy() env["GRAPHSCOPE_HOME"] = GRAPHSCOPE_HOME env.update(mpi_env) self._analytical_engine_process = subprocess.Popen( cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", errors="replace", universal_newlines=True, bufsize=1, ) stdout_watcher = PipeWatcher( self._analytical_engine_process.stdout, sys.stdout, drop=True ) stderr_watcher = PipeWatcher( self._analytical_engine_process.stderr, sys.stderr, drop=True ) setattr(self._analytical_engine_process, "stdout_watcher", stdout_watcher) setattr(self._analytical_engine_process, "stderr_watcher", stderr_watcher) def create_analytical_instance(self): pod_name_list, pod_ip_list, _ = self._allocate_analytical_engine() if not pod_name_list or not pod_ip_list: raise RuntimeError("Failed to allocate analytical engine.") self._distribute_analytical_process(pod_name_list, pod_ip_list) self._analytical_engine_endpoint = ( f"{self._pod_ip_list[0]}:{self._random_analytical_engine_rpc_port}" ) logger.info( "GAE rpc service is listening on %s ...", self._analytical_engine_endpoint ) def _delete_dangling_coordinator(self): # delete service try: self._core_api.delete_namespaced_service( self._coordinator_service_name, self._namespace ) except K8SApiException as ex: if ex.status == 404: logger.warning( "coordinator service %s not found", self._coordinator_service_name ) else: logger.exception( "Deleting dangling coordinator service %s failed", self._coordinator_service_name, ) try: self._apps_api.delete_namespaced_deployment( self._coordinator_name, self._namespace ) except K8SApiException as ex: if ex.status == 404: logger.warning( "coordinator deployment %s not found", self._coordinator_name ) else: logger.exception( "Deleting dangling coordinator %s failed", self._coordinator_name ) if self._waiting_for_delete: start_time = time.time() while True: try: self._apps_api.read_namespaced_deployment( self._coordinator_name, self._namespace ) except K8SApiException as ex: if ex.status != 404: logger.exception( "Deleting dangling coordinator %s failed", self._coordinator_name, ) break else: if time.time() - start_time > self._timeout_seconds: logger.error( "Deleting dangling coordinator %s timeout", self._coordinator_name, ) time.sleep(self._retry_time_seconds) def _get_owner_reference_as_json(self): if self._owner_references: owner_reference = [ { "apiVersion": self._owner_references[0].api_version, "kind": self._owner_references[0].kind, "name": self._owner_references[0].name, "uid": self._owner_references[0].uid, } ] owner_reference_json = json.dumps(owner_reference) else: owner_reference_json = json.dumps([]) return owner_reference_json def _check_if_vineyard_deployment_exist(self): if self._vineyard_deployment is None or self._vineyard_deployment == "": return False try: self._apps_api.read_namespaced_deployment( self._vineyard_deployment, self._namespace ) except K8SApiException: logger.info( "Vineyard deployment %s/%s not exist", self._namespace, self._vineyard_deployment, ) return False return True def _deploy_vineyard_deployment_if_not_exist(self): if not self._check_if_vineyard_deployment_exist(): self._deploy_vineyard_deployment() else: logger.info( "The external vineyard deployment %s is ready." "Please make sure the type of the vineyard rpc service is the same as %s.", self._vineyard_deployment, self._service_type, ) def _deploy_vineyard_deployment(self): import vineyard owner_reference_json = self._get_owner_reference_as_json() vineyard.deploy.vineyardctl.deploy.vineyard_deployment( kubeconfig=self._k8s_config_file, name=self._vineyard_deployment, namespace=self._namespace, replicas=self._num_workers, etcd_replicas=1, vineyardd_image=self._vineyard_image, vineyardd_memory=self._vineyard_mem, vineyardd_cpu=self._vineyard_cpu, vineyardd_service_type=self._service_type, owner_references=owner_reference_json, ) vineyard_pods = self._core_api.list_namespaced_pod( self._namespace, label_selector=f"app.kubernetes.io/instance={self._namespace}-{self._vineyard_deployment}", ) self._vineyard_pod_name_list.extend( [pod.metadata.name for pod in vineyard_pods.items] ) def start(self): if self._serving: return True try: if self._deploy_mode == "eager": self._create_services() self._waiting_for_services_ready() self._dump_resource_object() self._serving = True except Exception: # pylint: disable=broad-except time.sleep(1) logger.exception("Error when launching GraphScope on kubernetes cluster") self.stop() return False return True def stop(self, is_dangling=False): if self._serving: logger.info("Cleaning up kubernetes resources") for target in self._resource_object: delete_kubernetes_object( api_client=self._api_client, target=target, wait=self._waiting_for_delete, timeout_seconds=self._timeout_seconds, ) self._resource_object.clear() if is_dangling: logger.info("Dangling coordinator detected, cleaning up...") # delete everything inside namespace of graphscope instance if self._delete_namespace: # delete namespace created by graphscope self._core_api.delete_namespace(self._namespace) if self._waiting_for_delete: start_time = time.time() while True: try: self._core_api.read_namespace(self._namespace) except K8SApiException as ex: if ex.status != 404: logger.exception( "Deleting dangling namespace %s failed", self._namespace, ) break else: if time.time() - start_time > self._timeout_seconds: logger.error( "Deleting namespace %s timeout", self._namespace ) time.sleep(self._retry_time_seconds) else: # delete coordinator deployment and service self._delete_dangling_coordinator() self._serving = False logger.info("Kubernetes launcher stopped") def _allocate_graphlearn_engine(self, object_id): # check the graphlearn engine flag if not self._config.kubernetes_launcher.engine.enable_gle: raise NotImplementedError("GraphLearn engine not enabled") # allocate graphlearn engine based on the mode if self._deploy_mode == "eager": return self._pod_name_list, self._pod_ip_list, self._pod_host_ip_list return self.deploy_graphlearn_engine(object_id) def _allocate_graphlearn_torch_engine(self, object_id): # check the graphlearn torch engine flag if not self._config.kubernetes_launcher.engine.enable_glt: raise NotImplementedError("GraphLearn torch engine not enabled") # allocate graphlearn engine based on the mode if self._deploy_mode == "eager": return self._pod_name_list, self._pod_ip_list, self._pod_host_ip_list return self.deploy_graphlearn_torch_engine(object_id) def _distribute_graphlearn_process( self, pod_name_list, pod_host_ip_list, object_id, handle, config ): # allocate service for ports # prepare arguments handle = json.loads( base64.b64decode(handle.encode("utf-8", errors="ignore")).decode( "utf-8", errors="ignore" ) ) hosts = ",".join( [ f"{pod_name}:{port}" for pod_name, port in zip( pod_name_list, self._engine_cluster.get_graphlearn_ports( self._graphlearn_start_port ), ) ] ) handle["server"] = hosts handle = base64.b64encode( json.dumps(handle).encode("utf-8", errors="ignore") ).decode("utf-8", errors="ignore") # launch the server self._graphlearn_instance_processes[object_id] = [] for pod_index, pod in enumerate(self._pod_name_list): container = GRAPHLEARN_CONTAINER_NAME sub_cmd = f"python3 -m gscoordinator.launch_graphlearn {handle} {config} {pod_index}" cmd = f"kubectl -n {self._namespace} exec -it -c {container} {pod} -- {sub_cmd}" # logger.debug("launching learning server: %s", " ".join(cmd)) proc = subprocess.Popen( shlex.split(cmd), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8", errors="replace", universal_newlines=True, bufsize=1, ) stdout_watcher = PipeWatcher( proc.stdout, sys.stdout, drop=True, suppressed=(not logger.isEnabledFor(logging.DEBUG)), ) setattr(proc, "stdout_watcher", stdout_watcher) self._graphlearn_instance_processes[object_id].append(proc) # Create Service self._create_graphlearn_service(object_id) # update the port usage record self._graphlearn_start_port += len(pod_name_list) # parse the service hosts and ports return self._engine_cluster.get_graphlearn_service_endpoint( self._api_client, object_id, pod_host_ip_list ) def _distribute_graphlearn_torch_process( self, pod_name_list, pod_ip_list, object_id, handle, config ): # allocate service for ports # prepare arguments handle = json.loads( base64.b64decode(handle.encode("utf-8", errors="ignore")).decode( "utf-8", errors="ignore" ) ) ports = self._engine_cluster.get_graphlearn_torch_ports( self._graphlearn_torch_start_port ) handle["master_addr"] = pod_ip_list[0] handle["server_client_master_port"] = ports[0] server_list = [f"{pod_ip_list[0]}:{ports[i]}" for i in range(4)] server_handle = base64.b64encode( json.dumps(handle).encode("utf-8", errors="ignore") ).decode("utf-8", errors="ignore") # launch the server self._graphlearn_torch_instance_processes[object_id] = [] for pod_index, pod in enumerate(self._pod_name_list): container = GRAPHLEARN_TORCH_CONTAINER_NAME sub_cmd = f"env PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python \ python3 -m gscoordinator.launch_graphlearn_torch \ {server_handle} {config} {pod_index}" cmd = f"kubectl -n {self._namespace} exec -it -c {container} {pod} -- {sub_cmd}" # logger.debug("launching learning server: %s", " ".join(cmd)) proc = subprocess.Popen( shlex.split(cmd), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8", errors="replace", universal_newlines=True, bufsize=1, ) stdout_watcher = PipeWatcher( proc.stdout, sys.stdout, suppressed=(not logger.isEnabledFor(logging.DEBUG)), ) time.sleep(5) logger.debug("process status: %s", proc.poll()) setattr(proc, "stdout_watcher", stdout_watcher) self._graphlearn_torch_instance_processes[object_id].append(proc) # Create Service self._create_graphlearn_torch_service(object_id) # update the port usage record self._graphlearn_torch_start_port += len(pod_name_list) # prepare config map for client scripts config_map = kube_client.V1ConfigMap( api_version="v1", kind="ConfigMap", metadata=kube_client.V1ObjectMeta( name="graphlearn-torch-client-config", namespace=self._namespace, ), data=handle["client_content"], ) self._core_api.create_namespaced_config_map(self._namespace, config_map) # prepare the manifest pytorch_job_manifest = replace_string_in_dict( handle["manifest"], "${MASTER_ADDR}", handle["master_addr"] ) # parse the pytorchjob yaml group = pytorch_job_manifest["apiVersion"].split("/")[0] version = pytorch_job_manifest["apiVersion"].split("/")[1] name = pytorch_job_manifest["metadata"]["name"] namespace = pytorch_job_manifest["metadata"]["namespace"] plural = "pytorchjobs" # This is PyTorchJob CRD's plural name try: # create PyTorchJob api_response = self._pytorchjobs_api.create_namespaced_custom_object( group=group, version=version, namespace=namespace, plural=plural, body=pytorch_job_manifest, ) logger.info(api_response) except K8SApiException as e: logger.info( f"Exception when calling CustomObjectsApi->create_namespaced_custom_object: {e}" ) raise # set Watcher to monitor the state of the PyTorchJob w = kube_watch.Watch() # loop checking the state of PyTorchJob for event in w.stream( self._pytorchjobs_api.list_namespaced_custom_object, group, version, namespace, plural, ): pytorch_job = event["object"] if pytorch_job.get("metadata", {}).get("name") == name: status = pytorch_job.get("status", {}) if status: # check status existence conditions = status.get("conditions", []) for condition in conditions: if ( condition.get("type") == "Succeeded" and condition.get("status") == "True" ): logger.info(f"PyTorchJob {name} has succeeded!") w.stop() break elif ( condition.get("type") == "Failed" and condition.get("status") == "True" ): logger.info(f"PyTorchJob {name} has failed!") w.stop() break self.close_graphlearn_torch_client(group, name, version, plural, namespace) return server_list def create_learning_instance(self, object_id, handle, config, learning_backend): if learning_backend == message_pb2.LearningBackend.GRAPHLEARN: pod_name_list, _, pod_host_ip_list = self._allocate_graphlearn_engine( object_id ) if not pod_name_list or not pod_host_ip_list: raise RuntimeError("Failed to allocate learning engine") return self._distribute_graphlearn_process( pod_name_list, pod_host_ip_list, object_id, handle, config ) elif learning_backend == message_pb2.LearningBackend.GRAPHLEARN_TORCH: ( pod_name_list, pod_ip_list, pod_host_ip_list, ) = self._allocate_graphlearn_torch_engine(object_id) if not pod_name_list or not pod_host_ip_list: raise RuntimeError("Failed to allocate learning engine") return self._distribute_graphlearn_torch_process( pod_name_list, pod_ip_list, object_id, handle, config ) else: raise ValueError("invalid learning backend") def close_learning_instance(self, object_id, learning_backend): if learning_backend == message_pb2.LearningBackend.GRAPHLEARN: self.close_graphlearn_instance(object_id) elif learning_backend == message_pb2.LearningBackend.GRAPHLEARN_TORCH: self.close_graphlearn_torch_instance(object_id) else: raise ValueError("invalid learning backend") def close_graphlearn_instance(self, object_id): if self._deploy_mode == "lazy": self.delete_graphlearn_engine(object_id) return if object_id not in self._graphlearn_instance_processes: return # delete the services target = self._graphlearn_services[object_id] try: delete_kubernetes_object( api_client=self._api_client, target=target, wait=self._waiting_for_delete, timeout_seconds=self._timeout_seconds, ) except Exception: # pylint: disable=broad-except logger.exception("Failed to delete graphlearn service for %s", object_id) # terminate the process for proc in self._graphlearn_instance_processes[object_id]: try: proc.terminate() proc.wait(1) except Exception: # pylint: disable=broad-except logger.exception("Failed to terminate graphlearn server") self._graphlearn_instance_processes[object_id].clear() def close_graphlearn_torch_instance(self, object_id): if self._deploy_mode == "lazy": self.delete_graphlearn_torch_engine(object_id) return if object_id not in self._graphlearn_torch_instance_processes: return # delete the services target = self._graphlearn_torch_services[object_id] try: delete_kubernetes_object( api_client=self._api_client, target=target, wait=self._waiting_for_delete, timeout_seconds=self._timeout_seconds, ) except Exception: # pylint: disable=broad-except logger.exception( "Failed to delete graphlearn torch service for %s", object_id ) # terminate the process for proc in self._graphlearn_torch_instance_processes[object_id]: try: proc.terminate() proc.wait(1) except Exception: # pylint: disable=broad-except logger.exception("Failed to terminate graphlearn torch server") self._graphlearn_torch_instance_processes[object_id].clear() def close_graphlearn_torch_client(self, group, name, version, plural, namespace): # clear PyTorchJob logger.info(f"Deleting PyTorchJob {name}...") try: response = self._pytorchjobs_api.delete_namespaced_custom_object( group=group, name=name, version=version, plural=plural, namespace=namespace, body=kube_client.V1DeleteOptions( propagation_policy="Foreground", ), ) logger.info(f"PyTorchJob {name} deleted. Response: {response}") except K8SApiException as e: logger.info( f"Exception when calling CustomObjectsApi->delete_namespaced_custom_object: {e}" ) try: response = self._core_api.delete_namespaced_config_map( name="graphlearn-torch-client-config", namespace=self._namespace, ) logger.info( f"ConfigMap graphlearn-torch-client-config deleted. Response: {response}" ) except K8SApiException as e: logger.info( f"Exception when calling CoreV1Api->delete_namespaced_config_map: {e}" ) class ResourceManager(object): """A class to manager kubernetes object. Object managed by this class will dump meta info to disk file for pod preStop lifecycle management. meta info format: { "my-deployment": "Deployment", "my-service": "Service" } """ _resource_object_path = os.path.join(get_tempdir(), "resource_object") # fixed def __init__(self, api_client): self._api_client = api_client self._resource_object = [] self._meta_info = {} def append(self, target): self._resource_object.append(target) self._meta_info.update( get_kubernetes_object_info(api_client=self._api_client, target=target) ) self.dump() def extend(self, targets): self._resource_object.extend(targets) for target in targets: self._meta_info.update( get_kubernetes_object_info(api_client=self._api_client, target=target) ) self.dump() def clear(self): self._resource_object.clear() self._meta_info.clear() def __str__(self): return str(self._meta_info) def __getitem__(self, index): return self._resource_object[index] def dump(self, extra_resource=None): """Dump meta info to disk file. Args: extra_resource (dict): extra resource to dump. A typical scenario is dumping meta info of namespace for coordinator dangling processing. """ if extra_resource is not None: rlt = copy.deepcopy(self._meta_info) rlt.update(extra_resource) else: rlt = self._meta_info with open(self._resource_object_path, "w") as f: json.dump(rlt, f)