wehub-resource-sync
1470 lines
53 KiB
Python
1470 lines
53 KiB
Python
import logging
|
|
import threading
|
|
import urllib
|
|
import warnings
|
|
from contextlib import contextmanager
|
|
from dataclasses import fields
|
|
from typing import Any, Dict, Generator, List, Optional, Tuple, Union
|
|
|
|
import requests
|
|
|
|
import ray
|
|
from ray._private.authentication.http_token_authentication import (
|
|
get_auth_headers_if_auth_enabled,
|
|
)
|
|
from ray.dashboard.modules.dashboard_sdk import SubmissionClient
|
|
from ray.dashboard.utils import (
|
|
get_address_for_submission_client,
|
|
ray_address_to_api_server_url,
|
|
)
|
|
from ray.util.annotations import DeveloperAPI
|
|
from ray.util.state.common import (
|
|
DEFAULT_LIMIT,
|
|
DEFAULT_RPC_TIMEOUT,
|
|
ActorState,
|
|
ClusterEventState,
|
|
GetApiOptions,
|
|
GetLogOptions,
|
|
JobState,
|
|
ListApiOptions,
|
|
NodeState,
|
|
ObjectState,
|
|
PlacementGroupState,
|
|
PredicateType,
|
|
RuntimeEnvState,
|
|
StateResource,
|
|
SummaryApiOptions,
|
|
SummaryResource,
|
|
SupportedFilterType,
|
|
TaskState,
|
|
WorkerState,
|
|
dict_to_state,
|
|
)
|
|
from ray.util.state.exception import RayStateApiException, ServerUnavailable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
_MAX_HTTP_RESPONSE_EXCEPTION_TEXT = 500
|
|
|
|
|
|
@contextmanager
|
|
def warnings_on_slow_request(
|
|
*, address: str, endpoint: str, timeout: float, explain: bool
|
|
):
|
|
"""A context manager to print warnings if the request is replied slowly.
|
|
|
|
Warnings are printed 3 times
|
|
|
|
Args:
|
|
address: The address of the endpoint.
|
|
endpoint: The name of the endpoint.
|
|
timeout: Request timeout in seconds.
|
|
explain: Whether ot not it will print the warning.
|
|
"""
|
|
# Do nothing if explain is not specified.
|
|
if not explain:
|
|
yield
|
|
return
|
|
|
|
# Prepare timers to print warning.
|
|
# Print 3 times with exponential backoff. timeout / 2, timeout / 4, timeout / 8
|
|
def print_warning(elapsed: float):
|
|
logger.info(
|
|
f"({round(elapsed, 2)} / {timeout} seconds) "
|
|
"Waiting for the response from the API server "
|
|
f"address {address}{endpoint}.",
|
|
)
|
|
|
|
warning_timers = [
|
|
threading.Timer(timeout / i, print_warning, args=[timeout / i])
|
|
for i in [2, 4, 8]
|
|
]
|
|
|
|
try:
|
|
for timer in warning_timers:
|
|
timer.start()
|
|
yield
|
|
finally:
|
|
# Make sure all timers are cancelled once request is terminated.
|
|
for timer in warning_timers:
|
|
timer.cancel()
|
|
|
|
|
|
"""
|
|
This file contains API client and methods for querying ray state.
|
|
|
|
Usage:
|
|
1. [Recommended] With StateApiClient:
|
|
```
|
|
client = StateApiClient(address="auto")
|
|
data = client.list(StateResource.NODES)
|
|
...
|
|
```
|
|
|
|
2. With SDK APIs:
|
|
The API creates a `StateApiClient` for each invocation. So if multiple
|
|
invocations of listing are used, it is better to reuse the `StateApiClient`
|
|
as suggested above.
|
|
```
|
|
data = list_nodes(address="auto")
|
|
```
|
|
"""
|
|
|
|
|
|
@DeveloperAPI
|
|
class StateApiClient(SubmissionClient):
|
|
"""State API Client issues REST GET requests to the server for resource states."""
|
|
|
|
def __init__(
|
|
self,
|
|
address: Optional[str] = None,
|
|
cookies: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""Initialize a StateApiClient and check the connection to the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address (e.g. `127.0.0.0:6379`, `auto`), or Ray
|
|
Client address (e.g. `ray://<head-node-ip>:10001`), or Ray dashboard
|
|
address (e.g. `http://<head-node-ip>:8265`).
|
|
If not provided, it will be detected automatically from any running
|
|
local Ray cluster.
|
|
cookies: Cookies to use when sending requests to the HTTP job server.
|
|
headers: Headers to use when sending requests to the HTTP job server, used
|
|
for cases like authentication to a remote cluster.
|
|
"""
|
|
if requests is None:
|
|
raise RuntimeError(
|
|
"The Ray state CLI & SDK require the ray[default] "
|
|
"installation: `pip install 'ray[default']``"
|
|
)
|
|
if not headers:
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
# Resolve API server URL
|
|
api_server_url = get_address_for_submission_client(address)
|
|
|
|
super().__init__(
|
|
address=api_server_url,
|
|
create_cluster_if_needed=False,
|
|
headers=headers,
|
|
cookies=cookies,
|
|
)
|
|
|
|
@classmethod
|
|
def _make_param(cls, options: Union[ListApiOptions, GetApiOptions]) -> Dict:
|
|
options_dict = {}
|
|
for field in fields(options):
|
|
# TODO(rickyyx): We will need to find a way to pass server side timeout
|
|
# TODO(rickyyx): We will have to convert filter option
|
|
# slightly differently for now. But could we do k,v pair rather than this?
|
|
# I see we are also converting dict to XXXApiOptions later on, we could
|
|
# probably organize the marshaling a bit better.
|
|
if field.name == "filters":
|
|
options_dict["filter_keys"] = []
|
|
options_dict["filter_predicates"] = []
|
|
options_dict["filter_values"] = []
|
|
for filter in options.filters:
|
|
if len(filter) != 3:
|
|
raise ValueError(
|
|
f"The given filter has incorrect input type, {filter}. "
|
|
"Provide (key, predicate, value) tuples."
|
|
)
|
|
filter_k, filter_predicate, filter_val = filter
|
|
options_dict["filter_keys"].append(filter_k)
|
|
options_dict["filter_predicates"].append(filter_predicate)
|
|
options_dict["filter_values"].append(filter_val)
|
|
continue
|
|
|
|
option_val = getattr(options, field.name)
|
|
if option_val is not None:
|
|
options_dict[field.name] = option_val
|
|
|
|
return options_dict
|
|
|
|
def _make_http_get_request(
|
|
self,
|
|
endpoint: str,
|
|
params: Dict,
|
|
timeout: float,
|
|
_explain: bool = False,
|
|
) -> Dict:
|
|
with warnings_on_slow_request(
|
|
address=self._address, endpoint=endpoint, timeout=timeout, explain=_explain
|
|
):
|
|
# Send a request.
|
|
response = None
|
|
try:
|
|
response = self._do_request(
|
|
"GET",
|
|
endpoint,
|
|
timeout=timeout,
|
|
params=params,
|
|
)
|
|
# If we have a valid JSON error, don't raise a generic exception but
|
|
# instead let the caller parse it to raise a more precise exception.
|
|
if (
|
|
response.status_code == 500
|
|
and "application/json"
|
|
not in response.headers.get("Content-Type", "")
|
|
):
|
|
response.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
err_str = f"Failed to make request to {self._address}{endpoint}. "
|
|
|
|
# Best-effort to give hints to users on potential reasons of connection
|
|
# failure.
|
|
err_str += (
|
|
"Failed to connect to API server. Please check the API server "
|
|
"log for details. Make sure dependencies are installed with "
|
|
"`pip install ray[default]`. Please also check dashboard is "
|
|
"available, and included when starting ray cluster, "
|
|
"i.e. `ray start --include-dashboard=True --head`. "
|
|
)
|
|
if response is None:
|
|
raise ServerUnavailable(err_str)
|
|
|
|
err_str += f"Response(url={response.url},status={response.status_code})"
|
|
raise RayStateApiException(err_str) from e
|
|
|
|
try:
|
|
# Process the response.
|
|
response = response.json()
|
|
except requests.exceptions.JSONDecodeError as e:
|
|
raise RayStateApiException(
|
|
f"Failed to parse Response(url={response.url}, "
|
|
f"status={response.status_code}, text='{response.text[:_MAX_HTTP_RESPONSE_EXCEPTION_TEXT]}')"
|
|
) from e
|
|
if response["result"] is False:
|
|
raise RayStateApiException(
|
|
"API server internal error. See dashboard.log file for more details. "
|
|
f"Error: {response['msg']}"
|
|
)
|
|
|
|
# Dictionary of `ListApiResponse` or `SummaryApiResponse`
|
|
return response["data"]["result"]
|
|
|
|
def get(
|
|
self,
|
|
resource: StateResource,
|
|
id: str,
|
|
options: Optional[GetApiOptions],
|
|
_explain: bool = False,
|
|
) -> Optional[
|
|
Union[
|
|
ActorState,
|
|
PlacementGroupState,
|
|
NodeState,
|
|
WorkerState,
|
|
TaskState,
|
|
List[ObjectState],
|
|
JobState,
|
|
]
|
|
]:
|
|
"""Get resources states by id
|
|
|
|
Args:
|
|
resource: Resource names, i.e. 'workers', 'actors', 'nodes',
|
|
'placement_groups', 'tasks', 'objects'.
|
|
'jobs' and 'runtime-envs' are not supported yet.
|
|
id: ID for the resource, i.e. 'node_id' for nodes.
|
|
options: Get options. See `GetApiOptions` for details.
|
|
_explain: Print the API information such as API
|
|
latency or failed query information.
|
|
|
|
Returns:
|
|
None if not found, and if found:
|
|
- ActorState for actors
|
|
- PlacementGroupState for placement groups
|
|
- NodeState for nodes
|
|
- WorkerState for workers
|
|
- TaskState for tasks
|
|
- JobState for jobs
|
|
|
|
Empty list for objects if not found, or list of ObjectState for objects
|
|
|
|
Raises:
|
|
Exception: This doesn't catch any exceptions raised when the underlying request
|
|
call raises exceptions. For example, it could raise `requests.Timeout`
|
|
when timeout occurs.
|
|
ValueError:
|
|
if the resource could not be GET by id, i.e. jobs and runtime-envs.
|
|
|
|
"""
|
|
# TODO(rickyyx): Make GET not using filters on list operation
|
|
params = self._make_param(options)
|
|
|
|
RESOURCE_ID_KEY_NAME = {
|
|
StateResource.NODES: "node_id",
|
|
StateResource.ACTORS: "actor_id",
|
|
StateResource.PLACEMENT_GROUPS: "placement_group_id",
|
|
StateResource.WORKERS: "worker_id",
|
|
StateResource.TASKS: "task_id",
|
|
StateResource.OBJECTS: "object_id",
|
|
StateResource.JOBS: "submission_id",
|
|
}
|
|
if resource not in RESOURCE_ID_KEY_NAME:
|
|
raise ValueError(f"Can't get {resource.name} by id.")
|
|
|
|
params["filter_keys"] = [RESOURCE_ID_KEY_NAME[resource]]
|
|
params["filter_predicates"] = ["="]
|
|
params["filter_values"] = [id]
|
|
params["detail"] = True
|
|
endpoint = f"/api/v0/{resource.value}"
|
|
|
|
list_api_response = self._make_http_get_request(
|
|
endpoint=endpoint,
|
|
params=params,
|
|
timeout=options.timeout,
|
|
_explain=_explain,
|
|
)
|
|
result = list_api_response["result"]
|
|
|
|
# Empty result
|
|
if len(result) == 0:
|
|
return None
|
|
|
|
result = [dict_to_state(d, resource) for d in result]
|
|
if resource == StateResource.OBJECTS:
|
|
# NOTE(rickyyx):
|
|
# There might be multiple object entries for a single object id
|
|
# because a single object could be referenced at different places
|
|
# e.g. pinned as local variable, used as parameter
|
|
return result
|
|
|
|
if resource == StateResource.TASKS:
|
|
# There might be multiple task attempts given a task id due to
|
|
# task retries.
|
|
if len(result) == 1:
|
|
return result[0]
|
|
return result
|
|
|
|
# For the rest of the resources, there should only be a single entry
|
|
# for a particular id.
|
|
assert len(result) == 1
|
|
return result[0]
|
|
|
|
def _print_api_warning(
|
|
self,
|
|
resource: StateResource,
|
|
api_response: dict,
|
|
warn_data_source_not_available: bool = True,
|
|
warn_data_truncation: bool = True,
|
|
warn_limit: bool = True,
|
|
warn_server_side_warnings: bool = True,
|
|
):
|
|
"""Print the API warnings.
|
|
|
|
Args:
|
|
resource: Resource names, i.e. 'jobs', 'actors', 'nodes',
|
|
see `StateResource` for details.
|
|
api_response: The dictionarified `ListApiResponse` or `SummaryApiResponse`.
|
|
warn_data_source_not_available: Warn when some data sources
|
|
are not available.
|
|
warn_data_truncation: Warn when results were truncated at
|
|
the data source.
|
|
warn_limit: Warn when results were limited.
|
|
warn_server_side_warnings: Warn when the server side generates warnings
|
|
(E.g., when callsites not enabled for listing objects)
|
|
"""
|
|
# Print warnings if anything was given.
|
|
if warn_data_source_not_available:
|
|
warning_msgs = api_response.get("partial_failure_warning", None)
|
|
if warning_msgs:
|
|
warnings.warn(warning_msgs)
|
|
|
|
if warn_data_truncation:
|
|
# Print warnings if data is truncated at the data source.
|
|
num_after_truncation = api_response["num_after_truncation"]
|
|
total = api_response["total"]
|
|
if total > num_after_truncation:
|
|
# NOTE(rickyyx): For now, there's not much users
|
|
# could do (neither can we), with hard truncation.
|
|
# Unless we allow users to set a higher
|
|
# `RAY_MAX_LIMIT_FROM_DATA_SOURCE`, the data will
|
|
# always be truncated at the data source.
|
|
warnings.warn(
|
|
(
|
|
"The returned data may contain incomplete result. "
|
|
f"{num_after_truncation} ({total} total from the cluster) "
|
|
f"{resource.value} are retrieved from the data source. "
|
|
f"{total - num_after_truncation} entries have been truncated. "
|
|
f"Max of {num_after_truncation} entries are retrieved "
|
|
"from data source to prevent over-sized payloads."
|
|
),
|
|
)
|
|
|
|
if warn_limit:
|
|
# Print warnings if return data is limited at the API server due to
|
|
# limit enforced at the server side
|
|
num_filtered = api_response["num_filtered"]
|
|
data = api_response["result"]
|
|
if num_filtered > len(data):
|
|
warnings.warn(
|
|
(
|
|
f"Limit last {len(data)} entries "
|
|
f"(Total {num_filtered}). Use `--filter` to reduce "
|
|
"the amount of data to return or "
|
|
"setting a higher limit with `--limit` to see all data. "
|
|
),
|
|
)
|
|
|
|
if warn_server_side_warnings:
|
|
# Print the additional warnings.
|
|
warnings_to_print = api_response.get("warnings", [])
|
|
if warnings_to_print:
|
|
for warning_to_print in warnings_to_print:
|
|
warnings.warn(warning_to_print)
|
|
|
|
def _raise_on_missing_output(self, resource: StateResource, api_response: dict):
|
|
"""Raise an exception when the API resopnse contains a missing output.
|
|
|
|
Output can be missing if (1) Failures on some of data source queries (e.g.,
|
|
`ray list tasks` queries all raylets, and if some of queries fail, it will
|
|
contain missing output. If all queries fail, it will just fail). (2) Data
|
|
is truncated because the output is too large.
|
|
|
|
Args:
|
|
resource: Resource names, i.e. 'jobs', 'actors', 'nodes',
|
|
see `StateResource` for details.
|
|
api_response: The dictionarified `ListApiResponse` or `SummaryApiResponse`.
|
|
"""
|
|
# Raise an exception if there are partial failures that cause missing output.
|
|
warning_msgs = api_response.get("partial_failure_warning", None)
|
|
if warning_msgs:
|
|
raise RayStateApiException(
|
|
f"Failed to retrieve all {resource.value} from the cluster because"
|
|
"they are not reachable due to query failures to the data sources. "
|
|
"To avoid raising an exception and allow having missing output, "
|
|
"set `raise_on_missing_output=False`. "
|
|
)
|
|
# Raise an exception is there is data truncation that cause missing output.
|
|
total = api_response["total"]
|
|
num_after_truncation = api_response["num_after_truncation"]
|
|
|
|
if total != num_after_truncation:
|
|
raise RayStateApiException(
|
|
f"Failed to retrieve all {total} {resource.value} from the cluster "
|
|
"because they are not reachable due to data truncation. It happens "
|
|
"when the returned data is too large "
|
|
# When the data is truncated, the truncation
|
|
# threshold == num_after_truncation. We cannot set this to env
|
|
# var because the CLI side might not have the correct env var.
|
|
f"(> {num_after_truncation}) "
|
|
"To avoid raising an exception and allow having missing output, "
|
|
"set `raise_on_missing_output=False`. "
|
|
)
|
|
|
|
def list(
|
|
self,
|
|
resource: StateResource,
|
|
options: ListApiOptions,
|
|
raise_on_missing_output: bool,
|
|
_explain: bool = False,
|
|
) -> List[
|
|
Union[
|
|
ActorState,
|
|
JobState,
|
|
NodeState,
|
|
TaskState,
|
|
ObjectState,
|
|
PlacementGroupState,
|
|
RuntimeEnvState,
|
|
WorkerState,
|
|
ClusterEventState,
|
|
]
|
|
]:
|
|
"""List resources states
|
|
|
|
Args:
|
|
resource: Resource names, i.e. 'jobs', 'actors', 'nodes',
|
|
see `StateResource` for details.
|
|
options: List options. See `ListApiOptions` for details.
|
|
raise_on_missing_output: When True, raise an exception if the output
|
|
is incomplete. Output can be incomplete if
|
|
(1) there's a partial network failure when the source is distributed.
|
|
(2) data is truncated because it is too large.
|
|
Set it to False to avoid throwing an exception on missing data.
|
|
_explain: Print the API information such as API
|
|
latency or failed query information.
|
|
|
|
Returns:
|
|
A list of queried result from `ListApiResponse`,
|
|
|
|
Raises:
|
|
Exception: This doesn't catch any exceptions raised when the
|
|
underlying request call raises exceptions. For example, it could
|
|
raise `requests.Timeout` when timeout occurs.
|
|
|
|
"""
|
|
if options.has_conflicting_filters():
|
|
# return early with empty list when there are conflicting filters
|
|
return []
|
|
|
|
endpoint = f"/api/v0/{resource.value}"
|
|
params = self._make_param(options)
|
|
list_api_response = self._make_http_get_request(
|
|
endpoint=endpoint,
|
|
params=params,
|
|
timeout=options.timeout,
|
|
_explain=_explain,
|
|
)
|
|
if raise_on_missing_output:
|
|
self._raise_on_missing_output(resource, list_api_response)
|
|
if _explain:
|
|
self._print_api_warning(resource, list_api_response)
|
|
return [dict_to_state(d, resource) for d in list_api_response["result"]]
|
|
|
|
def summary(
|
|
self,
|
|
resource: SummaryResource,
|
|
*,
|
|
options: SummaryApiOptions,
|
|
raise_on_missing_output: bool,
|
|
_explain: bool = False,
|
|
) -> Dict:
|
|
"""Summarize resources states
|
|
|
|
Args:
|
|
resource: Resource names,
|
|
see `SummaryResource` for details.
|
|
options: summary options. See `SummaryApiOptions` for details.
|
|
raise_on_missing_output: Raise an exception if the output has missing data.
|
|
Output can have missing data if (1) there's a partial network failure
|
|
when the source is distributed. (2) data is truncated
|
|
because it is too large.
|
|
_explain: Print the API information such as API
|
|
latency or failed query information.
|
|
|
|
Returns:
|
|
A dictionary of queried result from `SummaryApiResponse`.
|
|
|
|
Raises:
|
|
Exception: This doesn't catch any exceptions raised when the
|
|
underlying request call raises exceptions. For example, it could
|
|
raise `requests.Timeout` when timeout occurs.
|
|
"""
|
|
params = {"timeout": options.timeout}
|
|
endpoint = f"/api/v0/{resource.value}/summarize"
|
|
summary_api_response = self._make_http_get_request(
|
|
endpoint=endpoint,
|
|
params=params,
|
|
timeout=options.timeout,
|
|
_explain=_explain,
|
|
)
|
|
if raise_on_missing_output:
|
|
self._raise_on_missing_output(resource, summary_api_response)
|
|
if _explain:
|
|
# There's no limit applied to summary, so we shouldn't warn.
|
|
self._print_api_warning(resource, summary_api_response, warn_limit=False)
|
|
return summary_api_response["result"]["node_id_to_summary"]
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_actor(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[ActorState]:
|
|
"""Get an actor by id.
|
|
|
|
Args:
|
|
id: Id of the actor
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state API requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if actor not found, or
|
|
:class:`ActorState <ray.util.state.common.ActorState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.ACTORS, id, GetApiOptions(timeout=timeout), _explain=_explain
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_job(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[JobState]:
|
|
"""Get a submission job detail by id.
|
|
|
|
Args:
|
|
id: Submission ID obtained from job API.
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state API requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if job not found, or
|
|
:class:`JobState <ray.util.state.common.JobState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.JOBS,
|
|
id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_placement_group(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[PlacementGroupState]:
|
|
"""Get a placement group by id.
|
|
|
|
Args:
|
|
id: Id of the placement group
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if actor not found, or
|
|
:class:`~ray.util.state.common.PlacementGroupState`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.PLACEMENT_GROUPS,
|
|
id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_node(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[NodeState]:
|
|
"""Get a node by id.
|
|
|
|
Args:
|
|
id: Id of the node.
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if actor not found, or
|
|
:class:`NodeState <ray.util.state.common.NodeState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI is failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.NODES,
|
|
id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_worker(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[WorkerState]:
|
|
"""Get a worker by id.
|
|
|
|
Args:
|
|
id: Id of the worker
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if actor not found, or
|
|
:class:`WorkerState <ray.util.state.common.WorkerState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.WORKERS,
|
|
id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_task(
|
|
id: Union[str, "ray.ObjectRef"],
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> Optional[TaskState]:
|
|
"""Get task attempts of a task by id.
|
|
|
|
Args:
|
|
id: String id of the task or ObjectRef that corresponds to task
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
None if task not found, or a list of
|
|
:class:`~ray.util.state.common.TaskState`
|
|
from the task attempts.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
str_id: str
|
|
if isinstance(id, str):
|
|
str_id = id
|
|
else:
|
|
str_id = id.task_id().hex()
|
|
return StateApiClient(address=address).get(
|
|
StateResource.TASKS,
|
|
str_id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_objects(
|
|
id: str,
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
_explain: bool = False,
|
|
) -> List[ObjectState]:
|
|
"""Get objects by id.
|
|
|
|
There could be more than 1 entry returned since an object could be
|
|
referenced at different places.
|
|
|
|
Args:
|
|
id: Id of the object
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`~ray.util.state.common.ObjectState`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).get(
|
|
StateResource.OBJECTS,
|
|
id,
|
|
GetApiOptions(timeout=timeout),
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_actors(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[ActorState]:
|
|
"""List actors in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("id", "=", "abcd")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `ActorState`)
|
|
will be queried and returned. See
|
|
:class:`ActorState <ray.util.state.common.ActorState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`ActorState <ray.util.state.common.ActorState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.ACTORS,
|
|
options=ListApiOptions(
|
|
limit=limit,
|
|
timeout=timeout,
|
|
filters=filters,
|
|
detail=detail,
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_placement_groups(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[PlacementGroupState]:
|
|
"""List placement groups in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("state", "=", "abcd")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `PlacementGroupState`)
|
|
will be queried and returned. See
|
|
:class:`~ray.util.state.common.PlacementGroupState`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of :class:`~ray.util.state.common.PlacementGroupState`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.PLACEMENT_GROUPS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_nodes(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[NodeState]:
|
|
"""List nodes in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("node_name", "=", "abcd")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `NodeState`)
|
|
will be queried and returned. See
|
|
:class:`NodeState <ray.util.state.common.NodeState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of dictionarified
|
|
:class:`NodeState <ray.util.state.common.NodeState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.NODES,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_jobs(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[JobState]:
|
|
"""List jobs submitted to the cluster by :ref:`ray job submission <jobs-overview>`.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("status", "=", "abcd")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `JobState`)
|
|
will be queried and returned. See
|
|
:class:`JobState <ray.util.state.common.JobState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of dictionarified
|
|
:class:`JobState <ray.util.state.common.JobState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.JOBS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_workers(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[WorkerState]:
|
|
"""List workers in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("is_alive", "=", "True")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `WorkerState`)
|
|
will be queried and returned. See
|
|
:class:`WorkerState <ray.util.state.common.WorkerState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`WorkerState <ray.util.state.common.WorkerState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.WORKERS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_tasks(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[TaskState]:
|
|
"""List tasks in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("is_alive", "=", "True")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `TaskState`)
|
|
will be queried and returned. See
|
|
:class:`TaskState <ray.util.state.common.TaskState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`TaskState <ray.util.state.common.TaskState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.TASKS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_objects(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[ObjectState]:
|
|
"""List objects in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("ip", "=", "0.0.0.0")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `ObjectState`)
|
|
will be queried and returned. See
|
|
:class:`ObjectState <ray.util.state.common.ObjectState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`ObjectState <ray.util.state.common.ObjectState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.OBJECTS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_runtime_envs(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[RuntimeEnvState]:
|
|
"""List runtime environments in the cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
filters: List of tuples of filter key, predicate (=, or !=), and
|
|
the filter value. E.g., `("node_id", "=", "abcdef")`
|
|
String filter values are case-insensitive.
|
|
limit: Max number of entries returned by the state backend.
|
|
timeout: Max timeout value for the state APIs requests made.
|
|
detail: When True, more details info (specified in `RuntimeEnvState`)
|
|
will be queried and returned. See
|
|
:class:`RuntimeEnvState <ray.util.state.common.RuntimeEnvState>`.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
List of
|
|
:class:`RuntimeEnvState <ray.util.state.common.RuntimeEnvState>`.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).list(
|
|
StateResource.RUNTIME_ENVS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_cluster_events(
|
|
address: Optional[str] = None,
|
|
filters: Optional[List[Tuple[str, PredicateType, SupportedFilterType]]] = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
detail: bool = False,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> List[Dict]:
|
|
return StateApiClient(address=address).list(
|
|
StateResource.CLUSTER_EVENTS,
|
|
options=ListApiOptions(
|
|
limit=limit, timeout=timeout, filters=filters, detail=detail
|
|
),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
"""
|
|
Log APIs
|
|
"""
|
|
|
|
|
|
@DeveloperAPI
|
|
def get_log(
|
|
address: Optional[str] = None,
|
|
node_id: Optional[str] = None,
|
|
node_ip: Optional[str] = None,
|
|
filename: Optional[str] = None,
|
|
actor_id: Optional[str] = None,
|
|
task_id: Optional[str] = None,
|
|
pid: Optional[int] = None,
|
|
follow: bool = False,
|
|
tail: int = -1,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
suffix: str = "out",
|
|
encoding: Optional[str] = "utf-8",
|
|
errors: Optional[str] = "strict",
|
|
submission_id: Optional[str] = None,
|
|
attempt_number: int = 0,
|
|
_interval: Optional[float] = None,
|
|
filter_ansi_code: bool = False,
|
|
) -> Generator[str, None, None]:
|
|
"""Retrieve log file based on file name or some entities ids (pid, actor id, task id).
|
|
|
|
Examples:
|
|
.. testcode::
|
|
:hide:
|
|
|
|
import ray
|
|
import time
|
|
|
|
ray.shutdown()
|
|
ray.init()
|
|
|
|
# Wait for the node to be registered to the dashboard
|
|
time.sleep(5)
|
|
|
|
.. testcode::
|
|
|
|
import ray
|
|
from ray.util.state import get_log
|
|
|
|
# Node id could be retrieved from list_nodes() or ray.nodes()
|
|
node_id = ray.nodes()[0]["NodeID"]
|
|
filename = "raylet.out"
|
|
for l in get_log(filename=filename, node_id=node_id):
|
|
print(l)
|
|
|
|
.. testoutput::
|
|
:options: +MOCK
|
|
|
|
[2023-05-19 12:35:18,347 I 4259 68399276] (raylet) io_service_pool.cc:35: IOServicePool is running with 1 io_service.
|
|
[2023-05-19 12:35:18,348 I 4259 68399276] (raylet) store_runner.cc:32: Allowing the Plasma store to use up to 2.14748GB of memory.
|
|
[2023-05-19 12:35:18,348 I 4259 68399276] (raylet) store_runner.cc:48: Starting object store with directory /tmp, fallback /tmp/ray, and huge page support disabled
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If not specified, it will be retrieved from the initialized ray cluster.
|
|
node_id: Id of the node containing the logs .
|
|
node_ip: Ip of the node containing the logs. (At least one of the node_id and
|
|
node_ip have to be supplied when identifying a node).
|
|
filename: Name of the file (relative to the ray log directory) to be retrieved.
|
|
actor_id: Id of the actor if getting logs from an actor.
|
|
task_id: Id of the task if getting logs from a non concurrent actor.
|
|
For concurrent actor, please query the log with actor_id.
|
|
pid: PID of the worker if getting logs generated by a worker. When querying
|
|
with pid, either node_id or node_ip must be supplied.
|
|
follow: When set to True, logs will be streamed and followed.
|
|
tail: Number of lines to get from the end of the log file. Set to -1 for getting
|
|
the entire log.
|
|
timeout: Max timeout for requests made when getting the logs.
|
|
suffix: The suffix of the log file if query by id of tasks/workers/actors. Default to "out".
|
|
encoding: The encoding used to decode the content of the log file. Default is
|
|
"utf-8". Use None to get binary data directly.
|
|
errors: The error handling scheme to use for decoding errors. Default is
|
|
"strict". See https://docs.python.org/3/library/codecs.html#error-handlers
|
|
submission_id: Job submission ID if getting log from a submission job.
|
|
attempt_number: The attempt number of the task if getting logs generated by a task.
|
|
_interval: The interval in secs to print new logs when `follow=True`.
|
|
filter_ansi_code: A boolean flag for determining whether to filter ANSI escape codes.
|
|
Setting to `True` removes ANSI escape codes from the output. The default value is `False`.
|
|
|
|
Yields:
|
|
str: A chunk of log content. When ``encoding`` is ``None`` the raw bytes
|
|
chunk is yielded instead.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
|
|
api_server_url = ray_address_to_api_server_url(address)
|
|
media_type = "stream" if follow else "file"
|
|
|
|
options = GetLogOptions(
|
|
node_id=node_id,
|
|
node_ip=node_ip,
|
|
filename=filename,
|
|
actor_id=actor_id,
|
|
task_id=task_id,
|
|
pid=pid,
|
|
lines=tail,
|
|
interval=_interval,
|
|
media_type=media_type,
|
|
timeout=timeout,
|
|
suffix=suffix,
|
|
submission_id=submission_id,
|
|
attempt_number=attempt_number,
|
|
)
|
|
options_dict = {}
|
|
for field in fields(options):
|
|
option_val = getattr(options, field.name)
|
|
if option_val is not None:
|
|
options_dict[field.name] = option_val
|
|
|
|
if filter_ansi_code is not None:
|
|
options_dict["filter_ansi_code"] = filter_ansi_code
|
|
|
|
with requests.get(
|
|
f"{api_server_url}/api/v0/logs/{media_type}?"
|
|
f"{urllib.parse.urlencode(options_dict)}",
|
|
stream=True,
|
|
headers=get_auth_headers_if_auth_enabled({}),
|
|
) as r:
|
|
if r.status_code != 200:
|
|
raise RayStateApiException(r.text)
|
|
for chunk in r.iter_content(chunk_size=None):
|
|
if encoding is not None:
|
|
chunk = chunk.decode(encoding=encoding, errors=errors)
|
|
yield chunk
|
|
|
|
|
|
@DeveloperAPI
|
|
def list_logs(
|
|
address: Optional[str] = None,
|
|
node_id: Optional[str] = None,
|
|
node_ip: Optional[str] = None,
|
|
glob_filter: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
) -> Dict[str, List[str]]:
|
|
"""Listing log files available.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If not specified, it will be retrieved from the initialized ray cluster.
|
|
node_id: Id of the node containing the logs.
|
|
node_ip: Ip of the node containing the logs.
|
|
glob_filter: Name of the file (relative to the ray log directory) to be
|
|
retrieved. E.g. `glob_filter="*worker*"` for all worker logs.
|
|
timeout: Max timeout for requests made when getting the logs.
|
|
|
|
Returns:
|
|
A dictionary where the keys are log groups (e.g. gcs, raylet, worker), and
|
|
values are list of log filenames.
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data, or ConnectionError if
|
|
failed to resolve the ray address.
|
|
""" # noqa: E501
|
|
assert (
|
|
node_ip is not None or node_id is not None
|
|
), "At least one of node ip and node id is required"
|
|
|
|
api_server_url = ray_address_to_api_server_url(address)
|
|
|
|
if not glob_filter:
|
|
glob_filter = "*"
|
|
|
|
options_dict = {}
|
|
if node_ip:
|
|
options_dict["node_ip"] = node_ip
|
|
if node_id:
|
|
options_dict["node_id"] = node_id
|
|
if glob_filter:
|
|
options_dict["glob"] = glob_filter
|
|
options_dict["timeout"] = timeout
|
|
|
|
r = requests.get(
|
|
f"{api_server_url}/api/v0/logs?{urllib.parse.urlencode(options_dict)}",
|
|
headers=get_auth_headers_if_auth_enabled({}),
|
|
)
|
|
# TODO(rickyx): we could do better at error handling here.
|
|
r.raise_for_status()
|
|
|
|
response = r.json()
|
|
if response["result"] is False:
|
|
raise RayStateApiException(
|
|
"API server internal error. See dashboard.log file for more details. "
|
|
f"Error: {response['msg']}"
|
|
)
|
|
return response["data"]["result"]
|
|
|
|
|
|
"""
|
|
Summary APIs
|
|
"""
|
|
|
|
|
|
@DeveloperAPI
|
|
def summarize_tasks(
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> Dict:
|
|
"""Summarize the tasks in cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout for requests made when getting the states.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
Dictionarified
|
|
:class:`~ray.util.state.common.TaskSummaries`
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI is failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).summary(
|
|
SummaryResource.TASKS,
|
|
options=SummaryApiOptions(timeout=timeout),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def summarize_actors(
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> Dict:
|
|
"""Summarize the actors in cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout for requests made when getting the states.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
Dictionarified
|
|
:class:`~ray.util.state.common.ActorSummaries`
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).summary(
|
|
SummaryResource.ACTORS,
|
|
options=SummaryApiOptions(timeout=timeout),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|
|
|
|
|
|
@DeveloperAPI
|
|
def summarize_objects(
|
|
address: Optional[str] = None,
|
|
timeout: int = DEFAULT_RPC_TIMEOUT,
|
|
raise_on_missing_output: bool = True,
|
|
_explain: bool = False,
|
|
) -> Dict:
|
|
"""Summarize the objects in cluster.
|
|
|
|
Args:
|
|
address: Ray bootstrap address, could be `auto`, `localhost:6379`.
|
|
If None, it will be resolved automatically from an initialized ray.
|
|
timeout: Max timeout for requests made when getting the states.
|
|
raise_on_missing_output: When True, exceptions will be raised if
|
|
there is missing data due to truncation/data source unavailable.
|
|
_explain: Print the API information such as API latency or
|
|
failed query information.
|
|
|
|
Returns:
|
|
Dictionarified :class:`~ray.util.state.common.ObjectSummaries`
|
|
|
|
Raises:
|
|
RayStateApiException: if the CLI failed to query the data.
|
|
""" # noqa: E501
|
|
return StateApiClient(address=address).summary(
|
|
SummaryResource.OBJECTS,
|
|
options=SummaryApiOptions(timeout=timeout),
|
|
raise_on_missing_output=raise_on_missing_output,
|
|
_explain=_explain,
|
|
)
|