wehub-resource-sync
2219 lines
89 KiB
Python
2219 lines
89 KiB
Python
import asyncio
|
|
import csv
|
|
import functools
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import string
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Set, Tuple
|
|
|
|
from jinja2 import Environment
|
|
|
|
import ray
|
|
from ray._common.network_utils import get_localhost_ip
|
|
from ray._common.utils import get_or_create_event_loop
|
|
from ray.serve._private.common import (
|
|
NodeId,
|
|
ReplicaID,
|
|
RequestMetadata,
|
|
RequestProtocol,
|
|
)
|
|
from ray.serve._private.constants import (
|
|
DRAINING_MESSAGE,
|
|
HEALTHY_MESSAGE,
|
|
NO_REPLICAS_MESSAGE,
|
|
NO_ROUTES_MESSAGE,
|
|
PROXY_MIN_DRAINING_PERIOD_S,
|
|
RAY_SERVE_ENABLE_HAPROXY_OPTIMIZED_CONFIG,
|
|
RAY_SERVE_HAPROXY_BALANCE_ALGORITHM,
|
|
RAY_SERVE_HAPROXY_BINARY_PATH,
|
|
RAY_SERVE_HAPROXY_BROADCAST_COALESCE_S,
|
|
RAY_SERVE_HAPROXY_CONFIG_FILE_LOC,
|
|
RAY_SERVE_HAPROXY_H2_BE_INITIAL_WINDOW_SIZE,
|
|
RAY_SERVE_HAPROXY_H2_BE_MAX_CONCURRENT_STREAMS,
|
|
RAY_SERVE_HAPROXY_H2_FE_INITIAL_WINDOW_SIZE,
|
|
RAY_SERVE_HAPROXY_H2_FE_MAX_CONCURRENT_STREAMS,
|
|
RAY_SERVE_HAPROXY_H2_MAX_FRAME_SIZE,
|
|
RAY_SERVE_HAPROXY_HARD_STOP_AFTER_S,
|
|
RAY_SERVE_HAPROXY_HEALTH_CHECK_DOWNINTER,
|
|
RAY_SERVE_HAPROXY_HEALTH_CHECK_FALL,
|
|
RAY_SERVE_HAPROXY_HEALTH_CHECK_FASTINTER,
|
|
RAY_SERVE_HAPROXY_HEALTH_CHECK_INTER,
|
|
RAY_SERVE_HAPROXY_HEALTH_CHECK_RISE,
|
|
RAY_SERVE_HAPROXY_INGRESS_REQUEST_ROUTER_BUFSIZE,
|
|
RAY_SERVE_HAPROXY_INGRESS_REQUEST_ROUTER_TIMEOUT_S,
|
|
RAY_SERVE_HAPROXY_INGRESS_RETRIES,
|
|
RAY_SERVE_HAPROXY_INGRESS_RETRY_ON,
|
|
RAY_SERVE_HAPROXY_INGRESS_TIMEOUT_SERVER_S,
|
|
RAY_SERVE_HAPROXY_LOG_TARGET,
|
|
RAY_SERVE_HAPROXY_MAXCONN,
|
|
RAY_SERVE_HAPROXY_METRICS_ENABLED,
|
|
RAY_SERVE_HAPROXY_METRICS_PORT,
|
|
RAY_SERVE_HAPROXY_METRICS_REPORT_INTERVAL_S,
|
|
RAY_SERVE_HAPROXY_METRICS_SOCKET_PATH,
|
|
RAY_SERVE_HAPROXY_NBTHREAD,
|
|
RAY_SERVE_HAPROXY_RETRIES,
|
|
RAY_SERVE_HAPROXY_RETRY_ON,
|
|
RAY_SERVE_HAPROXY_SERVER_STATE_BASE,
|
|
RAY_SERVE_HAPROXY_SERVER_STATE_FILE,
|
|
RAY_SERVE_HAPROXY_SOCKET_PATH,
|
|
RAY_SERVE_HAPROXY_STARTUP_TIMEOUT_S,
|
|
RAY_SERVE_HAPROXY_STATS_PORT,
|
|
RAY_SERVE_HAPROXY_TCP_NODELAY,
|
|
RAY_SERVE_HAPROXY_TIMEOUT_CLIENT_S,
|
|
RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S,
|
|
RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S,
|
|
RAY_SERVE_HAPROXY_TUNE_BUFSIZE,
|
|
RAY_SERVE_HAPROXY_UPDATE_LATENCY_BUCKETS_S,
|
|
RAY_SERVE_INGRESS_REQUEST_ROUTER_FORWARD_BODY,
|
|
RAY_SERVE_INGRESS_REQUEST_ROUTER_METRICS_ENABLED,
|
|
SERVE_CONTROLLER_NAME,
|
|
SERVE_LOGGER_NAME,
|
|
SERVE_NAMESPACE,
|
|
SERVE_SESSION_ID,
|
|
)
|
|
from ray.serve._private.haproxy_templates import (
|
|
HAPROXY_CONFIG_TEMPLATE,
|
|
HAPROXY_GRPC_HEALTHZ_RULES_TEMPLATE,
|
|
HAPROXY_HEALTHZ_RULES_TEMPLATE,
|
|
)
|
|
from ray.serve._private.logging_utils import get_component_logger_file_path
|
|
from ray.serve._private.long_poll import LongPollClient, LongPollNamespace
|
|
from ray.serve._private.proxy import (
|
|
ProxyActorInterface,
|
|
apply_per_node_port_overrides,
|
|
)
|
|
from ray.serve._private.utils import get_head_node_id, is_grpc_enabled
|
|
from ray.serve.config import HTTPOptions, gRPCOptions
|
|
from ray.serve.schema import (
|
|
LoggingConfig,
|
|
Target,
|
|
TargetGroup,
|
|
)
|
|
from ray.util import metrics
|
|
|
|
logger = logging.getLogger(SERVE_LOGGER_NAME)
|
|
|
|
|
|
@functools.cache
|
|
def _haproxy_fmt_literal(value: Any) -> str:
|
|
r"""Format a string as a double-quoted HAProxy log-format literal for a
|
|
`set-var-fmt` rule.
|
|
|
|
App, route, and deployment names can contain any character, so the value is
|
|
quoted and the characters that are special inside a HAProxy double-quoted
|
|
log-format string are escaped: `\` (escape), `"` (delimiter), `%`
|
|
(directive), and `$` (env var). HAProxy removes the escaping when it reads
|
|
the config, so the stored value is the original string.
|
|
|
|
Example: `a"b%c` becomes `"a\"b%%c"`.
|
|
"""
|
|
s = str(value)
|
|
s = s.replace("\\", "\\\\")
|
|
s = s.replace('"', '\\"')
|
|
s = s.replace("%", "%%")
|
|
s = s.replace("$", "\\$")
|
|
return '"' + s + '"'
|
|
|
|
|
|
def _load_lua_template() -> string.Template:
|
|
path = Path(__file__).parent / "ingress_request_router.lua.tmpl"
|
|
try:
|
|
return string.Template(path.read_text())
|
|
except FileNotFoundError as e:
|
|
raise FileNotFoundError(
|
|
f"{path.name} is missing. Ray Serve installation is incomplete."
|
|
) from e
|
|
|
|
|
|
def _routers_and_targets_by_backend(
|
|
backends: "List[BackendConfig]",
|
|
) -> "Tuple[Dict[str, ServerConfig], Dict[str, List[Tuple[str, str]]]]":
|
|
"""Per-backend router and replica map, restricted to backends with both.
|
|
|
|
Pick deterministically within each router pool to avoid the first-response
|
|
latency regression from cycling routers across requests.
|
|
"""
|
|
routers: Dict[str, ServerConfig] = {}
|
|
targets: Dict[str, List[Tuple[str, str]]] = {}
|
|
for backend in backends:
|
|
if not backend.ingress_request_router_servers:
|
|
continue
|
|
entries = [
|
|
(s.replica_id, s.name) for s in backend.servers if s.replica_id is not None
|
|
]
|
|
if not entries:
|
|
continue
|
|
# Host-first so co-located routers sort adjacent in debug output.
|
|
routers[backend.name] = min(
|
|
backend.ingress_request_router_servers, key=lambda s: (s.host, s.port)
|
|
)
|
|
targets[backend.name] = entries
|
|
return routers, targets
|
|
|
|
|
|
def _format_routers_lua(routers: "Dict[str, ServerConfig]") -> str:
|
|
"""Render {backend_name: ServerConfig} as a Lua table literal."""
|
|
body = ",\n".join(
|
|
f" [{json.dumps(name)}] = "
|
|
f"{{ host = {json.dumps(s.host)}, port = {s.port}, "
|
|
f"host_header = {json.dumps(f'{s.host}:{s.port}')} }}"
|
|
for name, s in routers.items()
|
|
)
|
|
return "{\n" + body + "\n}"
|
|
|
|
|
|
def _format_replica_targets_lua(
|
|
targets: "Dict[str, List[Tuple[str, str]]]",
|
|
) -> str:
|
|
"""Render {backend_name: {replica_id: server_name}} as nested Lua tables."""
|
|
backends_lua = []
|
|
for backend_name, entries in targets.items():
|
|
inner = ",\n".join(
|
|
f" [{json.dumps(rid)}] = {json.dumps(sname)}"
|
|
for rid, sname in entries
|
|
)
|
|
backends_lua.append(
|
|
f" [{json.dumps(backend_name)}] = " + "{\n" + inner + "\n }"
|
|
)
|
|
return "{\n" + ",\n".join(backends_lua) + "\n}"
|
|
|
|
|
|
def _write_if_changed(path: str, content: str) -> bool:
|
|
"""Write content to path only if it differs from what's there."""
|
|
try:
|
|
with open(path) as f:
|
|
if f.read() == content:
|
|
return False
|
|
except FileNotFoundError:
|
|
pass
|
|
with open(path, "w") as f:
|
|
f.write(content)
|
|
return True
|
|
|
|
|
|
def _tail_file(path: str, n_bytes: int = 4096) -> str:
|
|
"""Return up to the last ``n_bytes`` of ``path``, or "" on any error."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
f.seek(0, 2)
|
|
f.seek(max(0, f.tell() - n_bytes))
|
|
return f.read().decode("utf-8", errors="ignore").strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def get_haproxy_binary() -> str:
|
|
"""Return the path to the HAProxy binary.
|
|
|
|
Resolution order:
|
|
1. ``RAY_SERVE_HAPROXY_BINARY_PATH`` if explicitly set to an absolute path.
|
|
2. The binary bundled in the ``ray-haproxy`` package.
|
|
3. ``haproxy`` on the system PATH (fallback).
|
|
|
|
Raises ``FileNotFoundError`` if no usable binary is found.
|
|
"""
|
|
binary = _resolve_haproxy_binary()
|
|
logger.info(f"Using HAProxy binary: {binary}")
|
|
return binary
|
|
|
|
|
|
def _resolve_haproxy_binary() -> str:
|
|
# 1. Explicit RAY_SERVE_HAPROXY_BINARY_PATH override. An operator who sets
|
|
# this wants that specific binary, so it takes precedence over the bundled
|
|
# package.
|
|
if RAY_SERVE_HAPROXY_BINARY_PATH:
|
|
if os.path.isfile(RAY_SERVE_HAPROXY_BINARY_PATH) and os.access(
|
|
RAY_SERVE_HAPROXY_BINARY_PATH, os.X_OK
|
|
):
|
|
return RAY_SERVE_HAPROXY_BINARY_PATH
|
|
raise FileNotFoundError(
|
|
f"RAY_SERVE_HAPROXY_BINARY_PATH={RAY_SERVE_HAPROXY_BINARY_PATH!r} "
|
|
"does not point to an executable file."
|
|
)
|
|
|
|
# 2. Bundled binary from the ray-haproxy package, so ray[serve] installs work
|
|
# without extra configuration. Falls through if the package is missing or its
|
|
# binary is unusable.
|
|
try:
|
|
from ray_haproxy import get_haproxy_binary as _pip_binary
|
|
|
|
return _pip_binary()
|
|
except ImportError:
|
|
pass
|
|
except OSError:
|
|
pass
|
|
|
|
# 3. System PATH fallback.
|
|
system_haproxy = shutil.which("haproxy")
|
|
if system_haproxy:
|
|
return system_haproxy
|
|
|
|
raise FileNotFoundError(
|
|
"Could not find an HAProxy binary. "
|
|
"Install 'ray[serve]' on Linux for the bundled binary, "
|
|
"set RAY_SERVE_HAPROXY_BINARY_PATH to an executable, "
|
|
"or ensure 'haproxy' is on PATH."
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ServerStats:
|
|
"""Server statistics from HAProxy."""
|
|
|
|
backend: str # Which backend this server belongs to
|
|
server: str # Server name within the backend
|
|
status: str # Current status: "UP", "DOWN", "DRAIN", etc.
|
|
current_sessions: int # Active sessions (HAProxy 'scur')
|
|
queued: int # Queued requests (HAProxy 'qcur')
|
|
|
|
@property
|
|
def is_up(self) -> bool:
|
|
return self.status == "UP"
|
|
|
|
@property
|
|
def is_draining(self) -> bool:
|
|
return self.status in ["DRAIN", "NOLB"]
|
|
|
|
@property
|
|
def can_drain_safely(self) -> bool:
|
|
"""
|
|
Return True if the server can be drained safely based on the current load.
|
|
Safe to drain when:
|
|
- No current active sessions (0)
|
|
- No queued requests waiting
|
|
This ensures no active user sessions are disrupted during draining.
|
|
"""
|
|
return self.current_sessions == 0 and self.queued == 0
|
|
|
|
|
|
@dataclass
|
|
class HAProxyStats:
|
|
"""Complete HAProxy statistics with both individual server data and aggregate metrics."""
|
|
|
|
# Individual server statistics by backend and server name
|
|
backend_to_servers: Dict[str, Dict[str, ServerStats]] = field(default_factory=dict)
|
|
|
|
# Computed aggregate metrics (calculated from server data)
|
|
@property
|
|
def total_backends(self) -> int:
|
|
"""Total number of backends."""
|
|
return len(self.backend_to_servers)
|
|
|
|
@property
|
|
def total_servers(self) -> int:
|
|
"""Total number of servers across all backends."""
|
|
return sum(
|
|
len(backend_servers) for backend_servers in self.backend_to_servers.values()
|
|
)
|
|
|
|
@property
|
|
def active_servers(self) -> int:
|
|
"""Number of servers currently UP."""
|
|
return sum(
|
|
1
|
|
for backend_servers in self.backend_to_servers.values()
|
|
for server in backend_servers.values()
|
|
if server.is_up
|
|
)
|
|
|
|
@property
|
|
def draining_servers(self) -> int:
|
|
"""Number of servers currently draining."""
|
|
return sum(
|
|
1
|
|
for backend_servers in self.backend_to_servers.values()
|
|
for server in backend_servers.values()
|
|
if server.is_draining
|
|
)
|
|
|
|
@property
|
|
def total_active_sessions(self) -> int:
|
|
"""Sum of all active sessions across all servers."""
|
|
return sum(
|
|
server.current_sessions
|
|
for backend_servers in self.backend_to_servers.values()
|
|
for server in backend_servers.values()
|
|
)
|
|
|
|
@property
|
|
def total_queued_requests(self) -> int:
|
|
"""Sum of all queued requests across all servers."""
|
|
return sum(
|
|
server.queued
|
|
for backend_servers in self.backend_to_servers.values()
|
|
for server in backend_servers.values()
|
|
)
|
|
|
|
@property
|
|
def is_system_idle(self) -> bool:
|
|
"""Return True if the entire system has no active load."""
|
|
return self.total_active_sessions == 0 and self.total_queued_requests == 0
|
|
|
|
@property
|
|
def draining_progress_pct(self) -> float:
|
|
"""Return percentage of servers currently draining (0.0 to 100.0)."""
|
|
if self.total_servers == 0:
|
|
return 0.0
|
|
return (self.draining_servers / self.total_servers) * 100.0
|
|
|
|
|
|
@dataclass
|
|
class HealthRouteInfo:
|
|
"""Information regarding how proxy should respond to health and routes requests."""
|
|
|
|
healthy: bool = True
|
|
status: int = 200
|
|
health_message: str = HEALTHY_MESSAGE
|
|
routes_message: str = "{}"
|
|
routes_content_type: str = "application/json"
|
|
|
|
|
|
def build_grpc_healthcheck_request_hex(path: str) -> str:
|
|
"""Build the raw HTTP/2 bytes for a unary gRPC health-check request.
|
|
|
|
HAProxy's `http-check` cannot health-check a gRPC server: `http-check send`
|
|
truncates its body at the first NUL byte, and a gRPC length-prefixed message
|
|
always begins with the NUL compression-flag byte. With no message frame, the
|
|
server receives a message-less unary call and stalls until the check times
|
|
out. We therefore drive the check via `tcp-check send-binary`, which emits
|
|
exact bytes (NUL included), replaying a complete gRPC request: the HTTP/2
|
|
connection preface, an empty SETTINGS frame, a HEADERS frame (HPACK literal,
|
|
no Huffman/dynamic table), and a DATA frame carrying an empty message
|
|
(`00 00 00 00 00`) with END_STREAM. The peer responds with the real
|
|
`Healthz` body, which `tcp-check expect binary` matches against the healthy
|
|
message. Returns the request as a hex string for `tcp-check send-binary`.
|
|
"""
|
|
|
|
def hpack_str(s: str) -> bytes:
|
|
b = s.encode()
|
|
# Literal string, no Huffman; lengths here are always < 127.
|
|
assert len(b) < 127, f"header value too long for simple HPACK: {s!r}"
|
|
return bytes([len(b)]) + b
|
|
|
|
def hpack_header(name: str, value: str) -> bytes:
|
|
# 0x00 = literal header field without indexing, new name.
|
|
return b"\x00" + hpack_str(name) + hpack_str(value)
|
|
|
|
def h2_frame(frame_type: int, flags: int, stream_id: int, payload: bytes) -> bytes:
|
|
return (
|
|
len(payload).to_bytes(3, "big")
|
|
+ bytes([frame_type, flags])
|
|
+ stream_id.to_bytes(4, "big")
|
|
+ payload
|
|
)
|
|
|
|
preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
|
|
settings = h2_frame(0x4, 0x0, 0, b"") # empty SETTINGS
|
|
headers_block = b"".join(
|
|
[
|
|
hpack_header(":method", "POST"),
|
|
hpack_header(":scheme", "http"),
|
|
hpack_header(":path", path),
|
|
hpack_header(":authority", "ray-serve-grpc"),
|
|
hpack_header("te", "trailers"),
|
|
hpack_header("content-type", "application/grpc"),
|
|
]
|
|
)
|
|
headers = h2_frame(0x1, 0x4, 1, headers_block) # END_HEADERS (not END_STREAM)
|
|
data = h2_frame(0x0, 0x1, 1, b"\x00\x00\x00\x00\x00") # empty msg, END_STREAM
|
|
return (preface + settings + headers + data).hex()
|
|
|
|
|
|
@dataclass
|
|
class ServerConfig:
|
|
"""Configuration for a single server."""
|
|
|
|
name: str # Server identifier for HAProxy config
|
|
host: str # IP/hostname to connect to
|
|
port: int # Port to connect to
|
|
# Replica identifier returned by /internal/route.
|
|
replica_id: Optional[str] = None
|
|
|
|
def __str__(self) -> str:
|
|
return f"ServerConfig(name='{self.name}', host='{self.host}', port={self.port})"
|
|
|
|
def __repr__(self) -> str:
|
|
return str(self)
|
|
|
|
|
|
@dataclass
|
|
class BackendConfig:
|
|
"""Configuration for a single application backend."""
|
|
|
|
# Name of the target group.
|
|
name: str
|
|
|
|
# Path prefix for the target group. This will be used to route requests to the target group.
|
|
path_prefix: str
|
|
|
|
# Maximum time HAProxy will wait for a successful TCP connection to be established with the backend server.
|
|
timeout_connect_s: Optional[int] = None
|
|
|
|
# Maximum time that the backend server can be inactive while sending data back to HAProxy.
|
|
# This is also active during the initial response phase.
|
|
timeout_server_s: Optional[int] = None
|
|
|
|
# Maximum time that the client can be inactive while sending data to HAProxy.
|
|
# This is active during the initial request phase.
|
|
timeout_client_s: Optional[int] = None
|
|
timeout_http_request_s: Optional[int] = None
|
|
|
|
# Maximum time HAProxy will wait for a request in the queue.
|
|
timeout_queue_s: Optional[int] = None
|
|
|
|
# Maximum time HAProxy will keep the connection alive.
|
|
# This has to be the same or greater than the client side keep-alive timeout.
|
|
timeout_http_keep_alive_s: Optional[int] = None
|
|
|
|
# Control the inactivity timeout for established WebSocket connections.
|
|
# Without this setting, a WebSocket connection could be prematurely terminated by other,
|
|
# more general timeout settings like timeout client or timeout server,
|
|
# which are intended for the initial phases of a connection.
|
|
timeout_tunnel_s: Optional[int] = None
|
|
|
|
# The number of consecutive failed health checks that must occur before a service instance is marked as unhealthy
|
|
health_check_fall: Optional[int] = None
|
|
|
|
# Number of consecutive successful health checks required to mark an unhealthy service instance as healthy again
|
|
health_check_rise: Optional[int] = None
|
|
|
|
# Interval, or the amount of time, between each health check attempt
|
|
health_check_inter: Optional[str] = None
|
|
|
|
# The interval between two consecutive health checks when the server is in any of the transition states: UP - transitionally DOWN or DOWN - transitionally UP
|
|
health_check_fastinter: Optional[str] = None
|
|
|
|
# The interval between two consecutive health checks when the server is in the DOWN state
|
|
health_check_downinter: Optional[str] = None
|
|
|
|
# Endpoint path that the health check mechanism will send a request to.
|
|
http_health_check_path: Optional[str] = "/-/healthz"
|
|
grpc_health_check_path: Optional[str] = "/ray.serve.RayServeAPIService/Healthz"
|
|
|
|
# List of servers in this backend
|
|
servers: List[ServerConfig] = field(default_factory=list)
|
|
|
|
# Ingress request router servers. When populated, HAProxy Lua calls
|
|
# /internal/route on one of these to pick a data-plane replica.
|
|
ingress_request_router_servers: List[ServerConfig] = field(default_factory=list)
|
|
|
|
# The fallback server for this backend.
|
|
fallback_server: Optional[ServerConfig] = None
|
|
|
|
# The app name for this backend.
|
|
app_name: str = field(default_factory=str)
|
|
|
|
# Name of the app's ingress deployment. Rendered into the per-request HTTP
|
|
# metrics log line so HAProxy-emitted ingress metrics carry the deployment
|
|
# tag (matching what the Python proxy records). Empty when unknown.
|
|
ingress_deployment_name: str = field(default_factory=str)
|
|
|
|
# Protocol of this backend. HTTP backends are path-routed and use HTTP/1.1;
|
|
# gRPC backends are header-routed (by the `application` gRPC metadata) and
|
|
# speak HTTP/2 cleartext to replica direct-ingress gRPC servers.
|
|
protocol: RequestProtocol = RequestProtocol.HTTP
|
|
|
|
def build_health_check_config(self, global_config: "HAProxyConfig") -> dict:
|
|
"""Build health check configuration for HAProxy backend.
|
|
|
|
Returns a dict with:
|
|
- health_path: path for HTTP health checks (or None)
|
|
- default_server_directive: complete "default-server" line with all params
|
|
"""
|
|
# Resolve values: backend-specific overrides global defaults
|
|
fall = (
|
|
self.health_check_fall
|
|
if self.health_check_fall is not None
|
|
else global_config.health_check_fall
|
|
)
|
|
rise = (
|
|
self.health_check_rise
|
|
if self.health_check_rise is not None
|
|
else global_config.health_check_rise
|
|
)
|
|
inter = (
|
|
self.health_check_inter
|
|
if self.health_check_inter is not None
|
|
else global_config.health_check_inter
|
|
)
|
|
fastinter = (
|
|
self.health_check_fastinter
|
|
if self.health_check_fastinter is not None
|
|
else global_config.health_check_fastinter
|
|
)
|
|
downinter = (
|
|
self.health_check_downinter
|
|
if self.health_check_downinter is not None
|
|
else global_config.health_check_downinter
|
|
)
|
|
health_path = (
|
|
self.http_health_check_path
|
|
if self.http_health_check_path is not None
|
|
else global_config.http_health_check_path
|
|
)
|
|
if self.protocol == RequestProtocol.GRPC:
|
|
health_path = (
|
|
self.grpc_health_check_path
|
|
if self.grpc_health_check_path is not None
|
|
else global_config.grpc_health_check_path
|
|
)
|
|
|
|
# Build default-server directive
|
|
parts = []
|
|
|
|
# Add optional fastinter/downinter only if provided
|
|
if fastinter is not None:
|
|
parts.append(f"fastinter {fastinter}")
|
|
if downinter is not None:
|
|
parts.append(f"downinter {downinter}")
|
|
|
|
# Add required fall/rise/inter if any are set
|
|
if fall is not None:
|
|
parts.append(f"fall {fall}")
|
|
if rise is not None:
|
|
parts.append(f"rise {rise}")
|
|
if inter is not None:
|
|
parts.append(f"inter {inter}")
|
|
|
|
# Always add check at the end
|
|
parts.append("check")
|
|
|
|
default_server_directive = "default-server " + " ".join(parts)
|
|
|
|
result = {
|
|
"health_path": health_path,
|
|
"default_server_directive": default_server_directive,
|
|
}
|
|
|
|
# gRPC backends are health-checked via `tcp-check send-binary` (see
|
|
# build_grpc_healthcheck_request_hex for why `http-check` can't be used).
|
|
# Precompute the request bytes and the expected response marker so the
|
|
# template just emits them.
|
|
if self.protocol == RequestProtocol.GRPC:
|
|
result["grpc_healthcheck_request_hex"] = build_grpc_healthcheck_request_hex(
|
|
health_path
|
|
)
|
|
result["grpc_healthcheck_expect_hex"] = HEALTHY_MESSAGE.encode().hex()
|
|
|
|
return result
|
|
|
|
def __str__(self) -> str:
|
|
return f"BackendConfig(app_name='{self.app_name}', name='{self.name}', path_prefix='{self.path_prefix}', servers={self.servers}, ingress_request_router_servers={self.ingress_request_router_servers}, fallback_server={self.fallback_server}, protocol={self.protocol.value})"
|
|
|
|
def __repr__(self) -> str:
|
|
return str(self)
|
|
|
|
|
|
@dataclass
|
|
class HAProxyConfig:
|
|
"""Configuration for HAProxy."""
|
|
|
|
socket_path: str = RAY_SERVE_HAPROXY_SOCKET_PATH
|
|
server_state_base: str = RAY_SERVE_HAPROXY_SERVER_STATE_BASE
|
|
server_state_file: str = RAY_SERVE_HAPROXY_SERVER_STATE_FILE
|
|
# Enable HAProxy optimizations (server state persistence, etc.)
|
|
# Disabled by default to prevent test suite interference
|
|
enable_hap_optimization: bool = RAY_SERVE_ENABLE_HAPROXY_OPTIMIZED_CONFIG
|
|
maxconn: int = RAY_SERVE_HAPROXY_MAXCONN
|
|
nbthread: int = RAY_SERVE_HAPROXY_NBTHREAD
|
|
stats_port: int = RAY_SERVE_HAPROXY_STATS_PORT
|
|
stats_uri: str = "/stats"
|
|
metrics_port: int = RAY_SERVE_HAPROXY_METRICS_PORT
|
|
metrics_uri: str = "/metrics"
|
|
# All timeout values are in seconds
|
|
timeout_queue_s: Optional[int] = None
|
|
timeout_connect_s: Optional[int] = RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S
|
|
timeout_client_s: Optional[int] = RAY_SERVE_HAPROXY_TIMEOUT_CLIENT_S
|
|
timeout_server_s: Optional[int] = RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S
|
|
timeout_http_request_s: Optional[int] = None
|
|
hard_stop_after_s: Optional[int] = RAY_SERVE_HAPROXY_HARD_STOP_AFTER_S
|
|
custom_global: Dict[str, str] = field(default_factory=dict)
|
|
custom_defaults: Dict[str, str] = field(default_factory=dict)
|
|
inject_process_id_header: bool = False
|
|
reload_id: Optional[str] = None # Unique ID for each reload
|
|
tcp_nodelay: bool = RAY_SERVE_HAPROXY_TCP_NODELAY
|
|
enable_so_reuseport: bool = (
|
|
os.environ.get("SERVE_SOCKET_REUSE_PORT_ENABLED", "0") == "1"
|
|
)
|
|
has_received_routes: bool = False
|
|
has_received_servers: bool = False
|
|
pass_health_checks: bool = True
|
|
health_check_endpoint: str = "/-/healthz"
|
|
# Global health check parameters (used as defaults for backends)
|
|
# Number of consecutive failed health checks that must occur before a service instance is marked as unhealthy
|
|
health_check_fall: Optional[int] = RAY_SERVE_HAPROXY_HEALTH_CHECK_FALL
|
|
|
|
# Number of consecutive successful health checks required to mark an unhealthy service instance as healthy again
|
|
health_check_rise: Optional[int] = RAY_SERVE_HAPROXY_HEALTH_CHECK_RISE
|
|
|
|
# Interval, or the amount of time, between each health check attempt
|
|
health_check_inter: Optional[str] = RAY_SERVE_HAPROXY_HEALTH_CHECK_INTER
|
|
|
|
# The interval between two consecutive health checks when the server is in any of the transition states: UP - transitionally DOWN or DOWN - transitionally UP
|
|
health_check_fastinter: Optional[str] = RAY_SERVE_HAPROXY_HEALTH_CHECK_FASTINTER
|
|
|
|
# The interval between two consecutive health checks when the server is in the DOWN state
|
|
health_check_downinter: Optional[str] = RAY_SERVE_HAPROXY_HEALTH_CHECK_DOWNINTER
|
|
|
|
http_health_check_path: Optional[str] = "/-/healthz" # For HTTP health checks
|
|
grpc_health_check_path: Optional[
|
|
str
|
|
] = "/ray.serve.RayServeAPIService/Healthz" # For gRPC health checks
|
|
|
|
http_options: HTTPOptions = field(default_factory=HTTPOptions)
|
|
|
|
grpc_options: gRPCOptions = field(default_factory=gRPCOptions)
|
|
|
|
log_target: str = RAY_SERVE_HAPROXY_LOG_TARGET
|
|
|
|
# Per-request metrics for the ingress request router data path.
|
|
# When metrics are disabled, there is no additional overhead:
|
|
# 1) no metric log target / log-format-sd is rendered,
|
|
# 2) the Lua template skips the timing+truncation set_vars, and
|
|
# 3) HAProxyManager does not bind the dgram socket.
|
|
ingress_request_router_metrics_enabled: bool = (
|
|
RAY_SERVE_INGRESS_REQUEST_ROUTER_METRICS_ENABLED
|
|
)
|
|
|
|
# Per-request HTTP ingress metrics (serve_num_http_requests, latency,
|
|
# errors) emitted from HAProxy log datagrams -- the metrics the Python proxy
|
|
# emits in non-HAProxy mode. On by default whenever HAProxy metrics are
|
|
# enabled. When set, the http_frontend renders a per-request RFC 5424 log
|
|
# line to metrics_socket_path. gRPC is intentionally excluded due to
|
|
# limitations in parsing trailers.
|
|
metrics_enabled: bool = RAY_SERVE_HAPROXY_METRICS_ENABLED
|
|
metrics_socket_path: str = RAY_SERVE_HAPROXY_METRICS_SOCKET_PATH
|
|
|
|
balance_algorithm: str = RAY_SERVE_HAPROXY_BALANCE_ALGORITHM
|
|
|
|
# Global retry policy for the defaults block (inherited by every backend).
|
|
retry_on: str = RAY_SERVE_HAPROXY_RETRY_ON
|
|
retries: Optional[int] = RAY_SERVE_HAPROXY_RETRIES
|
|
|
|
# Per-ingress-router overrides; when None the ingress backend inherits the
|
|
# global policy above.
|
|
ingress_retry_on: Optional[str] = RAY_SERVE_HAPROXY_INGRESS_RETRY_ON
|
|
ingress_retries: Optional[int] = RAY_SERVE_HAPROXY_INGRESS_RETRIES
|
|
ingress_timeout_server_s: Optional[int] = RAY_SERVE_HAPROXY_INGRESS_TIMEOUT_SERVER_S
|
|
|
|
is_head: bool = False
|
|
|
|
# Tuning flags
|
|
bufsize: int = RAY_SERVE_HAPROXY_TUNE_BUFSIZE
|
|
h2_max_frame_size: int = RAY_SERVE_HAPROXY_H2_MAX_FRAME_SIZE
|
|
h2_be_initial_window_size: int = RAY_SERVE_HAPROXY_H2_BE_INITIAL_WINDOW_SIZE
|
|
h2_be_max_concurrent_streams: int = RAY_SERVE_HAPROXY_H2_BE_MAX_CONCURRENT_STREAMS
|
|
h2_fe_initial_window_size: int = RAY_SERVE_HAPROXY_H2_FE_INITIAL_WINDOW_SIZE
|
|
h2_fe_max_concurrent_streams: int = RAY_SERVE_HAPROXY_H2_FE_MAX_CONCURRENT_STREAMS
|
|
|
|
@property
|
|
def frontend_host(self) -> str:
|
|
if self.http_options.host is None or self.http_options.host in (
|
|
"0.0.0.0",
|
|
"::",
|
|
):
|
|
return "*"
|
|
|
|
return self.http_options.host
|
|
|
|
@property
|
|
def frontend_port(self) -> int:
|
|
return self.http_options.port
|
|
|
|
@property
|
|
def grpc_enabled(self) -> bool:
|
|
return is_grpc_enabled(self.grpc_options)
|
|
|
|
@property
|
|
def grpc_frontend_host(self) -> str:
|
|
# gRPCOptions has no host field; reuse the HTTP host so both proxies
|
|
# bind on the same interface.
|
|
return self.frontend_host
|
|
|
|
@property
|
|
def grpc_frontend_port(self) -> int:
|
|
return self.grpc_options.port
|
|
|
|
@property
|
|
def root_path(self) -> str:
|
|
"""Global root_path prefix, normalized without a trailing slash.
|
|
|
|
Empty when unset so the config template omits root_path handling.
|
|
"""
|
|
return (self.http_options.root_path or "").rstrip("/")
|
|
|
|
@property
|
|
def timeout_http_keep_alive_s(self) -> int:
|
|
return self.http_options.keep_alive_timeout_s
|
|
|
|
def build_health_route_info(self, backends: List[BackendConfig]) -> HealthRouteInfo:
|
|
if not self.has_received_routes:
|
|
router_ready_for_traffic = False
|
|
router_message = NO_ROUTES_MESSAGE
|
|
elif self.is_head or self.has_received_servers:
|
|
router_ready_for_traffic = True
|
|
router_message = ""
|
|
else:
|
|
router_ready_for_traffic = False
|
|
router_message = NO_REPLICAS_MESSAGE
|
|
|
|
if not self.pass_health_checks:
|
|
healthy = False
|
|
message = DRAINING_MESSAGE
|
|
elif not router_ready_for_traffic:
|
|
healthy = False
|
|
message = router_message
|
|
else:
|
|
healthy = True
|
|
message = HEALTHY_MESSAGE
|
|
|
|
if healthy:
|
|
# Build routes JSON mapping: {"<path_prefix>": "<app_name>", ...}
|
|
routes = {
|
|
be.path_prefix: be.app_name
|
|
for be in backends
|
|
if be.app_name and be.path_prefix
|
|
}
|
|
routes_json = json.dumps(routes, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
# Escape for haproxy double-quoted string literal
|
|
routes_message = routes_json.replace("\\", "\\\\").replace('"', '\\"')
|
|
else:
|
|
routes_message = message
|
|
|
|
return HealthRouteInfo(
|
|
healthy=healthy,
|
|
status=200 if healthy else 503,
|
|
health_message=message,
|
|
routes_message=routes_message,
|
|
routes_content_type="application/json" if healthy else "text/plain",
|
|
)
|
|
|
|
|
|
class ProxyApi(ABC):
|
|
"""Generic interface for load balancer management operations."""
|
|
|
|
@abstractmethod
|
|
async def start(self) -> None:
|
|
"""Initializes proxy configuration files."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_all_stats(self) -> Dict[str, Dict[str, ServerStats]]:
|
|
"""Get statistics for all servers in all backends."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def stop(self) -> None:
|
|
"""Stop the proxy."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def disable(self) -> None:
|
|
"""Disables the proxy from receiving any HTTP requests"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def enable(self) -> None:
|
|
"""Enables the proxy from receiving any HTTP requests"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def reload(self) -> None:
|
|
"""Gracefully reload the service."""
|
|
pass
|
|
|
|
|
|
class HAProxyApi(ProxyApi):
|
|
"""ProxyApi implementation for HAProxy."""
|
|
|
|
def __init__(
|
|
self,
|
|
cfg: HAProxyConfig,
|
|
backend_configs: Dict[str, BackendConfig] = None,
|
|
config_file_path: str = RAY_SERVE_HAPROXY_CONFIG_FILE_LOC,
|
|
):
|
|
self.cfg = cfg
|
|
self.backend_configs = backend_configs or {}
|
|
# Standalone gRPC fallback server used for requests (e.g. ListApplications)
|
|
# that can only be answered by the Serve proxy.
|
|
self.grpc_fallback_server: Optional[ServerConfig] = None
|
|
self.grpc_fallback_backend: Optional[BackendConfig] = None
|
|
self.config_file_path = config_file_path
|
|
# Lock to prevent concurrent config modifications
|
|
self._config_lock = asyncio.Lock()
|
|
self._proc = None
|
|
# Track old processes from graceful reloads that may still be draining
|
|
self._old_procs: List[asyncio.subprocess.Process] = []
|
|
# Per-spawn counter for stdout/stderr file names.
|
|
self._spawn_seq: int = 0
|
|
# Unique per actor incarnation. A restart resets _spawn_seq to 0, so
|
|
# without this a new incarnation would reuse — and (wb) truncate — the
|
|
# previous one's log files. The actor PID changes on every restart.
|
|
self._log_id: int = os.getpid()
|
|
# Debug ring of (stdout_path, stderr_path) for the most recently
|
|
# exited workers. Files for still-alive workers are never touched
|
|
# (their fds are still writing); once a worker exits its files are
|
|
# retired here and the oldest beyond the cap are deleted, so the
|
|
# file count stays bounded instead of growing with every reload.
|
|
self._retired_logs: "deque[Tuple[str, str]]" = deque()
|
|
self._max_retained_logs: int = 10
|
|
|
|
# Ensure required directories exist during initialization
|
|
self._initialize_directories_and_error_files()
|
|
|
|
def _initialize_directories_and_error_files(self) -> None:
|
|
"""
|
|
Ensures all required directories exist, creates a unified 500 error file,
|
|
and assigns its path to self.cfg.error_file_path. Called once during initialization.
|
|
"""
|
|
# Create a config file directory
|
|
config_dir = os.path.dirname(self.config_file_path)
|
|
os.makedirs(config_dir, exist_ok=True)
|
|
|
|
# Create a socket directory
|
|
socket_dir = os.path.dirname(self.cfg.socket_path)
|
|
os.makedirs(socket_dir, exist_ok=True)
|
|
|
|
# Sweep std-stream logs left by a previous incarnation on this node.
|
|
# No worker has spawned yet, so any match belongs to a dead actor.
|
|
self._sweep_stale_logs()
|
|
|
|
# Create a server state directory only if optimization is enabled
|
|
if self.cfg.enable_hap_optimization:
|
|
server_state_dir = os.path.dirname(self.cfg.server_state_file)
|
|
os.makedirs(server_state_dir, exist_ok=True)
|
|
|
|
# Create a single error file for both 502 and 504 errors
|
|
# Both will be normalized to 500 Internal Server Error
|
|
error_file_path = os.path.join(config_dir, "500.http")
|
|
with open(error_file_path, "w") as ef:
|
|
ef.write("HTTP/1.1 500 Internal Server Error\r\n")
|
|
ef.write("Content-Type: text/plain\r\n")
|
|
ef.write("Content-Length: 21\r\n")
|
|
ef.write("\r\n")
|
|
ef.write("Internal Server Error")
|
|
|
|
self.cfg.error_file_path = error_file_path
|
|
|
|
def _sweep_stale_logs(self) -> None:
|
|
"""Delete std-stream log files left by a previous actor incarnation.
|
|
|
|
Called at init before any worker spawns, so every match belongs to a
|
|
dead actor (a crash-restart can't run the normal prune on its way out).
|
|
"""
|
|
socket_path = Path(self.cfg.socket_path)
|
|
for pattern in (
|
|
f"{socket_path.name}.*.stdout.log",
|
|
f"{socket_path.name}.*.stderr.log",
|
|
):
|
|
for stale in socket_path.parent.glob(pattern):
|
|
try:
|
|
stale.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
def _prune_old_procs(self) -> None:
|
|
"""Drop exited workers from ``_old_procs`` and bound their log files.
|
|
|
|
A still-alive (draining) worker is kept untouched — its fd is still
|
|
writing to its file. When a worker has exited, its (stdout, stderr)
|
|
files are retired into a fixed-size ring kept for post-mortem
|
|
debugging; files that fall off the end of the ring are deleted, so the
|
|
file count stays bounded instead of growing with every reload.
|
|
"""
|
|
still_alive = []
|
|
for p in self._old_procs:
|
|
if p.returncode is None:
|
|
still_alive.append(p)
|
|
continue
|
|
self._retire_log_files(p)
|
|
self._old_procs = still_alive
|
|
|
|
def _soft_stop_old_procs(self) -> None:
|
|
"""Send SIGUSR1 to displaced workers so they close their listeners and
|
|
drain.
|
|
|
|
HAProxy's `-sf` flag already requests this, but its delivery can be
|
|
lost, so the manager sends the signal itself where it is reliable.
|
|
"""
|
|
self._prune_old_procs()
|
|
for proc in self._old_procs:
|
|
if not self._is_our_haproxy(proc.pid):
|
|
continue
|
|
try:
|
|
# This is a no-op if the worker is already draining
|
|
os.kill(proc.pid, signal.SIGUSR1)
|
|
except OSError:
|
|
pass
|
|
|
|
def _is_our_haproxy(self, pid: int) -> bool:
|
|
"""Whether `pid` is currently one of our haproxy workers.
|
|
|
|
Reads /proc to guard against signaling a recycled pid: a reaped or
|
|
recycled pid won't have our config_file_path in its cmdline and drops
|
|
out. Returns False on any non-Linux / no-/proc platform (fail-safe:
|
|
skips re-signaling rather than risk hitting the wrong process).
|
|
"""
|
|
try:
|
|
with open(f"/proc/{pid}/cmdline", "rb") as f:
|
|
return self.config_file_path.encode() in f.read().split(b"\0")
|
|
except OSError:
|
|
return False
|
|
|
|
def count_haproxy_processes(self) -> int:
|
|
"""Number of HAProxy processes on the node belonging to this manager.
|
|
|
|
Scans /proc for processes whose cmdline references our config file, so
|
|
the count spans the live worker, draining workers from prior reloads,
|
|
and any leaked/orphaned workers — making it a signal for process leaks.
|
|
Returns 0 on non-Linux / no-/proc platforms where /proc is unavailable.
|
|
"""
|
|
try:
|
|
entries = os.listdir("/proc")
|
|
except OSError:
|
|
return 0
|
|
|
|
processes = {
|
|
entry
|
|
for entry in entries
|
|
if entry.isdigit() and self._is_our_haproxy(int(entry))
|
|
}
|
|
return len(processes)
|
|
|
|
async def compute_target_mismatch(self) -> int:
|
|
"""Cardinality of the mismatch between the controller's broadcasted
|
|
targets and the targets HAProxy actually reports in its stats.
|
|
|
|
Broadcasted targets are the (backend, server) pairs the controller asked
|
|
this proxy to configure; reported targets are the (backend, server) pairs
|
|
present in HAProxy's live stats. The result is the symmetric difference,
|
|
so it counts both targets not yet applied to HAProxy and stale targets
|
|
HAProxy still reports.
|
|
"""
|
|
expected = set()
|
|
for backend_name, backend_config in self.backend_configs.items():
|
|
for server in backend_config.servers:
|
|
expected.add((backend_name, server.name))
|
|
# The generated config also renders the fallback server as a real
|
|
# `server ... backup` line in the same backend, so it appears in
|
|
# get_all_stats; count it as expected or the gauge never converges
|
|
# to zero for backends that have a fallback.
|
|
if backend_config.fallback_server is not None:
|
|
expected.add((backend_name, backend_config.fallback_server.name))
|
|
|
|
# Note that if an entire backend scaled down (when an app is removed),
|
|
# get_all_stats will filter out that entire backend, so there could be
|
|
# stale servers in that haproxy backend but not reported by this metric.
|
|
# This case is rare and not really something we worry about since requests
|
|
# shouldn't be hitting that backend or should return errors anyways.
|
|
stats = await self.get_all_stats()
|
|
reported = {
|
|
(backend_name, server_name)
|
|
for backend_name, servers in stats.items()
|
|
for server_name in servers
|
|
}
|
|
return len(expected ^ reported)
|
|
|
|
async def count_ongoing_http_requests(self) -> int:
|
|
"""Total in-flight HTTP requests across this node's HTTP app backends.
|
|
|
|
Sums each HTTP backend's aggregate `scur` (current sessions), which
|
|
counts request streams actively being served. Unlike the frontend scur
|
|
it does not count idle keep-alive connections, and unlike server-connection
|
|
counts it is unaffected by `http-reuse`. The aggregate row also includes
|
|
requests queued waiting for a server. The `-via-ingress-request-router`
|
|
backends are included (a router request lives in exactly one backend).
|
|
"""
|
|
suffix = "-via-ingress-request-router"
|
|
stats = self._parse_haproxy_csv_stats(
|
|
await self._send_socket_command("show stat")
|
|
)
|
|
total = 0
|
|
for backend_name, servers in stats.items():
|
|
# The via-router backend is rendered from its app's BackendConfig but
|
|
# under a suffixed name; strip it to look the config (and protocol) up.
|
|
base = (
|
|
backend_name[: -len(suffix)]
|
|
if backend_name.endswith(suffix)
|
|
else backend_name
|
|
)
|
|
backend_config = self.backend_configs.get(base)
|
|
if (
|
|
backend_config is None
|
|
or backend_config.protocol != RequestProtocol.HTTP
|
|
):
|
|
continue
|
|
backend_row = servers.get("BACKEND")
|
|
if backend_row is not None:
|
|
total += backend_row.current_sessions
|
|
return total
|
|
|
|
def _retire_log_files(self, proc: asyncio.subprocess.Process) -> None:
|
|
"""Move an exited proc's std-stream logs into the bounded debug ring,
|
|
deleting the oldest pair once the ring exceeds its cap. Only call this
|
|
for procs that have exited — their fds must be closed."""
|
|
self._retired_logs.append((proc._stdout_path, proc._stderr_path))
|
|
while len(self._retired_logs) > self._max_retained_logs:
|
|
for path in self._retired_logs.popleft():
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
|
|
def _is_running(self) -> bool:
|
|
"""Check if the HAProxy process is still running."""
|
|
return self._proc is not None and self._proc.returncode is None
|
|
|
|
async def _start_and_wait_for_haproxy(
|
|
self,
|
|
*extra_args: str,
|
|
timeout_s: int = RAY_SERVE_HAPROXY_STARTUP_TIMEOUT_S,
|
|
) -> asyncio.subprocess.Process:
|
|
# Build command args
|
|
haproxy_bin = get_haproxy_binary()
|
|
args = [haproxy_bin, "-db", "-f", self.config_file_path]
|
|
|
|
if not self.cfg.enable_so_reuseport:
|
|
args.append("-dR")
|
|
|
|
# Add any extra args (like -sf for graceful reload)
|
|
args.extend(extra_args)
|
|
|
|
# Redirect both std streams to files → no 64KB pipe buffer to deadlock on.
|
|
self._spawn_seq += 1
|
|
base = f"{self.cfg.socket_path}.{self._log_id}.{self._spawn_seq}"
|
|
stdout_path = f"{base}.stdout.log"
|
|
stderr_path = f"{base}.stderr.log"
|
|
with open(stdout_path, "wb", buffering=0) as stdout_file, open(
|
|
stderr_path, "wb", buffering=0
|
|
) as stderr_file:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*args,
|
|
stdout=stdout_file,
|
|
stderr=stderr_file,
|
|
)
|
|
proc._stdout_path = stdout_path
|
|
proc._stderr_path = stderr_path
|
|
logger.info(
|
|
f"Starting HAProxy (spawn #{self._spawn_seq}, pid={proc.pid}, "
|
|
f"stdout={stdout_path}, stderr={stderr_path}); args={args}"
|
|
)
|
|
|
|
try:
|
|
await self._wait_for_hap_availability(proc, timeout_s=timeout_s)
|
|
except Exception:
|
|
# If startup fails, ensure the process is killed to avoid orphaned processes
|
|
if proc.returncode is None:
|
|
proc.kill()
|
|
await proc.wait()
|
|
# The proc has exited; retire its log files so a failed start/reload
|
|
# doesn't orphan them outside the bounded ring.
|
|
self._retire_log_files(proc)
|
|
raise
|
|
|
|
return proc
|
|
|
|
async def _save_server_state(self) -> None:
|
|
"""Save the server state to the file."""
|
|
server_state = await self._send_socket_command("show servers state")
|
|
with open(self.cfg.server_state_file, "w") as f:
|
|
f.write(server_state)
|
|
|
|
async def _graceful_reload(self) -> None:
|
|
"""Perform a graceful reload of HAProxy by starting a new process with -sf."""
|
|
try:
|
|
old_proc = self._proc
|
|
await self._wait_for_hap_availability(old_proc)
|
|
|
|
# Save server state if optimization is enabled
|
|
if self.cfg.enable_hap_optimization:
|
|
await self._save_server_state()
|
|
|
|
# Start new HAProxy process with -sf flag to gracefully take over from old process
|
|
# Use -x socket transfer for seamless reloads if optimization is enabled
|
|
# Also re-signal still-alive displaced workers: heals any worker
|
|
# whose earlier -sf signal was lost (re-signaling a draining one is
|
|
# a no-op). The new process delivers SIGUSR1 only after startup, so
|
|
# filter to pids whose /proc cmdline is still one of our haproxy
|
|
# workers: a worker that exited and had its pid recycled in the
|
|
# meantime drops out, so the signal can't land on an unrelated
|
|
# process.
|
|
self._prune_old_procs()
|
|
live_old_pids = [
|
|
str(p.pid) for p in self._old_procs if self._is_our_haproxy(p.pid)
|
|
]
|
|
reload_args = ["-sf", str(old_proc.pid), *live_old_pids]
|
|
if self.cfg.enable_hap_optimization:
|
|
reload_args.extend(["-x", self.cfg.socket_path])
|
|
|
|
self._proc = await self._start_and_wait_for_haproxy(*reload_args)
|
|
|
|
# Track for shutdown cleanup; pruned at the next reload.
|
|
if old_proc is not None:
|
|
self._old_procs.append(old_proc)
|
|
|
|
# `-sf` only re-signals stranded workers on the next reload, and
|
|
# there is none after the final one. Re-send SIGUSR1 directly so a
|
|
# worker left serving stale config stops now instead of racing the
|
|
# client's next request.
|
|
self._soft_stop_old_procs()
|
|
|
|
logger.info(
|
|
"Successfully performed graceful HAProxy reload with process restart."
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"HAProxy graceful reload failed: {e}")
|
|
raise
|
|
|
|
async def _wait_for_hap_availability(
|
|
self,
|
|
proc: asyncio.subprocess.Process,
|
|
timeout_s: int = RAY_SERVE_HAPROXY_STARTUP_TIMEOUT_S,
|
|
) -> None:
|
|
start_time = time.time()
|
|
|
|
# TODO: update this to use health checks
|
|
while time.time() - start_time < timeout_s:
|
|
if proc.returncode is not None:
|
|
# Both streams were redirected to files at spawn; tail them.
|
|
stderr_text = _tail_file(proc._stderr_path)
|
|
stdout_text = _tail_file(proc._stdout_path)
|
|
output = stderr_text or stdout_text
|
|
|
|
raise RuntimeError(
|
|
f"HAProxy crashed during startup: {output or f'exit code {proc.returncode}'}"
|
|
)
|
|
|
|
# The socket path answers through its previous owner until the
|
|
# new process rebinds it, so require the answer to come from
|
|
# `proc` itself. Otherwise a spawn that dies before taking over is
|
|
# declared ready and its -sf target is stranded forever.
|
|
if await self._get_running_pid() == proc.pid:
|
|
return
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
raise RuntimeError(
|
|
f"HAProxy (pid={proc.pid}) did not take over the admin socket within "
|
|
f"{timeout_s} seconds. stderr: {_tail_file(proc._stderr_path)}"
|
|
)
|
|
|
|
def _write_ingress_request_router_lua(
|
|
self,
|
|
backends: List[BackendConfig],
|
|
) -> Optional[str]:
|
|
"""Render the ingress-request-router Lua action and write it to disk.
|
|
|
|
Returns the script path, or None if no backend has both ingress
|
|
request routers AND replicas with replica IDs.
|
|
"""
|
|
routers, targets = _routers_and_targets_by_backend(backends)
|
|
if not routers:
|
|
return None
|
|
|
|
# When metrics are enabled, render the two timing hooks and the
|
|
# truncation set_var; when disabled, all three substitute to empty
|
|
# strings to avoid any additional overhead.
|
|
if self.cfg.ingress_request_router_metrics_enabled:
|
|
metrics_pre = "local _metrics_t0 = core.now()"
|
|
metrics_post = (
|
|
"local _metrics_t1 = core.now(); "
|
|
'txn:set_var("txn.ingress_request_router_latency_us", '
|
|
"(_metrics_t1.sec - _metrics_t0.sec) * 1000000 "
|
|
"+ (_metrics_t1.usec - _metrics_t0.usec))"
|
|
)
|
|
metrics_set_truncated = (
|
|
'txn:set_var("txn.ingress_request_router_truncated_full_length", '
|
|
"truncated)"
|
|
)
|
|
else:
|
|
metrics_pre = ""
|
|
metrics_post = ""
|
|
metrics_set_truncated = ""
|
|
|
|
content = _load_lua_template().substitute(
|
|
TIMEOUT_S=RAY_SERVE_HAPROXY_INGRESS_REQUEST_ROUTER_TIMEOUT_S,
|
|
FORWARD_BODY=str(RAY_SERVE_INGRESS_REQUEST_ROUTER_FORWARD_BODY).lower(),
|
|
# HAProxy's req_get_headers() returns lowercase header keys,
|
|
# so lowercase here for the Lua lookup. Empty string disables
|
|
# forwarding entirely.
|
|
SESSION_HEADER=SERVE_SESSION_ID.lower(),
|
|
ROUTERS=_format_routers_lua(routers),
|
|
REPLICA_TARGETS=_format_replica_targets_lua(targets),
|
|
METRICS_PRE_CALL_ROUTER=metrics_pre,
|
|
METRICS_POST_CALL_ROUTER=metrics_post,
|
|
METRICS_SET_TRUNCATED=metrics_set_truncated,
|
|
)
|
|
|
|
lua_path = os.path.join(
|
|
os.path.dirname(self.config_file_path), "ingress_request_router.lua"
|
|
)
|
|
if _write_if_changed(lua_path, content):
|
|
logger.debug(f"Wrote Lua routing script to {lua_path}")
|
|
return lua_path
|
|
|
|
def _generate_config_file_internal(self) -> bool:
|
|
"""Internal config generation without locking (for use within locked sections)."""
|
|
try:
|
|
env = Environment()
|
|
# Escapes names before they are rendered into set-var-fmt values.
|
|
env.filters["haproxy_fmt"] = _haproxy_fmt_literal
|
|
|
|
# Backends are sorted in decreasing order of length of path prefix
|
|
# to ensure that the longest path prefix match is taken first.
|
|
# Equal lengthed prefixes are then sorted alphabetically.
|
|
backends = sorted(
|
|
self.backend_configs.values(),
|
|
key=lambda be: (-len(be.path_prefix), be.path_prefix),
|
|
)
|
|
|
|
# HTTP backends are path-routed; gRPC backends are header-routed
|
|
# (by the `application` gRPC metadata) and need a different
|
|
# frontend / backend rendering. Split them up here so the template
|
|
# can iterate each set independently.
|
|
http_backends = [b for b in backends if b.protocol == RequestProtocol.HTTP]
|
|
grpc_backends = [b for b in backends if b.protocol == RequestProtocol.GRPC]
|
|
|
|
# Derive from the write result: returns None when no backend has
|
|
# both routers and replicas with IDs (transient during scaling).
|
|
# The ingress request router is HTTP-only.
|
|
ingress_request_router_lua_path = self._write_ingress_request_router_lua(
|
|
http_backends
|
|
)
|
|
has_ingress_request_router = ingress_request_router_lua_path is not None
|
|
|
|
# Enrich HTTP backends with precomputed health check configuration strings
|
|
http_backends_with_health_config = [
|
|
{
|
|
"backend": backend,
|
|
"health_config": backend.build_health_check_config(self.cfg),
|
|
}
|
|
for backend in http_backends
|
|
]
|
|
|
|
# Enrich gRPC backends with precomputed health check configuration strings
|
|
grpc_backends_with_health_config = [
|
|
{
|
|
"backend": backend,
|
|
"health_config": backend.build_health_check_config(self.cfg),
|
|
}
|
|
for backend in grpc_backends
|
|
]
|
|
|
|
# Healthz / routes only consider HTTP backends. The HAProxy `/-/healthz`
|
|
# and `/-/routes` endpoints are HTTP-facing.
|
|
health_route_info = self.cfg.build_health_route_info(http_backends)
|
|
|
|
# Render healthz rules separately for readability/reuse
|
|
healthz_template = env.from_string(HAPROXY_HEALTHZ_RULES_TEMPLATE)
|
|
healthz_rules = healthz_template.render(
|
|
{
|
|
"config": self.cfg,
|
|
"backends": http_backends,
|
|
"health_info": health_route_info,
|
|
}
|
|
)
|
|
grpc_healthz_template = env.from_string(HAPROXY_GRPC_HEALTHZ_RULES_TEMPLATE)
|
|
grpc_healthz_rules = grpc_healthz_template.render(
|
|
{
|
|
"config": self.cfg,
|
|
"backends": grpc_backends,
|
|
"health_info": health_route_info,
|
|
}
|
|
)
|
|
|
|
# Generate the gRPC fallback backend with health check configuration, which only
|
|
# contains the head node serve proxy. This is only used for ListApplication requests
|
|
# since HAProxy can't construct the gRPC response itself, so it relies on the serve
|
|
# proxy to do so.
|
|
grpc_fallback_backend_with_health_config = None
|
|
if (
|
|
self.grpc_fallback_server is not None
|
|
and self.grpc_fallback_backend is not None
|
|
):
|
|
grpc_fallback_backend_with_health_config = {
|
|
"backend": self.grpc_fallback_backend,
|
|
"health_config": self.grpc_fallback_backend.build_health_check_config(
|
|
self.cfg
|
|
),
|
|
}
|
|
|
|
config_template = env.from_string(HAPROXY_CONFIG_TEMPLATE)
|
|
config_content = config_template.render(
|
|
{
|
|
"config": self.cfg,
|
|
"backends": http_backends,
|
|
"grpc_backends": grpc_backends,
|
|
"backends_with_health_config": http_backends_with_health_config,
|
|
"grpc_backends_with_health_config": (
|
|
grpc_backends_with_health_config
|
|
),
|
|
"healthz_rules": healthz_rules,
|
|
"grpc_healthz_rules": grpc_healthz_rules,
|
|
"route_info": health_route_info,
|
|
"has_ingress_request_router": has_ingress_request_router,
|
|
"ingress_request_router_lua_path": ingress_request_router_lua_path,
|
|
"ingress_request_router_timeout_s": (
|
|
RAY_SERVE_HAPROXY_INGRESS_REQUEST_ROUTER_TIMEOUT_S
|
|
),
|
|
"ingress_request_router_bufsize": (
|
|
RAY_SERVE_HAPROXY_INGRESS_REQUEST_ROUTER_BUFSIZE
|
|
),
|
|
"ingress_request_router_forward_body": (
|
|
RAY_SERVE_INGRESS_REQUEST_ROUTER_FORWARD_BODY
|
|
),
|
|
"ingress_request_router_metrics_enabled": self.cfg.ingress_request_router_metrics_enabled,
|
|
"metrics_enabled": self.cfg.metrics_enabled,
|
|
"metrics_socket_path": self.cfg.metrics_socket_path,
|
|
"grpc_fallback_backend_with_health_config": grpc_fallback_backend_with_health_config,
|
|
"healthy_message": HEALTHY_MESSAGE, # the message in the response body for healthy replicas
|
|
}
|
|
)
|
|
|
|
# Ensure the config ends with a newline
|
|
if not config_content.endswith("\n"):
|
|
config_content += "\n"
|
|
|
|
# Use file locking to prevent concurrent writes from multiple processes
|
|
# This is important in test environments where multiple nodes may run
|
|
# on the same machine
|
|
import fcntl
|
|
|
|
lock_file_path = self.config_file_path + ".lock"
|
|
with open(lock_file_path, "w") as lock_f:
|
|
fcntl.flock(lock_f.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
with open(self.config_file_path, "w") as f:
|
|
f.write(config_content)
|
|
finally:
|
|
fcntl.flock(lock_f.fileno(), fcntl.LOCK_UN)
|
|
|
|
logger.debug(
|
|
f"Succesfully generated HAProxy configuration: {self.config_file_path}."
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to create HAProxy configuration files: {e}")
|
|
raise
|
|
|
|
async def start(self) -> None:
|
|
"""
|
|
Generate HAProxy configuration files and start the HAProxy server process.
|
|
|
|
This method creates the necessary configuration files and launches the HAProxy
|
|
process in foreground mode, ensuring that the proxy is running with the latest
|
|
configuration and that the parent retains control of the subprocess handle.
|
|
"""
|
|
try:
|
|
async with self._config_lock:
|
|
# Set initial reload ID if header injection is enabled and ID is not set
|
|
if self.cfg.inject_process_id_header and self.cfg.reload_id is None:
|
|
self.cfg.reload_id = f"initial-{int(time.time() * 1000)}"
|
|
|
|
self._generate_config_file_internal()
|
|
logger.info("Successfully generated HAProxy config file.")
|
|
|
|
self._proc = await self._start_and_wait_for_haproxy()
|
|
logger.info("HAProxy started successfully.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize and start HAProxy configuration: {e}")
|
|
raise
|
|
|
|
async def get_all_stats(self) -> Dict[str, Dict[str, ServerStats]]:
|
|
"""Get statistics for all servers in all backends (implements abstract method).
|
|
|
|
Returns only application backends configured in self.backend_configs,
|
|
excluding HAProxy internal components (frontends, default_backend, stats).
|
|
Also excludes BACKEND aggregate entries, returning only individual servers.
|
|
"""
|
|
try:
|
|
stats_output = await self._send_socket_command("show stat")
|
|
all_stats = self._parse_haproxy_csv_stats(stats_output)
|
|
|
|
# Filter to only return application backends (ones in backend_configs)
|
|
# Exclude HAProxy internal components like frontends, default_backend, stats
|
|
# Also exclude BACKEND aggregate entries, keep only individual servers
|
|
return {
|
|
backend_name: {
|
|
server_name: stats
|
|
for server_name, stats in servers.items()
|
|
if server_name != "BACKEND"
|
|
}
|
|
for backend_name, servers in all_stats.items()
|
|
if backend_name in self.backend_configs
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Failed to get HAProxy stats: {e}")
|
|
return {}
|
|
|
|
async def get_haproxy_stats(self) -> HAProxyStats:
|
|
"""Get complete HAProxy statistics including both individual and aggregate data."""
|
|
server_stats = await self.get_all_stats()
|
|
return HAProxyStats(backend_to_servers=server_stats)
|
|
|
|
async def _send_socket_command(self, command: str) -> str:
|
|
"""Send a command to the HAProxy stats socket via Unix domain socket."""
|
|
try:
|
|
if not os.path.exists(self.cfg.socket_path):
|
|
raise RuntimeError(
|
|
f"HAProxy socket file does not exist: {self.cfg.socket_path}."
|
|
)
|
|
|
|
try:
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_unix_connection(self.cfg.socket_path),
|
|
timeout=5.0,
|
|
)
|
|
except asyncio.TimeoutError:
|
|
raise RuntimeError(
|
|
f"Timeout connecting to HAProxy socket: {self.cfg.socket_path}"
|
|
)
|
|
|
|
try:
|
|
writer.write(f"{command}\n".encode("utf-8"))
|
|
await writer.drain()
|
|
|
|
# Read until EOF (HAProxy closes connection after response)
|
|
try:
|
|
result_bytes = await asyncio.wait_for(reader.read(), timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
raise RuntimeError(
|
|
f"Timeout while sending command '{command}' to HAProxy socket"
|
|
)
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
result = result_bytes.decode("utf-8", errors="ignore")
|
|
logger.debug(f"Socket command '{command}' returned {len(result)} chars.")
|
|
return result
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to send socket command '{command}': {e}")
|
|
|
|
@staticmethod
|
|
def _parse_haproxy_csv_stats(
|
|
stats_output: str,
|
|
) -> Dict[str, Dict[str, ServerStats]]:
|
|
"""Parse HAProxy stats CSV output into structured data."""
|
|
if not stats_output or not stats_output.strip():
|
|
return {}
|
|
|
|
# HAProxy stats start with '#' comment - replace with nothing for CSV parsing
|
|
csv_data = stats_output.replace("# ", "", 1)
|
|
backend_stats: Dict[str, Dict[str, ServerStats]] = {}
|
|
|
|
def safe_int(v):
|
|
try:
|
|
return int(v)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
for row in csv.DictReader(io.StringIO(csv_data)):
|
|
backend = row.get("pxname", "").strip()
|
|
server = row.get("svname", "").strip()
|
|
status = row.get("status", "").strip() or "UNKNOWN"
|
|
|
|
if not backend or not server:
|
|
continue
|
|
|
|
backend_stats.setdefault(backend, {})
|
|
backend_stats[backend][server] = ServerStats(
|
|
backend=backend,
|
|
server=server,
|
|
status=status,
|
|
current_sessions=safe_int(row.get("scur")),
|
|
queued=safe_int(row.get("qcur")),
|
|
)
|
|
|
|
return backend_stats
|
|
|
|
async def stop(self) -> None:
|
|
proc = self._proc
|
|
if proc is None:
|
|
logger.info("HAProxy process not running, skipping shutdown.")
|
|
return
|
|
|
|
try:
|
|
# Kill the current process
|
|
if proc.returncode is None:
|
|
proc.kill()
|
|
await proc.wait()
|
|
self._proc = None
|
|
|
|
# Also kill any old processes from graceful reloads that might still be running
|
|
for old_proc in self._old_procs:
|
|
try:
|
|
if old_proc.returncode is None:
|
|
old_proc.kill()
|
|
await old_proc.wait()
|
|
logger.info(f"Killed old HAProxy process (PID: {old_proc.pid})")
|
|
except Exception as e:
|
|
logger.warning(f"Error killing old HAProxy process: {e}")
|
|
|
|
self._old_procs.clear()
|
|
|
|
logger.info("Stopped HAProxy process.")
|
|
except RuntimeError as e:
|
|
logger.error(f"Error during HAProxy shutdown: {e}")
|
|
|
|
async def reload(self) -> None:
|
|
try:
|
|
self._generate_config_file_internal()
|
|
await self._graceful_reload()
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to update and reload HAProxy: {e}")
|
|
|
|
def has_alive_old_procs(self) -> bool:
|
|
"""True if any soft-stopping HAProxy workers from prior reloads remain."""
|
|
self._prune_old_procs()
|
|
return len(self._old_procs) > 0
|
|
|
|
async def disable(self) -> None:
|
|
"""Force haproxy health checks to fail."""
|
|
try:
|
|
# Disable health checks (set to fail)
|
|
self.cfg.pass_health_checks = False
|
|
|
|
# Regenerate the config file with the deny rule
|
|
self._generate_config_file_internal()
|
|
|
|
# Perform a graceful reload to apply changes
|
|
await self._graceful_reload()
|
|
logger.info("Successfully disabled health checks.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to disable health checks: {e}")
|
|
raise
|
|
|
|
async def enable(self) -> None:
|
|
"""Force haproxy health checks to pass."""
|
|
try:
|
|
self.cfg.pass_health_checks = True
|
|
|
|
self._generate_config_file_internal()
|
|
# Perform a graceful reload to apply changes
|
|
await self._graceful_reload()
|
|
logger.info("Successfully enabled health checks.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to disable health checks: {e}")
|
|
raise
|
|
|
|
def set_backend_configs(
|
|
self,
|
|
backend_configs: Dict[str, BackendConfig],
|
|
) -> None:
|
|
if backend_configs:
|
|
self.cfg.has_received_routes = True
|
|
|
|
self.backend_configs = backend_configs
|
|
|
|
self.cfg.has_received_servers = self.cfg.has_received_servers or any(
|
|
len(bc.servers) > 0 for bc in backend_configs.values()
|
|
)
|
|
|
|
def set_grpc_fallback_server(self, server: Optional[ServerConfig]) -> None:
|
|
self.grpc_fallback_server = server
|
|
if server is not None:
|
|
self.grpc_fallback_backend = BackendConfig(
|
|
name="grpc_fallback_backend",
|
|
path_prefix="/", # this is unused for grpc
|
|
protocol=RequestProtocol.GRPC,
|
|
servers=[server],
|
|
)
|
|
else:
|
|
self.grpc_fallback_backend = None
|
|
|
|
async def is_grpc_fallback_ready(self) -> bool:
|
|
"""True iff the running HAProxy has an UP server in grpc_fallback_backend."""
|
|
if self.grpc_fallback_backend is None:
|
|
return False
|
|
try:
|
|
stats = self._parse_haproxy_csv_stats(
|
|
await self._send_socket_command("show stat")
|
|
)
|
|
except Exception:
|
|
return False
|
|
servers = stats.get("grpc_fallback_backend", {})
|
|
return any(server.is_up for server in servers.values())
|
|
|
|
async def _get_running_pid(self) -> Optional[int]:
|
|
"""Pid of the HAProxy process currently answering the admin socket,
|
|
or None if the socket is unavailable or the response is unparseable.
|
|
"""
|
|
try:
|
|
info = await self._send_socket_command("show info")
|
|
except Exception:
|
|
return None
|
|
for line in info.splitlines():
|
|
if line.startswith("Pid:"):
|
|
try:
|
|
return int(line.split(":", 1)[1])
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
async def is_running(self) -> bool:
|
|
try:
|
|
await self._send_socket_command("show info")
|
|
return True
|
|
except Exception:
|
|
# During reload or shutdown, socket can be temporarily unavailable.
|
|
# Treat as unhealthy instead of raising.
|
|
return False
|
|
|
|
|
|
@ray.remote(num_cpus=0)
|
|
class HAProxyManager(ProxyActorInterface):
|
|
def __init__(
|
|
self,
|
|
http_options: HTTPOptions,
|
|
grpc_options: gRPCOptions,
|
|
*,
|
|
node_id: NodeId,
|
|
node_ip_address: str,
|
|
logging_config: LoggingConfig,
|
|
long_poll_client: Optional[LongPollClient] = None,
|
|
): # noqa: F821
|
|
super().__init__(
|
|
node_id=node_id,
|
|
node_ip_address=node_ip_address,
|
|
logging_config=logging_config,
|
|
# HAProxyManager is not on the request path, so we can disable
|
|
# the buffer to ensure logs are immediately flushed.
|
|
log_buffer_size=1,
|
|
)
|
|
|
|
self._grpc_options = grpc_options
|
|
self._http_options = http_options
|
|
|
|
# The time when the node starts to drain.
|
|
# The node is not draining if it's None.
|
|
self._draining_start_time: Optional[float] = None
|
|
|
|
self.event_loop = get_or_create_event_loop()
|
|
|
|
self._target_groups: List[TargetGroup] = []
|
|
self._received_target_groups_broadcast = False
|
|
|
|
# Fallback targets.
|
|
self._http_fallback_target: Optional[Target] = None
|
|
self._grpc_fallback_target: Optional[Target] = None
|
|
|
|
# Lock to serialize HAProxy reloads and prevent concurrent reload operations
|
|
# which can cause race conditions with SO_REUSEPORT
|
|
self._reload_lock = asyncio.Lock()
|
|
|
|
# Coalescing state: each broadcast sets _update_pending=True and
|
|
# (re)arms _coalesce_task, which sleeps one window then applies.
|
|
# _coalesce_armed_at is the monotonic time of the first broadcast
|
|
# since the last apply, used to report end-to-end update latency.
|
|
self._update_pending: bool = False
|
|
self._coalesce_task: Optional[asyncio.Task] = None
|
|
self._coalesce_armed_at: Optional[float] = None
|
|
self._haproxy_update_latency = metrics.Histogram(
|
|
"serve_haproxy_update_latency_s",
|
|
description=(
|
|
"Seconds from the first coalesced controller broadcast to the "
|
|
"HAProxy reload completing. Includes the coalesce window, time "
|
|
"queued behind an in-flight reload, and the reload itself."
|
|
),
|
|
boundaries=RAY_SERVE_HAPROXY_UPDATE_LATENCY_BUCKETS_S,
|
|
tag_keys=("node_id",),
|
|
).set_default_tags({"node_id": node_id})
|
|
|
|
self.long_poll_client = long_poll_client or LongPollClient(
|
|
ray.get_actor(SERVE_CONTROLLER_NAME, namespace=SERVE_NAMESPACE),
|
|
{
|
|
LongPollNamespace.GLOBAL_LOGGING_CONFIG: self._update_logging_config,
|
|
LongPollNamespace.TARGET_GROUPS: self.update_target_groups,
|
|
LongPollNamespace.FALLBACK_TARGETS: self.update_fallback_targets,
|
|
},
|
|
call_in_event_loop=self.event_loop,
|
|
client_id=f"{type(self).__name__}:{ray.get_runtime_context().get_actor_id()}",
|
|
)
|
|
|
|
is_head = self._node_id == get_head_node_id()
|
|
apply_per_node_port_overrides(self._http_options, self._grpc_options, is_head)
|
|
|
|
startup_msg = f"HAProxy starting on node {self._node_id} (HTTP port: {self._http_options.port}"
|
|
if is_grpc_enabled(self._grpc_options):
|
|
startup_msg += f", gRPC port: {self._grpc_options.port})."
|
|
else:
|
|
startup_msg += ")."
|
|
logger.info(startup_msg)
|
|
logger.debug(
|
|
f"Configure HAProxyManager actor {ray.get_runtime_context().get_actor_id()} "
|
|
f"logger with logging config: {logging_config}"
|
|
)
|
|
|
|
# Scope paths under node_id so co-located managers (multi-raylet
|
|
# test clusters on one host) don't overwrite each other's files.
|
|
def _per_node(path: str) -> str:
|
|
d, f = os.path.split(path)
|
|
return os.path.join(d, self._node_id, f)
|
|
|
|
self._haproxy = HAProxyApi(
|
|
cfg=HAProxyConfig(
|
|
http_options=http_options,
|
|
grpc_options=grpc_options,
|
|
is_head=is_head,
|
|
socket_path=_per_node(RAY_SERVE_HAPROXY_SOCKET_PATH),
|
|
server_state_base=os.path.join(
|
|
RAY_SERVE_HAPROXY_SERVER_STATE_BASE, self._node_id
|
|
),
|
|
server_state_file=_per_node(RAY_SERVE_HAPROXY_SERVER_STATE_FILE),
|
|
metrics_socket_path=_per_node(RAY_SERVE_HAPROXY_METRICS_SOCKET_PATH),
|
|
),
|
|
config_file_path=_per_node(RAY_SERVE_HAPROXY_CONFIG_FILE_LOC),
|
|
)
|
|
|
|
self._metrics_collector = None
|
|
self._metrics_attach_task: Optional[asyncio.Task] = None
|
|
if RAY_SERVE_HAPROXY_METRICS_ENABLED:
|
|
from ray.serve._private.haproxy_metrics import HAProxyMetricsCollector
|
|
|
|
# The metrics collector owns all serve_haproxy_* metrics for this proxy.
|
|
# It is constructed if haproxy metrics are enabled. start() always begins
|
|
# node-level polling and binds the per-request dgram reader (where
|
|
# HAProxy writes one line per request). It returns its bind task (which
|
|
# ready() awaits to surface bind failures).
|
|
self._metrics_collector = HAProxyMetricsCollector(
|
|
haproxy_api=self._haproxy,
|
|
node_id=self._node_id,
|
|
node_ip_address=self._node_ip_address,
|
|
)
|
|
try:
|
|
self._metrics_attach_task = self._metrics_collector.start(
|
|
self.event_loop,
|
|
poll_interval_s=RAY_SERVE_HAPROXY_METRICS_REPORT_INTERVAL_S,
|
|
metrics_socket_path=self._haproxy.cfg.metrics_socket_path,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to start the per-request metrics datagram reader; "
|
|
"per-request metrics will not be emitted. Node-level metrics "
|
|
"and HAProxy continue normally."
|
|
)
|
|
|
|
self._haproxy_start_task = self.event_loop.create_task(self._haproxy.start())
|
|
|
|
async def shutdown(self) -> None:
|
|
"""Shutdown the HAProxyManager and clean up the HAProxy process.
|
|
|
|
This method should be called before the actor is killed to ensure
|
|
the HAProxy subprocess is properly terminated.
|
|
"""
|
|
try:
|
|
logger.info(
|
|
f"Shutting down HAProxyManager on node {self._node_id}.",
|
|
extra={"log_to_stderr": False},
|
|
)
|
|
|
|
await self._haproxy.stop()
|
|
|
|
logger.info(
|
|
f"Successfully stopped HAProxy process on node {self._node_id}.",
|
|
extra={"log_to_stderr": False},
|
|
)
|
|
except Exception as e:
|
|
raise RuntimeError(f"Error stopping HAProxy during shutdown: {e}")
|
|
finally:
|
|
if self._metrics_collector is not None:
|
|
self._metrics_collector.close()
|
|
self._metrics_collector = None
|
|
|
|
async def ready(self) -> str:
|
|
try:
|
|
# Wait for haproxy to start. Internally, this starts the process and
|
|
# waits for it to be running by querying the stats socket.
|
|
await self._haproxy_start_task
|
|
# Surface metrics bind failures here. HAProxy startup is much
|
|
# slower than the bind (subprocess vs. one syscall), so by the
|
|
# time we get here the attach task has either succeeded or
|
|
# raised; awaiting it is just collecting the result.
|
|
if self._metrics_attach_task is not None:
|
|
try:
|
|
await self._metrics_attach_task
|
|
except Exception:
|
|
# Only the dgram reader failed. Leave the collector intact
|
|
# so node-level gauges keep polling; the socket is left
|
|
# unbound (bind_and_attach is safe to have failed).
|
|
logger.exception(
|
|
"Failed to bind/attach metrics dgram reader; "
|
|
"per-request router metrics will not be drained."
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Failed to start HAProxy.")
|
|
raise e from None
|
|
|
|
# Return proxy metadata used by the controller.
|
|
# NOTE(zcin): We need to convert the metadata to a json string because
|
|
# of cross-language scenarios. Java can't deserialize a Python tuple.
|
|
return json.dumps(
|
|
[
|
|
ray.get_runtime_context().get_worker_id(),
|
|
get_component_logger_file_path(),
|
|
]
|
|
)
|
|
|
|
async def serving(self, wait_for_applications_running: bool = True) -> None:
|
|
"""Wait for the HAProxy process to be ready to serve requests."""
|
|
if not wait_for_applications_running:
|
|
return
|
|
|
|
ready_to_serve = False
|
|
while not ready_to_serve:
|
|
if self._is_draining():
|
|
return
|
|
|
|
# We have not received an update from the controller yet. We always
|
|
# get at least one broadcast even if target groups is empty (e.g.
|
|
# when the user sets route_prefix=None).
|
|
if not self._received_target_groups_broadcast:
|
|
await asyncio.sleep(0.2)
|
|
continue
|
|
|
|
# When gRPC is used, haproxy relies on the fallback serve proxy to
|
|
# handle ListApplications requests, so we block until HAProxy reprots
|
|
# an UP server in grpc_fallback_backend.
|
|
if (
|
|
is_grpc_enabled(self._grpc_options)
|
|
and not await self._haproxy.is_grpc_fallback_ready()
|
|
):
|
|
await asyncio.sleep(0.2)
|
|
continue
|
|
|
|
desired_backend_servers = {
|
|
self._generate_backend_name(tg): {
|
|
self._generate_server_name(target) for target in tg.targets
|
|
}
|
|
for tg in self._target_groups
|
|
}
|
|
fallback_servers = {
|
|
self._generate_server_name(target)
|
|
for target in (
|
|
self._http_fallback_target,
|
|
self._grpc_fallback_target,
|
|
)
|
|
if target is not None
|
|
}
|
|
|
|
try:
|
|
stats = await self._haproxy.get_all_stats()
|
|
ready_backends = set()
|
|
for backend, servers in stats.items():
|
|
desired_servers = desired_backend_servers.get(backend, set())
|
|
desired_servers.update(fallback_servers)
|
|
for server_name, server in servers.items():
|
|
if not server.is_up:
|
|
continue
|
|
|
|
if server_name in desired_servers:
|
|
ready_backends.add(backend)
|
|
break
|
|
|
|
ready_to_serve = set(desired_backend_servers) == ready_backends
|
|
except Exception:
|
|
pass
|
|
if not ready_to_serve:
|
|
await asyncio.sleep(0.2)
|
|
|
|
def _is_draining(self) -> bool:
|
|
"""Whether is haproxy is in the draining status or not."""
|
|
return self._draining_start_time is not None
|
|
|
|
async def update_draining(
|
|
self, draining: bool, _after: Optional[Any] = None
|
|
) -> None:
|
|
"""Update the draining status of the proxy.
|
|
|
|
This is called by the proxy state manager
|
|
to drain or un-drain the haproxy.
|
|
"""
|
|
|
|
if draining and (not self._is_draining()):
|
|
logger.info(
|
|
f"Start to drain the HAProxy on node {self._node_id}.",
|
|
extra={"log_to_stderr": False},
|
|
)
|
|
# Use the reload lock to serialize with other HAProxy reload operations
|
|
async with self._reload_lock:
|
|
await self._haproxy.disable()
|
|
self._draining_start_time = time.time()
|
|
if (not draining) and self._is_draining():
|
|
logger.info(
|
|
f"Stop draining the HAProxy on node {self._node_id}.",
|
|
extra={"log_to_stderr": False},
|
|
)
|
|
# Use the reload lock to serialize with other HAProxy reload operations
|
|
async with self._reload_lock:
|
|
await self._haproxy.enable()
|
|
self._draining_start_time = None
|
|
|
|
async def is_drained(self, _after: Optional[Any] = None) -> bool:
|
|
"""Drained iff past min drain period AND no old workers still serving
|
|
AND current worker reports idle.
|
|
|
|
The admin socket only sees the current (post-reload) worker; old
|
|
soft-stopping workers can still hold in-flight long-running requests
|
|
invisible to ``is_system_idle``. Gating on ``has_alive_old_procs``
|
|
prevents SIGKILL from severing those connections.
|
|
"""
|
|
if not self._is_draining():
|
|
return False
|
|
if (time.time() - self._draining_start_time) <= PROXY_MIN_DRAINING_PERIOD_S:
|
|
return False
|
|
if self._haproxy.has_alive_old_procs():
|
|
return False
|
|
haproxy_stats = await self._haproxy.get_haproxy_stats()
|
|
return haproxy_stats.is_system_idle
|
|
|
|
async def check_health(self) -> bool:
|
|
# If haproxy is already shutdown, return False.
|
|
if not self._haproxy or not self._haproxy._proc:
|
|
return False
|
|
|
|
logger.debug("Received health check.", extra={"log_to_stderr": False})
|
|
return await self._haproxy.is_running()
|
|
|
|
def pong(self) -> str:
|
|
pass
|
|
|
|
async def receive_asgi_messages(self, request_metadata: RequestMetadata) -> bytes:
|
|
raise NotImplementedError("Receive is handled by the ingress replicas.")
|
|
|
|
async def receive_grpc_messages(
|
|
self, session_id: str
|
|
) -> Tuple[bool, Optional[Any], bool]:
|
|
raise NotImplementedError("Receive is handled by the ingress replicas.")
|
|
|
|
def _get_http_options(self) -> HTTPOptions:
|
|
return self._http_options
|
|
|
|
def _get_logging_config(self) -> Optional[str]:
|
|
"""Get the logging configuration (for testing purposes)."""
|
|
log_file_path = None
|
|
for handler in logger.handlers:
|
|
if isinstance(handler, logging.handlers.MemoryHandler):
|
|
log_file_path = handler.target.baseFilename
|
|
|
|
return log_file_path
|
|
|
|
def _generate_server_name(self, target: Target) -> str:
|
|
# The server name is derived from the replica's actor name, with the
|
|
# format `SERVE_REPLICA::<app>#<deployment>#<replica_id>`, or the
|
|
# proxy's actor name, with the format `SERVE_PROXY_ACTOR-<node_id>`.
|
|
# Special characters in the names are converted to comply with haproxy
|
|
# config's allowed characters, e.g. `#` -> `-`.
|
|
return self.get_safe_name(target.name)
|
|
|
|
def _target_to_server(self, target: Target) -> ServerConfig:
|
|
"""Convert a target to a server."""
|
|
# Use localhost if target is on the same node as HAProxy
|
|
host = get_localhost_ip() if target.ip == self._node_ip_address else target.ip
|
|
return ServerConfig(
|
|
name=self._generate_server_name(target),
|
|
host=host,
|
|
port=target.port,
|
|
# Unsanitized actor name; matches what /internal/route returns.
|
|
replica_id=target.name,
|
|
)
|
|
|
|
def _generate_backend_name(self, target_group: TargetGroup) -> str:
|
|
# The name is lowercased and formatted as <protocol>-<app_name>. Special
|
|
# characters in the name are converted to comply with haproxy config's
|
|
# allowed characters, e.g. `#` -> `-`.
|
|
return self.get_safe_name(
|
|
f"{target_group.protocol.value.lower()}-{target_group.app_name}"
|
|
)
|
|
|
|
def _create_backend_config(
|
|
self,
|
|
target_group: TargetGroup,
|
|
fallback_target: Optional[Target],
|
|
) -> BackendConfig:
|
|
"""Create a backend configuration from a target group and fallback target."""
|
|
servers = [self._target_to_server(target) for target in target_group.targets]
|
|
|
|
ingress_request_router_servers = [
|
|
self._target_to_server(target)
|
|
for target in target_group.ingress_request_router_targets
|
|
]
|
|
|
|
fallback_server = None
|
|
if fallback_target is not None:
|
|
fallback_server = self._target_to_server(fallback_target)
|
|
|
|
return BackendConfig(
|
|
name=self._generate_backend_name(target_group),
|
|
path_prefix=target_group.route_prefix,
|
|
servers=servers,
|
|
ingress_request_router_servers=ingress_request_router_servers,
|
|
app_name=target_group.app_name,
|
|
ingress_deployment_name=target_group.ingress_deployment_name,
|
|
fallback_server=fallback_server,
|
|
protocol=target_group.protocol,
|
|
)
|
|
|
|
async def _reload_haproxy(self) -> None:
|
|
# To avoid dropping updates from a long poll, we wait until HAProxy
|
|
# is up and running before attempting to generate config and reload.
|
|
# Use lock to serialize reloads and prevent race conditions with SO_REUSEPORT
|
|
async with self._reload_lock:
|
|
await self._haproxy_start_task
|
|
await self._haproxy.reload()
|
|
|
|
async def _update_haproxy_backends(self) -> None:
|
|
backend_configs = []
|
|
for target_group in self._target_groups:
|
|
fallback_target = None
|
|
if target_group.protocol == RequestProtocol.HTTP:
|
|
fallback_target = self._http_fallback_target
|
|
elif target_group.protocol == RequestProtocol.GRPC:
|
|
fallback_target = self._grpc_fallback_target
|
|
|
|
backend_config = self._create_backend_config(target_group, fallback_target)
|
|
backend_configs.append(backend_config)
|
|
|
|
logger.info(
|
|
f"Got updated backend configs: {backend_configs}.",
|
|
extra={"log_to_stderr": True},
|
|
)
|
|
|
|
name_to_backend_configs = {
|
|
backend_config.name: backend_config for backend_config in backend_configs
|
|
}
|
|
|
|
self._haproxy.set_backend_configs(name_to_backend_configs)
|
|
|
|
grpc_fallback_server = (
|
|
self._target_to_server(self._grpc_fallback_target)
|
|
if self._grpc_fallback_target is not None
|
|
else None
|
|
)
|
|
self._haproxy.set_grpc_fallback_server(grpc_fallback_server)
|
|
|
|
await self._reload_haproxy()
|
|
|
|
def update_target_groups(self, target_groups: List[TargetGroup]) -> None:
|
|
self._target_groups = target_groups
|
|
self._received_target_groups_broadcast = True
|
|
self._schedule_haproxy_update()
|
|
|
|
def update_fallback_targets(
|
|
self,
|
|
fallback_targets: Dict[RequestProtocol, Target],
|
|
) -> None:
|
|
# Reset so that fallbacks are repopulated from LongPoll.
|
|
self._http_fallback_target = None
|
|
self._grpc_fallback_target = None
|
|
|
|
for protocol, target in fallback_targets.items():
|
|
if protocol == RequestProtocol.HTTP:
|
|
self._http_fallback_target = target
|
|
elif protocol == RequestProtocol.GRPC:
|
|
self._grpc_fallback_target = target
|
|
|
|
self._schedule_haproxy_update()
|
|
|
|
def _schedule_haproxy_update(self) -> None:
|
|
"""Mark state dirty and ensure the apply task is running.
|
|
|
|
Broadcasts within RAY_SERVE_HAPROXY_BROADCAST_COALESCE_S collapse
|
|
into a single _update_haproxy_backends call. A window of 0 disables
|
|
the coalesce delay (each broadcast applies immediately) but still
|
|
runs through the same task, so update latency is always recorded.
|
|
"""
|
|
self._update_pending = True
|
|
if self._coalesce_armed_at is None:
|
|
self._coalesce_armed_at = time.monotonic()
|
|
if self._coalesce_task is None or self._coalesce_task.done():
|
|
self._coalesce_task = self.event_loop.create_task(
|
|
self._coalesce_and_apply()
|
|
)
|
|
|
|
async def _coalesce_and_apply(self) -> None:
|
|
# Awaits the reload so failures propagate here and so broadcasts
|
|
# arriving during a slow reload coalesce into the next cycle
|
|
# instead of queueing behind the reload lock. On failure, re-arm
|
|
# for up to 2 retries; past that, wait for the next broadcast.
|
|
consecutive_failures = 0
|
|
while self._update_pending:
|
|
if RAY_SERVE_HAPROXY_BROADCAST_COALESCE_S > 0:
|
|
await asyncio.sleep(RAY_SERVE_HAPROXY_BROADCAST_COALESCE_S)
|
|
self._update_pending = False
|
|
armed_at = self._coalesce_armed_at
|
|
self._coalesce_armed_at = None
|
|
try:
|
|
await self._update_haproxy_backends()
|
|
if armed_at is not None:
|
|
self._haproxy_update_latency.observe(time.monotonic() - armed_at)
|
|
consecutive_failures = 0
|
|
except Exception:
|
|
logger.exception("Coalesced HAProxy update failed.")
|
|
consecutive_failures += 1
|
|
if consecutive_failures < 3:
|
|
self._update_pending = True
|
|
# Keep measuring from the original broadcast so a retry's
|
|
# latency still reflects the full delay.
|
|
self._coalesce_armed_at = armed_at
|
|
|
|
def get_target_groups(self) -> List[TargetGroup]:
|
|
"""Get current target groups."""
|
|
return self._target_groups
|
|
|
|
@staticmethod
|
|
def get_safe_name(name: str) -> str:
|
|
"""Get a safe label name for the haproxy config."""
|
|
name = name.replace("#", "-").replace("/", ".")
|
|
# replace all remaining non-alphanumeric and non-{".", "_", "-"} with "_"
|
|
return re.sub(r"[^A-Za-z0-9._-]+", "_", name)
|
|
|
|
def _dump_ingress_replicas_for_testing(self, route: str) -> Set[ReplicaID]:
|
|
"""Return the set of replica IDs for targets matching the given route.
|
|
|
|
Args:
|
|
route: The route prefix to match against target groups.
|
|
|
|
Returns:
|
|
Set of ReplicaID objects for targets in the matching target group.
|
|
"""
|
|
replica_ids = set()
|
|
|
|
if self._target_groups is None:
|
|
return replica_ids
|
|
|
|
for target_group in self._target_groups:
|
|
if target_group.route_prefix == route:
|
|
for target in target_group.targets:
|
|
# Target names are in the format "SERVE_REPLICA::<app>#<deployment>#<replica_id>"
|
|
if ReplicaID.is_full_id_str(target.name):
|
|
replica_id = ReplicaID.from_full_id_str(target.name)
|
|
replica_ids.add(replica_id)
|
|
|
|
return replica_ids
|
|
|
|
def _dump_ingress_cache_for_testing(self, route: str) -> Set[ReplicaID]:
|
|
"""Return replica IDs that are cached/ready for the given route (for testing).
|
|
|
|
For HAProxy, all registered replicas are immediately ready for routing
|
|
(no warm-up cache like the internal router), so this returns the same
|
|
set as _dump_ingress_replicas_for_testing.
|
|
|
|
Args:
|
|
route: The route prefix to match against target groups.
|
|
|
|
Returns:
|
|
Set of ReplicaID objects for targets in the matching target group.
|
|
"""
|
|
return self._dump_ingress_replicas_for_testing(route)
|