Conversation
Cleaned up changed code EC2 path working now
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughRuntime execution metadata is written to Redis, extracted into normalized rows, and upserted into a SQLAlchemy ChangesRuntime persistence flow
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant WorkflowHandler
participant Future
participant LocalController
participant GlobalController
participant Redis
participant SQLAlchemy
WorkflowHandler->>Redis: store request workflow
Future->>Redis: store future created_at
LocalController->>Redis: store execution metadata
GlobalController->>Redis: read future hashes
GlobalController->>SQLAlchemy: upsert runtime rows with resources
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
ventis/controller/local_controller.py (1)
415-417: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winClear the thread-local request ID after each pooled task.
ThreadPoolExecutorreuses worker threads, andventis_contextonly exposesset_request_id/get_request_id, so a request ID set here can leak into the next task on the same thread and misattribute spawned futures. Wrap this intry/finallyand clear or restore the prior value on exit.🤖 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/local_controller.py` around lines 415 - 417, Update the worker-task flow around ventis_context.set_request_id to restore the thread’s prior request ID or clear it in a finally block after each pooled task completes. Preserve request_id propagation during task execution while ensuring reused ThreadPoolExecutor workers cannot retain the previous task’s context.
🤖 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 `@tests/test_runtime_sqlalchemy.py`:
- Around line 47-54: Update the test fixture’s setUp and tearDown methods to
preserve the pre-test VENTIS_DATABASE_URL value before overriding it, then
restore that value afterward or remove the variable if it was originally unset.
Keep the existing database engine reset and temporary-file cleanup behavior
unchanged.
In `@ventis/controller/global_controller.py`:
- Around line 405-409: Update the request completion and cleanup flow around
pull_data and the cleanup thread so completed future:<id> hashes are not deleted
until persistence is acknowledged for each request. Either record persistence at
completion time or add per-request acknowledgment gating, ensuring pull_data can
always observe and persist runtime metadata before cleanup removes it.
- Around line 405-409: Update _poll_controllers around the send_data call to
catch database or persistence exceptions from pull_data/send_data, report the
failure through the existing logging mechanism, and continue into the controller
health-check flow so run() remains alive.
In `@ventis/controller/local_controller_frontend.py`:
- Around line 45-47: Validate that future_id is present and valid before the
metadata writes in the request-processing flow, rejecting the payload before
either self.redis.hset call. Update the logic around future_id and
_process_request() so malformed requests cannot create a future:None entry,
while preserving normal processing for valid IDs.
In `@ventis/controller/local_controller.py`:
- Around line 464-470: Update the duration calculation in the future completion
flow to convert the values returned by Redis hget for finished_at and created_at
into numeric timestamps before subtracting them, then persist the resulting
execution_time_(s) without changing the existing finished_at write.
In `@ventis/controller/utils/sqlalchemy.py`:
- Around line 67-68: Update send_data() around the workflow lookup so it reads
request:<session_id>:workflow from the canonical Redis client used by deploy(),
rather than the polled node’s Redis client. Alternatively, propagate the
workflow metadata into each node’s request context before send_data() runs,
ensuring distributed executions persist the actual workflow value instead of
NULL.
- Around line 32-38: Update _get_engine to ensure the runtime_information schema
is created or migrated immediately after initializing the engine, before it can
be used for polling or inserts. Reuse the project’s existing migration/schema
initialization mechanism if available, and preserve the current engine caching
and database URL behavior.
---
Outside diff comments:
In `@ventis/controller/local_controller.py`:
- Around line 415-417: Update the worker-task flow around
ventis_context.set_request_id to restore the thread’s prior request ID or clear
it in a finally block after each pooled task completes. Preserve request_id
propagation during task execution while ensuring reused ThreadPoolExecutor
workers cannot retain the previous task’s context.
🪄 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: c4f8e2ff-43ec-4bba-9665-5aa174fab72e
📒 Files selected for processing (8)
requirements.txttests/test_runtime_ec2.pytests/test_runtime_sqlalchemy.pyventis/controller/global_controller.pyventis/controller/local_controller.pyventis/controller/local_controller_frontend.pyventis/controller/utils/sqlalchemy.pyventis/deploy.py
💤 Files with no reviewable changes (1)
- tests/test_runtime_ec2.py
| os.environ["VENTIS_DATABASE_URL"] = f"sqlite:///{self.db.name}" | ||
| sqlmod._engine = None | ||
| with create_engine(os.environ["VENTIS_DATABASE_URL"]).begin() as conn: | ||
| conn.execute(text(_CREATE)) | ||
|
|
||
| def tearDown(self): | ||
| sqlmod._engine = None | ||
| os.unlink(self.db.name) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Restore VENTIS_DATABASE_URL after the test.
The suite leaves the environment pointing to a deleted SQLite file, which can break tests that run afterward. Preserve the previous value in setUp() and restore or remove it in tearDown().
🤖 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 `@tests/test_runtime_sqlalchemy.py` around lines 47 - 54, Update the test
fixture’s setUp and tearDown methods to preserve the pre-test
VENTIS_DATABASE_URL value before overriding it, then restore that value
afterward or remove the variable if it was originally unset. Keep the existing
database engine reset and temporary-file cleanup behavior unchanged.
| send_data( | ||
| pull_data(node_redis), | ||
| {c["name"]: c.get("resources", {}) for c in self.controllers}, | ||
| node_redis, | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Do not clean up futures before persistence is acknowledged.
The cleanup thread deletes future:<id> hashes for completed requests, while this code only samples them periodically. If cleanup runs after completion but before the next poll, pull_data() never sees the row and its runtime metadata is lost. Gate cleanup on a per-request persistence acknowledgment, or persist at completion time.
🤖 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/global_controller.py` around lines 405 - 409, Update the
request completion and cleanup flow around pull_data and the cleanup thread so
completed future:<id> hashes are not deleted until persistence is acknowledged
for each request. Either record persistence at completion time or add
per-request acknowledgment gating, ensuring pull_data can always observe and
persist runtime metadata before cleanup removes it.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Keep controller health polling alive when persistence fails.
send_data() runs before the health check and its database exceptions are uncaught; _poll_controllers() then escapes run(), stopping the controller daemon. Isolate this call with error handling and continue health polling while reporting the persistence failure.
🤖 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/global_controller.py` around lines 405 - 409, Update
_poll_controllers around the send_data call to catch database or persistence
exceptions from pull_data/send_data, report the failure through the existing
logging mechanism, and continue into the controller health-check flow so run()
remains alive.
| future_id = data.get("future_id") | ||
| self.redis.hset(f"future:{future_id}", "created_at", time.time()) | ||
| self.redis.hset(f"future:{future_id}", "agent", data.get("service")) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Validate future_id before writing metadata.
A malformed payload creates future:None; pull_data() treats that as a normal future and can upsert a bogus runtime row. _process_request() rejects missing IDs only after this mutation. Reject the request before either HSET.
🤖 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/local_controller_frontend.py` around lines 45 - 47,
Validate that future_id is present and valid before the metadata writes in the
request-processing flow, rejecting the payload before either self.redis.hset
call. Update the logic around future_id and _process_request() so malformed
requests cannot create a future:None entry, while preserving normal processing
for valid IDs.
| self.redis.hset(f"future:{future_id}", "finished_at", time.time()) | ||
| self.redis.hset( | ||
| f"future:{future_id}", | ||
| "execution_time_(s)", | ||
| self.redis.hget(f"future:{future_id}", "finished_at") | ||
| - self.redis.hget(f"future:{future_id}", "created_at"), | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Convert Redis timestamps before calculating duration.
hget() returns decoded strings, so Line 468 subtracts two strings and raises TypeError. The result may already be written, but finished_at and execution_time_(s) are never persisted.
Proposed fix
- self.redis.hset(f"future:{future_id}", "finished_at", time.time())
+ finished_at = time.time()
+ self.redis.hset(f"future:{future_id}", "finished_at", finished_at)
self.redis.hset(
f"future:{future_id}",
"execution_time_(s)",
- self.redis.hget(f"future:{future_id}", "finished_at")
- - self.redis.hget(f"future:{future_id}", "created_at"),
+ finished_at - float(self.redis.hget(f"future:{future_id}", "created_at")),
)📝 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.
| self.redis.hset(f"future:{future_id}", "finished_at", time.time()) | |
| self.redis.hset( | |
| f"future:{future_id}", | |
| "execution_time_(s)", | |
| self.redis.hget(f"future:{future_id}", "finished_at") | |
| - self.redis.hget(f"future:{future_id}", "created_at"), | |
| ) | |
| finished_at = time.time() | |
| self.redis.hset(f"future:{future_id}", "finished_at", finished_at) | |
| self.redis.hset( | |
| f"future:{future_id}", | |
| "execution_time_(s)", | |
| finished_at - float(self.redis.hget(f"future:{future_id}", "created_at")), | |
| ) |
🤖 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/local_controller.py` around lines 464 - 470, Update the
duration calculation in the future completion flow to convert the values
returned by Redis hget for finished_at and created_at into numeric timestamps
before subtracting them, then persist the resulting execution_time_(s) without
changing the existing finished_at write.
| def _get_engine(): | ||
| global _engine | ||
| if _engine is None: | ||
| _engine = create_engine( | ||
| os.environ.get("VENTIS_DATABASE_URL", "sqlite:///ventis_runtime.db") | ||
| ) | ||
| return _engine |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Create or migrate runtime_information before polling.
This only opens the database; production never creates the table. A fresh default sqlite:///ventis_runtime.db therefore fails on the first insert with “no such table: runtime_information.” The test currently masks this by creating the table itself. Add a migration or explicit bootstrap schema initialization.
🤖 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/utils/sqlalchemy.py` around lines 32 - 38, Update
_get_engine to ensure the runtime_information schema is created or migrated
immediately after initializing the engine, before it can be used for polling or
inserts. Reuse the project’s existing migration/schema initialization mechanism
if available, and preserve the current engine caching and database URL behavior.
| session_id = raw.get("request_id") | ||
| workflow = redis_client.get(f"request:{session_id}:workflow") |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Read workflow metadata from its canonical Redis location.
send_data() receives the polled node’s Redis client, but deploy() writes request:<id>:workflow to the deploy process’s Redis. The local-controller flow copies request context to node Redis, not workflow metadata, so distributed executions persist NULL workflow values. Propagate workflow into each node’s request metadata or query the canonical Redis client here.
🤖 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/utils/sqlalchemy.py` around lines 67 - 68, Update
send_data() around the workflow lookup so it reads request:<session_id>:workflow
from the canonical Redis client used by deploy(), rather than the polled node’s
Redis client. Alternatively, propagate the workflow metadata into each node’s
request context before send_data() runs, ensuring distributed executions persist
the actual workflow value instead of NULL.
| return rows | ||
|
|
||
|
|
||
| def send_data(rows, resources_by_agent=None, redis_client=None): |
There was a problem hiding this comment.
We eventually need actual CPU/GPU utilization from here
|
In commit 42cb13e, I changed the database URL to be manually configurable from global_controller.yaml. I also simplified the execution_time calculation, calculating it right before it gets pushed to the database instead of calculating it in the local controller and storing it in the local controller store. |
fixed the lint/type checking errors
|
In commit 4ebe96f, I fixed the linting/type errors the CI/test was displaying. No major/critical changes. Other than linting and formatting:
|
iidsample
left a comment
iidsample
left a comment
There was a problem hiding this comment.
Approved changes to push data to sql
2 participants
Added the ORM, with the global controller periodically polling each local controller for its requests, gathering those requests, then pushing to a SQL database.
Current code connects to an arbitrary sqlite db
Below is the schema used:
Table runtime_information {
session_id: uuid [primary key]
workflow: varchar
agent: varchar
execution_time: integer
cpu_resource: float
gpu_resource: float
created_at: timestamp
updated_at: timestamp
}
Summary by CodeRabbit
created_at; local execution/frontend record additional runtime metadata includingrequest_idandagentdetails.finished_at/updated timestamps).