feature: added threading to ventis deploy | bug fix: fixed schema misalignment for runtime_information table#16
Conversation
…alignment for runtime_information table
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughInstance provisioning now handles missing replicas concurrently, reserves local ports through Redis, and preserves existing instances. Runtime telemetry timestamps use numeric storage, PostgreSQL dependencies are declared, and EC2 image transfers now include Bash pipefail execution and logging. ChangesRuntime operations
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant InstanceManager
participant Redis
participant ThreadPoolExecutor
participant Runtime
participant Controller
InstanceManager->>Redis: reserve local host port
InstanceManager->>ThreadPoolExecutor: submit missing replica jobs
ThreadPoolExecutor->>Runtime: provision replica with port callback
Runtime-->>ThreadPoolExecutor: return provisioned instance
ThreadPoolExecutor-->>InstanceManager: return completed jobs
InstanceManager->>Controller: register instances and runtime IDs
InstanceManager->>Redis: publish routing snapshot
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
ventis/controller/instance_manager.py (1)
178-198: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPartial reservation record persisted under the final instance key can break downstream consumers that assume a complete record.
_next_host_portnowhset_multiplesagent_name/provider/replica_index/host/host_portdirectly into the same Redis key that_write_instancewill later fully populate. If the job subsequently fails (or hangs) before_write_instanceruns, this half-written record (noruntime_id,endpoint,container_port, etc.) is left behind and is fully visible tolist_instances(),publish_routing_snapshot(), andremove_instance()— none of which check forruntime_idpresence. In particular,remove_instanceat Line 145/148 (instance['agent_name'],instance["runtime_id"]) and_destroy_runtimewill raiseKeyError/fail if invoked against such a partial record, since onlyensure_instanceswas updated to treat missingruntime_idas "not yet provisioned."🔧 Proposed guard
def remove_instance(self, instance_id): key = f"agent_instance:{instance_id}" instance = self.redis.hgetall(key) - if not instance: + if not instance or not instance.get("runtime_id"): return🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/instance_manager.py` around lines 178 - 198, Update _next_host_port so port reservation does not persist a partial instance record under the final key used by _write_instance. Store the reservation separately or defer writing instance fields until _write_instance can populate the complete record, while preserving unique host-port allocation and ensuring list_instances, publish_routing_snapshot, and remove_instance only see fully provisioned records.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ventis/controller/cloud_provider_logic/EC2/_runtime.py`:
- Around line 234-236: Update the SSH command in the image transfer flow to
remove StrictHostKeyChecking=no and enable strict host-key verification.
Configure and reference a managed known_hosts file or entry for the target host,
preserving the existing key-based authentication and docker save/load pipeline.
- Around line 242-244: Update the logging around the docker save/load result to
sanitize and size-limit result.stdout and result.stderr before INFO logging,
escaping multiline content and truncating oversized output. Preserve complete
transfer output only when the explicitly enabled diagnostic logging level is
active.
- Line 231: Update the logging around the image transfer in the relevant EC2
runtime flow to include structured, redacted SSH command details, including the
destination and options, while explicitly omitting the private-key path.
Preserve the existing image and host context in the log message and ensure
sensitive command arguments are not emitted.
- Around line 232-241: Add a configured timeout to the image-transfer
subprocess.run call in the image-transfer flow, and catch
subprocess.TimeoutExpired to convert the timeout into the same failure path used
for other transfer errors so instance cleanup still executes. Preserve the
existing command and successful transfer behavior.
In `@ventis/controller/instance_manager.py`:
- Around line 74-91: Update the concurrent provisioning flow around
_provision_one and future.result() to isolate failures per job: apply the
existing provisioning timeout policy, catch and record/log each future’s
exception without aborting ensure_instances, and continue aggregating successful
instances so _add_instance_to_agent, _track_runtime, and the later routing
snapshot still run for completed jobs.
---
Outside diff comments:
In `@ventis/controller/instance_manager.py`:
- Around line 178-198: Update _next_host_port so port reservation does not
persist a partial instance record under the final key used by _write_instance.
Store the reservation separately or defer writing instance fields until
_write_instance can populate the complete record, while preserving unique
host-port allocation and ensuring list_instances, publish_routing_snapshot, and
remove_instance only see fully provisioned records.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: add3b464-d4d9-4942-8682-ff43a15bf541
📒 Files selected for processing (8)
pyproject.tomlrequirements.txttests/test_instance_manager_runtime.pytests/test_runtime_sqlalchemy.pyventis/controller/cloud_provider_logic/EC2/_runtime.pyventis/controller/instance_manager.pyventis/controller/local_controller.pyventis/controller/utils/sqlalchemy.py
| if spec.get("type") == "workflow": | ||
| port_args += ["-p", f"{spec.get('api_port', 8080)}:8080"] | ||
|
|
||
| logger.info("Transferring image %s to %s", image, host) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Log a redacted transfer command if command-level diagnostics are required.
This message records only the image and host, not the actual SSH destination/options. Add structured, redacted command details while omitting the private-key path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/cloud_provider_logic/EC2/_runtime.py` at line 231, Update
the logging around the image transfer in the relevant EC2 runtime flow to
include structured, redacted SSH command details, including the destination and
options, while explicitly omitting the private-key path. Preserve the existing
image and host context in the log message and ensure sensitive command arguments
are not emitted.
| result = subprocess.run( | ||
| "set -o pipefail; " | ||
| f"docker save {shlex.quote(image)} | ssh -o StrictHostKeyChecking=no " | ||
| f"-o IdentitiesOnly=yes -i {shlex.quote(key)} " | ||
| f"{shlex.quote(f'{ssh_user}@{host}')} 'sudo docker load'", | ||
| shell=True, | ||
| capture_output=True, | ||
| text=True, | ||
| executable="/bin/bash", | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Add a timeout to the image-transfer subprocess.
A stalled SSH handshake, sudo docker load, or pipeline can block a ThreadPoolExecutor worker indefinitely and prevent provisioning from completing. Pass a configured timeout and convert subprocess.TimeoutExpired into a failure so the existing instance-cleanup path runs.
🧰 Tools
🪛 OpenGrep (1.25.0)
[ERROR] 232-241: Dynamic command passed to subprocess with shell=True. Use a command list without shell=True, or use shlex.quote() to sanitize input.
(coderabbit.command-injection.python-shell-true)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/cloud_provider_logic/EC2/_runtime.py` around lines 232 -
241, Add a configured timeout to the image-transfer subprocess.run call in the
image-transfer flow, and catch subprocess.TimeoutExpired to convert the timeout
into the same failure path used for other transfer errors so instance cleanup
still executes. Preserve the existing command and successful transfer behavior.
| f"docker save {shlex.quote(image)} | ssh -o StrictHostKeyChecking=no " | ||
| f"-o IdentitiesOnly=yes -i {shlex.quote(key)} " | ||
| f"{shlex.quote(f'{ssh_user}@{host}')} 'sudo docker load'", |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Restore SSH host-key verification.
StrictHostKeyChecking=no accepts any server host key, so the image can be transferred to an unintended endpoint without cryptographic host authentication. Configure a managed known_hosts entry and use strict verification for this provisioning path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/cloud_provider_logic/EC2/_runtime.py` around lines 234 -
236, Update the SSH command in the image transfer flow to remove
StrictHostKeyChecking=no and enable strict host-key verification. Configure and
reference a managed known_hosts file or entry for the target host, preserving
the existing key-based authentication and docker save/load pipeline.
| logger.info( | ||
| "docker save|load returncode=%s stdout=%s stderr=%s", | ||
| result.returncode, result.stdout, result.stderr, |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
Bound and sanitize transfer output before logging.
Remote stdout and stderr are logged verbatim at INFO without a size limit, allowing large or multiline output to flood or obscure logs. Truncate and escape the captured output, while retaining full details only at an explicitly enabled diagnostic level.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/cloud_provider_logic/EC2/_runtime.py` around lines 242 -
244, Update the logging around the docker save/load result to sanitize and
size-limit result.stdout and result.stderr before INFO logging, escaping
multiline content and truncating oversized output. Preserve complete transfer
output only when the explicitly enabled diagnostic logging level is active.
| max_workers = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8)) | ||
| provisioned = [] | ||
| if jobs: | ||
| with ThreadPoolExecutor(max_workers=max_workers) as executor: | ||
| future_to_job = { | ||
| executor.submit(self._provision_one, job): job for job in jobs | ||
| } | ||
| for future in as_completed(future_to_job): | ||
| job = future_to_job[future] | ||
| instance = future.result() | ||
| provisioned.append( | ||
| (job["agent_name"], job["instance_id"], instance) | ||
| ) | ||
| self._write_instance(instance) | ||
|
|
||
| self._add_instance_to_agent(agent_name, instance_id) | ||
| self._track_runtime(agent_name, instance["runtime_id"]) | ||
| instances.append(instance) | ||
| for agent_name, instance_id, instance in existing + provisioned: | ||
| self._add_instance_to_agent(agent_name, instance_id) | ||
| self._track_runtime(agent_name, instance["runtime_id"]) | ||
| instances.append(instance) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unhandled/timeout-less future.result() aborts the whole batch on a single failure.
future.result() is called with no timeout and no per-future exception handling. If any one job (a slow EC2 boot, SSH hang, or docker failure) raises or blocks, it either hangs ensure_instances indefinitely or propagates immediately and skips _add_instance_to_agent/_track_runtime/publish_routing_snapshot for every other job that already completed successfully in this call (their Redis records are written via _write_instance inside _provision_one, but the aggregation/registration step at Line 88 is never reached). Callers get an unhandled exception instead of a partial result, and the routing snapshot is never republished even though other replicas came up fine.
🔧 Proposed fix: isolate failures per job
for future in as_completed(future_to_job):
job = future_to_job[future]
- instance = future.result()
- provisioned.append(
- (job["agent_name"], job["instance_id"], instance)
- )
+ try:
+ instance = future.result()
+ except Exception:
+ logger.exception(
+ "Provisioning failed for %s replica %s",
+ job["agent_name"], job["replica_index"],
+ )
+ continue
+ provisioned.append(
+ (job["agent_name"], job["instance_id"], instance)
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| max_workers = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8)) | |
| provisioned = [] | |
| if jobs: | |
| with ThreadPoolExecutor(max_workers=max_workers) as executor: | |
| future_to_job = { | |
| executor.submit(self._provision_one, job): job for job in jobs | |
| } | |
| for future in as_completed(future_to_job): | |
| job = future_to_job[future] | |
| instance = future.result() | |
| provisioned.append( | |
| (job["agent_name"], job["instance_id"], instance) | |
| ) | |
| self._write_instance(instance) | |
| self._add_instance_to_agent(agent_name, instance_id) | |
| self._track_runtime(agent_name, instance["runtime_id"]) | |
| instances.append(instance) | |
| for agent_name, instance_id, instance in existing + provisioned: | |
| self._add_instance_to_agent(agent_name, instance_id) | |
| self._track_runtime(agent_name, instance["runtime_id"]) | |
| instances.append(instance) | |
| max_workers = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8)) | |
| provisioned = [] | |
| if jobs: | |
| with ThreadPoolExecutor(max_workers=max_workers) as executor: | |
| future_to_job = { | |
| executor.submit(self._provision_one, job): job for job in jobs | |
| } | |
| for future in as_completed(future_to_job): | |
| job = future_to_job[future] | |
| try: | |
| instance = future.result() | |
| except Exception: | |
| logger.exception( | |
| "Provisioning failed for %s replica %s", | |
| job["agent_name"], job["replica_index"], | |
| ) | |
| continue | |
| provisioned.append( | |
| (job["agent_name"], job["instance_id"], instance) | |
| ) | |
| for agent_name, instance_id, instance in existing + provisioned: | |
| self._add_instance_to_agent(agent_name, instance_id) | |
| self._track_runtime(agent_name, instance["runtime_id"]) | |
| instances.append(instance) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/instance_manager.py` around lines 74 - 91, Update the
concurrent provisioning flow around _provision_one and future.result() to
isolate failures per job: apply the existing provisioning timeout policy, catch
and record/log each future’s exception without aborting ensure_instances, and
continue aggregating successful instances so _add_instance_to_agent,
_track_runtime, and the later routing snapshot still run for completed jobs.
2 participants
Parallelize agent instance provisioning in ventis deploy
This PR replaces the sequential per-replica provisioning loop in InstanceManager.ensure_instances() with a threaded implementation. Each replica's provision_instance/bootstrap_instance call (a blocking docker run locally, or an EC2 launch/SSH/health-check sequence) is I/O-bound, so a ThreadPoolExecutor (sized via VENTIS_MAX_AGENT_INSTANCES, matching the existing pattern in local_controller.py) now runs replicas concurrently instead of one at a time, cutting deploy time roughly to the slowest single replica instead of the sum of all of them.
To avoid two local replicas racing to the same port, port assignment happens sequentially during a planning pass before any threads start, with each reservation persisted to Redis immediately (_next_host_port) so later replicas see it as taken. The existing-instance check now requires a runtime_id to treat a record as complete, so a replica left half-provisioned by a failed bootstrap is correctly retried rather than skipped.
Also included: Postgres support for the runtime-tracking ORM (psycopg driver, a TABLE_NAME constant, schema-compatibility fixes for execution_time/created_at/updated_at as floats, and skipping rows without a valid session), a small EC2 image-transfer robustness fix (pipefail + logging), and updated/added tests covering the new threading and DB behavior.
Summary by CodeRabbit