wehub-resource-sync
381 lines
15 KiB
Python
381 lines
15 KiB
Python
import asyncio
|
|
import inspect
|
|
import logging
|
|
from types import FunctionType
|
|
from typing import Any, Dict, List, Tuple, Union
|
|
|
|
from pydantic import BaseModel
|
|
|
|
import ray
|
|
from ray import ObjectRef
|
|
from ray._common.usage import usage_lib
|
|
from ray.actor import ActorHandle
|
|
from ray.serve._private.client import ServeControllerClient
|
|
from ray.serve._private.constants import (
|
|
HTTP_PROXY_TIMEOUT,
|
|
SERVE_LOGGER_NAME,
|
|
SERVE_NAMESPACE,
|
|
)
|
|
from ray.serve._private.default_impl import get_controller_impl
|
|
from ray.serve.config import (
|
|
ControllerOptions,
|
|
HTTPOptions,
|
|
ProxyLocation,
|
|
gRPCOptions,
|
|
)
|
|
from ray.serve.context import (
|
|
_check_cached_client_alive,
|
|
_get_global_client,
|
|
_set_global_client,
|
|
)
|
|
from ray.serve.deployment import Application
|
|
from ray.serve.exceptions import RayServeException
|
|
from ray.serve.schema import LoggingConfig
|
|
|
|
logger = logging.getLogger(SERVE_LOGGER_NAME)
|
|
|
|
|
|
def _coerce_controller_options(
|
|
controller_options: Union[None, dict, ControllerOptions],
|
|
) -> ControllerOptions:
|
|
"""Normalize an optional dict / model into a validated ControllerOptions."""
|
|
if controller_options is None:
|
|
return ControllerOptions()
|
|
if isinstance(controller_options, ControllerOptions):
|
|
return controller_options
|
|
if isinstance(controller_options, dict):
|
|
return ControllerOptions.model_validate(controller_options)
|
|
raise TypeError(
|
|
"controller_options must be a dict, ControllerOptions, or None; got "
|
|
f"{type(controller_options).__name__}."
|
|
)
|
|
|
|
|
|
def _check_http_options(
|
|
client: ServeControllerClient, http_options: Union[dict, HTTPOptions]
|
|
) -> None:
|
|
if http_options:
|
|
client_http_options = client.http_config
|
|
new_http_options = (
|
|
http_options
|
|
if isinstance(http_options, HTTPOptions)
|
|
else HTTPOptions.model_validate(http_options)
|
|
)
|
|
different_fields = []
|
|
all_http_option_fields = new_http_options.__dict__
|
|
for field in all_http_option_fields:
|
|
if getattr(new_http_options, field) != getattr(client_http_options, field):
|
|
different_fields.append(field)
|
|
|
|
if len(different_fields):
|
|
logger.warning(
|
|
"The new client HTTP config differs from the existing one "
|
|
f"in the following fields: {different_fields}. "
|
|
"The new HTTP config is ignored."
|
|
)
|
|
|
|
|
|
def _create_controller_and_proxy_refs(
|
|
http_options: Union[None, dict, HTTPOptions],
|
|
grpc_options: Union[None, dict, gRPCOptions],
|
|
global_logging_config: Union[None, dict, LoggingConfig],
|
|
controller_options: ControllerOptions,
|
|
proxy_location: Union[None, str, ProxyLocation] = None,
|
|
**kwargs,
|
|
) -> Tuple[ActorHandle, List[ObjectRef]]:
|
|
"""Create the detached ServeController actor in the caller process.
|
|
|
|
Runs everything the old ``_start_controller`` remote task did, but inline:
|
|
ray.init if needed, arg validation, controller actor creation, and resolving
|
|
``get_proxies``. Returns the controller handle (owned locally by the caller,
|
|
following normal ref-counting — no cross-process transfer) plus the list of
|
|
proxy-readiness ObjectRefs the caller must still wait on.
|
|
|
|
Caller resolves ``proxy_ready_refs`` synchronously (``ray.get``) or
|
|
asynchronously (``await asyncio.gather``). Splitting the wait out of this
|
|
helper is what lets sync and async share one creation path without
|
|
duplication (per @edoakes's suggestion on #63597).
|
|
|
|
Parameters are same as ray.serve._private.api.serve_start(), except
|
|
``controller_options`` must already be a validated ``ControllerOptions``
|
|
(callers coerce eagerly so a bad value raises locally, not from a task).
|
|
"""
|
|
|
|
# Initialize ray if needed.
|
|
ray._private.worker.global_worker._filter_logs_by_job = False
|
|
if not ray.is_initialized():
|
|
ray.init(namespace=SERVE_NAMESPACE)
|
|
|
|
# Legacy http proxy actor check
|
|
http_deprecated_args = ["http_host", "http_port", "http_middlewares"]
|
|
for key in http_deprecated_args:
|
|
if key in kwargs:
|
|
raise ValueError(
|
|
f"{key} is deprecated, please use serve.start(http_options="
|
|
f'{{"{key}": {kwargs[key]}}}) instead.'
|
|
)
|
|
|
|
if isinstance(http_options, dict):
|
|
http_options = HTTPOptions.model_validate(http_options)
|
|
if http_options is None:
|
|
http_options = HTTPOptions()
|
|
|
|
proxy_location = ProxyLocation._normalize(proxy_location)
|
|
|
|
if isinstance(grpc_options, dict):
|
|
grpc_options = gRPCOptions(**grpc_options)
|
|
|
|
if global_logging_config is None:
|
|
global_logging_config = LoggingConfig()
|
|
elif isinstance(global_logging_config, dict):
|
|
global_logging_config = LoggingConfig(**global_logging_config)
|
|
|
|
controller_impl = get_controller_impl(controller_options=controller_options)
|
|
controller = controller_impl.remote(
|
|
http_options=http_options,
|
|
grpc_options=grpc_options,
|
|
global_logging_config=global_logging_config,
|
|
proxy_location=proxy_location,
|
|
)
|
|
|
|
proxy_handles = ray.get(controller.get_proxies.remote())
|
|
proxy_ready_refs = (
|
|
[handle.ready.remote() for handle in proxy_handles.values()]
|
|
if len(proxy_handles) > 0
|
|
else []
|
|
)
|
|
return controller, proxy_ready_refs
|
|
|
|
|
|
async def serve_start_async(
|
|
http_options: Union[None, dict, HTTPOptions] = None,
|
|
grpc_options: Union[None, dict, gRPCOptions] = None,
|
|
global_logging_config: Union[None, dict, LoggingConfig] = None,
|
|
controller_options: Union[None, dict, ControllerOptions] = None,
|
|
proxy_location: Union[None, str, ProxyLocation] = None,
|
|
**kwargs,
|
|
) -> ServeControllerClient:
|
|
"""Initialize a serve instance asynchronously.
|
|
|
|
This function is not thread-safe. The caller should maintain the async lock in order
|
|
to start the serve instance asynchronously.
|
|
|
|
This function has the same functionality as ray.serve._private.api.serve_start().
|
|
|
|
Parameters & Returns are same as ray.serve._private.api.serve_start().
|
|
"""
|
|
|
|
usage_lib.record_library_usage("serve")
|
|
|
|
# Validate eagerly in the caller so a bad ``controller_options`` raises
|
|
# here, at the call site, rather than after a round-trip into the helper.
|
|
controller_options = _coerce_controller_options(controller_options)
|
|
|
|
client, _ = _check_cached_client_alive()
|
|
if client is None:
|
|
try:
|
|
client = _get_global_client()
|
|
except RayServeException:
|
|
client = None
|
|
if client is not None:
|
|
logger.info(
|
|
f'Connecting to existing Serve app in namespace "{SERVE_NAMESPACE}".'
|
|
" New http_options/controller_options will not be applied."
|
|
)
|
|
if http_options:
|
|
_check_http_options(client, http_options)
|
|
return client
|
|
|
|
# Run the blocking controller-creation helper in a worker thread so its
|
|
# ray.get(controller.get_proxies.remote()) does not stall this event loop.
|
|
# serve_start_async exists (vs serve_start) precisely to keep long-lived
|
|
# callers like the Dashboard Agent responsive while Serve starts up.
|
|
# Safe because the helper only touches ray.init/ray.get/actor.remote, all
|
|
# thread-safe via the CoreWorker C++ API, and the Dashboard Agent uses sync
|
|
# ray.init (so the helper's ray.init branch is skipped). If the dashboard
|
|
# ever switched to async-mode ray.init this would need revisiting.
|
|
controller, proxy_ready_refs = await asyncio.to_thread(
|
|
_create_controller_and_proxy_refs,
|
|
http_options,
|
|
grpc_options,
|
|
global_logging_config,
|
|
controller_options,
|
|
proxy_location,
|
|
**kwargs,
|
|
)
|
|
|
|
if proxy_ready_refs:
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.gather(*proxy_ready_refs),
|
|
timeout=HTTP_PROXY_TIMEOUT,
|
|
)
|
|
except asyncio.TimeoutError:
|
|
raise TimeoutError(
|
|
f"HTTP proxies not available after {HTTP_PROXY_TIMEOUT}s."
|
|
)
|
|
|
|
client = ServeControllerClient(controller)
|
|
_set_global_client(client)
|
|
logger.info(f'Started Serve in namespace "{SERVE_NAMESPACE}".')
|
|
return client
|
|
|
|
|
|
def serve_start(
|
|
http_options: Union[None, dict, HTTPOptions] = None,
|
|
grpc_options: Union[None, dict, gRPCOptions] = None,
|
|
global_logging_config: Union[None, dict, LoggingConfig] = None,
|
|
controller_options: Union[None, dict, ControllerOptions] = None,
|
|
proxy_location: Union[None, str, ProxyLocation] = None,
|
|
**kwargs,
|
|
) -> ServeControllerClient:
|
|
"""Initialize a serve instance.
|
|
|
|
By default, the instance will be scoped to the lifetime of the returned
|
|
Client object (or when the script exits). This is
|
|
only relevant if connecting to a long-running Ray cluster (e.g., with
|
|
ray.init(address="auto") or ray.init("ray://<remote_addr>")).
|
|
|
|
Args:
|
|
http_options: Configuration options for HTTP proxy. You can pass in a
|
|
dictionary or HTTPOptions object with fields:
|
|
|
|
- host(str, None): Host for HTTP servers to listen on. Defaults to
|
|
localhost. To expose Serve publicly, you probably want to set
|
|
this to "0.0.0.0" for IPv4 or "::" for IPv6.
|
|
- port(int): Port for HTTP server. Defaults to 8000.
|
|
- root_path(str): Root path to mount the serve application
|
|
(for example, "/serve"). All deployment routes will be prefixed
|
|
with this path. Defaults to "".
|
|
- middlewares(list): A list of Starlette middlewares that will be
|
|
applied to the HTTP servers in the cluster. Defaults to [].
|
|
- location(str, serve.config.ProxyLocation): The deployment
|
|
location of HTTP servers:
|
|
|
|
- "HeadOnly": start one HTTP server on the head node. Serve
|
|
assumes the head node is the node you executed serve.start
|
|
on. This is the default.
|
|
- "EveryNode": start one HTTP server per node.
|
|
- "Disabled" (or legacy "NoServer") or None: disable HTTP server.
|
|
- num_cpus (int): The number of CPU cores to reserve for each
|
|
internal Serve HTTP proxy actor. Defaults to 0.
|
|
grpc_options: Configuration options for gRPC proxy.
|
|
You can pass in a gRPCOptions object with fields:
|
|
|
|
- port(int): Port for gRPC server. Defaults to 9000.
|
|
- grpc_servicer_functions(list): List of import paths for gRPC
|
|
`add_servicer_to_server` functions to add to Serve's gRPC proxy.
|
|
Default empty list, meaning not to start the gRPC server.
|
|
global_logging_config: Optional ``LoggingConfig`` (or dict) applied as
|
|
the default logging configuration for the Serve controller and all
|
|
proxies/replicas in this Serve instance.
|
|
controller_options: Optional ``ControllerOptions`` (or dict) for the
|
|
Serve controller actor. Currently only ``runtime_env.env_vars``
|
|
is honored; see ``ray.serve.config.ControllerOptions``. Only
|
|
applied on first controller creation -- ignored if the controller
|
|
is already running in this Ray cluster (a log line is emitted).
|
|
proxy_location: Where to run proxies that handle ingress traffic to
|
|
the cluster. See ``ProxyLocation`` for supported options. Defaults
|
|
to ``EveryNode`` when unspecified. An explicit (deprecated)
|
|
``HTTPOptions.location`` overrides this.
|
|
**kwargs: Reserved for forwarding to internal controller-start hooks;
|
|
no public keys are currently supported and unknown keys may raise.
|
|
|
|
Returns:
|
|
A ``ServeControllerClient`` connected to the Serve controller (either
|
|
newly started or an already-running one in this Ray cluster).
|
|
"""
|
|
|
|
usage_lib.record_library_usage("serve")
|
|
|
|
controller_options = _coerce_controller_options(controller_options)
|
|
|
|
client, _ = _check_cached_client_alive()
|
|
if client is None:
|
|
try:
|
|
client = _get_global_client()
|
|
except RayServeException:
|
|
client = None
|
|
if client is not None:
|
|
logger.info(
|
|
f'Connecting to existing Serve app in namespace "{SERVE_NAMESPACE}".'
|
|
" New http_options/controller_options will not be applied."
|
|
)
|
|
if http_options:
|
|
_check_http_options(client, http_options)
|
|
return client
|
|
|
|
controller, proxy_ready_refs = _create_controller_and_proxy_refs(
|
|
http_options,
|
|
grpc_options,
|
|
global_logging_config,
|
|
controller_options,
|
|
proxy_location,
|
|
**kwargs,
|
|
)
|
|
|
|
if proxy_ready_refs:
|
|
try:
|
|
ray.get(proxy_ready_refs, timeout=HTTP_PROXY_TIMEOUT)
|
|
except ray.exceptions.GetTimeoutError:
|
|
raise TimeoutError(
|
|
f"HTTP proxies not available after {HTTP_PROXY_TIMEOUT}s."
|
|
)
|
|
|
|
client = ServeControllerClient(controller)
|
|
_set_global_client(client)
|
|
logger.info(f'Started Serve in namespace "{SERVE_NAMESPACE}".')
|
|
return client
|
|
|
|
|
|
def call_user_app_builder_with_args_if_necessary(
|
|
builder: Union[Application, FunctionType],
|
|
args: Dict[str, Any],
|
|
) -> Application:
|
|
"""Calls a user-provided function that returns Serve application.
|
|
|
|
If an Application object is passed, this is a no-op.
|
|
|
|
Else, we validate the signature of the function, convert the args dictionary to
|
|
the user-annotated Pydantic model if provided, and call the function.
|
|
|
|
The output of the function is returned (must be an Application).
|
|
"""
|
|
if isinstance(builder, Application):
|
|
if len(args) > 0:
|
|
raise ValueError(
|
|
"Arguments can only be passed to an application builder function, "
|
|
"not an already built application."
|
|
)
|
|
return builder
|
|
elif not isinstance(builder, FunctionType):
|
|
raise TypeError(
|
|
"Expected a built Serve application or an application builder function "
|
|
f"but got: {type(builder)}."
|
|
)
|
|
|
|
# Check that the builder only takes a single argument.
|
|
# TODO(edoakes): we may want to loosen this to allow optional kwargs in the future.
|
|
signature = inspect.signature(builder)
|
|
if len(signature.parameters) != 1:
|
|
raise TypeError(
|
|
"Application builder functions should take exactly one parameter, "
|
|
"a dictionary containing the passed arguments."
|
|
)
|
|
|
|
# If the sole argument to the builder is a pydantic model, convert the args dict to
|
|
# that model. This will perform standard pydantic validation (e.g., raise an
|
|
# exception if required fields are missing).
|
|
param = signature.parameters[list(signature.parameters.keys())[0]]
|
|
if inspect.isclass(param.annotation) and issubclass(param.annotation, BaseModel):
|
|
args = param.annotation.model_validate(args)
|
|
|
|
app = builder(args)
|
|
if not isinstance(app, Application):
|
|
raise TypeError(
|
|
"Application builder functions must return an `Application` returned "
|
|
f"`from `Deployment.bind()`, but got: {type(app)}."
|
|
)
|
|
|
|
return app
|