feat: expose endpoint to actor(alibaba#515) alibaba#514 (alibaba#566)

* feat: expose metrics_endpoint to actor (alibaba#515)

* add runtime config to abstract operator

* fix test case

Author: FangwenDave

rock/sandbox/base_actor.py

@@ -35,6 +35,7 @@ class BaseActor:
     _user_id: str = "default"
     _experiment_id: str = "default"
     _namespace = "default"
+    _metrics_endpoint = ""
 
     def __init__(
         self,
@@ -80,7 +81,8 @@ def _init_monitor(self):
         role = self._role
         self.host = host
         logger.info(f"Initializing MetricsCollector with host={host}, port={port}, " f"env={env}, role={role}")
-        self.otlp_exporter = OTLPMetricExporter(endpoint=f"http://{host}:{port}/v1/metrics")
+        endpoint = self._metrics_endpoint or f"http://{host}:{port}/v1/metrics"
+        self.otlp_exporter = OTLPMetricExporter(endpoint=endpoint)
         self.metric_reader = PeriodicExportingMetricReader(
             self.otlp_exporter,
             export_interval_millis=self._export_interval_millis,
@@ -182,6 +184,12 @@ def __del__(self):
             logger.error(f"Error stopping monitoring: {e}")
             pass
 
+    def set_metrics_endpoint(self, metrics_endpoint: str):
+        self._metrics_endpoint = metrics_endpoint
+
+    def get_metrics_endpoint(self) -> str:
+        return self._metrics_endpoint
+
     async def get_sandbox_statistics(self):
         """Get sandbox statistics - default implementation returns None"""
         if isinstance(self._deployment, DockerDeployment):

rock/sandbox/operator/abstract.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 
 from rock.actions.sandbox.sandbox_info import SandboxInfo
+from rock.config import RuntimeConfig
 from rock.deployments.config import DeploymentConfig
 from rock.utils.providers.nacos_provider import NacosConfigProvider
 from rock.utils.providers.redis_provider import RedisProvider
@@ -9,6 +10,7 @@
 class AbstractOperator(ABC):
     _redis_provider: RedisProvider | None = None
     _nacos_provider: NacosConfigProvider | None = None
+    _runtime_config: RuntimeConfig | None = None
 
     @abstractmethod
     async def submit(self, config: DeploymentConfig, user_info: dict = {}) -> SandboxInfo:

rock/sandbox/operator/factory.py

@@ -4,7 +4,7 @@
 from typing import Any
 
 from rock.admin.core.ray_service import RayService
-from rock.config import RuntimeConfig, K8sConfig
+from rock.config import K8sConfig, RuntimeConfig
 from rock.logger import init_logger
 from rock.sandbox.operator.abstract import AbstractOperator
 from rock.sandbox.operator.k8s.operator import K8sOperator
@@ -58,7 +58,7 @@ def create_operator(context: OperatorContext) -> AbstractOperator:
             if context.ray_service is None:
                 raise ValueError("RayService is required for RayOperator")
             logger.info("Creating RayOperator")
-            ray_operator = RayOperator(ray_service=context.ray_service)
+            ray_operator = RayOperator(ray_service=context.ray_service, runtime_config=context.runtime_config)
             if context.nacos_provider is not None:
                 ray_operator.set_nacos_provider(context.nacos_provider)
             return ray_operator

rock/sandbox/operator/ray.py

@@ -8,6 +8,7 @@
 from rock.admin.core.ray_service import RayService
 from rock.admin.core.redis_key import alive_sandbox_key
 from rock.common.constants import GET_STATUS_SWITCH
+from rock.config import RuntimeConfig
 from rock.deployments.config import DockerDeploymentConfig
 from rock.deployments.constants import Port
 from rock.deployments.docker import DockerDeployment
@@ -25,8 +26,9 @@
 
 
 class RayOperator(AbstractOperator):
-    def __init__(self, ray_service: RayService):
+    def __init__(self, ray_service: RayService, runtime_config: RuntimeConfig):
         self._ray_service = ray_service
+        self._runtime_config = runtime_config
 
     def _get_actor_name(self, sandbox_id: str) -> str:
         return f"sandbox-{sandbox_id}"
@@ -54,6 +56,7 @@ async def submit(self, config: DockerDeploymentConfig, user_info: dict = {}) ->
             sandbox_id = config.container_name
             logger.info(f"[{sandbox_id}] start_async params:{json.dumps(config.model_dump(), indent=2)}")
             sandbox_actor: SandboxActor = await self.create_actor(config)
+            sandbox_actor.set_metrics_endpoint.remote(self._runtime_config.metrics_endpoint)
             sandbox_actor.start.remote()
             user_id = user_info.get("user_id", "default")
             experiment_id = user_info.get("experiment_id", "default")

tests/unit/conftest.py

@@ -1,15 +1,13 @@
 import asyncio
 import time
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import pytest
 import ray
 from fakeredis import aioredis
-from ray.util.state import list_actors
-
-from unittest.mock import MagicMock
-
 from kubernetes import client
+from ray.util.state import list_actors
 
 from rock.admin.core.ray_service import RayService
 from rock.config import K8sConfig, RockConfig
@@ -69,8 +67,13 @@ def ray_service(rock_config: RockConfig, ray_init_shutdown):
 
 
 @pytest.fixture
-def ray_operator(ray_service):
-    ray_operator = RayOperator(ray_service)
+def runtime_config(rock_config: RockConfig):
+    return rock_config.runtime
+
+
+@pytest.fixture
+def ray_operator(ray_service, runtime_config):
+    ray_operator = RayOperator(ray_service, runtime_config)
     ray_operator.set_nacos_provider(None)
     return ray_operator
 

tests/unit/sandbox/operator/test_ray_operator.py

@@ -9,17 +9,17 @@
 
 
 @pytest.mark.need_ray
-def test_use_rocklet_returns_false_when_nacos_provider_is_none(ray_service):
+def test_use_rocklet_returns_false_when_nacos_provider_is_none(ray_service, runtime_config):
     """When _nacos_provider is None, use_rocklet should return False"""
-    operator = RayOperator(ray_service=ray_service)
+    operator = RayOperator(ray_service=ray_service, runtime_config=runtime_config)
     operator.set_nacos_provider(None)
     assert operator.use_rocklet() is False
 
 
 @pytest.mark.need_ray
-def test_use_rocklet_returns_false_when_switch_is_off(ray_service):
+def test_use_rocklet_returns_false_when_switch_is_off(ray_service, runtime_config):
     """When switch status is False, use_rocklet should return False"""
-    operator = RayOperator(ray_service=ray_service)
+    operator = RayOperator(ray_service=ray_service, runtime_config=runtime_config)
     mock_nacos_provider = MagicMock()
     mock_nacos_provider.get_switch_status.return_value = False
     operator.set_nacos_provider(mock_nacos_provider)
@@ -28,20 +28,21 @@ def test_use_rocklet_returns_false_when_switch_is_off(ray_service):
 
 
 @pytest.mark.need_ray
-def test_use_rocklet_returns_true_when_switch_is_on(ray_service):
+def test_use_rocklet_returns_true_when_switch_is_on(ray_service, runtime_config):
     """When switch status is True, use_rocklet should return True"""
-    operator = RayOperator(ray_service=ray_service)
+    operator = RayOperator(ray_service=ray_service, runtime_config=runtime_config)
     mock_nacos_provider = MagicMock()
     mock_nacos_provider.get_switch_status.return_value = True
     operator.set_nacos_provider(mock_nacos_provider)
     assert operator.use_rocklet() is True
     mock_nacos_provider.get_switch_status.assert_called_once_with(GET_STATUS_SWITCH)
 
+
 @pytest.mark.need_docker
 @pytest.mark.need_ray
 @pytest.mark.asyncio
-async def test_ray_operator(ray_service):
-    operator = RayOperator(ray_service=ray_service)
+async def test_ray_operator(ray_service, runtime_config):
+    operator = RayOperator(ray_service=ray_service, runtime_config=runtime_config)
     operator.set_nacos_provider(None)
     start_response: SandboxInfo = await operator.submit(DockerDeploymentConfig(container_name="test"))
     assert start_response.get("sandbox_id") == "test"

tests/unit/test_base_actor.py

@@ -0,0 +1,59 @@
+import pytest
+import ray
+
+from rock.deployments.config import LocalDeploymentConfig
+from rock.logger import init_logger
+from rock.sandbox.sandbox_actor import SandboxActor
+
+logger = init_logger(__name__)
+
+
+@pytest.mark.need_ray
+@pytest.mark.asyncio
+async def test_set_and_get_metrics_endpoint(ray_init_shutdown):
+    """Test setting and getting metrics endpoint together using Ray actor"""
+    sandbox_config = LocalDeploymentConfig(role="test", env="dev")
+    actor_name = "test-set-and-get-metrics-endpoint"
+
+    # Create SandboxActor using Ray
+    sandbox_actor = SandboxActor.options(name=actor_name, lifetime="detached").remote(
+        sandbox_config, sandbox_config.get_deployment()
+    )
+
+    try:
+        # Test initial setting
+        test_endpoint = "http://test-host:9090/v1/metrics"
+        ray.get(sandbox_actor.set_metrics_endpoint.remote(test_endpoint))
+        result = ray.get(sandbox_actor.get_metrics_endpoint.remote())
+        assert result == test_endpoint
+        logger.info(f"Initial endpoint set successfully: {result}")
+
+        # Test updating the endpoint
+        new_endpoint = "http://new-host:5000/v1/metrics"
+        ray.get(sandbox_actor.set_metrics_endpoint.remote(new_endpoint))
+        result = ray.get(sandbox_actor.get_metrics_endpoint.remote())
+        assert result == new_endpoint
+        logger.info(f"Updated endpoint successfully: {result}")
+    finally:
+        ray.kill(sandbox_actor)
+
+
+@pytest.mark.need_ray
+@pytest.mark.asyncio
+async def test_get_metrics_endpoint_default(ray_init_shutdown):
+    """Test getting metrics endpoint with default empty value using Ray actor"""
+    sandbox_config = LocalDeploymentConfig(role="test", env="dev")
+    actor_name = "test-get-metrics-endpoint-default"
+
+    # Create SandboxActor using Ray
+    sandbox_actor = SandboxActor.options(name=actor_name, lifetime="detached").remote(
+        sandbox_config, sandbox_config.get_deployment()
+    )
+
+    try:
+        # Get the default endpoint (should be empty string)
+        result = ray.get(sandbox_actor.get_metrics_endpoint.remote())
+        assert result == ""
+        logger.info(f"Default metrics endpoint is empty as expected: '{result}'")
+    finally:
+        ray.kill(sandbox_actor)