import http import json import os import sys import threading import time from concurrent.futures import ThreadPoolExecutor from typing import Dict, Optional import grpc import httpx import pytest from fastapi import FastAPI, WebSocket from starlette.requests import Request from starlette.responses import PlainTextResponse from websockets.exceptions import ConnectionClosed from websockets.sync.client import connect import ray from ray import serve from ray._common.network_utils import parse_address from ray._common.test_utils import ( PrometheusTimeseries, SignalActor, fetch_prometheus_metric_timeseries, wait_for_condition, ) from ray.serve._private.constants import ( RAY_SERVE_CONTROLLER_METRICS_INCLUDE_HIGH_CARDINALITY_TAGS, RAY_SERVE_ENABLE_DIRECT_INGRESS, RAY_SERVE_ENABLE_HA_PROXY, ) from ray.serve._private.test_utils import ( PROMETHEUS_METRICS_TIMEOUT_S, TEST_METRICS_EXPORT_PORT, check_metric_float_eq, get_application_url, get_metric_dictionaries, get_metric_float, ping_grpc_call_method, ping_grpc_list_applications, ) from ray.serve._private.utils import block_until_http_ready from ray.serve.config import RequestRouterConfig from ray.serve.generated import serve_pb2, serve_pb2_grpc def extract_tags(line: str) -> Dict[str, str]: """Extracts any tags from the metrics line.""" try: tags_string = line.replace("{", "}").split("}")[1] except IndexError: # No tags were found in this line. return {} detected_tags = {} for tag_pair in tags_string.split(","): sanitized_pair = tag_pair.replace('"', "") tag, value = sanitized_pair.split("=") detected_tags[tag] = value return detected_tags def check_sum_metric_eq( metric_name: str, expected: float, tags: Optional[Dict[str, str]] = None, timeseries: Optional[PrometheusTimeseries] = None, ) -> bool: if tags is None: tags = {} if timeseries is None: timeseries = PrometheusTimeseries() metrics = fetch_prometheus_metric_timeseries( [f"localhost:{TEST_METRICS_EXPORT_PORT}"], timeseries, timeout=PROMETHEUS_METRICS_TIMEOUT_S, ) metrics = {k: v for k, v in metrics.items() if "ray_serve_" in k} metric_samples = metrics.get(metric_name, None) if metric_samples is None: metric_sum = 0 else: metric_samples = [ sample for sample in metric_samples if tags.items() <= sample.labels.items() ] metric_sum = sum(sample.value for sample in metric_samples) # Check the metrics sum to the expected number assert float(metric_sum) == float(expected), ( f"The following metrics don't sum to {expected}: " f"{json.dumps(metric_samples, indent=4)}\n." f"All metrics: {json.dumps(metrics, indent=4)}" ) # # For debugging if metric_samples: print(f"The following sum to {expected} for '{metric_name}' and tags {tags}:") for sample in metric_samples: print(sample) return True def test_serve_metrics_for_successful_connection(metrics_start_shutdown): @serve.deployment(name="metrics") async def f(request): return "hello" app_name = "app1" handle = serve.run(target=f.bind(), name=app_name) http_url = get_application_url(app_name=app_name) # send 10 concurrent requests ray.get([block_until_http_ready.remote(http_url) for _ in range(10)]) [handle.remote(http_url) for _ in range(10)] # Ping gPRC proxy grpc_url = get_application_url("gRPC", app_name=app_name) channel = grpc.insecure_channel(grpc_url) wait_for_condition( ping_grpc_list_applications, channel=channel, app_names=[app_name] ) def verify_metrics(do_assert=False): try: resp = httpx.get("http://127.0.0.1:9999").text # Requests will fail if we are crashing the controller except httpx.HTTPError: return False # NOTE: These metrics should be documented at # https://docs.ray.io/en/latest/serve/monitoring.html#metrics # Any updates to here should be reflected there too. expected_metrics = [ # counter "serve_num_router_requests", "serve_num_http_requests", "serve_num_grpc_requests", "serve_deployment_queued_queries", "serve_deployment_request_counter", "serve_deployment_replica_starts", # histogram "serve_deployment_processing_latency_ms_bucket", "serve_deployment_processing_latency_ms_count", "serve_deployment_processing_latency_ms_sum", "serve_deployment_processing_latency_ms", # gauge "serve_replica_processing_queries", "serve_deployment_replica_healthy", # handle "serve_handle_request_counter", ] for metric in expected_metrics: # For the final error round if do_assert: assert metric in resp # For the wait_for_condition else: if metric not in resp: return False return True try: wait_for_condition(verify_metrics, retry_interval_ms=500) except RuntimeError: verify_metrics(do_assert=True) def test_http_replica_gauge_metrics(metrics_start_shutdown): """Test http replica gauge metrics""" signal = SignalActor.remote() @serve.deployment(graceful_shutdown_timeout_s=0.0001) class A: async def __call__(self): await signal.wait.remote() handle = serve.run(A.bind(), name="app1") _ = handle.remote() processing_requests = get_metric_dictionaries( "ray_serve_replica_processing_queries", timeout=5 ) assert len(processing_requests) == 1 assert processing_requests[0]["deployment"] == "A" assert processing_requests[0]["application"] == "app1" print("serve_replica_processing_queries exists.") def ensure_request_processing(): resp = httpx.get("http://127.0.0.1:9999").text resp = resp.split("\n") for metrics in resp: if "# HELP" in metrics or "# TYPE" in metrics: continue if "serve_replica_processing_queries" in metrics: assert "1.0" in metrics return True wait_for_condition(ensure_request_processing, timeout=5) def test_proxy_metrics_not_found(metrics_start_shutdown): # NOTE: These metrics should be documented at # https://docs.ray.io/en/latest/serve/monitoring.html#metrics # Any updates here should be reflected there too. expected_metrics = [ "serve_num_http_requests", "serve_num_grpc_requests", "serve_num_http_error_requests", "serve_num_grpc_error_requests", "serve_num_deployment_http_error_requests", "serve_http_request_latency_ms", "serve_num_deployment_grpc_error_requests", "serve_grpc_request_latency_ms", ] def verify_metrics(_expected_metrics, do_assert=False): try: resp = httpx.get("http://127.0.0.1:9999").text # Requests will fail if we are crashing the controller except httpx.HTTPError: return False for metric in _expected_metrics: if do_assert: assert metric in resp if metric not in resp: return False return True # Create a dummy app so that there is a replica to hit for direct ingress tests. @serve.deployment() def f(*args): return "Hi" app_name = "app" serve.run(f.bind(), name=app_name, route_prefix="/app") serve.run(f.bind(), name="app2", route_prefix="/app2") # Trigger HTTP 404 error http_url = get_application_url("HTTP", app_name=app_name, exclude_route_prefix=True) httpx.get(f"{http_url}/B/") httpx.get(f"{http_url}/B/") # Ping gPRC proxy grpc_url = get_application_url("gRPC", app_name=app_name) channel = grpc.insecure_channel(grpc_url) ping_grpc_call_method(channel=channel, app_name="foo", test_not_found=True) # Ensure all expected metrics are present. try: wait_for_condition( verify_metrics, retry_interval_ms=1000, timeout=10, expected_metrics=expected_metrics, ) except RuntimeError: verify_metrics(expected_metrics, True) def verify_error_count(do_assert=False): resp = httpx.get("http://127.0.0.1:9999").text resp = resp.split("\n") for metrics in resp: if "# HELP" in metrics or "# TYPE" in metrics: continue if "serve_num_http_error_requests" in metrics: # route "/B/" should have error count 2 if do_assert: assert "2.0" in metrics if "2.0" not in metrics: return False elif "serve_num_deployment_http_error_requests" in metrics: # deployment B should have error count 2 if do_assert: assert 'error_code="404"' in metrics and "2.0" in metrics if 'error_code="404"' not in metrics or "2.0" not in metrics: return False elif "serve_num_grpc_error_requests" in metrics: # gRPC pinged "B" once if do_assert: assert "1.0" in metrics if "1.0" not in metrics: return False elif "serve_num_deployment_grpc_error_requests" in metrics: # gRPC pinged "B" once if do_assert: assert 'error_code="NOT_FOUND"' in metrics and "1.0" in metrics if 'error_code="NOT_FOUND"' not in metrics or "1.0" not in metrics: return False return True # There is a latency in updating the counter try: wait_for_condition(verify_error_count, retry_interval_ms=1000, timeout=10) except RuntimeError: verify_error_count(do_assert=True) def test_proxy_metrics_internal_error(metrics_start_shutdown): # This test kills the replica process so metrics are not emitted. if RAY_SERVE_ENABLE_DIRECT_INGRESS and not RAY_SERVE_ENABLE_HA_PROXY: pytest.skip() # NOTE: These metrics should be documented at # https://docs.ray.io/en/latest/serve/monitoring.html#metrics # Any updates here should be reflected there too. expected_metrics = [ "serve_num_http_requests", "serve_num_grpc_requests", "serve_num_http_error_requests", "serve_num_grpc_error_requests", "serve_num_deployment_http_error_requests", "serve_http_request_latency_ms", "serve_num_deployment_grpc_error_requests", "serve_grpc_request_latency_ms", ] def verify_metrics(_expected_metrics, do_assert=False): try: resp = httpx.get("http://127.0.0.1:9999", timeout=None).text # Requests will fail if we are crashing the controller except httpx.HTTPError: return False for metric in _expected_metrics: if do_assert: assert metric in resp if metric not in resp: return False return True @serve.deployment(name="A") class A: async def __init__(self): pass async def __call__(self, *args): # Trigger RayActorError os._exit(0) app_name = "app" serve.run(A.bind(), name=app_name) http_url = get_application_url("HTTP", app_name=app_name) _ = httpx.get(http_url, timeout=None) _ = httpx.get(http_url, timeout=None) grpc_url = get_application_url("gRPC", app_name=app_name) channel = grpc.insecure_channel(grpc_url) with pytest.raises(grpc.RpcError): ping_grpc_call_method(channel=channel, app_name=app_name) # Ensure all expected metrics are present. try: wait_for_condition( verify_metrics, retry_interval_ms=1000, timeout=10, expected_metrics=expected_metrics, ) except RuntimeError: verify_metrics(expected_metrics, True) def verify_error_count(do_assert=False): resp = httpx.get("http://127.0.0.1:9999", timeout=None).text resp = resp.split("\n") for metrics in resp: if "# HELP" in metrics or "# TYPE" in metrics: continue if "serve_num_http_error_requests" in metrics: # route "/A/" should have error count 2 if do_assert: assert "2.0" in metrics if "2.0" not in metrics: return False elif "serve_num_deployment_http_error_requests" in metrics: # deployment A should have error count 2 if do_assert: assert 'deployment="A"' in metrics and "2.0" in metrics if 'deployment="A"' not in metrics or "2.0" not in metrics: return False elif "serve_num_grpc_error_requests" in metrics: # gRPC pinged "A" once if do_assert: assert "1.0" in metrics if "1.0" not in metrics: return False elif "serve_num_deployment_grpc_error_requests" in metrics: # gRPC pinged "A" once if do_assert: assert 'deployment="A"' in metrics and "1.0" in metrics if 'deployment="A"' not in metrics or "1.0" not in metrics: return False return True # There is a latency in updating the counter try: wait_for_condition(verify_error_count, retry_interval_ms=1000, timeout=10) except RuntimeError: verify_error_count(do_assert=True) def test_proxy_metrics_fields_not_found(metrics_start_shutdown): """Tests the proxy metrics' fields' behavior for not found.""" # Create dummy apps so that there is a replica to hit for direct ingress tests. @serve.deployment() def f(*args): return "Hi" app_name = "app" serve.run(f.bind(), name=app_name, route_prefix="/app") serve.run(f.bind(), name="app2", route_prefix="/app2") # Should generate 404 responses app_url = get_application_url("HTTP", app_name=app_name, exclude_route_prefix=True) broken_url = f"{app_url}/fake_route" _ = httpx.get(broken_url).text print("Sent requests to broken URL.") # Ping gRPC proxy for not existing application. grpc_url = get_application_url("gRPC", app_name=app_name) channel = grpc.insecure_channel(grpc_url) fake_app_name = "fake-app" ping_grpc_call_method(channel=channel, app_name=fake_app_name, test_not_found=True) num_requests = get_metric_dictionaries("ray_serve_num_http_requests_total") assert len(num_requests) == 1 assert num_requests[0]["route"] == "" assert num_requests[0]["method"] == "GET" assert num_requests[0]["application"] == "" assert num_requests[0]["status_code"] == "404" print("serve_num_http_requests working as expected.") num_requests = get_metric_dictionaries("ray_serve_num_grpc_requests_total") assert len(num_requests) == 1 assert num_requests[0]["route"] == "" assert num_requests[0]["method"] == "/ray.serve.UserDefinedService/__call__" assert num_requests[0]["application"] == "" assert num_requests[0]["status_code"] == grpc.StatusCode.NOT_FOUND.name print("serve_num_grpc_requests working as expected.") num_errors = get_metric_dictionaries("ray_serve_num_http_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "" assert num_errors[0]["error_code"] == "404" assert num_errors[0]["method"] == "GET" print("serve_num_http_error_requests working as expected.") num_errors = get_metric_dictionaries("ray_serve_num_grpc_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "" assert num_errors[0]["error_code"] == grpc.StatusCode.NOT_FOUND.name assert num_errors[0]["method"] == "/ray.serve.UserDefinedService/__call__" print("serve_num_grpc_error_requests working as expected.") @pytest.mark.parametrize( "metrics_start_shutdown", [ 1, ], indirect=True, ) def test_proxy_timeout_metrics(metrics_start_shutdown): """Test that HTTP timeout metrics are reported correctly.""" signal = SignalActor.remote() @serve.deployment async def return_status_code_with_timeout(request: Request): await signal.wait.remote() return serve.run( return_status_code_with_timeout.bind(), route_prefix="/status_code_timeout", name="status_code_timeout", ) http_url = get_application_url("HTTP", app_name="status_code_timeout") r = httpx.get(http_url) assert r.status_code == 408 ray.get(signal.send.remote(clear=True)) # make grpc call grpc_url = get_application_url("gRPC", app_name="status_code_timeout") channel = grpc.insecure_channel(grpc_url) with pytest.raises(grpc.RpcError): ping_grpc_call_method(channel=channel, app_name="status_code_timeout") num_errors = get_metric_dictionaries("ray_serve_num_http_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "/status_code_timeout" assert num_errors[0]["error_code"] == "408" assert num_errors[0]["method"] == "GET" assert num_errors[0]["application"] == "status_code_timeout" num_errors = get_metric_dictionaries("ray_serve_num_grpc_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "status_code_timeout" assert num_errors[0]["error_code"] == grpc.StatusCode.DEADLINE_EXCEEDED.name assert num_errors[0]["method"] == "/ray.serve.UserDefinedService/__call__" assert num_errors[0]["application"] == "status_code_timeout" @pytest.mark.skipif(sys.platform == "win32", reason="Flaky on Windows") def test_proxy_disconnect_http_metrics(metrics_start_shutdown): """Test that HTTP disconnect metrics are reported correctly.""" signal = SignalActor.remote() @serve.deployment class Disconnect: async def __call__(self, request: Request): await signal.wait.remote() return serve.run( Disconnect.bind(), route_prefix="/disconnect", name="disconnect", ) # Simulate an HTTP disconnect http_url = get_application_url("HTTP", app_name="disconnect") ip_port = http_url.replace("http://", "").split("/")[0] # remove the route prefix ip, port = parse_address(ip_port) conn = http.client.HTTPConnection(ip, int(port)) conn.request("GET", "/disconnect") wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 1, timeout=10 ) conn.close() # Forcefully close the connection ray.get(signal.send.remote(clear=True)) num_errors = get_metric_dictionaries("ray_serve_num_http_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "/disconnect" assert num_errors[0]["error_code"] == "499" assert num_errors[0]["method"] == "GET" assert num_errors[0]["application"] == "disconnect" def test_proxy_disconnect_grpc_metrics(metrics_start_shutdown): """Test that gRPC disconnect metrics are reported correctly.""" signal = SignalActor.remote() @serve.deployment class Disconnect: async def __call__(self, request: Request): await signal.wait.remote() return serve.run( Disconnect.bind(), route_prefix="/disconnect", name="disconnect", ) # make grpc call grpc_url = get_application_url("gRPC", app_name="disconnect") channel = grpc.insecure_channel(grpc_url) stub = serve_pb2_grpc.UserDefinedServiceStub(channel) request = serve_pb2.UserDefinedMessage(name="foo", num=30, foo="bar") metadata = (("application", "disconnect"),) def make_request(): try: response = stub.__call__( request, metadata=metadata ) # Long-running RPC call print("Response received:", response) except grpc.RpcError as e: print("Client disconnected:", e.code(), e.details()) thread = threading.Thread(target=make_request) thread.start() # Wait briefly, then forcefully close the channel wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 1, timeout=10 ) channel.close() # Forcefully close the channel, simulating a client disconnect thread.join() ray.get(signal.send.remote(clear=True)) num_errors = get_metric_dictionaries("ray_serve_num_grpc_error_requests_total") assert len(num_errors) == 1 assert num_errors[0]["route"] == "disconnect" assert num_errors[0]["error_code"] == grpc.StatusCode.CANCELLED.name assert num_errors[0]["method"] == "/ray.serve.UserDefinedService/__call__" assert num_errors[0]["application"] == "disconnect" def test_proxy_metrics_fields_internal_error(metrics_start_shutdown): """Tests the proxy metrics' fields' behavior for internal error.""" @serve.deployment() def f(*args): return 1 / 0 real_app_name = "app" real_app_name2 = "app2" serve.run(f.bind(), name=real_app_name, route_prefix="/real_route") serve.run(f.bind(), name=real_app_name2, route_prefix="/real_route2") # Deployment should generate divide-by-zero errors correct_url = get_application_url("HTTP", real_app_name) _ = httpx.get(correct_url).text print("Sent requests to correct URL.") # Ping gPRC proxy for broken app grpc_url = get_application_url("gRPC", app_name=real_app_name) channel = grpc.insecure_channel(grpc_url) with pytest.raises(grpc.RpcError): ping_grpc_call_method(channel=channel, app_name=real_app_name) num_deployment_errors = get_metric_dictionaries( "ray_serve_num_deployment_http_error_requests_total" ) assert len(num_deployment_errors) == 1 assert num_deployment_errors[0]["deployment"] == "f" assert num_deployment_errors[0]["error_code"] == "500" assert num_deployment_errors[0]["method"] == "GET" assert num_deployment_errors[0]["application"] == "app" print("serve_num_deployment_http_error_requests working as expected.") num_deployment_errors = get_metric_dictionaries( "ray_serve_num_deployment_grpc_error_requests_total" ) assert len(num_deployment_errors) == 1 assert num_deployment_errors[0]["deployment"] == "f" assert num_deployment_errors[0]["error_code"] == grpc.StatusCode.INTERNAL.name assert ( num_deployment_errors[0]["method"] == "/ray.serve.UserDefinedService/__call__" ) assert num_deployment_errors[0]["application"] == real_app_name print("serve_num_deployment_grpc_error_requests working as expected.") latency_metrics = get_metric_dictionaries("ray_serve_http_request_latency_ms_sum") assert len(latency_metrics) == 1 assert latency_metrics[0]["method"] == "GET" assert latency_metrics[0]["route"] == "/real_route" assert latency_metrics[0]["application"] == "app" assert latency_metrics[0]["status_code"] == "500" print("serve_http_request_latency_ms working as expected.") latency_metrics = get_metric_dictionaries("ray_serve_grpc_request_latency_ms_sum") assert len(latency_metrics) == 1 assert latency_metrics[0]["method"] == "/ray.serve.UserDefinedService/__call__" assert latency_metrics[0]["route"] == real_app_name assert latency_metrics[0]["application"] == real_app_name assert latency_metrics[0]["status_code"] == grpc.StatusCode.INTERNAL.name print("serve_grpc_request_latency_ms_sum working as expected.") @pytest.mark.skipif(sys.platform == "win32", reason="Flaky on Windows") def test_proxy_metrics_http_status_code_is_error(metrics_start_shutdown): """Verify that 2xx and 3xx status codes aren't errors, others are.""" # TODO(eicherseiji): Remove skip when HAProxy is open-sourced. if RAY_SERVE_ENABLE_HA_PROXY: pytest.skip() def check_request_count_metrics( expected_error_count: int, expected_success_count: int, ): resp = httpx.get("http://127.0.0.1:9999").text error_count = 0 success_count = 0 for line in resp.split("\n"): if line.startswith("ray_serve_num_http_error_requests_total"): error_count += int(float(line.split(" ")[-1])) if line.startswith("ray_serve_num_http_requests_total"): success_count += int(float(line.split(" ")[-1])) assert error_count == expected_error_count assert success_count == expected_success_count return True @serve.deployment async def return_status_code(request: Request): code = int((await request.body()).decode("utf-8")) return PlainTextResponse("", status_code=code) serve.run(return_status_code.bind()) http_url = get_application_url("HTTP") # 200 is not an error. r = httpx.request("GET", http_url, content=b"200") assert r.status_code == 200 wait_for_condition( check_request_count_metrics, expected_error_count=0, expected_success_count=1, ) # 2xx is not an error. r = httpx.request("GET", http_url, content=b"250") assert r.status_code == 250 wait_for_condition( check_request_count_metrics, expected_error_count=0, expected_success_count=2, ) # 3xx is not an error. r = httpx.request("GET", http_url, content=b"300") assert r.status_code == 300 wait_for_condition( check_request_count_metrics, expected_error_count=0, expected_success_count=3, ) # 4xx is an error. r = httpx.request("GET", http_url, content=b"400") assert r.status_code == 400 wait_for_condition( check_request_count_metrics, expected_error_count=1, expected_success_count=4, ) # 5xx is an error. r = httpx.request("GET", http_url, content=b"500") assert r.status_code == 500 wait_for_condition( check_request_count_metrics, expected_error_count=2, expected_success_count=5, ) def test_proxy_metrics_websocket_status_code_is_error(metrics_start_shutdown): """Verify that status codes aisde from 1000 or 1001 are errors.""" def check_request_count_metrics( expected_error_count: int, expected_success_count: int, ): resp = httpx.get("http://127.0.0.1:9999").text error_count = 0 success_count = 0 for line in resp.split("\n"): if line.startswith("ray_serve_num_http_error_requests_total"): error_count += int(float(line.split(" ")[-1])) if line.startswith("ray_serve_num_http_requests_total"): success_count += int(float(line.split(" ")[-1])) assert error_count == expected_error_count assert success_count == expected_success_count return True fastapi_app = FastAPI() @serve.deployment @serve.ingress(fastapi_app) class WebSocketServer: @fastapi_app.websocket("/") async def accept_then_close(self, ws: WebSocket): await ws.accept() code = int(await ws.receive_text()) await ws.close(code=code) serve.run(WebSocketServer.bind()) # Regular disconnect (1000) is not an error. with connect("ws://localhost:8000/") as ws: with pytest.raises(ConnectionClosed): ws.send("1000") ws.recv() wait_for_condition( check_request_count_metrics, expected_error_count=0, expected_success_count=1, ) # Goaway disconnect (1001) is not an error. with connect("ws://localhost:8000/") as ws: with pytest.raises(ConnectionClosed): ws.send("1001") ws.recv() wait_for_condition( check_request_count_metrics, expected_error_count=0, expected_success_count=2, ) # Other codes are errors. with connect("ws://localhost:8000/") as ws: with pytest.raises(ConnectionClosed): ws.send("1011") ws.recv() wait_for_condition( check_request_count_metrics, expected_error_count=1, expected_success_count=3, ) # Other codes are errors. with connect("ws://localhost:8000/") as ws: with pytest.raises(ConnectionClosed): ws.send("3000") ws.recv() wait_for_condition( check_request_count_metrics, expected_error_count=2, expected_success_count=4, ) def test_replica_metrics_fields(metrics_start_shutdown): """Test replica metrics fields""" @serve.deployment def f(): return "hello" @serve.deployment def g(): return "world" serve.run(f.bind(), name="app1", route_prefix="/f") serve.run(g.bind(), name="app2", route_prefix="/g") url_f = get_application_url("HTTP", "app1") url_g = get_application_url("HTTP", "app2") assert "hello" == httpx.get(url_f).text assert "world" == httpx.get(url_g).text wait_for_condition( lambda: len( get_metric_dictionaries( "ray_serve_deployment_request_counter_total", wait=False ) ) == 2, timeout=40, ) metrics = get_metric_dictionaries("ray_serve_deployment_request_counter_total") assert len(metrics) == 2 expected_output = { ("/f", "f", "app1"), ("/g", "g", "app2"), } assert { ( metric["route"], metric["deployment"], metric["application"], ) for metric in metrics } == expected_output start_metrics = get_metric_dictionaries("ray_serve_deployment_replica_starts_total") assert len(start_metrics) == 2 expected_output = {("f", "app1"), ("g", "app2")} assert { (start_metric["deployment"], start_metric["application"]) for start_metric in start_metrics } == expected_output # Latency metrics wait_for_condition( lambda: len( get_metric_dictionaries( "ray_serve_deployment_processing_latency_ms_count", wait=False ) ) == 2, timeout=40, ) for metric_name in [ "ray_serve_deployment_processing_latency_ms_count", "ray_serve_deployment_processing_latency_ms_sum", ]: latency_metrics = get_metric_dictionaries(metric_name) print(f"checking metric {metric_name}, {latency_metrics}") assert len(latency_metrics) == 2 expected_output = {("f", "app1"), ("g", "app2")} assert { (latency_metric["deployment"], latency_metric["application"]) for latency_metric in latency_metrics } == expected_output wait_for_condition( lambda: len( get_metric_dictionaries("ray_serve_replica_processing_queries", wait=False) ) == 2, timeout=40, ) processing_queries = get_metric_dictionaries("ray_serve_replica_processing_queries") expected_output = {("f", "app1"), ("g", "app2")} assert { (processing_query["deployment"], processing_query["application"]) for processing_query in processing_queries } == expected_output @serve.deployment def h(): return 1 / 0 serve.run(h.bind(), name="app3", route_prefix="/h") url_h = get_application_url("HTTP", "app3") assert 500 == httpx.get(url_h).status_code wait_for_condition( lambda: len( get_metric_dictionaries( "ray_serve_deployment_error_counter_total", wait=False ) ) == 1, timeout=40, ) err_requests = get_metric_dictionaries("ray_serve_deployment_error_counter_total") assert len(err_requests) == 1 expected_output = ("/h", "h", "app3") assert ( err_requests[0]["route"], err_requests[0]["deployment"], err_requests[0]["application"], ) == expected_output assert err_requests[0]["exception_type"] == "ZeroDivisionError" expected_deployments = {("f", "app1"), ("g", "app2"), ("h", "app3")} health_timeseries = PrometheusTimeseries() def _check_replica_healthy(): metrics = get_metric_dictionaries( "ray_serve_deployment_replica_healthy", wait=False, timeseries=health_timeseries, ) return { (m["deployment"], m["application"]) for m in metrics } >= expected_deployments wait_for_condition(_check_replica_healthy, timeout=40) health_metrics = get_metric_dictionaries( "ray_serve_deployment_replica_healthy", timeseries=health_timeseries ) assert { (health_metric["deployment"], health_metric["application"]) for health_metric in health_metrics } >= expected_deployments def test_deployment_error_counter_exception_type(metrics_start_shutdown): """Test that ray_serve_deployment_error_counter_total captures exception_type tag.""" @serve.deployment def raises_value_error(): raise ValueError("intentional test error") serve.run( raises_value_error.bind(), name="value_error_app", route_prefix="/value_error" ) url = get_application_url("HTTP", "value_error_app") + "/value_error" assert httpx.get(url).status_code == 500 def check_metric(): err_metrics = get_metric_dictionaries( "ray_serve_deployment_error_counter_total", wait=False ) value_error_metrics = [ m for m in err_metrics if m.get("exception_type") == "ValueError" ] if len(value_error_metrics) == 1: assert value_error_metrics[0]["exception_type"] == "ValueError" return True return False wait_for_condition(check_metric, timeout=40) def test_queue_wait_time_metric(metrics_start_shutdown): """Test that queue wait time metric is recorded correctly.""" signal = SignalActor.remote() @serve.deployment(max_ongoing_requests=1) class SlowDeployment: async def __call__(self): await signal.wait.remote() return "done" handle = serve.run(SlowDeployment.bind(), name="app1", route_prefix="/slow") futures = [handle.remote() for _ in range(2)] wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 1, timeout=10 ) time.sleep(0.5) ray.get(signal.send.remote()) [f.result() for f in futures] timeseries = PrometheusTimeseries() def check_queue_wait_time_metric(): metrics = get_metric_dictionaries( "ray_serve_request_router_fulfillment_time_ms_sum", timeseries=timeseries, wait=False, ) if not metrics: return False for metric in metrics: if ( metric.get("deployment") == "SlowDeployment" and metric.get("application") == "app1" ): return True return False wait_for_condition(check_queue_wait_time_metric, timeout=10) def check_queue_wait_time_metric_value(): value = get_metric_float( "ray_serve_request_router_fulfillment_time_ms_sum", timeseries=timeseries, expected_tags={"deployment": "SlowDeployment", "application": "app1"}, ) assert value > 400, f"Queue wait time should be greater than 500ms, got {value}" return True wait_for_condition(check_queue_wait_time_metric_value, timeout=10) wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 0, timeout=10 ) def test_router_queue_len_metric(metrics_start_shutdown): """Test that router queue length metric is recorded correctly per replica.""" signal = SignalActor.remote() @serve.deployment(max_ongoing_requests=10) class TestDeployment: async def __call__(self, request: Request): await signal.wait.remote() return "done" serve.run(TestDeployment.bind(), name="app1", route_prefix="/test") # Send a request that will block with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(httpx.get, "http://localhost:8000/test", timeout=60) # Wait for request to reach the replica wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 1, timeout=15 ) timeseries = PrometheusTimeseries() # Check that the router queue length metric appears with correct tags def check_router_queue_len(): metrics = get_metric_dictionaries( "ray_serve_request_router_queue_len", timeseries=timeseries, wait=False ) if not metrics: return False # Find metric for our deployment with replica_id tag for metric in metrics: if ( metric.get("deployment") == "TestDeployment" and metric.get("application") == "app1" and "replica_id" in metric ): # Check that required tags are present assert ( "handle_source" in metric ), "handle_source tag should be present" print(f"Found router queue len metric: {metric}") return True return False wait_for_condition(check_router_queue_len, timeout=30) wait_for_condition( check_metric_float_eq, timeout=15, metric="ray_serve_request_router_queue_len", expected=1, expected_tags={"deployment": "TestDeployment", "application": "app1"}, timeseries=timeseries, ) print("Router queue len metric verified.") # Release request ray.get(signal.send.remote()) future.result() @pytest.mark.skipif(sys.platform == "win32", reason="Flaky on Windows") def test_multiplexed_metrics(metrics_start_shutdown): """Tests multiplexed API corresponding metrics.""" @serve.deployment class Model: @serve.multiplexed(max_num_models_per_replica=2) async def get_model(self, model_id: str): return model_id async def __call__(self, model_id: str): await self.get_model(model_id) return # Multiplexing is not supported on the ingress deployment when direct ingress / # HAProxy is enabled, so keep the multiplexed deployment downstream of a plain # ingress. @serve.deployment class Ingress: def __init__(self, model): self._model = model async def __call__(self, model_id: str): await self._model.remote(model_id) handle = serve.run(Ingress.bind(Model.bind()), name="app", route_prefix="/app") handle.remote("model1") handle.remote("model2") # Trigger model eviction. handle.remote("model3") expected_metrics = [ "serve_multiplexed_model_load_latency_ms", "serve_multiplexed_model_unload_latency_ms", "serve_num_multiplexed_models", "serve_multiplexed_models_load_counter", "serve_multiplexed_models_unload_counter", ] def verify_metrics(): try: resp = httpx.get("http://127.0.0.1:9999").text # Requests will fail if we are crashing the controller except httpx.HTTPError: return False for metric in expected_metrics: assert metric in resp return True wait_for_condition( verify_metrics, timeout=40, retry_interval_ms=1000, ) @pytest.mark.parametrize("use_factory_pattern", [False, True]) def test_proxy_metrics_with_route_patterns(metrics_start_shutdown, use_factory_pattern): """Test that proxy metrics use specific route patterns for FastAPI apps. This test verifies that: 1. Route patterns are extracted from FastAPI apps at replica initialization 2. Proxy metrics use parameterized patterns (e.g., /api/users/{user_id}) instead of just route prefixes (e.g., /api) 3. Individual request paths don't appear in metrics (avoiding high cardinality) 4. Multiple requests to the same pattern are grouped together 5. Both normal pattern and factory pattern work correctly """ if use_factory_pattern: # Factory pattern: callable returns FastAPI app at runtime def create_app(): app = FastAPI() @app.get("/") def root(): return {"message": "root"} @app.get("/users/{user_id}") def get_user(user_id: str): return {"user_id": user_id} @app.get("/items/{item_id}/details") def get_item(item_id: str): return {"item_id": item_id} return app @serve.deployment @serve.ingress(create_app) class APIServer: pass else: # Normal pattern: routes defined in deployment class app = FastAPI() @serve.deployment @serve.ingress(app) class APIServer: @app.get("/") def root(self): return {"message": "root"} @app.get("/users/{user_id}") def get_user(self, user_id: str): return {"user_id": user_id} @app.get("/items/{item_id}/details") def get_item(self, item_id: str): return {"item_id": item_id} serve.run(APIServer.bind(), name="api_app", route_prefix="/api") # Make requests to different route patterns with various parameter values base_url = get_application_url("HTTP", app_name="api_app") assert httpx.get(f"{base_url}/").status_code == 200 assert httpx.get(f"{base_url}/users/123").status_code == 200 assert httpx.get(f"{base_url}/users/456").status_code == 200 assert httpx.get(f"{base_url}/users/789").status_code == 200 assert httpx.get(f"{base_url}/items/abc/details").status_code == 200 assert httpx.get(f"{base_url}/items/xyz/details").status_code == 200 # Wait for metrics to be updated def metrics_available(): metrics = get_metric_dictionaries( "ray_serve_num_http_requests_total", wait=False ) api_metrics = [m for m in metrics if m.get("application") == "api_app"] return len(api_metrics) >= 3 wait_for_condition(metrics_available, timeout=20) # Verify metrics use route patterns, not individual paths metrics = get_metric_dictionaries("ray_serve_num_http_requests_total") api_metrics = [m for m in metrics if m.get("application") == "api_app"] routes = {m["route"] for m in api_metrics} print(f"Routes found in metrics: {routes}") # Should contain the route patterns (parameterized), not just the prefix # The root might be either "/api/" or "/api" depending on normalization assert any( r in routes for r in ["/api/", "/api"] ), f"Root route not found. Routes: {routes}" # Should contain parameterized user route assert ( "/api/users/{user_id}" in routes ), f"User route pattern not found. Routes: {routes}" # Should contain nested parameterized route assert ( "/api/items/{item_id}/details" in routes ), f"Item details route pattern not found. Routes: {routes}" # Should NOT contain individual request paths (that would be high cardinality) # These should not appear as they would create unbounded cardinality assert ( "/api/users/123" not in routes ), "Individual user path found - high cardinality issue!" assert ( "/api/users/456" not in routes ), "Individual user path found - high cardinality issue!" assert ( "/api/users/789" not in routes ), "Individual user path found - high cardinality issue!" assert ( "/api/items/abc/details" not in routes ), "Individual item path found - high cardinality issue!" assert ( "/api/items/xyz/details" not in routes ), "Individual item path found - high cardinality issue!" # Verify that multiple requests to the same pattern are grouped user_route_metrics = [ m for m in api_metrics if m["route"] == "/api/users/{user_id}" ] assert ( len(user_route_metrics) == 1 ), "Multiple metrics entries for same route pattern - should be grouped!" # Optionally verify the counter value if we can parse it from the metrics endpoint metrics_text = httpx.get("http://127.0.0.1:9999").text for line in metrics_text.split("\n"): if "serve_num_http_requests" in line and "/api/users/{user_id}" in line: # Extract the value from the prometheus format line value_str = line.split()[-1] user_metric_value = float(value_str) assert ( user_metric_value == 3 ), f"Expected exactly 3 requests to user route, got {user_metric_value}" break # Verify error metrics also use route patterns num_errors = get_metric_dictionaries("ray_serve_http_request_latency_ms_sum") api_latency_metrics = [m for m in num_errors if m.get("application") == "api_app"] latency_routes = {m["route"] for m in api_latency_metrics} # Latency metrics should also use patterns assert ( "/api/users/{user_id}" in latency_routes or "/api/" in latency_routes ), f"Latency metrics should use route patterns. Found: {latency_routes}" def _check_controller_high_cardinality_metric_tags(include_high_cardinality: bool): """Test controller metrics respect high-cardinality tag config.""" @ray.remote class ReplicaHealthState: def __init__(self): self.replica_ids = set() self.failures_enabled = False self.failing_replica_id = None def get_num_registered_replicas(self) -> int: return len(self.replica_ids) def enable_failures(self): self.failures_enabled = True def register_and_should_fail_health_check(self, replica_id: str) -> bool: self.replica_ids.add(replica_id) if not self.failures_enabled: return False if self.failing_replica_id is None: self.failing_replica_id = replica_id return replica_id == self.failing_replica_id signal = SignalActor.remote() replica_health_state = ReplicaHealthState.remote() @serve.deployment( name="autoscaling_metrics_model", autoscaling_config={ "min_replicas": 1, "max_replicas": 5, "target_ongoing_requests": 2, "metrics_interval_s": 0.1, "upscale_delay_s": 0, "downscale_delay_s": 5, "look_back_period_s": 1, }, max_ongoing_requests=10, graceful_shutdown_timeout_s=0.1, ) class AutoscalingModel: async def __call__(self): await signal.wait.remote() return "hello" async def record_autoscaling_stats(self): return {"custom_metric": 1} @serve.deployment( name="lifecycle_metrics_model", num_replicas=2, health_check_period_s=0.1, health_check_timeout_s=1, graceful_shutdown_timeout_s=0.1, ) class LifecycleModel: async def __call__(self): return serve.get_replica_context().replica_tag async def check_health(self): replica_id = serve.get_replica_context().replica_tag should_fail_health_check = ( await replica_health_state.register_and_should_fail_health_check.remote( replica_id ) ) if should_fail_health_check: raise RuntimeError("Intentional health check failure.") autoscaling_app_name = "autoscaling_metrics_app" autoscaling_deployment_name = "autoscaling_metrics_model" lifecycle_app_name = "lifecycle_metrics_app" lifecycle_deployment_name = "lifecycle_metrics_model" serve.run( AutoscalingModel.bind(), name=autoscaling_app_name, route_prefix="/autoscaling", ) serve.run( LifecycleModel.bind(), name=lifecycle_app_name, route_prefix="/lifecycle", ) wait_for_condition( lambda: ray.get(replica_health_state.get_num_registered_replicas.remote()) == 2, timeout=60, ) timeseries = PrometheusTimeseries() def get_health_status_value(deployment: str, application: str) -> float: return get_metric_float( "ray_serve_deployment_replica_healthy", { "deployment": deployment, "application": application, }, timeseries=timeseries, timeout=PROMETHEUS_METRICS_TIMEOUT_S, ) if not include_high_cardinality: wait_for_condition( lambda: get_health_status_value( lifecycle_deployment_name, lifecycle_app_name ) == 2, timeout=60, ) ray.get(replica_health_state.enable_failures.remote()) handle = serve.get_deployment_handle( autoscaling_deployment_name, autoscaling_app_name ) [handle.remote() for _ in range(10)] def get_matching_metrics(metric_name: str, deployment: str, application: str): return [ metric for metric in get_metric_dictionaries( metric_name, timeseries=timeseries, wait=False ) if metric.get("deployment") == deployment and metric.get("application") == application ] def assert_high_cardinality_tag(metric, tag): assert (tag in metric) is include_high_cardinality def check_controller_metric_tags(): health_failure_metrics = get_matching_metrics( "ray_serve_health_check_failures_total", lifecycle_deployment_name, lifecycle_app_name, ) health_status_metrics = get_matching_metrics( "ray_serve_deployment_replica_healthy", lifecycle_deployment_name, lifecycle_app_name, ) if not health_failure_metrics or not health_status_metrics: return False for metric in health_failure_metrics: assert_high_cardinality_tag(metric, "replica") for metric in health_status_metrics: assert_high_cardinality_tag(metric, "replica") return True try: wait_for_condition(check_controller_metric_tags, timeout=60) finally: ray.get(signal.send.remote()) @pytest.mark.skipif( not RAY_SERVE_CONTROLLER_METRICS_INCLUDE_HIGH_CARDINALITY_TAGS, reason="controller metric high-cardinality tags are disabled", ) def test_controller_high_cardinality_metric_tags(metrics_start_shutdown): _check_controller_high_cardinality_metric_tags(include_high_cardinality=True) @pytest.mark.skipif( RAY_SERVE_CONTROLLER_METRICS_INCLUDE_HIGH_CARDINALITY_TAGS, reason="controller metric high-cardinality tags are enabled", ) def test_disable_high_cardinality_controller_metrics(metrics_start_shutdown): _check_controller_high_cardinality_metric_tags(include_high_cardinality=False) def test_routing_stats_delay_metric(metrics_start_shutdown): """Test that routing stats delay metric is reported correctly.""" @serve.deployment class Model: def __call__(self): return "hello" async def record_routing_stats(self): return {} serve.run(Model.bind(), name="app") timeseries = PrometheusTimeseries() # Wait for routing stats delay metric to be reported # This metric is recorded when the controller polls routing stats from replicas def check_routing_stats_delay_metric(): metrics = get_metric_dictionaries( "ray_serve_routing_stats_delay_ms_count", timeseries=timeseries, wait=False ) if not metrics: return False # Check that at least one metric has expected tags (no per-replica label) for metric in metrics: assert metric["deployment"] == "Model" assert metric["application"] == "app" assert "replica" not in metric return True return False wait_for_condition(check_routing_stats_delay_metric, timeout=60) # Verify the metric value is greater than 0 def check_routing_stats_delay_metric_value(): value = get_metric_float( "ray_serve_routing_stats_delay_ms_count", timeseries=timeseries, expected_tags={ "deployment": "Model", "application": "app", }, ) return value > 0 wait_for_condition(check_routing_stats_delay_metric_value, timeout=60) def test_routing_stats_error_metric(metrics_start_shutdown): """Test that routing stats error metric is reported on exception and timeout.""" signal = SignalActor.remote() @serve.deployment( request_router_config=RequestRouterConfig( request_routing_stats_period_s=0.1, request_routing_stats_timeout_s=0.5 ) ) class FailingModel: def __init__(self, signal_actor): self.should_fail = False self.should_hang = False self.signal = signal_actor async def record_routing_stats(self): if self.should_hang: await self.signal.wait.remote() if self.should_fail: raise Exception("Intentional failure for testing") return {} def __call__(self): return "hello" def set_should_fail(self, value: bool): self.should_fail = value def set_should_hang(self, value: bool): self.should_hang = value handle = serve.run(FailingModel.bind(signal), name="error_app") timeseries = PrometheusTimeseries() # Make a request to ensure deployment is running handle.remote().result() # Trigger exception in record_routing_stats handle.set_should_fail.remote(True).result() # Make requests to trigger routing stats collection for _ in range(5): handle.remote().result() # Check that error metric with error_type="exception" is reported def check_exception_error_metric(): metrics = get_metric_dictionaries( "ray_serve_routing_stats_error_total", timeseries=timeseries, wait=False ) for metric in metrics: if ( metric.get("deployment") == "FailingModel" and metric.get("application") == "error_app" and metric.get("error_type") == "exception" ): assert "replica" in metric return True return False wait_for_condition(check_exception_error_metric, timeout=30) print("Exception error metric verified.") # Now test timeout case handle.set_should_fail.remote(False).result() handle.set_should_hang.remote(True).result() # Make requests to trigger routing stats timeout for _ in range(5): handle.remote().result() # Check that error metric with error_type="timeout" is reported def check_timeout_error_metric(): metrics = get_metric_dictionaries( "ray_serve_routing_stats_error_total", timeseries=timeseries, wait=False ) for metric in metrics: if ( metric.get("deployment") == "FailingModel" and metric.get("application") == "error_app" and metric.get("error_type") == "timeout" ): assert "replica" in metric return True return False wait_for_condition(check_timeout_error_metric, timeout=30) print("Timeout error metric verified.") ray.get(signal.send.remote(clear=True)) def test_replica_utilization_metric(metrics_start_shutdown): """Test that the replica utilization metric is correctly reported. This test verifies that: 1. The serve_replica_utilization_percent metric is emitted 2. It has the correct tags (deployment, application, replica) 3. The value is within the expected range (0-100) The utilization window and report interval are configured via env vars in BUILD.bazel (RAY_SERVE_REPLICA_UTILIZATION_WINDOW_S, etc.). With max_ongoing_requests=1 and continuous 800ms-sleep requests the replica is busy ~80% of the time in steady state. We wait for one full window duration so the window is saturated, then assert >= 70%. """ @serve.deployment(name="UtilizationTest", max_ongoing_requests=1) class SlowDeployment: def __call__(self): # Sleep for 800ms per request to generate ~80% utilization. time.sleep(0.8) return "ok" app_name = "utilization_app" handle = serve.run(SlowDeployment.bind(), name=app_name) # Continuously send requests in a background thread to maintain utilization # throughout the rolling window while we poll for the metric. stop_sending = threading.Event() def _send_requests_forever(): while not stop_sending.is_set(): try: handle.remote().result() except Exception: pass sender = threading.Thread(target=_send_requests_forever, daemon=True) sender.start() try: # Wait for the rolling window to fill up with continuous requests so # we observe steady-state utilization rather than a ramp-up value. window_s = float(os.environ.get("RAY_SERVE_REPLICA_UTILIZATION_WINDOW_S", "5")) time.sleep(window_s) timeseries = PrometheusTimeseries() # Wait for the utilization metric to be reported def check_utilization_metric_exists(): metrics = get_metric_dictionaries( "ray_serve_replica_utilization_percent", timeseries=timeseries, wait=False, ) if not metrics: return False # Check that at least one metric has the expected tags for metric in metrics: if ( metric.get("deployment") == "UtilizationTest" and metric.get("application") == app_name and "replica" in metric ): return True return False wait_for_condition(check_utilization_metric_exists, timeout=30) print("Replica utilization metric exists with correct tags.") # Verify the metric value is within expected range def check_utilization_metric_value(): value = get_metric_float( "ray_serve_replica_utilization_percent", timeseries=timeseries, expected_tags={ "deployment": "UtilizationTest", "application": app_name, }, ) # Value should be between 0 and 100 assert 0 <= value <= 100, f"Utilization should be 0-100, got {value}" # With continuous 800ms requests and max_ongoing_requests=1 the # theoretical steady-state utilization is ~80%. After sleeping for # one full window duration the window should be saturated. assert ( value >= 70 ), f"Utilization should be >= 70 at steady state, got {value}" print(f"Replica utilization value: {value}%") return True wait_for_condition(check_utilization_metric_value, timeout=30) print("Replica utilization metric value verified.") finally: stop_sending.set() sender.join(timeout=10) def test_max_processing_latency_metric(metrics_start_shutdown): """Test that the max processing latency metric is correctly reported per route. This test verifies that: 1. The serve_deployment_max_processing_latency_ms metric is emitted 2. Separate max values are tracked per route tag 3. Each route's max reflects its actual maximum latency 4. A fast route has a lower max than a slow route Uses a FastAPI app with two routes having different sleep durations to produce distinct per-route max latencies. """ app = FastAPI() @serve.deployment(name="MaxLatencyTest", max_ongoing_requests=2) @serve.ingress(app) class MultiRouteDeployment: @app.get("/fast") def fast(self): time.sleep(0.1) return {"route": "fast"} @app.get("/slow") def slow(self): time.sleep(0.8) return {"route": "slow"} app_name = "max_latency_app" serve.run(MultiRouteDeployment.bind(), name=app_name, route_prefix="/api") stop_sending = threading.Event() http_url = get_application_url("HTTP", app_name=app_name) def _send_requests_forever(route: str): while not stop_sending.is_set(): try: httpx.get(f"{http_url}{route}", timeout=5) except Exception: pass fast_sender = threading.Thread( target=_send_requests_forever, args=("/fast",), daemon=True ) slow_sender = threading.Thread( target=_send_requests_forever, args=("/slow",), daemon=True ) fast_sender.start() slow_sender.start() try: report_interval_s = float( os.environ.get( "RAY_SERVE_REPLICA_MAX_PROCESSING_LATENCY_REPORT_INTERVAL_S", "10" ) ) time.sleep(report_interval_s + 2) timeseries = PrometheusTimeseries() def check_both_routes_reported(): metrics = get_metric_dictionaries( "ray_serve_deployment_max_processing_latency_ms", timeseries=timeseries, wait=False, ) if not metrics: return False routes_found = set() for metric in metrics: if ( metric.get("deployment") == "MaxLatencyTest" and metric.get("application") == app_name ): route = metric.get("route", "") routes_found.add(route) return "/api/fast" in routes_found and "/api/slow" in routes_found wait_for_condition(check_both_routes_reported, timeout=30) def check_per_route_values(): fast_value = get_metric_float( "ray_serve_deployment_max_processing_latency_ms", timeseries=timeseries, expected_tags={ "deployment": "MaxLatencyTest", "application": app_name, "route": "/api/fast", }, ) slow_value = get_metric_float( "ray_serve_deployment_max_processing_latency_ms", timeseries=timeseries, expected_tags={ "deployment": "MaxLatencyTest", "application": app_name, "route": "/api/slow", }, ) assert fast_value >= 0, f"Fast max latency should be >= 0, got {fast_value}" assert slow_value >= 0, f"Slow max latency should be >= 0, got {slow_value}" # /fast sleeps 100ms, /slow sleeps 800ms. assert ( fast_value >= 80 ), f"Fast route max should be >= 80ms with 100ms sleep, got {fast_value}" assert ( slow_value >= 600 ), f"Slow route max should be >= 600ms with 800ms sleep, got {slow_value}" assert ( slow_value > fast_value ), f"Slow route ({slow_value}ms) should exceed fast route ({fast_value}ms)" return True wait_for_condition(check_per_route_values, timeout=30) finally: stop_sending.set() fast_sender.join(timeout=10) slow_sender.join(timeout=10) def test_objref_resolution_latency_metric(metrics_start_shutdown): """Test that objref resolution latency metric is emitted when a DeploymentResponse is passed as an argument to another handle call. """ signal = SignalActor.remote() @serve.deployment class Upstream: async def __call__(self): await signal.wait.remote() return "upstream_result" @serve.deployment class Downstream: def __call__(self, val: str): return f"got_{val}" @serve.deployment class Router: def __init__(self, upstream, downstream): self._upstream = upstream self._downstream = downstream async def __call__(self): upstream_resp = self._upstream.remote() downstream_resp = self._downstream.remote(upstream_resp) return await downstream_resp serve.run( Router.bind(Upstream.bind(), Downstream.bind()), name="app1", route_prefix="/chain", ) url = get_application_url("HTTP", "app1") + "/chain" with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(httpx.get, url, timeout=30) wait_for_condition( lambda: ray.get(signal.cur_num_waiters.remote()) == 1, timeout=10 ) time.sleep(0.5) ray.get(signal.send.remote()) resp = future.result() assert resp.status_code == 200 assert resp.text == "got_upstream_result" timeseries = PrometheusTimeseries() def check_objref_resolution_metric(): metrics = get_metric_dictionaries( "ray_serve_router_args_resolution_latency_ms_count", timeseries=timeseries, wait=False, ) if not metrics: return False for metric in metrics: if ( metric.get("deployment") == "Downstream" and metric.get("application") == "app1" ): handle = metric.get("handle", "") actor_id = metric.get("actor_id", "") if handle and actor_id: return True return False wait_for_condition(check_objref_resolution_metric, timeout=30) def check_objref_resolution_metric_value(): value = get_metric_float( "ray_serve_router_args_resolution_latency_ms_sum", timeseries=timeseries, expected_tags={"deployment": "Downstream", "application": "app1"}, ) if value < 0: return False assert value >= 400, f"Resolution latency should be >= 400ms got {value}" return True wait_for_condition(check_objref_resolution_metric_value, timeout=30) if __name__ == "__main__": sys.exit(pytest.main(["-v", "-s"] + sys.argv[1:] + [__file__]))