[Serve] Enable serve metrics lib working in ray actor (ray-project#33717

)

Make sure ray.serve.lib working with ray.actor without serve context.
```
@ray.remote
class MyActor:
    def __init__(self):
        self.my_counter = metrics.Counter(
            "my_ray_actor",
            description=("The number of requests to this deployment."),
            tag_keys=("my_tag",),
        )
    def test(self):
        self.my_counter.inc(tags={"my_tag": "value"})
        return "hello"

@serve.deployment(num_replicas=2)
class Model:
    def __init__(self, model_name):
        self.my_actor = MyActor.remote()

    async def __call__(self, req: starlette.requests.Request):
        await self.my_actor.test.remote()
        return
```

Signed-off-by: Jack He <[email protected]>

python/ray/serve/metrics.py

@@ -8,6 +8,8 @@
 
 def _add_serve_metric_tags(tag_keys: Optional[Tuple[str]] = None):
     """Add serve context tags to the tag_keys"""
+    if context.get_internal_replica_context() is None:
+        return tag_keys
     if DEPLOYMENT_TAG in tag_keys:
         raise ValueError(f"'{DEPLOYMENT_TAG}' tag is reserved for Ray Serve metrics")
     if REPLICA_TAG in tag_keys:
@@ -22,6 +24,8 @@ def _add_serve_metric_tags(tag_keys: Optional[Tuple[str]] = None):
 
 def _add_serve_metric_default_tags(default_tags: Dict[str, str]):
     """Add serve context tags and values to the default_tags"""
+    if context.get_internal_replica_context() is None:
+        return default_tags
     if DEPLOYMENT_TAG in default_tags:
         raise ValueError(f"'{DEPLOYMENT_TAG}' tag is reserved for Ray Serve metrics")
     if REPLICA_TAG in default_tags:

python/ray/serve/tests/test_metrics.py

@@ -442,6 +442,120 @@ def __call__(self):
         histogram_metrics[0]["replica"] == replica_tag
         histogram_metrics[0]["deployment"] == deployment_name
 
+    @pytest.mark.parametrize("use_actor", [False, True])
+    def test_serve_metrics_outside_serve(self, use_actor, serve_start_shutdown):
+        """Make sure ray.serve.metrics work in ray actor"""
+        if use_actor:
+
+            @ray.remote
+            class MyActor:
+                def __init__(self):
+                    self.counter = Counter(
+                        "my_counter",
+                        description="my counter metrics",
+                        tag_keys=(
+                            "my_static_tag",
+                            "my_runtime_tag",
+                        ),
+                    )
+                    self.counter.set_default_tags({"my_static_tag": "static_value"})
+                    self.histogram = Histogram(
+                        "my_histogram",
+                        description=("my histogram "),
+                        boundaries=DEFAULT_LATENCY_BUCKET_MS,
+                        tag_keys=(
+                            "my_static_tag",
+                            "my_runtime_tag",
+                        ),
+                    )
+                    self.histogram.set_default_tags({"my_static_tag": "static_value"})
+                    self.gauge = Gauge(
+                        "my_gauge",
+                        description=("my_gauge"),
+                        tag_keys=(
+                            "my_static_tag",
+                            "my_runtime_tag",
+                        ),
+                    )
+                    self.gauge.set_default_tags({"my_static_tag": "static_value"})
+
+                def test(self):
+                    self.counter.inc(tags={"my_runtime_tag": "100"})
+                    self.histogram.observe(200, tags={"my_runtime_tag": "200"})
+                    self.gauge.set(300, tags={"my_runtime_tag": "300"})
+                    return "hello"
+
+        else:
+            counter = Counter(
+                "my_counter",
+                description="my counter metrics",
+                tag_keys=(
+                    "my_static_tag",
+                    "my_runtime_tag",
+                ),
+            )
+            histogram = Histogram(
+                "my_histogram",
+                description=("my histogram "),
+                boundaries=DEFAULT_LATENCY_BUCKET_MS,
+                tag_keys=(
+                    "my_static_tag",
+                    "my_runtime_tag",
+                ),
+            )
+            gauge = Gauge(
+                "my_gauge",
+                description=("my_gauge"),
+                tag_keys=(
+                    "my_static_tag",
+                    "my_runtime_tag",
+                ),
+            )
+
+            @ray.remote
+            def fn():
+                counter.set_default_tags({"my_static_tag": "static_value"})
+                histogram.set_default_tags({"my_static_tag": "static_value"})
+                gauge.set_default_tags({"my_static_tag": "static_value"})
+                counter.inc(tags={"my_runtime_tag": "100"})
+                histogram.observe(200, tags={"my_runtime_tag": "200"})
+                gauge.set(300, tags={"my_runtime_tag": "300"})
+                return "hello"
+
+        @serve.deployment
+        class Model:
+            def __init__(self):
+                if use_actor:
+                    self.my_actor = MyActor.remote()
+
+            async def __call__(self):
+                if use_actor:
+                    return await self.my_actor.test.remote()
+                else:
+                    return await fn.remote()
+
+        serve.run(Model.bind(), name="app", route_prefix="/app")
+        resp = requests.get("http://127.0.0.1:8000/app")
+        assert resp.text == "hello"
+        wait_for_condition(
+            lambda: len(get_metric_dictionaries("my_gauge")) == 1,
+            timeout=20,
+        )
+        counter_metrics = get_metric_dictionaries("my_counter")
+        assert len(counter_metrics) == 1
+        counter_metrics[0]["my_static_tag"] == "static_value"
+        counter_metrics[0]["my_runtime_tag"] == "100"
+
+        gauge_metrics = get_metric_dictionaries("my_gauge")
+        assert len(counter_metrics) == 1
+        gauge_metrics[0]["my_static_tag"] == "static_value"
+        gauge_metrics[0]["my_runtime_tag"] == "300"
+
+        histogram_metrics = get_metric_dictionaries("my_histogram_sum")
+        assert len(histogram_metrics) == 1
+        histogram_metrics[0]["my_static_tag"] == "static_value"
+        histogram_metrics[0]["my_runtime_tag"] == "200"
+
 
 def test_actor_summary(serve_instance):
     @serve.deployment