diff --git a/doc/source/serve/advanced.rst b/doc/source/serve/advanced.rst
index 5ea7b8dae..e0a2de7a0 100644
--- a/doc/source/serve/advanced.rst
+++ b/doc/source/serve/advanced.rst
@@ -293,10 +293,58 @@ Monitoring
 ==========
 
 Ray Serve exposes important system metrics like the number of successful and
-errored requests through the Ray metrics monitoring infrastructure. By default,
-the metrics are exposed in Prometheus format on each node. See the
-`Ray Monitoring documentation <../ray-metrics.html>`__ for more information.
+errored requests through the `Ray metrics monitoring infrastructure <../ray-metrics.html>`__. By default,
+the metrics are exposed in Prometheus format on each node.
 
+The following metrics are exposed by Ray Serve:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Name
+     - Description
+   * - ``serve_backend_request_counter``
+     - The number of queries that have been processed in this replica.
+   * - ``serve_backend_error_counter``
+     - The number of exceptions that have occurred in the backend.
+   * - ``serve_backend_replica_starts``
+     - The number of times this replica has been restarted due to failure.
+   * - ``serve_backend_queuing_latency_ms``
+     - The latency for queries in the replica's queue waiting to be processed or batched.
+   * - ``serve_backend_processing_latency_ms``
+     - The latency for queries to be processed.
+   * - ``serve_replica_queued_queries``
+     - The current number of queries queued in the backend replicas.
+   * - ``serve_replica_processing_queries``
+     - The current number of queries being processed.
+   * - ``serve_num_http_requests``
+     - The number of HTTP requests processed.
+   * - ``serve_num_router_requests``
+     - The number of requests processed by the router.
+
+To see this in action, run ``ray start --head --metrics-export-port=8080`` in your terminal, and then run the following script:
+
+.. literalinclude:: ../../../python/ray/serve/examples/doc/snippet_metrics.py
+
+In your web browser, navigate to ``localhost:8080``.
+In the output there, you can search for ``serve_`` to locate the metrics above.
+The metrics are updated once every ten seconds, and you will need to refresh the page to see the new values.
+
+For example, after running the script for some time and refreshing ``localhost:8080`` you might see something that looks like::
+
+  ray_serve_backend_processing_latency_ms_count{...,backend="f",...} 99.0
+  ray_serve_backend_processing_latency_ms_sum{...,backend="f",...} 99279.30498123169
+
+which indicates that the average processing latency is just over one second, as expected.
+
+You can even define a `custom metric <..ray-metrics.html#custom-metrics>`__ to use in your backend, and tag it with the current backend or replica.
+Here's an example:
+
+.. literalinclude:: ../../../python/ray/serve/examples/doc/snippet_custom_metric.py
+  :lines: 11-23
+
+See the
+`Ray Monitoring documentation <../ray-metrics.html>`__ for more details, including instructions for scraping these metrics using Prometheus.
 
 Reconfiguring Backends (Experimental)
 =====================================
diff --git a/doc/source/serve/package-ref.rst b/doc/source/serve/package-ref.rst
index b7014ab45..3df9c2915 100644
--- a/doc/source/serve/package-ref.rst
+++ b/doc/source/serve/package-ref.rst
@@ -17,6 +17,10 @@ Backend Configuration
 
 .. autoclass:: ray.serve.CondaEnv
 
+.. autofunction:: ray.serve.get_current_backend_tag
+
+.. autofunction:: ray.serve.get_current_replica_tag
+
 .. _`servehandle-api`:
 
 ServeHandle API
@@ -35,4 +39,5 @@ Batching Requests
 .. autofunction:: ray.serve.accept_batch
 
 Built-in Backends
+-----------------
 .. autoclass:: ray.serve.backends.ImportedBackend
diff --git a/python/ray/serve/__init__.py b/python/ray/serve/__init__.py
index ee23980d1..61aa92acc 100644
--- a/python/ray/serve/__init__.py
+++ b/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",
 ]
diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py
index 05ce4576e..483917182 100644
--- a/python/ray/serve/api.py
+++ b/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.
 
diff --git a/python/ray/serve/backend_worker.py b/python/ray/serve/backend_worker.py
index 93904ce94..197ee302c 100644
--- a/python/ray/serve/backend_worker.py
+++ b/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,
diff --git a/python/ray/serve/examples/doc/snippet_custom_metric.py b/python/ray/serve/examples/doc/snippet_custom_metric.py
new file mode 100644
index 000000000..cfe5dd16f
--- /dev/null
+++ b/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)
diff --git a/python/ray/serve/examples/doc/snippet_metrics.py b/python/ray/serve/examples/doc/snippet_metrics.py
new file mode 100644
index 000000000..391289e5f
--- /dev/null
+++ b/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())
diff --git a/python/ray/serve/http_proxy.py b/python/ray/serve/http_proxy.py
index 77448860a..b6a551023 100644
--- a/python/ray/serve/http_proxy.py
+++ b/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):
diff --git a/python/ray/serve/router.py b/python/ray/serve/router.py
index 8d0f578f7..1ee2e3c59 100644
--- a/python/ray/serve/router.py
+++ b/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):
diff --git a/python/ray/serve/utils.py b/python/ray/serve/utils.py
index a7d271592..550ccbb45 100644
--- a/python/ray/serve/utils.py
+++ b/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,