diff --git a/examples/helloworld/config/global_controller.yaml b/examples/helloworld/config/global_controller.yaml index 0b9c194..5f6c0cc 100644 --- a/examples/helloworld/config/global_controller.yaml +++ b/examples/helloworld/config/global_controller.yaml @@ -10,7 +10,7 @@ agents: cpu: 1 memory: 512 entrypoint: agents/example_agent.py - provider: EC2 + provider: local - name: VllmAgent replicas: 1 @@ -19,7 +19,7 @@ agents: cpu: 2 memory: 2048 entrypoint: agents/vllm_agent.py - provider: EC2 + provider: local instance_type: t3.micro - name: Workflow @@ -28,7 +28,7 @@ agents: redis_port: 6379 api_port: 8080 # Only needed for workflows, defaults to 8080 if not filled workflow_file: workflow/example_workflow.py - provider: EC2 + provider: local instance_type: t3.micro poll_interval: 5 diff --git a/tests/test_instance_manager_runtime.py b/tests/test_instance_manager_runtime.py index df35efe..13f9876 100644 --- a/tests/test_instance_manager_runtime.py +++ b/tests/test_instance_manager_runtime.py @@ -138,7 +138,7 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self): "host_port": "8000", "container_port": "50051", "endpoint": "localhost:8000", - "redis_host": "host.docker.internal", + "redis_host": "ventis-redis-localhost", "redis_port": "6379", "runtime_id": "ventis-local-alpha-0", }, @@ -154,17 +154,18 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self): "run", "-d", "-it", - "--add-host=host.docker.internal:host-gateway", + "--network", + "ventis-local", "--name", "ventis-local-alpha-0", "-p", "8000:50051", "-e", - "VENTIS_AGENT_PORT=8000", + "VENTIS_AGENT_PORT=50051", "-e", - "VENTIS_AGENT_HOST=host.docker.internal", + "VENTIS_AGENT_HOST=ventis-local-alpha-0", "-e", - "VENTIS_REDIS_HOST=host.docker.internal", + "VENTIS_REDIS_HOST=ventis-redis-localhost", "-e", "VENTIS_REDIS_PORT=6379", "-e", @@ -209,17 +210,18 @@ def test_local_workflow_and_resource_flags_stay_the_same(self): "run", "-d", "-it", - "--add-host=host.docker.internal:host-gateway", + "--network", + "ventis-local", "--name", "ventis-local-workflow-0", "-p", "8000:50051", "-e", - "VENTIS_AGENT_PORT=8000", + "VENTIS_AGENT_PORT=50051", "-e", - "VENTIS_AGENT_HOST=host.docker.internal", + "VENTIS_AGENT_HOST=ventis-local-workflow-0", "-e", - "VENTIS_REDIS_HOST=host.docker.internal", + "VENTIS_REDIS_HOST=ventis-redis-localhost", "-e", "VENTIS_REDIS_PORT=6379", "-e", @@ -255,7 +257,7 @@ def test_agent_id_is_published_under_the_controller_endpoint_key(self): alpha = manager.ensure_instances([{"name": "Alpha", "provider": "local"}])[0] self.assertEqual( - controller.redis.get("controller:host.docker.internal:8000:agent_id"), + controller.redis.get("controller:ventis-local-alpha-0:50051:agent_id"), alpha["agent_id"], ) diff --git a/ventis/controller/cloud_provider_logic/Local/_runtime.py b/ventis/controller/cloud_provider_logic/Local/_runtime.py index 1a84e56..dda4807 100644 --- a/ventis/controller/cloud_provider_logic/Local/_runtime.py +++ b/ventis/controller/cloud_provider_logic/Local/_runtime.py @@ -16,6 +16,7 @@ CONTAINER_PORT = 50051 PROVIDER = "local" MAX_PORT_ATTEMPTS = 50 +NETWORK = "ventis-local" _controller = None @@ -29,10 +30,6 @@ def _is_local_host(host): return host in {"localhost", "127.0.0.1"} -def _container_routing_host(host): - return "host.docker.internal" if _is_local_host(host) else host - - def validate_config(): return None @@ -46,7 +43,7 @@ def provision_instance(spec, replica_index, next_host_port): "provider": PROVIDER, "host": host, "host_port": host_port, - "redis_host": _container_routing_host(host), + "redis_host": f"ventis-redis-{host.replace('.', '-')}", "runtime_id": f"ventis-{PROVIDER}-{agent_name.lower()}-{replica_index}", "user": spec.get("user"), } @@ -80,15 +77,16 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id): "run", "-d", "-it", - "--add-host=host.docker.internal:host-gateway", + "--network", + NETWORK, "--name", runtime_id, "-p", f"{host_port}:{CONTAINER_PORT}", "-e", - f"VENTIS_AGENT_PORT={host_port}", + f"VENTIS_AGENT_PORT={CONTAINER_PORT}", "-e", - f"VENTIS_AGENT_HOST={redis_host}", + f"VENTIS_AGENT_HOST={runtime_id}", "-e", f"VENTIS_REDIS_HOST={redis_host}", "-e", @@ -138,7 +136,7 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id): f"{MAX_PORT_ATTEMPTS} attempts" ) - endpoint = f"{_container_routing_host(host)}:{host_port}" + endpoint = f"{runtime_id}:{CONTAINER_PORT}" _require_controller().redis.set(f"controller:{endpoint}:agent_id", agent_id) instance = { @@ -174,6 +172,4 @@ def terminate_instance(instance): def routing_endpoint_for(instance): - host = instance.get("host") - port = instance["host_port"] - return f"{_container_routing_host(host)}:{port}" + return f"{instance['runtime_id']}:{CONTAINER_PORT}" diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index c85dcbf..4bf109a 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -44,15 +44,13 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) +LOCAL_NETWORK = "ventis-local" + def _is_local_host(host): return host in {"localhost", "127.0.0.1"} -def _container_routing_host(host): - return "host.docker.internal" if _is_local_host(host) else host - - class GlobalController(object): """ Daemon that manages a routing table across multiple local controller instances. @@ -406,12 +404,16 @@ def _launch_redis_containers(self): logger.info("Reusing existing Redis container %s on %s", container_name, host) self.redis_containers[host] = container_name else: + if _is_local_host(host): + self._run_cmd(["docker", "network", "create", LOCAL_NETWORK], host, user) + network_args = ["--network", LOCAL_NETWORK] if _is_local_host(host) else [] cmd = [ "docker", "run", "-d", "--name", container_name, + *network_args, "-p", f"{redis_port}:6379", "redis:alpine", @@ -484,10 +486,6 @@ def _get_node_redis_for(self, host): """Get the Redis client for a given host, falling back to self.redis.""" return self.node_redis.get(host, self.redis) - def _agent_host_key(self, host): - """Return the host string as seen by Docker containers (for status key matching).""" - return _container_routing_host(host) - def _wait_for_healthy(self, timeout=30, interval=2): """ Block until all controllers report healthy in Redis, or until timeout. @@ -497,10 +495,7 @@ def _wait_for_healthy(self, timeout=30, interval=2): interval: Seconds between checks. """ deadline = time.time() + timeout - pending = [ - (instance["agent_name"], instance["host"], instance["host_port"]) - for instance in self.instance_manager.list_instances() - ] + pending = self.instance_manager.list_instances() logger.info( "Waiting for %d replica(s) to become healthy (timeout=%ds)...", @@ -510,26 +505,29 @@ def _wait_for_healthy(self, timeout=30, interval=2): while pending and time.time() < deadline: still_pending = [] - for name, host, port in pending: + for instance in pending: + name = instance["agent_name"] + host = instance["host"] + port = instance["host_port"] node_redis = self._get_node_redis_for(host) - agent_host = self._agent_host_key(host) - status = node_redis.get(f"controller:{agent_host}:{port}:status") + endpoint = self.instance_manager._routing_endpoint_for(instance) + status = node_redis.get(f"controller:{endpoint}:status") if status == "healthy": logger.info("Controller %s (%s:%s) is ready.", name, host, port) self._last_status[(host, port)] = "healthy" else: - still_pending.append((name, host, port)) + still_pending.append(instance) pending = still_pending if pending: time.sleep(interval) if pending: - for name, host, port in pending: + for instance in pending: logger.warning( "Controller %s (%s:%s) not ready after %ds.", - name, - host, - port, + instance["agent_name"], + instance["host"], + instance["host_port"], timeout, ) @@ -601,9 +599,9 @@ def _poll_one_instance(self, instance): port, e, ) - agent_host = self._agent_host_key(host) - status_key = f"controller:{agent_host}:{port}:status" - metrics_key = f"controller:{agent_host}:{port}:metrics" + endpoint = self.instance_manager._routing_endpoint_for(instance) + status_key = f"controller:{endpoint}:status" + metrics_key = f"controller:{endpoint}:metrics" # Getting metrics from local controllers # See LocalController._execute_locally