Skip to content

Route local-provider agents over a dedicated docker network #71

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Saaketh0 wants to merge 1 commit into from
Open

Route local-provider agents over a dedicated docker network #71

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions examples/helloworld/config/global_controller.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ agents:
cpu: 1
memory: 512
entrypoint: agents/example_agent.py
provider: EC2
provider: local

- name: VllmAgent
replicas: 1
Expand All @@ -19,7 +19,7 @@ agents:
cpu: 2
memory: 2048
entrypoint: agents/vllm_agent.py
provider: EC2
provider: local
instance_type: t3.micro

- name: Workflow
Expand All @@ -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
Expand Down
22 changes: 12 additions & 10 deletions tests/test_instance_manager_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
},
Expand All @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"],
)

Expand Down
20 changes: 8 additions & 12 deletions ventis/controller/cloud_provider_logic/Local/_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
CONTAINER_PORT = 50051
PROVIDER = "local"
MAX_PORT_ATTEMPTS = 50
NETWORK = "ventis-local"
_controller = None


Expand All @@ -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

Expand All @@ -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"),
}
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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 = {
Expand Down Expand Up @@ -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}"
44 changes: 21 additions & 23 deletions ventis/controller/global_controller.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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.
Expand All @@ -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)...",
Expand All @@ -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,
)

Expand Down Expand Up @@ -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
Expand Down