From 9bbc35d2206016e72b65061579d7bf2e62098c47 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Fri, 28 Aug 2026 13:58:05 -0700 Subject: [PATCH 1/5] rough draft --- .../portfolio/config/global_controller.yaml | 25 +++++---- .../portfolio/workflow/portfolio_workflow.py | 1 - pyproject.toml | 2 +- requirements.txt | 2 - tests/test_otel_exporter_fanout.py | 2 +- tests/test_otel_exporter_fields.py | 2 +- .../OTLP_Exporter}/DESIGN.md | 6 +- ventis/OTLP_Exporter/SCHEMA.md | 56 +++++++++++++++++++ .../OTLP_Exporter}/__init__.py | 0 .../OTLP_Exporter}/convert.py | 24 ++++++-- {OTel_Exporter => ventis/OTLP_Exporter}/db.py | 4 +- .../OTLP_Exporter}/otel_exporter.py | 0 ventis/OTLP_Exporter/otel_queue.db | 0 ventis/cli.py | 21 +++++-- .../cloud_provider_logic/EC2/_runtime.py | 5 +- ventis/controller/global_controller.py | 25 +++++---- ventis/controller/utils/process_supervisor.py | 2 +- ventis/stub_generator.py | 40 +++++++++++-- 18 files changed, 166 insertions(+), 51 deletions(-) rename {OTel_Exporter => ventis/OTLP_Exporter}/DESIGN.md (98%) create mode 100644 ventis/OTLP_Exporter/SCHEMA.md rename {OTel_Exporter => ventis/OTLP_Exporter}/__init__.py (100%) rename {OTel_Exporter => ventis/OTLP_Exporter}/convert.py (72%) rename {OTel_Exporter => ventis/OTLP_Exporter}/db.py (98%) rename {OTel_Exporter => ventis/OTLP_Exporter}/otel_exporter.py (100%) create mode 100644 ventis/OTLP_Exporter/otel_queue.db diff --git a/examples/portfolio/config/global_controller.yaml b/examples/portfolio/config/global_controller.yaml index ab5af4d..edee85f 100644 --- a/examples/portfolio/config/global_controller.yaml +++ b/examples/portfolio/config/global_controller.yaml @@ -83,18 +83,18 @@ otel: destinations: - name: railway protocol: grpc - endpoint: yamanote.proxy.rlwy.net:19803 + endpoint: ${RAILWAY_OTLP_ENDPOINT} insecure: true headers: {} - - name: langfuse - protocol: http - endpoint: ${LANGFUSE_BASE_URL}/api/public/otel/v1/traces - headers: {} - name: grafana protocol: http endpoint: ${GRAFANA_OTLP_ENDPOINT}/v1/traces headers: Authorization: Basic ${GRAFANA_OTLP_HEADERS} +# - name: langfuse +# protocol: http +# endpoint: ${LANGFUSE_BASE_URL}/api/public/otel/v1/traces +# headers: {} # Polling interval in seconds poll_interval: 5 @@ -107,10 +107,13 @@ redis: # EC2 defaults for `provider: EC2` replicas. ec2: - region: us-east-1 - ami_id: ami-031ff6df47f26b546 - subnet_id: subnet-0638ac6d79d488124 + region: ${EC2_REGION} + ami_id: ${EC2_AMI_ID} + subnet_id: ${EC2_SUBNET_ID} security_group_ids: - - sg-025daf3a98e06cef3 - ssh_user: ubuntu - ssh_private_key_path: ~/.ssh/ventis_ec2 + - ${EC2_SECURITY_GROUP_ID} + ssh_user: ${EC2_SSH_USER} + ssh_private_key_path: ${EC2_SSH_PRIVATE_KEY_PATH} + +database: + url: ${DATABASE_URL} diff --git a/examples/portfolio/workflow/portfolio_workflow.py b/examples/portfolio/workflow/portfolio_workflow.py index 299a3f5..b8b684a 100644 --- a/examples/portfolio/workflow/portfolio_workflow.py +++ b/examples/portfolio/workflow/portfolio_workflow.py @@ -46,7 +46,6 @@ def main( advisor = AdvisorAgent() # Stage 0: parse the free-text request into structured holdings + window. - intent = intent_agent.parse(query=query).value() # parse() returns a dict, but a Future's .value() only ever gives back the # raw string ventis stored in Redis -- same deserialization requirement as # every other dict-returning agent call below. diff --git a/pyproject.toml b/pyproject.toml index 9ae1234..40cb43a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ requires = ["setuptools>=64"] build-backend = "setuptools.build_meta" [tool.setuptools.packages.find] -include = ["ventis*", "OTel_Exporter*"] +include = ["ventis*"] [tool.setuptools.package-data] ventis = [ diff --git a/requirements.txt b/requirements.txt index dd7a254..f06e0b0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,8 +4,6 @@ grpcio-tools redis pyyaml flask -ipdb -ipython sqlalchemy psycopg[binary] psutil diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index ed1b423..0691d61 100644 --- a/tests/test_otel_exporter_fanout.py +++ b/tests/test_otel_exporter_fanout.py @@ -13,7 +13,7 @@ ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) # ``otel_exporter.py`` is also executed as a script from its own directory and # therefore imports ``convert`` and ``db`` as top-level modules. -sys.path.insert(0, os.path.join(ROOT, "OTel_Exporter")) +sys.path.insert(0, os.path.join(ROOT, "ventis", "OTLP_Exporter")) import db # noqa: E402 import otel_exporter # noqa: E402 diff --git a/tests/test_otel_exporter_fields.py b/tests/test_otel_exporter_fields.py index 4a62c58..ff8cad6 100644 --- a/tests/test_otel_exporter_fields.py +++ b/tests/test_otel_exporter_fields.py @@ -5,7 +5,7 @@ import unittest from unittest.mock import patch -from OTel_Exporter import convert, db +from ventis.OTLP_Exporter import convert, db class OTelExporterFieldTests(unittest.TestCase): diff --git a/OTel_Exporter/DESIGN.md b/ventis/OTLP_Exporter/DESIGN.md similarity index 98% rename from OTel_Exporter/DESIGN.md rename to ventis/OTLP_Exporter/DESIGN.md index 86a2a95..287fffd 100644 --- a/OTel_Exporter/DESIGN.md +++ b/ventis/OTLP_Exporter/DESIGN.md @@ -48,7 +48,7 @@ Decisions (final status): Configuration is read at exporter startup; changing it requires a GlobalController/exporter restart. - **Data source**: NOT `runtime_information` — a dedicated `waiting` table in its own - SQLite file (`OTel_Exporter/otel_queue.db`, see `db.py`), written by GC's existing + SQLite file (`ventis/OTLP_Exporter/otel_queue.db`, see `db.py`), written by GC's existing `_poll_controllers` *alongside* (not instead of) the existing `send_runtime_information` write. Keeps this pipeline's schema/state fully decoupled from the dashboard/cost table. @@ -95,7 +95,7 @@ endpoint and headers to the SDK. `BatchSpanProcessor(..., schedule_delay_millis= `max_export_batch_size` is left at the SDK default (512), which already approximates the original "500 spans" batching ask without any override needed. -### 2. `OTel_Exporter/otel_exporter.py` +### 2. `ventis/OTLP_Exporter/otel_exporter.py` A plain loop, polling every `POLL_INTERVAL_SECONDS` (5s, checked every 1s so SIGTERM stays responsive), calling `_send_pending()` each tick. At startup it constructs one independent OTLP exporter and `BatchSpanProcessor` for each configured destination; @@ -111,7 +111,7 @@ each pair may use a different protocol, endpoint, and headers: since spans are hand-built and handed straight to the processors via `on_end()`. - Every processor is shut down on exit, flushing its pending batch independently. -### 3. Future row → OTel span conversion (`OTel_Exporter/convert.py`) +### 3. Future row → OTel span conversion (`ventis/OTLP_Exporter/convert.py`) `future_id` maps to OTel `span_id`, not `trace_id` — `session_id` (== `request_id`) is the one that maps to `trace_id`. Both are `uuid4().hex` (32 hex chars / 16 bytes); OTel `trace_id` is 128-bit (16 bytes, fits directly) and `span_id` is 64-bit (8 bytes, needs diff --git a/ventis/OTLP_Exporter/SCHEMA.md b/ventis/OTLP_Exporter/SCHEMA.md new file mode 100644 index 0000000..38f8664 --- /dev/null +++ b/ventis/OTLP_Exporter/SCHEMA.md @@ -0,0 +1,56 @@ +# OTel span storage schema + +Each emitted OTel span is stored as one `otel_spans` record. This chart also defines a +one-to-one `otel_span_attributes` projection, linked by the same `span_id`, for the +known exporter attributes. The current receiver retains the raw `attributes` JSONB map; +the child table is the relational schema described in `otel_spans_schema.txt`. + +```text +┌───────────────────────────┐ ┌────────────────────────────────┐ +│ otel_spans │ │ otel_span_attributes │ +├───────────────────────────┤ ├────────────────────────────────┤ +│ PK span_id │────1:1───│ PK/FK span_id │ +│ trace_id │ │ model and agent ID │ +│ parent_span_id │ │ CPU and GPU │ +│ name │ │ timing and token usage │ +│ kind │ │ input and output │ +│ start/end time (ns) │ │ project and error count │ +│ status code/message │ │ server/token/total cost │ +│ attributes (JSONB) │ │ cache tokens/hit ratio │ +│ events (JSONB) │ └────────────────────────────────┘ +└───────────────────────────┘ +``` + +## `otel_spans` + +| Column | Meaning | +| --- | --- | +| `span_id` | Unique identifier for this span. | +| `trace_id` | Identifier shared by all spans in the same trace. | +| `parent_span_id` | Parent span; empty for a root span. | +| `name` | Operation name, such as an agent method. | +| `kind` | OTel role of the work; current exporter spans are `SPAN_KIND_INTERNAL`. | +| `start_time_unix_nano` / `end_time_unix_nano` | Raw Unix timestamps in nanoseconds. | +| `status_code` / `status_message` | OTel outcome: normally `STATUS_CODE_UNSET`, or `STATUS_CODE_ERROR` with an error message. | +| `attributes` | Complete raw OTel attribute map (JSONB). | +| `events` | OTel events, including any `exception` event (JSONB). | + +## `otel_span_attributes` + +This one-to-one projection mirrors every attribute currently emitted by +`convert.waiting_row_to_span()`. Fields are nullable because OTel omits an attribute +whose source value is `None`. + +| Group | Columns | +| --- | --- | +| Model and agent | `gen_ai.request.model`, `gen_ai.agent.id` | +| Resources and timing | `cpu`, `gpu`, `execution_time_ms`, `queue_time_ms` | +| Token usage | `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `token_count`, `gen_ai.usage.cache_read.input_tokens` | +| Input and output | `langfuse.observation.input`, `langfuse.observation.output` | +| Project and errors | `project_id`, `error_count` | +| Costs | `server_cost`, `token_cost`, `gen_ai.usage.cost` | +| Cache | `cache_hit_ratio` | + +The DBML source is [`../otel_spans_schema.txt`](../otel_spans_schema.txt). + +Successful spans use `STATUS_CODE_UNSET` rather than `STATUS_CODE_OK` because OpenTelemetry reserves `OK` for application- or operator-validated success, while instrumentation normally sets a status only when it records an error. diff --git a/OTel_Exporter/__init__.py b/ventis/OTLP_Exporter/__init__.py similarity index 100% rename from OTel_Exporter/__init__.py rename to ventis/OTLP_Exporter/__init__.py diff --git a/OTel_Exporter/convert.py b/ventis/OTLP_Exporter/convert.py similarity index 72% rename from OTel_Exporter/convert.py rename to ventis/OTLP_Exporter/convert.py index b7fb493..66da972 100644 --- a/OTel_Exporter/convert.py +++ b/ventis/OTLP_Exporter/convert.py @@ -3,7 +3,7 @@ Pure function, no I/O, no batching, no network calls. Builds ReadableSpan objects directly instead of going through Tracer.start_span() -- there's no live tracer here, futures already finished (sometimes in another process), so this is a historical-row -conversion, not live tracing. See OTel_Exporter/DESIGN.md for why this deviates from +conversion, not live tracing. See ventis/OTLP_Exporter/DESIGN.md for why this deviates from the SDK's usual advice against constructing ReadableSpan by hand. """ @@ -64,9 +64,17 @@ def waiting_row_to_span(row): ) status = Status(StatusCode.ERROR, description=row.get("error_message")) - # Model and token usage use OTel GenAI semantic-convention names. Observation - # input/output use Langfuse's documented JSON-string attributes. The remaining - # Ventis infrastructure values have no GenAI equivalent, so they keep plain names. + # Model, token, agent, and cache-read usage use OTel GenAI semantic-convention + # names (see https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/). + # total_cost uses gen_ai.usage.cost, which isn't yet an official semconv attribute + # but is the name Langfuse's OTel ingestion actually reads (its JSON cost_details + # attribute is currently broken -- see langfuse/langfuse#11030). Observation + # input/output use Langfuse's documented JSON-string attributes. `errors` is named + # error_count, not "errors"/"error", to avoid colliding with OTel's reserved + # error.* namespace (error.type etc.), which describes a single error, not a + # count. The remaining Ventis-specific values (project_id, server/token cost + # breakdown, cache_hit_ratio) have no GenAI or Langfuse equivalent, so they keep + # plain names. attributes = { k: v for k, v in { @@ -80,6 +88,14 @@ def waiting_row_to_span(row): "token_count": row.get("token_count"), "langfuse.observation.input": row.get("input"), "langfuse.observation.output": row.get("output"), + "project_id": row.get("project_id"), + "gen_ai.agent.id": row.get("agent_id"), + "error_count": row.get("errors"), + "server_cost": row.get("server_cost"), + "token_cost": row.get("token_cost"), + "gen_ai.usage.cost": row.get("total_cost"), + "gen_ai.usage.cache_read.input_tokens": row.get("cached_tokens"), + "cache_hit_ratio": row.get("cache_hit_ratio"), }.items() if v is not None } diff --git a/OTel_Exporter/db.py b/ventis/OTLP_Exporter/db.py similarity index 98% rename from OTel_Exporter/db.py rename to ventis/OTLP_Exporter/db.py index a8a675f..e6a1834 100644 --- a/OTel_Exporter/db.py +++ b/ventis/OTLP_Exporter/db.py @@ -130,7 +130,7 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH if not fid or not session_id: continue agent_id = raw.get("agent") - started_at = float(raw.get("created_at") or 0) or None + started_at = float(raw.get("created_at") or 0) finished_at = float(raw["finished_at"]) if raw.get("finished_at") else None execution_time_ms = ( round((finished_at - started_at) * 1000) @@ -160,7 +160,7 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH * _TOKEN_COST_MULTIPLIER ) # Server cost needs an elapsed duration -- only available once finished. - if finished_at and started_at: + if finished_at is not None: server_cost = ( pricing.compute_server_cost( redis_client.get(f"agent:{agent_id}:instance_type") diff --git a/OTel_Exporter/otel_exporter.py b/ventis/OTLP_Exporter/otel_exporter.py similarity index 100% rename from OTel_Exporter/otel_exporter.py rename to ventis/OTLP_Exporter/otel_exporter.py diff --git a/ventis/OTLP_Exporter/otel_queue.db b/ventis/OTLP_Exporter/otel_queue.db new file mode 100644 index 0000000..e69de29 diff --git a/ventis/cli.py b/ventis/cli.py index 920df15..31a32e0 100644 --- a/ventis/cli.py +++ b/ventis/cli.py @@ -238,6 +238,15 @@ def cmd_build(args): if name: yaml_by_name[name] = yaml_path + # Maps each generated stub's basename to its agent's entrypoint path, so a + # stub can also be placed at its nested, entrypoint-mirrored location. + entrypoints_by_name = {a["name"]: a.get("entrypoint") for a in agents} + stub_entrypoints = { + f"{os.path.splitext(os.path.basename(p))[0]}.py": entrypoints_by_name[n] + for n, p in yaml_by_name.items() + if entrypoints_by_name.get(n) + } + stub_paths = [] for yaml_path in yaml_files: base_name = os.path.splitext(os.path.basename(yaml_path))[0] @@ -302,9 +311,9 @@ def cmd_build(args): api_port=agent_cfg.get("api_port", 8080), requirements=_normalize_requirements(agent_cfg), project_dir=project_dir, - # A workflow script imports stubs by flat module name (e.g. `from - # intent_agent import ...`), not by the agent's own entrypoint path. - stub_entrypoints=None, + # Stubs are placed both flat and at their entrypoint-mirrored path, + # so both flat and nested import styles resolve to the stub. + stub_entrypoints=stub_entrypoints, ) else: @@ -340,9 +349,9 @@ def cmd_build(args): stub_files=stub_paths, requirements=_normalize_requirements(agent_cfg), project_dir=project_dir, - # Same reasoning as the workflow call above: this project's agents - # import each other's stubs by flat module name, not entrypoint path. - stub_entrypoints=None, + # Same reasoning as the workflow call above: stubs are placed both + # flat and at their entrypoint-mirrored path. + stub_entrypoints=stub_entrypoints, ) bake_targets.append( diff --git a/ventis/controller/cloud_provider_logic/EC2/_runtime.py b/ventis/controller/cloud_provider_logic/EC2/_runtime.py index 9955fa2..7e4e9ab 100644 --- a/ventis/controller/cloud_provider_logic/EC2/_runtime.py +++ b/ventis/controller/cloud_provider_logic/EC2/_runtime.py @@ -239,11 +239,12 @@ def _bootstrap_instance(host, spec, replica_index, cfg, redis_host, redis_port, logger.info("Transferring image %s to %s", image, host) result = subprocess.run( "set -o pipefail; " - f"docker save {shlex.quote(image)} | ssh -o StrictHostKeyChecking=no " + f"docker save {shlex.quote(image)} | zstd -T0 | ssh -o StrictHostKeyChecking=no " f"-o IdentitiesOnly=yes -o ConnectTimeout=10 " f"-o ServerAliveInterval=10 -o ServerAliveCountMax=3 " f"-i {shlex.quote(key)} " - f"{shlex.quote(f'{ssh_user}@{host}')} 'sudo docker load'", + f"{shlex.quote(f'{ssh_user}@{host}')} " + "'set -o pipefail; zstd -d | sudo docker load'", shell=True, capture_output=True, text=True, diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index bdec683..16e0e9c 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -110,16 +110,16 @@ def __init__(self, config_path): len(self.controllers), ) - # Start background cleanup thread, woken by each poll tick rather than its own timer -- see OTel_Exporter/DESIGN.md. + # Start background cleanup thread, woken by each poll tick rather than its own timer -- see ventis/OTLP_Exporter/DESIGN.md. self._cleanup_ready = threading.Event() self._cleanup_thread = threading.Thread(target=self._cleanup_loop, daemon=True) self._cleanup_thread.start() - # Spawn the OTLP exporter as a separate process (see OTel_Exporter/DESIGN.md), + # Spawn the OTLP exporter as a separate process (see ventis/OTLP_Exporter/DESIGN.md), # supervised so it gets restarted if it ever exits unexpectedly. otel_exporter_dir = os.path.join( - os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), - "OTel_Exporter", + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "OTLP_Exporter", ) otel_exporter_script = os.path.join(otel_exporter_dir, "otel_exporter.py") self.process_supervisor = ProcessSupervisor() @@ -191,7 +191,12 @@ def _load_config(config_path): project_root = os.path.abspath(os.path.join(os.path.dirname(config_path), "..")) GlobalController._load_dotenv(os.path.join(project_root, ".env")) with open(config_path, "r") as f: - return yaml.safe_load(f) + config = yaml.safe_load(f) + if "ec2" in config: + config["ec2"] = GlobalController._expand_env_value(config["ec2"]) + if "database" in config: + config["database"] = GlobalController._expand_env_value(config["database"]) + return config @staticmethod def _load_dotenv(path): @@ -212,13 +217,13 @@ def _load_dotenv(path): os.environ[key] = value @staticmethod - def _expand_otel_value(value): + def _expand_env_value(value): if isinstance(value, str): return re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", lambda m: os.environ.get(m.group(1), m.group(0)), value) if isinstance(value, dict): - return {key: GlobalController._expand_otel_value(item) for key, item in value.items()} + return {key: GlobalController._expand_env_value(item) for key, item in value.items()} if isinstance(value, list): - return [GlobalController._expand_otel_value(item) for item in value] + return [GlobalController._expand_env_value(item) for item in value] return value @staticmethod @@ -242,7 +247,7 @@ def _otel_exporter_env(otel_cfg): ) if "destinations" in otel_cfg: - destinations = GlobalController._expand_otel_value(otel_cfg["destinations"]) + destinations = GlobalController._expand_env_value(otel_cfg["destinations"]) for destination in destinations: if destination.get("name") == "langfuse": public_key = os.environ.get("LANGFUSE_PUBLIC_KEY") @@ -639,7 +644,7 @@ def _poll_controllers(self): # Guarded on self.running: a SIGTERM can interrupt mid-tick and run stop() (which # terminates every managed process) via the signal handler before this line is # reached -- without the guard, this could respawn a process just intentionally - # killed. See OTel_Exporter/DESIGN.md. + # killed. See ventis/OTLP_Exporter/DESIGN.md. if self.running: self.process_supervisor.check_and_respawn() diff --git a/ventis/controller/utils/process_supervisor.py b/ventis/controller/utils/process_supervisor.py index 8f5724c..8c061bc 100644 --- a/ventis/controller/utils/process_supervisor.py +++ b/ventis/controller/utils/process_supervisor.py @@ -3,7 +3,7 @@ register() + start_all() spawn processes; check_and_respawn() (call from GC's existing poll tick) restarts any that exit unexpectedly; terminate_all() (call from GC's shutdown path) stops them all cleanly. Deliberately GC-agnostic -- callers are responsible for not -calling check_and_respawn() during their own shutdown (see OTel_Exporter/DESIGN.md's +calling check_and_respawn() during their own shutdown (see ventis/OTLP_Exporter/DESIGN.md's shutdown-race note). """ diff --git a/ventis/stub_generator.py b/ventis/stub_generator.py index 803fc2d..4647619 100644 --- a/ventis/stub_generator.py +++ b/ventis/stub_generator.py @@ -17,9 +17,7 @@ import yaml # Packages every agent container needs regardless of its specific business logic. -# grpcio-tools/pyyaml/ipdb/ipython aren't needed/used, but keeping to keep the scope constrained right now -# - Leave a comment if you want me to remove these, I kept them in since you originally had them but they aren't used -BASE_AGENT_REQUIREMENTS = ["grpcio", "grpcio-tools", "redis", "pyyaml", "psutil", "ipdb", "ipython", "boto3"] +BASE_AGENT_REQUIREMENTS = ["grpcio", "grpcio-tools", "redis", "pyyaml", "psutil", "boto3"] # Workflow will always require these BASE_WORKFLOW_REQUIREMENTS = BASE_AGENT_REQUIREMENTS + ["flask", "sqlalchemy", "psycopg[binary]"] @@ -322,6 +320,21 @@ def _copy_files(output_dir, files_to_copy): shutil.copy2(src, dest_path) +def _write_entrypoint_file(src, dest_path, project_dir): + """Copy an entrypoint file to dest_path, injecting a sys.path entry for its + original sibling directory so a co-located, non-stub helper import still resolves.""" + original_dir = os.path.dirname(os.path.relpath(src, project_dir)) if project_dir else "" + if not original_dir: + shutil.copy2(src, dest_path) + return + injection = ( + f"import sys, os\n" + f"sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), {original_dir!r}))\n" + ) + with open(src) as f, open(dest_path, "w") as out: + out.write(injection + f.read()) + + def generate_docker( yaml_path, agent_file, @@ -405,8 +418,7 @@ def generate_docker( _stub_destination(stub_file, stub_entrypoints or {}), ) ) - - files_to_copy.append((os.path.abspath(agent_file), os.path.basename(agent_file))) + files_to_copy.append((os.path.abspath(stub_file), os.path.basename(stub_file))) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -416,6 +428,14 @@ def generate_docker( _copy_files(output_dir, files_to_copy) + # Copy the agent's own real file last, so it wins over any same-named stub + # copy above; inject a sys.path entry so its own sibling helpers still resolve. + _write_entrypoint_file( + os.path.abspath(agent_file), + os.path.join(output_dir, os.path.basename(agent_file)), + project_dir, + ) + # Copy the YAML definition too shutil.copy2( os.path.abspath(yaml_path), @@ -499,7 +519,6 @@ def generate_workflow_docker( files_to_copy = _sweep_py_files(project_dir) if project_dir else [] files_to_copy += [ - (os.path.abspath(workflow_file), workflow_basename), (os.path.join(script_dir, "future.py"), "future.py"), (os.path.join(script_dir, "ventis_context.py"), "ventis_context.py"), (os.path.join(script_dir, "deploy.py"), "deploy.py"), @@ -527,6 +546,7 @@ def generate_workflow_docker( _stub_destination(stub_file, stub_entrypoints or {}), ) ) + files_to_copy.append((os.path.abspath(stub_file), os.path.basename(stub_file))) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -536,6 +556,14 @@ def generate_workflow_docker( _copy_files(output_dir, files_to_copy) + # Copy the workflow's own real file last, so it wins over any same-named stub + # copy above; inject a sys.path entry so its own sibling helpers still resolve. + _write_entrypoint_file( + os.path.abspath(workflow_file), + os.path.join(output_dir, workflow_basename), + project_dir, + ) + # ---- workflow_launcher.py -------------------------------------------- launcher = f"""import threading import time From 8baed6c324536bc88c0446b16376782c5b5cd9e4 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 31 Aug 2026 11:23:57 -0700 Subject: [PATCH 2/5] Parallelize per-instance polling in _poll_controllers Metrics/telemetry latency scaled with instance count x per-instance round-trip time since every instance was polled sequentially, one blocking the next, with the following tick only starting after the whole pass finished. Extracted the per-instance body into _poll_one_instance (whole body wrapped in one top-level try/except, since ThreadPoolExecutor.map() re-raises on first exception when results are consumed) and run all instances concurrently via the same ThreadPoolExecutor pattern _trigger_cleanup already used. Co-Authored-By: Claude Sonnet 5 --- ventis/controller/global_controller.py | 217 +++++++++++++------------ 1 file changed, 115 insertions(+), 102 deletions(-) diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index 16e0e9c..6584ef8 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -648,118 +648,131 @@ def _poll_controllers(self): if self.running: self.process_supervisor.check_and_respawn() - for instance in self.instance_manager.list_instances(): + # Polled in parallel, one instance's slow Redis/Postgres round-trip no longer + # gates every other instance's poll -- see ventis/OTLP_Exporter/DESIGN.md. + instances = self.instance_manager.list_instances() + if instances: + with ThreadPoolExecutor(max_workers=len(instances)) as executor: + list(executor.map(self._poll_one_instance, instances)) + + def _poll_one_instance(self, instance): + """Poll and persist one instance's runtime/metrics/health data; never raises.""" + try: name = instance["agent_name"] host = instance["host"] port = instance["host_port"] node_redis = self._get_node_redis_for(host) - try: - future_rows = pull_runtime_information(node_redis) - self._otel_db.write_waiting_rows( - future_rows, node_redis, self.config.get("project_id", 0) - ) - send_runtime_information( - future_rows, - node_redis, - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write runtime information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, + except Exception as e: + logger.warning("Failed to poll instance %s: %s", instance, e) + return + + try: + future_rows = pull_runtime_information(node_redis) + self._otel_db.write_waiting_rows( + future_rows, node_redis, self.config.get("project_id", 0) + ) + send_runtime_information( + future_rows, + node_redis, + self.config.get("database", {}).get("url"), + ) + except Exception as e: + logger.warning( + "Failed to write runtime information for instance %s (%s:%s) " + "(non-fatal): %s", + name, + host, + 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" + + # Getting metrics from local controllers + # See LocalController._execute_locally + try: + metrics = node_redis.hgetall(metrics_key) + if metrics: + now = time.time() + requests_served = int(float(metrics.get("requests_served") or 0)) + elapsed = now - self._last_metrics_poll_time.get( + (host, port), now - self.poll_interval ) - agent_host = self._agent_host_key(host) - status_key = f"controller:{agent_host}:{port}:status" - metrics_key = f"controller:{agent_host}:{port}:metrics" + throughput = requests_served / elapsed if elapsed > 0 else 0.0 + self._last_metrics_poll_time[(host, port)] = now - # Getting metrics from local controllers - # See LocalController._execute_locally - try: - metrics = node_redis.hgetall(metrics_key) - if metrics: - now = time.time() - requests_served = int(float(metrics.get("requests_served") or 0)) - elapsed = now - self._last_metrics_poll_time.get( - (host, port), now - self.poll_interval + try: + send_agent_information( + [ + { + **instance, + **metrics, + "requests_served": requests_served, + "throughput": throughput, + } + ], + self.config.get("database", {}).get("url"), ) - throughput = requests_served / elapsed if elapsed > 0 else 0.0 - self._last_metrics_poll_time[(host, port)] = now - - try: - send_agent_information( - [ - { - **instance, - **metrics, - "requests_served": requests_served, - "throughput": throughput, - } - ], - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write agent information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, - ) - else: - # Only clear the accumulated counters once they've actually been persisted - node_redis.hset_multiple( - metrics_key, - {"full_failures": 0, "error_count": 0, "requests_served": 0}, - ) - except Exception as e: - logger.warning( - "Failed to poll metrics for instance %s (%s:%s): %s", - name, - host, - port, - e, - ) + except Exception as e: + logger.warning( + "Failed to write agent information for instance %s (%s:%s) " + "(non-fatal): %s", + name, + host, + port, + e, + ) + else: + # Only clear the accumulated counters once they've actually been persisted + node_redis.hset_multiple( + metrics_key, + {"full_failures": 0, "error_count": 0, "requests_served": 0}, + ) + except Exception as e: + logger.warning( + "Failed to poll metrics for instance %s (%s:%s): %s", + name, + host, + port, + e, + ) - try: - status = node_redis.get(status_key) or "unknown" - prev = self._last_status.get((host, port)) + try: + status = node_redis.get(status_key) or "unknown" + prev = self._last_status.get((host, port)) - if status != prev: - if status == "healthy": - logger.info( - "Controller %s (%s:%s) is now healthy.", name, host, port - ) - self._on_controller_healthy(name, host, port) - else: - logger.warning( - "Controller %s (%s:%s) status changed: %s -> %s", - name, - host, - port, - prev or "(none)", - status, - ) - self._on_controller_unhealthy(name, host, port) - self._last_status[(host, port)] = status + if status != prev: + if status == "healthy": + logger.info( + "Controller %s (%s:%s) is now healthy.", name, host, port + ) + self._on_controller_healthy(name, host, port) else: - # No change — healthy stays quiet, unhealthy stays quiet too - if status == "healthy": - self._on_controller_healthy(name, host, port) - else: - self._on_controller_unhealthy(name, host, port) - except Exception as e: - logger.warning( - "Failed to poll status for instance %s (%s:%s): %s", - name, - host, - port, - e, - ) + logger.warning( + "Controller %s (%s:%s) status changed: %s -> %s", + name, + host, + port, + prev or "(none)", + status, + ) + self._on_controller_unhealthy(name, host, port) + self._last_status[(host, port)] = status + else: + # No change — healthy stays quiet, unhealthy stays quiet too + if status == "healthy": + self._on_controller_healthy(name, host, port) + else: + self._on_controller_unhealthy(name, host, port) + except Exception as e: + logger.warning( + "Failed to poll status for instance %s (%s:%s): %s", + name, + host, + port, + e, + ) # ------------------------------------------------------------------ # # Extensibility hooks — override in subclasses # From 1e6a6d8d3aad5e692b00a76c7cf9ea8db2d445d0 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 31 Aug 2026 14:25:29 -0700 Subject: [PATCH 3/5] Align with feature/otel-exporter: use updated langfuse config example and remove _write_entrypoint_file - Fixed langfuse example to use generic env-var headers pattern - Removed _write_entrypoint_file (directory structure preservation via _sweep_py_files is cleaner) --- .../portfolio/config/global_controller.yaml | 4 -- ventis/stub_generator.py | 40 +++---------------- 2 files changed, 6 insertions(+), 38 deletions(-) diff --git a/examples/portfolio/config/global_controller.yaml b/examples/portfolio/config/global_controller.yaml index edee85f..96f371c 100644 --- a/examples/portfolio/config/global_controller.yaml +++ b/examples/portfolio/config/global_controller.yaml @@ -91,10 +91,6 @@ otel: endpoint: ${GRAFANA_OTLP_ENDPOINT}/v1/traces headers: Authorization: Basic ${GRAFANA_OTLP_HEADERS} -# - name: langfuse -# protocol: http -# endpoint: ${LANGFUSE_BASE_URL}/api/public/otel/v1/traces -# headers: {} # Polling interval in seconds poll_interval: 5 diff --git a/ventis/stub_generator.py b/ventis/stub_generator.py index 9fe38a5..33824ff 100644 --- a/ventis/stub_generator.py +++ b/ventis/stub_generator.py @@ -17,7 +17,9 @@ import yaml # Packages every agent container needs regardless of its specific business logic. -BASE_AGENT_REQUIREMENTS = ["grpcio", "grpcio-tools", "redis", "pyyaml", "psutil", "boto3"] +# grpcio-tools/pyyaml/ipdb/ipython aren't needed/used, but keeping to keep the scope constrained right now +# - Leave a comment if you want me to remove these, I kept them in since you originally had them but they aren't used +BASE_AGENT_REQUIREMENTS = ["grpcio", "grpcio-tools", "redis", "pyyaml", "psutil", "ipdb", "ipython", "boto3"] # Workflow will always require these BASE_WORKFLOW_REQUIREMENTS = BASE_AGENT_REQUIREMENTS + ["flask", "sqlalchemy", "psycopg[binary]"] @@ -321,21 +323,6 @@ def _copy_files(output_dir, files_to_copy): shutil.copy2(src, dest_path) -def _write_entrypoint_file(src, dest_path, project_dir): - """Copy an entrypoint file to dest_path, injecting a sys.path entry for its - original sibling directory so a co-located, non-stub helper import still resolves.""" - original_dir = os.path.dirname(os.path.relpath(src, project_dir)) if project_dir else "" - if not original_dir: - shutil.copy2(src, dest_path) - return - injection = ( - f"import sys, os\n" - f"sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), {original_dir!r}))\n" - ) - with open(src) as f, open(dest_path, "w") as out: - out.write(injection + f.read()) - - def generate_docker( yaml_path, agent_file, @@ -419,7 +406,8 @@ def generate_docker( _stub_destination(stub_file, stub_entrypoints or {}), ) ) - files_to_copy.append((os.path.abspath(stub_file), os.path.basename(stub_file))) + + files_to_copy.append((os.path.abspath(agent_file), os.path.basename(agent_file))) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -429,14 +417,6 @@ def generate_docker( _copy_files(output_dir, files_to_copy) - # Copy the agent's own real file last, so it wins over any same-named stub - # copy above; inject a sys.path entry so its own sibling helpers still resolve. - _write_entrypoint_file( - os.path.abspath(agent_file), - os.path.join(output_dir, os.path.basename(agent_file)), - project_dir, - ) - # Copy the YAML definition too shutil.copy2( os.path.abspath(yaml_path), @@ -520,6 +500,7 @@ def generate_workflow_docker( files_to_copy = _sweep_py_files(project_dir) if project_dir else [] files_to_copy += [ + (os.path.abspath(workflow_file), workflow_basename), (os.path.join(script_dir, "future.py"), "future.py"), (os.path.join(script_dir, "ventis_context.py"), "ventis_context.py"), (os.path.join(script_dir, "deploy.py"), "deploy.py"), @@ -547,7 +528,6 @@ def generate_workflow_docker( _stub_destination(stub_file, stub_entrypoints or {}), ) ) - files_to_copy.append((os.path.abspath(stub_file), os.path.basename(stub_file))) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -557,14 +537,6 @@ def generate_workflow_docker( _copy_files(output_dir, files_to_copy) - # Copy the workflow's own real file last, so it wins over any same-named stub - # copy above; inject a sys.path entry so its own sibling helpers still resolve. - _write_entrypoint_file( - os.path.abspath(workflow_file), - os.path.join(output_dir, workflow_basename), - project_dir, - ) - # ---- workflow_launcher.py -------------------------------------------- launcher = f"""import threading import time From ede3facbdcc4ed3ac64ffb3b72d621609b9a74f7 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 31 Aug 2026 14:27:17 -0700 Subject: [PATCH 4/5] Restore parallelized polling implementation Re-applied the parallel instance polling that was lost during conflict resolution. The _poll_controllers method now uses ThreadPoolExecutor to poll all instances concurrently via _poll_one_instance, preventing one slow instance's Redis/Postgres round-trip from blocking the entire poll tick. --- ventis/controller/global_controller.py | 222 +++++++++++++------------ 1 file changed, 118 insertions(+), 104 deletions(-) diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index ba53881..b6b3948 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -530,124 +530,138 @@ def _poll_controllers(self): Check the health of each registered controller replica via its node's Redis. Also retrieves the request calls made in each instance. """ - # Prevents a process from restarting if a deliberate kill-cmd happens + # Guarded on self.running: a SIGTERM can interrupt mid-tick and run stop() (which + # terminates every managed process) via the signal handler before this line is + # reached -- without the guard, this could respawn a process just intentionally + # killed. See ventis/OTLP_Exporter/DESIGN.md. if self.running: self.process_supervisor.check_and_respawn() - for instance in self.instance_manager.list_instances(): + # Polled in parallel, one instance's slow Redis/Postgres round-trip no longer + # gates every other instance's poll -- see ventis/OTLP_Exporter/DESIGN.md. + instances = self.instance_manager.list_instances() + if instances: + with ThreadPoolExecutor(max_workers=len(instances)) as executor: + list(executor.map(self._poll_one_instance, instances)) + + def _poll_one_instance(self, instance): + """Poll and persist one instance's runtime/metrics/health data; never raises.""" + try: name = instance["agent_name"] host = instance["host"] port = instance["host_port"] node_redis = self._get_node_redis_for(host) - try: - future_rows = pull_runtime_information(node_redis) - self._otel_db.write_waiting_rows( - future_rows, node_redis, self.config.get("project_id", 0) - ) + except Exception as e: + logger.warning("Failed to poll instance %s: %s", instance, e) + return - # This is now legacy, keeping it for now, but will remove this later - send_runtime_information( - future_rows, - node_redis, - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write runtime information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, + try: + future_rows = pull_runtime_information(node_redis) + self._otel_db.write_waiting_rows( + future_rows, node_redis, self.config.get("project_id", 0) + ) + send_runtime_information( + future_rows, + node_redis, + self.config.get("database", {}).get("url"), + ) + except Exception as e: + logger.warning( + "Failed to write runtime information for instance %s (%s:%s) " + "(non-fatal): %s", + name, + host, + 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" + + # Getting metrics from local controllers + # See LocalController._execute_locally + try: + metrics = node_redis.hgetall(metrics_key) + if metrics: + now = time.time() + requests_served = int(float(metrics.get("requests_served") or 0)) + elapsed = now - self._last_metrics_poll_time.get( + (host, port), now - self.poll_interval ) - agent_host = self._agent_host_key(host) - status_key = f"controller:{agent_host}:{port}:status" - metrics_key = f"controller:{agent_host}:{port}:metrics" + throughput = requests_served / elapsed if elapsed > 0 else 0.0 + self._last_metrics_poll_time[(host, port)] = now - # Getting metrics from local controllers - # See LocalController._execute_locally - try: - metrics = node_redis.hgetall(metrics_key) - if metrics: - now = time.time() - requests_served = int(float(metrics.get("requests_served") or 0)) - elapsed = now - self._last_metrics_poll_time.get( - (host, port), now - self.poll_interval + try: + send_agent_information( + [ + { + **instance, + **metrics, + "requests_served": requests_served, + "throughput": throughput, + } + ], + self.config.get("database", {}).get("url"), ) - throughput = requests_served / elapsed if elapsed > 0 else 0.0 - self._last_metrics_poll_time[(host, port)] = now - - try: - send_agent_information( - [ - { - **instance, - **metrics, - "requests_served": requests_served, - "throughput": throughput, - } - ], - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write agent information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, - ) - else: - # Only clear the accumulated counters once they've actually been persisted - node_redis.hset_multiple( - metrics_key, - {"full_failures": 0, "error_count": 0, "requests_served": 0}, - ) - except Exception as e: - logger.warning( - "Failed to poll metrics for instance %s (%s:%s): %s", - name, - host, - port, - e, - ) + except Exception as e: + logger.warning( + "Failed to write agent information for instance %s (%s:%s) " + "(non-fatal): %s", + name, + host, + port, + e, + ) + else: + # Only clear the accumulated counters once they've actually been persisted + node_redis.hset_multiple( + metrics_key, + {"full_failures": 0, "error_count": 0, "requests_served": 0}, + ) + except Exception as e: + logger.warning( + "Failed to poll metrics for instance %s (%s:%s): %s", + name, + host, + port, + e, + ) - try: - status = node_redis.get(status_key) or "unknown" - prev = self._last_status.get((host, port)) + try: + status = node_redis.get(status_key) or "unknown" + prev = self._last_status.get((host, port)) - if status != prev: - if status == "healthy": - logger.info( - "Controller %s (%s:%s) is now healthy.", name, host, port - ) - self._on_controller_healthy(name, host, port) - else: - logger.warning( - "Controller %s (%s:%s) status changed: %s -> %s", - name, - host, - port, - prev or "(none)", - status, - ) - self._on_controller_unhealthy(name, host, port) - self._last_status[(host, port)] = status + if status != prev: + if status == "healthy": + logger.info( + "Controller %s (%s:%s) is now healthy.", name, host, port + ) + self._on_controller_healthy(name, host, port) else: - # No change — healthy stays quiet, unhealthy stays quiet too - if status == "healthy": - self._on_controller_healthy(name, host, port) - else: - self._on_controller_unhealthy(name, host, port) - except Exception as e: - logger.warning( - "Failed to poll status for instance %s (%s:%s): %s", - name, - host, - port, - e, - ) + logger.warning( + "Controller %s (%s:%s) status changed: %s -> %s", + name, + host, + port, + prev or "(none)", + status, + ) + self._on_controller_unhealthy(name, host, port) + self._last_status[(host, port)] = status + else: + # No change — healthy stays quiet, unhealthy stays quiet too + if status == "healthy": + self._on_controller_healthy(name, host, port) + else: + self._on_controller_unhealthy(name, host, port) + except Exception as e: + logger.warning( + "Failed to poll status for instance %s (%s:%s): %s", + name, + host, + port, + e, + ) # ------------------------------------------------------------------ # # Extensibility hooks — override in subclasses # From ea91ff9f776014bd85ef4ef0417f01d07f3d93e6 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Mon, 31 Aug 2026 14:32:42 -0700 Subject: [PATCH 5/5] added concurrent polling --- ventis/controller/global_controller.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index b6b3948..6bdbd1a 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -530,10 +530,7 @@ def _poll_controllers(self): Check the health of each registered controller replica via its node's Redis. Also retrieves the request calls made in each instance. """ - # Guarded on self.running: a SIGTERM can interrupt mid-tick and run stop() (which - # terminates every managed process) via the signal handler before this line is - # reached -- without the guard, this could respawn a process just intentionally - # killed. See ventis/OTLP_Exporter/DESIGN.md. + # Prevents a process from restarting if a deliberate kill-cmd happens if self.running: self.process_supervisor.check_and_respawn() @@ -560,6 +557,7 @@ def _poll_one_instance(self, instance): self._otel_db.write_waiting_rows( future_rows, node_redis, self.config.get("project_id", 0) ) + # This is now legacy, keeping it for now, but will remove this later send_runtime_information( future_rows, node_redis,