wehub-resource-sync
1273 lines
43 KiB
Python
1273 lines
43 KiB
Python
import asyncio
|
|
import concurrent.futures
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
from typing import Dict, List
|
|
|
|
import httpx
|
|
import pytest
|
|
import redis
|
|
from starlette.requests import Request
|
|
|
|
import ray
|
|
from ray import serve
|
|
from ray._common.test_utils import PrometheusTimeseries, SignalActor, wait_for_condition
|
|
from ray.serve._private.common import DeploymentID
|
|
from ray.serve._private.constants import (
|
|
RAY_SERVE_RUN_ROUTER_IN_SEPARATE_LOOP,
|
|
RAY_SERVE_RUN_USER_CODE_IN_SEPARATE_THREAD,
|
|
SERVE_CONTROLLER_NAME,
|
|
SERVE_NAMESPACE,
|
|
)
|
|
from ray.serve._private.long_poll import LongPollClient, LongPollHost, UpdatedObject
|
|
from ray.serve._private.queue_monitor import (
|
|
create_queue_monitor_actor,
|
|
kill_queue_monitor_actor,
|
|
)
|
|
from ray.serve._private.test_utils import (
|
|
check_metric_float_eq,
|
|
get_application_url,
|
|
get_metric_dictionaries,
|
|
get_metric_float,
|
|
)
|
|
from ray.tests.conftest import external_redis # noqa: F401
|
|
from ray.util.state import list_actors
|
|
|
|
|
|
def test_deployment_and_application_status_metrics(metrics_start_shutdown):
|
|
"""Test that deployment and application status metrics are exported correctly.
|
|
|
|
These metrics track the numeric status of deployments and applications:
|
|
- serve_deployment_status: 0=UNKNOWN, 1=DEPLOY_FAILED, 2=UNHEALTHY,
|
|
3=UPDATING, 4=UPSCALING, 5=DOWNSCALING, 6=HEALTHY
|
|
- serve_application_status: 0=UNKNOWN, 1=NOT_STARTED, 2=DEPLOYING,
|
|
3=DEPLOY_FAILED, 4=RUNNING, 5=UNHEALTHY, 6=DELETING
|
|
"""
|
|
|
|
signal = SignalActor.remote()
|
|
|
|
@serve.deployment(name="deployment_a")
|
|
class DeploymentA:
|
|
async def __init__(self):
|
|
await signal.wait.remote()
|
|
|
|
async def __call__(self):
|
|
return "hello"
|
|
|
|
@serve.deployment
|
|
def deployment_b():
|
|
return "world"
|
|
|
|
# Deploy two applications with different deployments
|
|
serve._run(DeploymentA.bind(), name="app1", route_prefix="/app1", _blocking=False)
|
|
serve._run(deployment_b.bind(), name="app2", route_prefix="/app2", _blocking=False)
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
|
|
# Wait for deployments to become healthy
|
|
def check_status_metrics():
|
|
# Check deployment status metrics
|
|
deployment_metrics = get_metric_dictionaries(
|
|
"ray_serve_deployment_status", timeseries=timeseries, wait=False
|
|
)
|
|
if len(deployment_metrics) < 2:
|
|
return False
|
|
|
|
# Check application status metrics
|
|
app_metrics = get_metric_dictionaries(
|
|
"ray_serve_application_status", timeseries=timeseries, wait=False
|
|
)
|
|
if len(app_metrics) < 2:
|
|
return False
|
|
|
|
return True
|
|
|
|
wait_for_condition(check_status_metrics, timeout=30)
|
|
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_deployment_status",
|
|
expected=3, # UPDATING
|
|
expected_tags={"deployment": "deployment_a", "application": "app1"},
|
|
timeseries=timeseries,
|
|
)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_application_status",
|
|
expected=5, # DEPLOYING
|
|
expected_tags={"application": "app1"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_deployment_status",
|
|
expected=6,
|
|
expected_tags={"deployment": "deployment_b", "application": "app2"},
|
|
timeseries=timeseries,
|
|
)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_application_status",
|
|
expected=6,
|
|
expected_tags={"application": "app2"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
ray.get(signal.send.remote())
|
|
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_deployment_status",
|
|
expected=6,
|
|
expected_tags={"deployment": "deployment_a", "application": "app1"},
|
|
timeseries=timeseries,
|
|
)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
metric="ray_serve_application_status",
|
|
expected=6,
|
|
expected_tags={"application": "app1"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
|
|
def test_replica_startup_and_initialization_latency_metrics(metrics_start_shutdown):
|
|
"""Test that replica startup and initialization latency metrics are recorded."""
|
|
|
|
@serve.deployment(num_replicas=2)
|
|
class MyDeployment:
|
|
def __init__(self):
|
|
time.sleep(1)
|
|
|
|
def __call__(self):
|
|
return "hello"
|
|
|
|
serve.run(MyDeployment.bind(), name="app", route_prefix="/f")
|
|
url = get_application_url("HTTP", "app")
|
|
assert "hello" == httpx.get(url).text
|
|
|
|
# Verify startup latency: two replicas aggregate into one time series (_count == 2).
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=20,
|
|
metric="ray_serve_replica_startup_latency_ms_count",
|
|
expected=2,
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
|
|
# Verify initialization latency _count matches (one observation per replica).
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=20,
|
|
metric="ray_serve_replica_initialization_latency_ms_count",
|
|
expected=2,
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
|
|
# Verify initialization latency metric value is greater than 500ms
|
|
def check_initialization_latency_value():
|
|
value = get_metric_float(
|
|
"ray_serve_replica_initialization_latency_ms_sum",
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
assert (
|
|
value > 500
|
|
), f"Initialization latency value is {value}, expected to be greater than 500ms"
|
|
return True
|
|
|
|
wait_for_condition(check_initialization_latency_value, timeout=20)
|
|
|
|
# One aggregated time series per deployment (no per-replica label).
|
|
def check_single_series_no_replica_label():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_replica_initialization_latency_ms_count",
|
|
wait=False,
|
|
)
|
|
assert len(metrics) == 1, f"Expected 1 metric series, got {len(metrics)}"
|
|
assert metrics[0]["deployment"] == "MyDeployment"
|
|
assert metrics[0]["application"] == "app"
|
|
assert "replica" not in metrics[0]
|
|
return True
|
|
|
|
wait_for_condition(check_single_series_no_replica_label, timeout=20)
|
|
|
|
|
|
def test_replica_reconfigure_latency_metrics(metrics_start_shutdown):
|
|
"""Test that replica reconfigure latency metrics are recorded when user_config changes."""
|
|
|
|
@serve.deployment
|
|
class Configurable:
|
|
def __init__(self):
|
|
self.config = None
|
|
|
|
def reconfigure(self, config):
|
|
time.sleep(1)
|
|
self.config = config
|
|
|
|
def __call__(self):
|
|
return self.config
|
|
|
|
# Use an internal code version to exercise in-place reconfigure.
|
|
Configurable = Configurable.options(_internal=True, version="1")
|
|
serve.run(
|
|
Configurable.options(user_config={"version": 1}).bind(),
|
|
name="app",
|
|
route_prefix="/config",
|
|
)
|
|
url = get_application_url("HTTP", "app")
|
|
assert httpx.get(url).json() == {"version": 1}
|
|
|
|
# Update user_config to trigger in-place reconfigure (same version, different config)
|
|
serve.run(
|
|
Configurable.options(user_config={"version": 2}).bind(),
|
|
name="app",
|
|
route_prefix="/config",
|
|
)
|
|
|
|
# Wait for the new config to take effect
|
|
def config_updated():
|
|
return httpx.get(url).json() == {"version": 2}
|
|
|
|
wait_for_condition(config_updated, timeout=20)
|
|
|
|
# Verify reconfigure latency metric count is exactly 1 (one reconfigure happened)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=20,
|
|
metric="ray_serve_replica_reconfigure_latency_ms_count",
|
|
expected=1,
|
|
expected_tags={"deployment": "Configurable", "application": "app"},
|
|
)
|
|
|
|
# Verify reconfigure latency metric value is greater than 500ms (we slept for 1s)
|
|
def check_reconfigure_latency_value():
|
|
value = get_metric_float(
|
|
"ray_serve_replica_reconfigure_latency_ms_sum",
|
|
expected_tags={"deployment": "Configurable", "application": "app"},
|
|
)
|
|
assert value > 500, f"Reconfigure latency value is {value}, expected > 500ms"
|
|
return True
|
|
|
|
wait_for_condition(check_reconfigure_latency_value, timeout=20)
|
|
|
|
|
|
def test_health_check_latency_metrics(metrics_start_shutdown):
|
|
"""Test that health check latency metrics are recorded."""
|
|
|
|
@serve.deployment(health_check_period_s=1)
|
|
class MyDeployment:
|
|
def __call__(self):
|
|
return "hello"
|
|
|
|
def check_health(self):
|
|
time.sleep(1)
|
|
|
|
serve.run(MyDeployment.bind(), name="app", route_prefix="/f")
|
|
url = get_application_url("HTTP", "app")
|
|
assert "hello" == httpx.get(url).text
|
|
|
|
# Wait for at least one health check to complete and verify metric is recorded
|
|
def check_health_check_latency_metrics():
|
|
value = get_metric_float(
|
|
"ray_serve_health_check_latency_ms_count",
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
# Health check count should be at least 1
|
|
assert value >= 1, f"Health check count is {value}, expected to be 1"
|
|
return True
|
|
|
|
wait_for_condition(check_health_check_latency_metrics, timeout=30)
|
|
|
|
# Verify health check latency metric value is greater than 500ms
|
|
def check_health_check_latency_value():
|
|
value = get_metric_float(
|
|
"ray_serve_health_check_latency_ms_sum",
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
assert (
|
|
value > 500
|
|
), f"Health check latency value is {value}, expected to be greater than 500ms"
|
|
return True
|
|
|
|
wait_for_condition(check_health_check_latency_value, timeout=30)
|
|
|
|
|
|
def test_health_check_failures_metrics(metrics_start_shutdown):
|
|
"""Test that health check failure metrics are recorded when health checks fail."""
|
|
|
|
@serve.deployment(health_check_period_s=1, health_check_timeout_s=2)
|
|
class FailingHealthCheck:
|
|
def __init__(self):
|
|
self.should_fail = False
|
|
|
|
async def check_health(self):
|
|
if self.should_fail:
|
|
raise Exception("Health check failed!")
|
|
|
|
async def __call__(self, request):
|
|
action = (await request.body()).decode("utf-8")
|
|
if action == "fail":
|
|
self.should_fail = True
|
|
return "ok"
|
|
|
|
serve.run(FailingHealthCheck.bind(), name="app", route_prefix="/health")
|
|
url = get_application_url("HTTP", "app")
|
|
|
|
# Verify deployment is healthy initially
|
|
assert httpx.get(url).text == "ok"
|
|
|
|
# Trigger health check failure
|
|
httpx.request("GET", url, content=b"fail")
|
|
|
|
# Wait for at least one health check failure to be recorded
|
|
def check_health_check_failure_metrics():
|
|
value = get_metric_float(
|
|
"ray_serve_health_check_failures_total",
|
|
expected_tags={"deployment": "FailingHealthCheck", "application": "app"},
|
|
)
|
|
# Should have at least 1 failure
|
|
return value >= 1
|
|
|
|
wait_for_condition(check_health_check_failure_metrics, timeout=30)
|
|
|
|
|
|
def test_replica_shutdown_duration_metrics(metrics_start_shutdown):
|
|
"""Test that replica shutdown duration metrics are recorded."""
|
|
|
|
@serve.deployment
|
|
class MyDeployment:
|
|
def __call__(self):
|
|
return "hello"
|
|
|
|
def __del__(self):
|
|
time.sleep(1)
|
|
|
|
# Deploy the application
|
|
serve.run(MyDeployment.bind(), name="app", route_prefix="/f")
|
|
url = get_application_url("HTTP", "app")
|
|
assert "hello" == httpx.get(url).text
|
|
|
|
# Delete the application to trigger shutdown
|
|
serve.delete("app", _blocking=True)
|
|
|
|
# Verify shutdown duration metric count is exactly 1 (one replica stopped)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=30,
|
|
metric="ray_serve_replica_shutdown_duration_ms_count",
|
|
expected=1,
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
print("serve_replica_shutdown_duration_ms working as expected.")
|
|
|
|
# Verify shutdown duration metric value is greater than 500ms
|
|
def check_shutdown_duration_value():
|
|
value = get_metric_float(
|
|
"ray_serve_replica_shutdown_duration_ms_sum",
|
|
expected_tags={"deployment": "MyDeployment", "application": "app"},
|
|
)
|
|
assert (
|
|
value > 500
|
|
), f"Shutdown duration value is {value}, expected to be greater than 500ms"
|
|
return True
|
|
|
|
wait_for_condition(check_shutdown_duration_value, timeout=30)
|
|
|
|
|
|
def test_batching_metrics(metrics_start_shutdown):
|
|
@serve.deployment
|
|
class BatchedDeployment:
|
|
@serve.batch(max_batch_size=4, batch_wait_timeout_s=0.5)
|
|
async def batch_handler(self, requests: List[str]) -> List[str]:
|
|
# Simulate some processing time
|
|
await asyncio.sleep(0.05)
|
|
return [f"processed:{r}" for r in requests]
|
|
|
|
async def __call__(self, request: Request):
|
|
data = await request.body()
|
|
return await self.batch_handler(data.decode())
|
|
|
|
app_name = "batched_app"
|
|
serve.run(BatchedDeployment.bind(), name=app_name, route_prefix="/batch")
|
|
|
|
http_url = "http://localhost:8000/batch"
|
|
|
|
# Send multiple concurrent requests to trigger batching
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
|
|
futures = [
|
|
executor.submit(lambda i=i: httpx.post(http_url, content=f"req{i}"))
|
|
for i in range(8)
|
|
]
|
|
results = [f.result() for f in futures]
|
|
|
|
# Verify all requests succeeded
|
|
assert all(r.status_code == 200 for r in results)
|
|
|
|
# Verify specific metric values and tags
|
|
timeseries = PrometheusTimeseries()
|
|
expected_tags = {
|
|
"deployment": "BatchedDeployment",
|
|
"application": app_name,
|
|
"function_name": "batch_handler",
|
|
}
|
|
|
|
# Check batches_processed_total counter exists and has correct tags
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_batches_processed_total",
|
|
expected=2,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
# Check batch_wait_time_ms histogram was recorded for 2 batches
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_batch_wait_time_ms_count",
|
|
expected=2,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
# Check batch_execution_time_ms histogram was recorded for 2 batches
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_batch_execution_time_ms_count",
|
|
expected=2,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
# Check batch_utilization_percent histogram: 2 batches at 100% each = 200 sum
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_batch_utilization_percent_count",
|
|
expected=2,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
# Check actual_batch_size histogram: 2 batches of 4 requests each = 8 sum
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_actual_batch_size_count",
|
|
expected=2,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
# Check batch_queue_length gauge exists (should be 0 after processing)
|
|
wait_for_condition(
|
|
lambda: check_metric_float_eq(
|
|
"ray_serve_batch_queue_length",
|
|
expected=0,
|
|
expected_tags=expected_tags,
|
|
timeseries=timeseries,
|
|
),
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
def test_autoscaling_metrics(metrics_start_shutdown):
|
|
"""Test that autoscaling metrics are emitted correctly.
|
|
|
|
This tests the following metrics:
|
|
- ray_serve_autoscaling_target_replicas: Target number of replicas
|
|
Tags: deployment, application
|
|
- ray_serve_autoscaling_desired_replicas: Raw decision before bounds
|
|
Tags: deployment, application
|
|
- ray_serve_autoscaling_total_requests: Total requests seen by autoscaler
|
|
Tags: deployment, application
|
|
- ray_serve_autoscaling_policy_execution_time_ms: Policy execution time
|
|
Tags: deployment, application, policy_scope
|
|
- ray_serve_autoscaling_target_ongoing_requests: Configured target ongoing
|
|
requests per replica
|
|
Tags: deployment, application
|
|
- ray_serve_autoscaling_replica_metrics_delay_ms: Replica metrics delay
|
|
Tags: deployment, application
|
|
- ray_serve_autoscaling_handle_metrics_delay_ms: Handle metrics delay
|
|
Tags: deployment, application
|
|
"""
|
|
signal = SignalActor.remote()
|
|
|
|
@serve.deployment(
|
|
autoscaling_config={
|
|
"metrics_interval_s": 0.1,
|
|
"min_replicas": 1,
|
|
"max_replicas": 5,
|
|
"target_ongoing_requests": 2,
|
|
"upscale_delay_s": 0,
|
|
"downscale_delay_s": 5,
|
|
"look_back_period_s": 1,
|
|
},
|
|
max_ongoing_requests=10,
|
|
graceful_shutdown_timeout_s=0.1,
|
|
)
|
|
class AutoscalingDeployment:
|
|
async def __call__(self):
|
|
await signal.wait.remote()
|
|
|
|
serve.run(AutoscalingDeployment.bind(), name="autoscaling_app")
|
|
|
|
# Send requests to trigger autoscaling
|
|
handle = serve.get_deployment_handle("AutoscalingDeployment", "autoscaling_app")
|
|
[handle.remote() for _ in range(10)]
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
base_tags = {
|
|
"deployment": "AutoscalingDeployment",
|
|
"application": "autoscaling_app",
|
|
}
|
|
|
|
# Test 1: Check that target_replicas metric is 5 (10 requests / target_ongoing_requests=2)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_autoscaling_target_replicas",
|
|
expected=5,
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
print("Target replicas metric verified.")
|
|
|
|
# Test 2: Check that autoscaling decision metric is 5 (10 requests / target_ongoing_requests=2)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_autoscaling_desired_replicas",
|
|
expected=5,
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
print("Autoscaling decision metric verified.")
|
|
|
|
# Test 3: Check that total requests metric is 10
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_autoscaling_total_requests",
|
|
expected=10,
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
print("Total requests metric verified.")
|
|
|
|
# Test 4: Check that policy execution time metric is emitted with policy_scope=deployment
|
|
def check_policy_execution_time_metric():
|
|
value = get_metric_float(
|
|
"ray_serve_autoscaling_policy_execution_time_ms",
|
|
expected_tags={**base_tags, "policy_scope": "deployment"},
|
|
timeseries=timeseries,
|
|
)
|
|
assert value >= 0
|
|
return True
|
|
|
|
wait_for_condition(check_policy_execution_time_metric, timeout=15)
|
|
print("Policy execution time metric verified.")
|
|
|
|
# Test 5: Check that target_ongoing_requests metric is 2 (matches config)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_autoscaling_target_ongoing_requests",
|
|
expected=2,
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
print("Target ongoing requests metric verified.")
|
|
|
|
# Test 6: Check that the metrics delay histograms are emitted. These are
|
|
# aggregated by Prometheus across all sources, so they carry no
|
|
# per-replica/handle tag.
|
|
def check_metrics_delay_metrics():
|
|
found = False
|
|
for metric_name in (
|
|
"ray_serve_autoscaling_handle_metrics_delay_ms_count",
|
|
"ray_serve_autoscaling_replica_metrics_delay_ms_count",
|
|
):
|
|
metrics_dicts = get_metric_dictionaries(
|
|
metric_name,
|
|
timeout=5,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics_dicts:
|
|
if (
|
|
m.get("deployment") == "AutoscalingDeployment"
|
|
and m.get("application") == "autoscaling_app"
|
|
):
|
|
assert "replica" not in m
|
|
assert "handle" not in m
|
|
found = True
|
|
return found
|
|
|
|
wait_for_condition(check_metrics_delay_metrics, timeout=15)
|
|
print("Metrics delay metrics verified.")
|
|
|
|
# Release signal to complete requests
|
|
ray.get(signal.send.remote())
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "win32",
|
|
reason="Async Inference feature testing is flaky on Windows.",
|
|
)
|
|
def test_async_inference_task_queue_metrics_delay(
|
|
metrics_start_shutdown, external_redis # noqa: F811
|
|
):
|
|
"""Test that async inference task queue metrics delay is emitted correctly.
|
|
|
|
This tests the metric:
|
|
- ray_serve_autoscaling_async_inference_task_queue_metrics_delay_ms
|
|
Tags: deployment, application
|
|
|
|
The QueueMonitor periodically pushes queue length metrics to the controller,
|
|
and the controller records the delay between when the metrics were collected
|
|
and when they were received.
|
|
"""
|
|
# Setup Redis client
|
|
redis_address = os.environ.get("RAY_REDIS_ADDRESS")
|
|
host, port = redis_address.split(":")
|
|
redis_client = redis.Redis(host=host, port=int(port), db=0)
|
|
redis_broker_url = f"redis://{redis_address}/0"
|
|
|
|
test_deployment_id = DeploymentID("test_deployment", "test_app")
|
|
test_queue_name = "test_metrics_queue"
|
|
|
|
try:
|
|
# Create QueueMonitor with the Serve controller
|
|
controller = ray.get_actor(SERVE_CONTROLLER_NAME, namespace=SERVE_NAMESPACE)
|
|
queue_monitor = create_queue_monitor_actor(
|
|
deployment_id=test_deployment_id,
|
|
broker_url=redis_broker_url,
|
|
queue_name=test_queue_name,
|
|
controller_handle=controller,
|
|
namespace=SERVE_NAMESPACE,
|
|
)
|
|
|
|
# Push some messages to the queue so the metrics pusher has data to report
|
|
for i in range(5):
|
|
redis_client.lpush(test_queue_name, f"message_{i}")
|
|
|
|
# Wait for the queue length to be picked up
|
|
def check_length():
|
|
return ray.get(queue_monitor.get_queue_length.remote()) == 5
|
|
|
|
wait_for_condition(check_length, timeout=30)
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
base_tags = {
|
|
"deployment": test_deployment_id.name,
|
|
"application": test_deployment_id.app_name,
|
|
}
|
|
|
|
# Wait for the metrics delay metric to be emitted with correct tags and value
|
|
def check_metrics_delay_metric():
|
|
value = get_metric_float(
|
|
"ray_serve_autoscaling_async_inference_task_queue_metrics_delay_ms",
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
if value <= 0:
|
|
return False
|
|
|
|
# Verify correct tags are attached
|
|
metrics_dicts = get_metric_dictionaries(
|
|
"ray_serve_autoscaling_async_inference_task_queue_metrics_delay_ms",
|
|
timeout=5,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics_dicts:
|
|
if (
|
|
m.get("deployment") == test_deployment_id.name
|
|
and m.get("application") == test_deployment_id.app_name
|
|
):
|
|
# Verify both required tags exist
|
|
assert "deployment" in m, "Missing 'deployment' tag"
|
|
assert "application" in m, "Missing 'application' tag"
|
|
return True
|
|
return False
|
|
|
|
wait_for_condition(check_metrics_delay_metric, timeout=30)
|
|
|
|
finally:
|
|
# Cleanup
|
|
redis_client.delete(test_queue_name)
|
|
redis_client.close()
|
|
try:
|
|
kill_queue_monitor_actor(test_deployment_id, namespace=SERVE_NAMESPACE)
|
|
except ValueError:
|
|
pass # Actor may already be killed
|
|
|
|
|
|
def test_user_autoscaling_stats_metrics(metrics_start_shutdown):
|
|
"""Test that user-defined autoscaling stats metrics are emitted correctly.
|
|
|
|
This tests the following metrics:
|
|
- ray_serve_user_autoscaling_stats_latency_ms: Time to execute user stats function
|
|
Tags: application, deployment, replica
|
|
- ray_serve_record_autoscaling_stats_failed_total: Failed stats collection
|
|
Tags: application, deployment, replica, exception_name
|
|
"""
|
|
|
|
@serve.deployment(
|
|
autoscaling_config={
|
|
"metrics_interval_s": 0.1,
|
|
"min_replicas": 1,
|
|
"max_replicas": 5,
|
|
"target_ongoing_requests": 2,
|
|
},
|
|
)
|
|
class DeploymentWithCustomStats:
|
|
def __init__(self):
|
|
self.call_count = 0
|
|
|
|
async def record_autoscaling_stats(self):
|
|
"""Custom autoscaling stats function."""
|
|
self.call_count += 1
|
|
return {"custom_metric": self.call_count}
|
|
|
|
def __call__(self):
|
|
return "ok"
|
|
|
|
serve.run(DeploymentWithCustomStats.bind(), name="custom_stats_app")
|
|
|
|
# Make a request to ensure the deployment is running
|
|
handle = serve.get_deployment_handle(
|
|
"DeploymentWithCustomStats", "custom_stats_app"
|
|
)
|
|
handle.remote().result()
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
base_tags = {
|
|
"deployment": "DeploymentWithCustomStats",
|
|
"application": "custom_stats_app",
|
|
}
|
|
|
|
# Test: Check that user autoscaling stats latency metric is emitted
|
|
def check_user_stats_latency_metric():
|
|
value = get_metric_float(
|
|
"ray_serve_user_autoscaling_stats_latency_ms_sum",
|
|
expected_tags=base_tags,
|
|
timeseries=timeseries,
|
|
)
|
|
if value >= 0:
|
|
# Verify replica tag exists
|
|
metrics_dicts = get_metric_dictionaries(
|
|
"ray_serve_user_autoscaling_stats_latency_ms_sum",
|
|
timeout=5,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics_dicts:
|
|
if (
|
|
m.get("deployment") == "DeploymentWithCustomStats"
|
|
and m.get("application") == "custom_stats_app"
|
|
):
|
|
assert m.get("replica") is not None
|
|
print(
|
|
f"User stats latency metric verified with replica tag: {m.get('replica')}"
|
|
)
|
|
return True
|
|
return False
|
|
|
|
wait_for_condition(check_user_stats_latency_metric, timeout=15)
|
|
print("User autoscaling stats latency metric verified.")
|
|
|
|
|
|
def test_user_autoscaling_stats_failure_metrics(metrics_start_shutdown):
|
|
"""Test that user autoscaling stats failure metrics are emitted on error."""
|
|
|
|
@serve.deployment(
|
|
autoscaling_config={
|
|
"metrics_interval_s": 0.1,
|
|
"min_replicas": 1,
|
|
"max_replicas": 5,
|
|
"target_ongoing_requests": 2,
|
|
},
|
|
)
|
|
class DeploymentWithFailingStats:
|
|
async def record_autoscaling_stats(self):
|
|
"""Custom autoscaling stats function that raises an error."""
|
|
raise ValueError("Intentional error for testing")
|
|
|
|
def __call__(self):
|
|
return "ok"
|
|
|
|
serve.run(DeploymentWithFailingStats.bind(), name="failing_stats_app")
|
|
|
|
# Make a request to ensure the deployment is running
|
|
handle = serve.get_deployment_handle(
|
|
"DeploymentWithFailingStats", "failing_stats_app"
|
|
)
|
|
handle.remote().result()
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
|
|
# Test: Check that failure counter is incremented
|
|
def check_stats_failure_metric():
|
|
metrics_dicts = get_metric_dictionaries(
|
|
"ray_serve_record_autoscaling_stats_failed_total",
|
|
timeout=5,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics_dicts:
|
|
if (
|
|
m.get("deployment") == "DeploymentWithFailingStats"
|
|
and m.get("application") == "failing_stats_app"
|
|
):
|
|
assert m.get("replica") is not None
|
|
assert m.get("exception_name") == "ValueError"
|
|
print(
|
|
f"Stats failure metric verified with exception_name: {m.get('exception_name')}"
|
|
)
|
|
return True
|
|
return False
|
|
|
|
wait_for_condition(check_stats_failure_metric, timeout=15)
|
|
print("User autoscaling stats failure metric verified.")
|
|
|
|
|
|
def test_long_poll_pending_clients_metric(metrics_start_shutdown):
|
|
"""Check that pending clients gauge is tracked correctly."""
|
|
timeseries = PrometheusTimeseries()
|
|
|
|
# Create a LongPollHost with a longer timeout so we can observe pending state
|
|
host = ray.remote(LongPollHost).remote(
|
|
listen_for_change_request_timeout_s=(5.0, 5.0)
|
|
)
|
|
|
|
# Write initial values
|
|
ray.get(host.notify_changed.remote({"key_1": 100}))
|
|
ray.get(host.notify_changed.remote({"key_2": 200}))
|
|
|
|
# Get the current snapshot IDs
|
|
result = ray.get(host.listen_for_change.remote({"key_1": -1, "key_2": -1}))
|
|
key_1_snapshot_id = result["key_1"].snapshot_id
|
|
key_2_snapshot_id = result["key_2"].snapshot_id
|
|
|
|
# Start a listen call that will block waiting for updates
|
|
# (since we're using up-to-date snapshot IDs)
|
|
pending_ref = host.listen_for_change.remote(
|
|
{"key_1": key_1_snapshot_id, "key_2": key_2_snapshot_id}
|
|
)
|
|
|
|
# Check that pending clients gauge shows 1 for each key
|
|
# (wait_for_condition will retry until the metric is available)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_pending_clients",
|
|
expected=1,
|
|
expected_tags={"namespace": "key_1"},
|
|
timeseries=timeseries,
|
|
)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_pending_clients",
|
|
expected=1,
|
|
expected_tags={"namespace": "key_2"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
# Trigger an update for key_1
|
|
ray.get(host.notify_changed.remote({"key_1": 101}))
|
|
|
|
# Wait for the pending call to complete
|
|
ray.get(pending_ref)
|
|
|
|
# After update, pending clients for key_1 should be 0
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_pending_clients",
|
|
expected=0,
|
|
expected_tags={"namespace": "key_1"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
|
|
def test_long_poll_latency_metric(metrics_start_shutdown):
|
|
"""Check that long poll latency histogram is recorded on the client side."""
|
|
timeseries = PrometheusTimeseries()
|
|
|
|
# Create a LongPollHost
|
|
host = ray.remote(LongPollHost).remote(
|
|
listen_for_change_request_timeout_s=(0.5, 0.5)
|
|
)
|
|
|
|
# Write initial value so the key exists
|
|
ray.get(host.notify_changed.remote({"test_key": "initial_value"}))
|
|
|
|
# Track received updates
|
|
received_updates = []
|
|
update_event = threading.Event()
|
|
|
|
def on_update(value):
|
|
received_updates.append(value)
|
|
update_event.set()
|
|
|
|
# Create event loop for the client
|
|
loop = asyncio.new_event_loop()
|
|
|
|
def run_loop():
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_forever()
|
|
|
|
loop_thread = threading.Thread(target=run_loop, daemon=True)
|
|
loop_thread.start()
|
|
|
|
# Create the LongPollClient
|
|
client = LongPollClient(
|
|
host_actor=host,
|
|
key_listeners={"test_key": on_update},
|
|
call_in_event_loop=loop,
|
|
client_id="test_metrics_client",
|
|
)
|
|
|
|
# Wait for initial update (client starts with snapshot_id -1)
|
|
assert update_event.wait(timeout=10), "Timed out waiting for initial update"
|
|
assert len(received_updates) == 1
|
|
assert received_updates[0] == "initial_value"
|
|
|
|
# Clear event and trigger another update
|
|
update_event.clear()
|
|
ray.get(host.notify_changed.remote({"test_key": "updated_value"}))
|
|
|
|
# Wait for the update to be received
|
|
assert update_event.wait(timeout=10), "Timed out waiting for update"
|
|
assert len(received_updates) == 2
|
|
assert received_updates[1] == "updated_value"
|
|
|
|
# Stop the client
|
|
client.stop()
|
|
loop.call_soon_threadsafe(loop.stop)
|
|
loop_thread.join(timeout=5)
|
|
|
|
# Check that latency metric was recorded
|
|
# The metric should have at least 2 observations (initial + update)
|
|
def check_latency_metric_exists():
|
|
metric_value = get_metric_float(
|
|
"ray_serve_long_poll_latency_ms_count",
|
|
expected_tags={"namespace": "test_key"},
|
|
timeseries=timeseries,
|
|
)
|
|
# Should have at least 2 observations
|
|
return metric_value == 2
|
|
|
|
wait_for_condition(check_latency_metric_exists, timeout=15)
|
|
|
|
# Verify the latency sum is positive (latency > 0)
|
|
latency_sum = get_metric_float(
|
|
"ray_serve_long_poll_latency_ms_sum",
|
|
expected_tags={"namespace": "test_key"},
|
|
timeseries=timeseries,
|
|
)
|
|
assert latency_sum > 0, "Latency sum should be positive"
|
|
|
|
|
|
def test_long_poll_host_sends_counted(metrics_start_shutdown):
|
|
"""Check that the transmissions by the long_poll are counted."""
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
host = ray.remote(LongPollHost).remote(
|
|
listen_for_change_request_timeout_s=(0.01, 0.01)
|
|
)
|
|
|
|
# Write a value.
|
|
ray.get(host.notify_changed.remote({"key_1": 999}))
|
|
object_ref = host.listen_for_change.remote({"key_1": -1})
|
|
|
|
# Check that the result's size is reported.
|
|
result_1: Dict[str, UpdatedObject] = ray.get(object_ref)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_host_transmission_counter_total",
|
|
expected=1,
|
|
expected_tags={"namespace_or_state": "key_1"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
# Write two new values.
|
|
ray.get(host.notify_changed.remote({"key_1": 1000}))
|
|
ray.get(host.notify_changed.remote({"key_2": 1000}))
|
|
object_ref = host.listen_for_change.remote(
|
|
{"key_1": result_1["key_1"].snapshot_id, "key_2": -1}
|
|
)
|
|
|
|
# Check that the new objects are transmitted.
|
|
result_2: Dict[str, UpdatedObject] = ray.get(object_ref)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_host_transmission_counter_total",
|
|
expected=1,
|
|
expected_tags={"namespace_or_state": "key_2"},
|
|
timeseries=timeseries,
|
|
)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_host_transmission_counter_total",
|
|
expected=2,
|
|
expected_tags={"namespace_or_state": "key_1"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
# Check that a timeout result is counted.
|
|
object_ref = host.listen_for_change.remote({"key_2": result_2["key_2"].snapshot_id})
|
|
_ = ray.get(object_ref)
|
|
wait_for_condition(
|
|
check_metric_float_eq,
|
|
timeout=15,
|
|
metric="ray_serve_long_poll_host_transmission_counter_total",
|
|
expected=1,
|
|
expected_tags={"namespace_or_state": "TIMEOUT"},
|
|
timeseries=timeseries,
|
|
)
|
|
|
|
|
|
def test_event_loop_monitoring_metrics(metrics_start_shutdown):
|
|
"""Test that event loop monitoring metrics are emitted correctly.
|
|
|
|
This tests the following metrics:
|
|
- serve_event_loop_scheduling_latency_ms: Event loop lag in milliseconds
|
|
Tags: component, loop_type, actor_id
|
|
- serve_event_loop_monitoring_iterations: Heartbeat counter
|
|
Tags: component, loop_type, actor_id
|
|
- serve_event_loop_tasks: Number of pending asyncio tasks
|
|
Tags: component, loop_type, actor_id
|
|
|
|
Components monitored:
|
|
- Proxy: main loop only
|
|
- Replica: main loop + user_code loop (when separate thread enabled)
|
|
- Router: router loop (when separate loop enabled, runs on replica)
|
|
"""
|
|
|
|
@serve.deployment(name="g")
|
|
class ChildDeployment:
|
|
def __call__(self):
|
|
return "child"
|
|
|
|
@serve.deployment(name="f")
|
|
class SimpleDeployment:
|
|
def __init__(self, child):
|
|
self.child = child
|
|
|
|
async def __call__(self):
|
|
return await self.child.remote()
|
|
|
|
serve.run(
|
|
SimpleDeployment.bind(ChildDeployment.bind()), name="app", route_prefix="/test"
|
|
)
|
|
|
|
# Make a request to ensure everything is running
|
|
url = get_application_url("HTTP", "app")
|
|
assert httpx.get(url).text == "child"
|
|
|
|
timeseries = PrometheusTimeseries()
|
|
|
|
# Test 1: Check proxy main loop metrics
|
|
def check_proxy_main_loop_metrics():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_monitoring_iterations_total",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics:
|
|
if m.get("component") == "proxy" and m.get("loop_type") == "main":
|
|
assert "actor_id" in m, "actor_id tag should be present"
|
|
print(f"Proxy main loop metric found: {m}")
|
|
return True
|
|
return False
|
|
|
|
wait_for_condition(check_proxy_main_loop_metrics, timeout=30)
|
|
print("Proxy main loop monitoring metrics verified.")
|
|
|
|
# Test 1a: Check proxy router loop metrics
|
|
def check_proxy_router_loop_metrics():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_monitoring_iterations_total",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics:
|
|
if m.get("component") == "proxy" and m.get("loop_type") == "router":
|
|
assert "actor_id" in m, "actor_id tag should be present"
|
|
print(f"Proxy router loop metric found: {m}")
|
|
return True
|
|
return False
|
|
|
|
if RAY_SERVE_RUN_ROUTER_IN_SEPARATE_LOOP:
|
|
wait_for_condition(check_proxy_router_loop_metrics, timeout=30)
|
|
print("Proxy router loop monitoring metrics verified.")
|
|
else:
|
|
print("Proxy router loop monitoring metrics not verified.")
|
|
|
|
# Test 2: Check replica main loop metrics
|
|
def check_replica_main_loop_metrics():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_monitoring_iterations_total",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics:
|
|
if m.get("component") == "replica" and m.get("loop_type") == "main":
|
|
assert "actor_id" in m, "actor_id tag should be present"
|
|
assert m.get("deployment") in [
|
|
"f",
|
|
"g",
|
|
], "deployment tag should be 'f' or 'g'"
|
|
assert m.get("application") == "app", "application tag should be 'app'"
|
|
print(f"Replica main loop metric found: {m}")
|
|
return True
|
|
return False
|
|
|
|
wait_for_condition(check_replica_main_loop_metrics, timeout=30)
|
|
print("Replica main loop monitoring metrics verified.")
|
|
|
|
# Test 3: Check replica user_code loop metrics (enabled by default)
|
|
def check_replica_user_code_loop_metrics():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_monitoring_iterations_total",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics:
|
|
if m.get("component") == "replica" and m.get("loop_type") == "user_code":
|
|
assert "actor_id" in m, "actor_id tag should be present"
|
|
assert m.get("deployment") in [
|
|
"f",
|
|
"g",
|
|
], "deployment tag should be 'f' or 'g'"
|
|
assert m.get("application") == "app", "application tag should be 'app'"
|
|
print(f"Replica user_code loop metric found: {m}")
|
|
return True
|
|
return False
|
|
|
|
if RAY_SERVE_RUN_USER_CODE_IN_SEPARATE_THREAD:
|
|
wait_for_condition(check_replica_user_code_loop_metrics, timeout=30)
|
|
print("Replica user_code loop monitoring metrics verified.")
|
|
else:
|
|
print("Replica user_code loop monitoring metrics not verified.")
|
|
|
|
# Test 4: Check router loop metrics (enabled by default)
|
|
def check_router_loop_metrics():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_monitoring_iterations_total",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
for m in metrics:
|
|
if m.get("component") == "replica" and m.get("loop_type") == "router":
|
|
assert "actor_id" in m, "actor_id tag should be present"
|
|
print(f"Router loop metric found: {m}")
|
|
return True
|
|
return False
|
|
|
|
if RAY_SERVE_RUN_ROUTER_IN_SEPARATE_LOOP:
|
|
wait_for_condition(check_router_loop_metrics, timeout=30)
|
|
print("Router loop monitoring metrics verified.")
|
|
else:
|
|
print("Router loop monitoring metrics not verified.")
|
|
|
|
# Test 5: Check that scheduling latency histogram exists and has reasonable values
|
|
def check_scheduling_latency_metric():
|
|
# Check for the histogram count metric
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_scheduling_latency_ms_count",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
# Should have metrics for proxy main, replica main, replica user_code, router
|
|
component_loop_pairs = set()
|
|
for m in metrics:
|
|
component = m.get("component")
|
|
loop_type = m.get("loop_type")
|
|
if component and loop_type:
|
|
component_loop_pairs.add((component, loop_type))
|
|
|
|
expected_pairs = {
|
|
("proxy", "main"),
|
|
("replica", "main"),
|
|
}
|
|
if RAY_SERVE_RUN_USER_CODE_IN_SEPARATE_THREAD:
|
|
expected_pairs.add(("replica", "user_code"))
|
|
if RAY_SERVE_RUN_ROUTER_IN_SEPARATE_LOOP:
|
|
expected_pairs.add(("replica", "router"))
|
|
expected_pairs.add(("proxy", "router"))
|
|
return expected_pairs.issubset(component_loop_pairs)
|
|
|
|
wait_for_condition(check_scheduling_latency_metric, timeout=30)
|
|
print("Scheduling latency histogram metrics verified.")
|
|
|
|
# Test 6: Check that tasks gauge exists
|
|
def check_tasks_gauge_metric():
|
|
metrics = get_metric_dictionaries(
|
|
"ray_serve_event_loop_tasks",
|
|
timeout=10,
|
|
timeseries=timeseries,
|
|
wait=False,
|
|
)
|
|
# Should have metrics for proxy main, replica main, replica user_code, router
|
|
component_loop_pairs = set()
|
|
for m in metrics:
|
|
component = m.get("component")
|
|
loop_type = m.get("loop_type")
|
|
if component and loop_type:
|
|
component_loop_pairs.add((component, loop_type))
|
|
|
|
expected_pairs = {
|
|
("proxy", "main"),
|
|
("replica", "main"),
|
|
}
|
|
if RAY_SERVE_RUN_USER_CODE_IN_SEPARATE_THREAD:
|
|
expected_pairs.add(("replica", "user_code"))
|
|
if RAY_SERVE_RUN_ROUTER_IN_SEPARATE_LOOP:
|
|
expected_pairs.add(("replica", "router"))
|
|
expected_pairs.add(("proxy", "router"))
|
|
return expected_pairs.issubset(component_loop_pairs)
|
|
|
|
wait_for_condition(check_tasks_gauge_metric, timeout=30)
|
|
print("Event loop tasks gauge metrics verified.")
|
|
|
|
|
|
def test_actor_summary(serve_instance):
|
|
@serve.deployment
|
|
def f():
|
|
pass
|
|
|
|
serve.run(f.bind(), name="app")
|
|
actors = list_actors(filters=[("state", "=", "ALIVE")])
|
|
class_names = {actor.class_name for actor in actors}
|
|
assert class_names.issuperset(
|
|
{"ServeController", "ProxyActor", "ServeReplica:app:f"}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", "-s", __file__]))
|