[Serve] Collect telemetry data for multiplexed api (ray-project#38091)

python/ray/serve/_private/utils.py

@@ -519,6 +519,7 @@ def check_obj_ref_ready_nowait(obj_ref: ObjectRef) -> bool:
  "SERVE_DEPLOYMENT_HANDLE_TO_OBJECT_REF_API_USED": (
  TagKey.SERVE_DEPLOYMENT_HANDLE_TO_OBJECT_REF_API_USED
  ),
+ "SERVE_MULTIPLEXED_API_USED": TagKey.SERVE_MULTIPLEXED_API_USED,
 }
 
 

python/ray/serve/multiplex.py

@@ -16,7 +16,7 @@
  get_internal_replica_context,
 )
 from ray.serve._private.common import MultiplexedReplicaInfo
-from ray.serve._private.utils import MetricsPusher
+from ray.serve._private.utils import MetricsPusher, record_serve_tag
 from ray.serve import metrics
 
 
@@ -52,6 +52,9 @@ def __init__(
  current replica. If it is -1, there is no limit for the number of models
  per replica.
  """
+
+ record_serve_tag("SERVE_MULTIPLEXED_API_USED", "1")
+
  self.models = OrderedDict()
  self._func: Callable = model_load_func
  self.self_arg: Any = self_arg

python/ray/serve/tests/test_telemetry.py

@@ -15,7 +15,11 @@
 from ray.dag.input_node import InputNode
 from ray.serve.drivers import DefaultgRPCDriver, DAGDriver
 from ray.serve.http_adapters import json_request
-from ray.serve._private.constants import SERVE_NAMESPACE, SERVE_DEFAULT_APP_NAME
+from ray.serve._private.constants import (
+ SERVE_NAMESPACE,
+ SERVE_DEFAULT_APP_NAME,
+ SERVE_MULTIPLEXED_MODEL_ID,
+)
 from ray.serve.context import get_global_client
 from ray.serve._private.common import ApplicationStatus
 from ray.serve.schema import ServeDeploySchema
@@ -640,5 +644,57 @@ def check_telemetry(tag_should_be_set: bool):
  wait_for_condition(check_telemetry, tag_should_be_set=True)
 
 
+def test_multiplexed_detect(manage_ray):
+ """Check that multiplexed api is detected by telemetry."""
+
+ subprocess.check_output(["ray", "start", "--head"])
+ wait_for_condition(check_ray_started, timeout=5)
+
+ @serve.deployment
+ class Model:
+ @serve.multiplexed(max_num_models_per_replica=1)
+ async def get_model(self, tag):
+ return tag
+
+ async def __call__(self, request):
+ tag = serve.get_multiplexed_model_id()
+ await self.get_model(tag)
+ return tag
+
+ serve.run(Model.bind(), name="app", route_prefix="/app")
+
+ storage_handle = start_telemetry_app()
+ wait_for_condition(
+ lambda: ray.get(storage_handle.get_reports_received.remote()) > 0, timeout=5
+ )
+ report = ray.get(storage_handle.get_report.remote())
+ assert "serve_multiplexed_api_used" not in report["extra_usage_tags"]
+
+ client = get_global_client()
+ wait_for_condition(
+ lambda: client.get_serve_status("app").app_status.status
+ == ApplicationStatus.RUNNING,
+ timeout=15,
+ )
+
+ wait_for_condition(
+ lambda: ray.get(storage_handle.get_reports_received.remote()) > 0, timeout=5
+ )
+
+ headers = {SERVE_MULTIPLEXED_MODEL_ID: "1"}
+ resp = requests.get("http:https://localhost:8000/app", headers=headers)
+ assert resp.status_code == 200
+
+ wait_for_condition(
+ lambda: int(
+ ray.get(storage_handle.get_report.remote())["extra_usage_tags"][
+ "serve_multiplexed_api_used"
+ ]
+ )
+ == 1,
+ timeout=5,
+ )
+
+
 if __name__ == "__main__":
  sys.exit(pytest.main(["-v", "-s", __file__]))

src/ray/protobuf/usage.proto

@@ -75,6 +75,8 @@ enum TagKey {
  // Whether the `to_object_ref` or any of its variants were used in the
  // `DeploymentHandle` API.
  SERVE_DEPLOYMENT_HANDLE_TO_OBJECT_REF_API_USED = 21;
+ // Whether multuplexed API is was used.
+ SERVE_MULTIPLEXED_API_USED = 22;
 
  // Ray Core State API
  // NOTE(rickyxx): Currently only setting "1" for tracking existence of