odps/mars_extension/legacy/deploy/client.py [224:334]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def rescale_workers(self, new_scale, min_workers=None, wait=True, timeout=None): self._mars_session._sess.rescale_workers( new_scale, min_workers=min_workers, wait=wait, timeout=timeout ) def get_logview_address(self): return self._kube_instance.get_logview_address() def get_mars_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, CUPID_APP_NAME ) def get_notebook_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, NOTEBOOK_NAME ) def get_graphscope_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, GS_COORDINATOR_NAME ) def get_req_session(self): from ....rest import RestClient if options.mars.use_common_proxy: return RestClient( self._odps.account, self._endpoint, self._odps.project, tag="MARS" ) else: return requests.Session() def check_instance_status(self): if self._kube_instance.is_terminated(): for task_name, task in (self._kube_instance.get_task_statuses()).items(): exc = None if task.status == Instance.Task.TaskStatus.FAILED: exc = errors.parse_instance_error( self._kube_instance.get_task_result(task_name) ) elif task.status != Instance.Task.TaskStatus.SUCCESS: exc = errors.ODPSError( "%s, status=%s" % (task_name, task.status.value) ) if exc: exc.instance_id = self._kube_instance.id raise exc def _post_pyodps_api(self, **data): r = self._req_session.post( self._endpoint.rstrip("/") + "/api/pyodps", data=data ) try: r.raise_for_status() except errors.InvalidStateSetting: if not self._kube_instance.is_successful(): raise def wait_for_success(self, min_worker_num=0, create_session=True): while True: self.check_instance_status() with warnings.catch_warnings(): warnings.filterwarnings("ignore") try: if self._endpoint is None: self._endpoint = self.get_mars_endpoint() logger.info("Mars UI: " + self._endpoint) self._req_session = self.get_req_session() self._post_pyodps_api( action="write_log", content="Mars UI from client: " + self._endpoint, ) if self._with_notebook and self._notebook_endpoint is None: self._notebook_endpoint = self.get_notebook_endpoint() logger.info("Notebook UI: " + self._notebook_endpoint) self._post_pyodps_api( action="write_log", content="Notebook UI from client: " + self._notebook_endpoint, ) if self._with_graphscope and self._graphscope_endpoint is None: self._graphscope_endpoint = self.get_graphscope_endpoint() logger.info("Graphscope endpoint: " + self._graphscope_endpoint) self._post_pyodps_api( action="write_log", content="Graphscope endpoint from client: " + self._graphscope_endpoint, ) except KeyboardInterrupt: raise except: time.sleep(1) continue try: if not self.check_service_ready(): continue if self.count_workers() >= min_worker_num: break except: continue finally: time.sleep(1) if create_session: try: self._mars_session = new_session( - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - odps/mars_extension/oscar/deploy/client.py [237:347]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def rescale_workers(self, new_scale, min_workers=None, wait=True, timeout=None): self._mars_session._sess.rescale_workers( new_scale, min_workers=min_workers, wait=wait, timeout=timeout ) def get_logview_address(self): return self._kube_instance.get_logview_address() def get_mars_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, CUPID_APP_NAME ) def get_notebook_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, NOTEBOOK_NAME ) def get_graphscope_endpoint(self): return self._cupid_session.get_proxied_url( self._kube_instance.id, GS_COORDINATOR_NAME ) def get_req_session(self): from ....rest import RestClient if options.mars.use_common_proxy: return RestClient( self._odps.account, self._endpoint, self._odps.project, tag="MARS" ) else: return requests.Session() def check_instance_status(self): if self._kube_instance.is_terminated(): for task_name, task in (self._kube_instance.get_task_statuses()).items(): exc = None if task.status == Instance.Task.TaskStatus.FAILED: exc = errors.parse_instance_error( self._kube_instance.get_task_result(task_name) ) elif task.status != Instance.Task.TaskStatus.SUCCESS: exc = errors.ODPSError( "%s, status=%s" % (task_name, task.status.value) ) if exc: exc.instance_id = self._kube_instance.id raise exc def _post_pyodps_api(self, **data): r = self._req_session.post( self._endpoint.rstrip("/") + "/api/pyodps", data=data ) try: r.raise_for_status() except errors.InvalidStateSetting: if not self._kube_instance.is_successful(): raise def wait_for_success(self, min_worker_num=0, create_session=True): while True: self.check_instance_status() with warnings.catch_warnings(): warnings.filterwarnings("ignore") try: if self._endpoint is None: self._endpoint = self.get_mars_endpoint() logger.info("Mars UI: " + self._endpoint) self._req_session = self.get_req_session() self._post_pyodps_api( action="write_log", content="Mars UI from client: " + self._endpoint, ) if self._with_notebook and self._notebook_endpoint is None: self._notebook_endpoint = self.get_notebook_endpoint() logger.info("Notebook UI: " + self._notebook_endpoint) self._post_pyodps_api( action="write_log", content="Notebook UI from client: " + self._notebook_endpoint, ) if self._with_graphscope and self._graphscope_endpoint is None: self._graphscope_endpoint = self.get_graphscope_endpoint() logger.info("Graphscope endpoint: " + self._graphscope_endpoint) self._post_pyodps_api( action="write_log", content="Graphscope endpoint from client: " + self._graphscope_endpoint, ) except KeyboardInterrupt: raise except: time.sleep(1) continue try: if not self.check_service_ready(): continue if self.count_workers() >= min_worker_num: break except: continue finally: time.sleep(1) if create_session: try: self._mars_session = new_session( - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -