Route local-provider agents over a dedicated docker network

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

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"],
         )
 

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}"

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