Serve metrics docs (#13096)

python/ray/serve/__init__.py

@@ -1,4 +1,6 @@
-from ray.serve.api import (accept_batch, Client, connect, start)  # noqa: F401
+from ray.serve.api import (accept_batch, Client, connect,
+                           get_current_backend_tag, get_current_replica_tag,
+                           start)
 from ray.serve.config import BackendConfig
 from ray.serve.env import CondaEnv
 
@@ -13,5 +15,7 @@ __all__ = [
     "CondaEnv",
     "connect",
     "Client",
+    "get_current_backend_tag",
+    "get_current_replica_tag",
     "start",
 ]

python/ray/serve/api.py

@@ -6,11 +6,12 @@ import os
 from uuid import UUID
 import threading
 from typing import Any, Callable, Coroutine, Dict, List, Optional, Type, Union
+from dataclasses import dataclass
 
 import ray
 from ray.serve.constants import (DEFAULT_HTTP_HOST, DEFAULT_HTTP_PORT,
                                  SERVE_CONTROLLER_NAME, HTTP_PROXY_TIMEOUT)
-from ray.serve.controller import ServeController
+from ray.serve.controller import ServeController, BackendTag, ReplicaTag
 from ray.serve.handle import RayServeHandle, RayServeSyncHandle
 from ray.serve.utils import (block_until_http_ready, format_actor_name,
                              get_random_letters, logger, get_conda_env_dir)
@@ -21,11 +22,18 @@ from ray.serve.env import CondaEnv
 from ray.serve.router import RequestMetadata, Router
 from ray.actor import ActorHandle
 
-_INTERNAL_CONTROLLER_NAME = None
-
+_INTERNAL_REPLICA_CONTEXT = None
 global_async_loop = None
 
 
+@dataclass
+class InternalReplicaContext:
+    """Stores data for Serve API calls from within the user's backend code."""
+    backend_tag: BackendTag
+    replica_tag: ReplicaTag
+    controller_name: str
+
+
 def create_or_get_async_loop_in_thread():
     global global_async_loop
     if global_async_loop is None:
@@ -38,9 +46,10 @@ def create_or_get_async_loop_in_thread():
     return global_async_loop
 
 
-def _set_internal_controller_name(name):
-    global _INTERNAL_CONTROLLER_NAME
-    _INTERNAL_CONTROLLER_NAME = name
+def _set_internal_replica_context(backend_tag, replica_tag, controller_name):
+    global _INTERNAL_REPLICA_CONTEXT
+    _INTERNAL_REPLICA_CONTEXT = InternalReplicaContext(
+        backend_tag, replica_tag, controller_name)
 
 
 def _ensure_connected(f: Callable) -> Callable:
@@ -598,12 +607,12 @@ def connect() -> Client:
     if not ray.is_initialized():
         ray.init()
 
-    # When running inside of a backend, _INTERNAL_CONTROLLER_NAME is set to
+    # When running inside of a backend, _INTERNAL_REPLICA_CONTEXT is set to
     # ensure that the correct instance is connected to.
-    if _INTERNAL_CONTROLLER_NAME is None:
+    if _INTERNAL_REPLICA_CONTEXT is None:
         controller_name = SERVE_CONTROLLER_NAME
     else:
-        controller_name = _INTERNAL_CONTROLLER_NAME
+        controller_name = _INTERNAL_REPLICA_CONTEXT.controller_name
 
     # Try to get serve controller if it exists
     try:
@@ -617,6 +626,37 @@ def connect() -> Client:
     return Client(controller, controller_name, detached=True)
 
 
+def get_current_backend_tag() -> BackendTag:
+    """When called from within a backend, return its backend tag.
+
+    Raises:
+        RayServeException if not called from within a Ray Serve backend.
+    """
+    if _INTERNAL_REPLICA_CONTEXT is None:
+        raise RayServeException("`serve.get_current_backend_tag()`"
+                                "may only be called from within a"
+                                "Ray Serve backend.")
+    else:
+        return _INTERNAL_REPLICA_CONTEXT.backend_tag
+
+
+def get_current_replica_tag() -> ReplicaTag:
+    """When called from within a backend, return its replica tag.
+
+    A replica tag uniquely identifies a single replica (a process)
+    for a Ray Serve backend.
+
+    Raises:
+        RayServeException if not called from within a Ray Serve backend.
+    """
+    if _INTERNAL_REPLICA_CONTEXT is None:
+        raise RayServeException("`serve.get_current_replica_tag()`"
+                                "may only be called from within a"
+                                "Ray Serve backend.")
+    else:
+        return _INTERNAL_REPLICA_CONTEXT.replica_tag
+
+
 def accept_batch(f: Callable) -> Callable:
     """Annotation to mark that a serving function accepts batches of requests.
 

python/ray/serve/backend_worker.py

@@ -109,7 +109,8 @@ def create_backend_replica(func_or_class: Union[Callable, Type[Callable]]):
             # Set the controller name so that serve.connect() in the user's
             # backend code will connect to the instance that this backend is
             # running in.
-            ray.serve.api._set_internal_controller_name(controller_name)
+            ray.serve.api._set_internal_replica_context(
+                backend_tag, replica_tag, controller_name)
             if is_function:
                 _callable = func_or_class
             else:
@@ -117,9 +118,8 @@ def create_backend_replica(func_or_class: Union[Callable, Type[Callable]]):
 
             assert controller_name, "Must provide a valid controller_name"
             controller_handle = ray.get_actor(controller_name)
-            self.backend = RayServeReplica(backend_tag, replica_tag, _callable,
-                                           backend_config, is_function,
-                                           controller_handle)
+            self.backend = RayServeReplica(_callable, backend_config,
+                                           is_function, controller_handle)
 
         async def handle_request(self, request):
             return await self.backend.handle_request(request)
@@ -153,11 +153,10 @@ def ensure_async(func: Callable) -> Callable:
 class RayServeReplica:
     """Handles requests with the provided callable."""
 
-    def __init__(self, backend_tag: str, replica_tag: str, _callable: Callable,
-                 backend_config: BackendConfig, is_function: bool,
-                 controller_handle: ActorHandle) -> None:
-        self.backend_tag = backend_tag
-        self.replica_tag = replica_tag
+    def __init__(self, _callable: Callable, backend_config: BackendConfig,
+                 is_function: bool, controller_handle: ActorHandle) -> None:
+        self.backend_tag = ray.serve.api.get_current_backend_tag()
+        self.replica_tag = ray.serve.api.get_current_replica_tag()
         self.callable = _callable
         self.is_function = is_function
 
@@ -169,9 +168,9 @@ class RayServeReplica:
         self.num_ongoing_requests = 0
 
         self.request_counter = metrics.Count(
-            "backend_request_counter",
-            description=("Number of queries that have been "
-                         "processed in this replica"),
+            "serve_backend_request_counter",
+            description=("The number of queries that have been "
+                         "processed in this replica."),
             tag_keys=("backend", ))
         self.request_counter.set_default_tags({"backend": self.backend_tag})
 
@@ -180,15 +179,15 @@ class RayServeReplica:
         })
 
         self.error_counter = metrics.Count(
-            "backend_error_counter",
-            description=("Number of exceptions that have "
-                         "occurred in the backend"),
+            "serve_backend_error_counter",
+            description=("The number of exceptions that have "
+                         "occurred in the backend."),
             tag_keys=("backend", ))
         self.error_counter.set_default_tags({"backend": self.backend_tag})
 
         self.restart_counter = metrics.Count(
-            "backend_replica_starts",
-            description=("The number of time this replica "
+            "serve_backend_replica_starts",
+            description=("The number of times this replica "
                          "has been restarted due to failure."),
             tag_keys=("backend", "replica"))
         self.restart_counter.set_default_tags({
@@ -197,10 +196,9 @@ class RayServeReplica:
         })
 
         self.queuing_latency_tracker = metrics.Histogram(
-            "backend_queuing_latency_ms",
-            description=(
-                "The latency for queries waiting in the replica's queue "
-                "waiting to be processed or batched."),
+            "serve_backend_queuing_latency_ms",
+            description=("The latency for queries in the replica's queue "
+                         "waiting to be processed or batched."),
             boundaries=DEFAULT_LATENCY_BUCKET_MS,
             tag_keys=("backend", "replica"))
         self.queuing_latency_tracker.set_default_tags({
@@ -209,8 +207,8 @@ class RayServeReplica:
         })
 
         self.processing_latency_tracker = metrics.Histogram(
-            "backend_processing_latency_ms",
-            description="The latency for queries to be processed",
+            "serve_backend_processing_latency_ms",
+            description="The latency for queries to be processed.",
             boundaries=DEFAULT_LATENCY_BUCKET_MS,
             tag_keys=("backend", "replica", "batch_size"))
         self.processing_latency_tracker.set_default_tags({
@@ -219,9 +217,9 @@ class RayServeReplica:
         })
 
         self.num_queued_items = metrics.Gauge(
-            "replica_queued_queries",
-            description=("Current number of queries queued in the "
-                         "the backend replicas"),
+            "serve_replica_queued_queries",
+            description=("The current number of queries queued in "
+                         "the backend replicas."),
             tag_keys=("backend", "replica"))
         self.num_queued_items.set_default_tags({
             "backend": self.backend_tag,
@@ -229,8 +227,8 @@ class RayServeReplica:
         })
 
         self.num_processing_items = metrics.Gauge(
-            "replica_processing_queries",
-            description="Current number of queries being processed",
+            "serve_replica_processing_queries",
+            description="The current number of queries being processed.",
             tag_keys=("backend", "replica"))
         self.num_processing_items.set_default_tags({
             "backend": self.backend_tag,

python/ray/serve/examples/doc/snippet_custom_metric.py

@@ -0,0 +1,32 @@
+import ray
+from ray import serve
+from ray.util import metrics
+
+import time
+
+ray.init(address="auto")
+client = serve.start()
+
+
+class MyBackendClass:
+    def __init__(self):
+        self.my_counter = metrics.Count(
+            "my_counter",
+            description=("The number of excellent requests to this backend."),
+            tag_keys=("backend", ))
+        self.my_counter.set_default_tags({
+            "backend": serve.get_current_backend_tag()
+        })
+
+    def __call__(self, request):
+        if "excellent" in request.query_params:
+            self.my_counter.record(1)
+
+
+client.create_backend("my_backend", MyBackendClass)
+client.create_endpoint("my_endpoint", backend="my_backend")
+
+handle = client.get_handle("my_endpoint")
+while (True):
+    ray.get(handle.remote(excellent=True))
+    time.sleep(1)

python/ray/serve/examples/doc/snippet_metrics.py

@@ -0,0 +1,19 @@
+import ray
+from ray import serve
+
+import time
+
+ray.init(address="auto")
+client = serve.start()
+
+
+def f(request):
+    time.sleep(1)
+
+
+client.create_backend("f", f)
+client.create_endpoint("f", backend="f")
+
+handle = client.get_handle("f")
+while (True):
+    ray.get(handle.remote())

python/ray/serve/http_proxy.py

@@ -28,7 +28,8 @@ class HTTPProxy:
     def __init__(self, controller_name):
         # Set the controller name so that serve.connect() will connect to the
         # controller instance this proxy is running in.
-        ray.serve.api._set_internal_controller_name(controller_name)
+        ray.serve.api._set_internal_replica_context(None, None,
+                                                    controller_name)
         self.client = ray.serve.connect()
 
         controller = ray.get_actor(controller_name)
@@ -39,8 +40,8 @@ class HTTPProxy:
         })
 
         self.request_counter = metrics.Count(
-            "num_http_requests",
-            description="The number of HTTP requests processed",
+            "serve_num_http_requests",
+            description="The number of HTTP requests processed.",
             tag_keys=("route", ))
 
     async def setup(self):

python/ray/serve/router.py

@@ -172,8 +172,8 @@ class Router:
 
         # -- Metrics Registration -- #
         self.num_router_requests = metrics.Count(
-            "num_router_requests",
-            description="Number of requests processed by the router.",
+            "serve_num_router_requests",
+            description="The number of requests processed by the router.",
             tag_keys=("endpoint", ))
 
     async def setup_in_async_loop(self):

python/ray/serve/utils.py

@@ -374,8 +374,8 @@ class MockImportedBackend:
     def __call__(self, *args):
         return {"arg": self.arg, "config": self.config}
 
-    def other_method(self, request):
-        return request.data
+    async def other_method(self, request):
+        return await request.body()
 
 
 def compute_iterable_delta(old: Iterable,