(self)
| 889 | return fetch_handler.wrap_results(response) |
| 890 | |
| 891 | def _connect(self): |
| 892 | if self._config.coordinator.endpoint is not None: |
| 893 | # try to connect to exist coordinator |
| 894 | self._coordinator_endpoint = self._config.coordinator.endpoint |
| 895 | elif self._cluster_type == types_pb2.K8S: |
| 896 | self._launcher = KubernetesClusterLauncher( |
| 897 | config=self._config, api_client=self._get_api_client() |
| 898 | ) |
| 899 | elif self._cluster_type == types_pb2.HOSTS: |
| 900 | # launch coordinator with hosts |
| 901 | self._launcher = HostsClusterLauncher(config=self._config) |
| 902 | else: |
| 903 | raise RuntimeError( |
| 904 | f"Unrecognized cluster type {types_pb2.ClusterType.Name(self._cluster_type)}." |
| 905 | ) |
| 906 | |
| 907 | # launching graphscope service |
| 908 | if self._launcher is not None: |
| 909 | self._launcher.start() |
| 910 | self._coordinator_endpoint = self._launcher.coordinator_endpoint |
| 911 | |
| 912 | # waiting service ready |
| 913 | self._grpc_client = GRPCClient( |
| 914 | self._launcher, self._coordinator_endpoint, self._config.session.reconnect |
| 915 | ) |
| 916 | self._grpc_client.waiting_service_ready( |
| 917 | timeout_seconds=self._config.session.timeout_seconds, |
| 918 | ) |
| 919 | |
| 920 | # connect and fetch logs from rpc server |
| 921 | try: |
| 922 | ( |
| 923 | self._session_id, |
| 924 | self._cluster_type, |
| 925 | self._config.session.num_workers, |
| 926 | self._config.kubernetes_launcher.namespace, |
| 927 | self._engine_config, |
| 928 | pod_name_list, |
| 929 | ) = self._grpc_client.connect( |
| 930 | cleanup_instance=not bool(self._config.coordinator.endpoint), |
| 931 | dangling_timeout_seconds=self._config.session.dangling_timeout_seconds, |
| 932 | ) |
| 933 | self._pod_name_list = list(pod_name_list) |
| 934 | |
| 935 | # fetch logs |
| 936 | if self._config.coordinator.endpoint or self._cluster_type == types_pb2.K8S: |
| 937 | self._grpc_client.fetch_logs() |
| 938 | _session_dict[self._session_id] = self |
| 939 | |
| 940 | # Launch analytical engine right after session connected. |
| 941 | # This may be changed to on demand launching in the future |
| 942 | if not self._engine_config and not self._pod_name_list: |
| 943 | ( |
| 944 | self._engine_config, |
| 945 | pod_name_list, |
| 946 | ) = self._grpc_client.create_analytical_instance() |
| 947 | self._pod_name_list = list(pod_name_list) |
| 948 | except Exception: |
no test coverage detected