diff --git a/canyonos_core/OTLP_Exporter/DESIGN.md b/canyonos_core/OTLP_Exporter/DESIGN.md index 15f0163..b33e457 100644 --- a/canyonos_core/OTLP_Exporter/DESIGN.md +++ b/canyonos_core/OTLP_Exporter/DESIGN.md @@ -2,10 +2,10 @@ Status: **implemented (single-table design; multi-destination fan-out in progress)**. `GlobalController` writes futures into a `waiting` table (SQLite); a GC-supervised, GC-restarted OTel Exporter process reads -finished/unsent rows, converts each to an OTel span, and hands it to a real -`BatchSpanProcessor`/`OTLPSpanExporter`. Batching, serialization, and sending are all -OTel SDK code — the only custom pieces are the row→span conversion and durable -sent-tracking. This doc is a design/rationale reference; the actual files +finished/unsent rows, converts them to OTel spans, and exports them synchronously via +`OTLPSpanExporter.export()`, marking rows sent only on a `SpanExportResult.SUCCESS` from +every destination. Serialization, transport, and transient-failure retry are all OTel SDK +code — the only custom pieces are the row→span conversion and durable sent-tracking. This doc is a design/rationale reference; the actual files (`otel_exporter.py`, `db.py`, `convert.py`, `canyonos/controller/utils/process_supervisor.py`) are the source of truth for current behavior. @@ -13,7 +13,7 @@ are the source of truth for current behavior. CanyonOS futures need to reach an external OTLP-compatible tracing backend. Design: a separate OTLP Exporter process, spawned and supervised by GlobalController, that reads unsent finished future rows from a local SQLite DB, converts them into OTel spans, and -hands them to the OTel SDK's own batching/export machinery, which ships them to an +exports them through the OTel SDK's own OTLP exporters, which ship them to an external OTLP Receiver (out of scope here — assumed to be a separate, already-addressable service). @@ -31,7 +31,7 @@ Decisions (final status): config-aware. `GlobalController` serializes that list to JSON and passes it to the exporter subprocess as a single `CANYONOS_OTEL_DESTINATIONS` env var via `ProcessSupervisor.register(..., env=...)`. The exporter builds one independent - exporter/`BatchSpanProcessor` pair per destination, picking the gRPC vs HTTP + OTLP exporter per destination, picking the gRPC vs HTTP exporter class from each destination's `protocol` field. gRPC and HTTP destinations may be mixed in the same list. Deliberately vendor-neutral: no backend name (Postgres, Langfuse, or otherwise) appears anywhere in `otel_exporter.py`; the @@ -41,19 +41,23 @@ Decisions (final status): own idiomatic mechanism, so no exporter-side config plumbing was added, only a GC-side YAML→env-var translation. If `otel.destinations` is absent, GlobalController logs that no OTel metrics collection will happen and skips starting the exporter - subprocess entirely. Configuration is read at exporter startup; changing it requires - a GlobalController/exporter restart. + subprocess entirely. Configuration now reaches the exporter through the + `otel:destinations` Redis key rather than the `CANYONOS_OTEL_DESTINATIONS` env var + described above (kept for history): GlobalController writes that key, and every poll + tick re-reads it and rebuilds the exporters if it changed, so a config reload (SIGHUP) + reaches this process without a restart. An invalid value is logged and ignored, + leaving the previous working exporters in place. - **Data source**: NOT `runtime_information` — a dedicated `waiting` table in its own SQLite file (`canyonos/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. - **Two tables collapsed into one**: an earlier version of this design had a second - `queue` table (`waiting` → promote → `queue` → drain → send). Collapsed once it became - clear `BatchSpanProcessor` already provides its own in-memory queue — the only thing a - second table added was durability across the exporter's own process restarts, which a - `sent` column on `waiting` alone provides just as well, with less code. See `db.py`'s - module docstring. + `queue` table (`waiting` → promote → `queue` → drain → send). Collapsed to a single + `sent` column on `waiting`, which provides the same durability with less code. (The + original rationale cited `BatchSpanProcessor`'s in-memory queue; that processor has + since been removed — see "Synchronous export" below — and `waiting` is now the only + queue in the pipeline, which is what makes the durability property hold at all.) - **Span construction**: settled — spans are built as `ReadableSpan` objects directly (bypassing `Tracer`/`TracerProvider` entirely, no `IdGenerator` workaround needed for either `trace_id` or `span_id`). Confirmed working via `ConsoleSpanExporter` during @@ -89,26 +93,78 @@ and raises if invoked directly without `CANYONOS_OTEL_DESTINATIONS` set). `otel_exporter.py` parses the destination configuration at startup and constructs the appropriate OTLP exporter for each entry (gRPC or HTTP), passing that destination's -endpoint and headers to the SDK. `BatchSpanProcessor(..., schedule_delay_millis=1000)` -— the flush delay is explicitly overridden from the SDK default (5000ms) to 1000ms; -`max_export_batch_size` is left at the SDK default (512), which already approximates the -original "500 spans" batching ask without any override needed. +endpoint, headers, and timeout to the SDK. Each destination's `timeout` bounds how long a +single failing export blocks the poll loop, so it is worth setting explicitly rather than +leaning on the SDK default of 10s. -### 2. `canyonos/OTLP_Exporter/otel_exporter.py` +### 2. `canyonos/OTLP_Exporter/otel_exporter.py` — synchronous export 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; -each pair may use a different protocol, endpoint, and headers: -- `SELECT * FROM waiting WHERE finished_at IS NOT NULL AND (sent IS NULL OR sent = 0)`. -- Per row, each isolated in its own try/except (one malformed row is logged and skipped, - never blocks the rest of the batch): `convert.waiting_row_to_span(row)` → - `on_end(span)` on every configured processor → `db.mark_sent(future_id)` immediately. - The row is marked after it has been queued to all processors. `sent` therefore means - **queued to every configured destination**, not remotely acknowledged; this is the - initial best-effort delivery contract and retains the existing single boolean schema. -- Each processor is constructed once at startup; no `TracerProvider` is used at all, - 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. +independent OTLP exporter per configured destination; each may use a different protocol, +endpoint, headers, and timeout: +- `SELECT * FROM waiting WHERE finished_at IS NOT NULL AND (sent IS NULL OR sent = 0)`, + bounded by `MAX_SPANS_PER_POLL` so one tick's export request stays a sane size; a + backlog is drained over successive polls. +- Conversion is per row, isolated in its own try/except, and each span is then + test-encoded individually (`_reject_unexportable`). Encoding is what actually rejects + a bad row — an out-of-range id, an unencodable attribute — and it would otherwise + happen inside the batched `export()` call, where one row's failure discards every + other span in the batch and the error names only the destination. Rejecting per row + keeps one bad future from blocking everything behind it and names the offending + `future_id`. +- After `MAX_ROW_EXPORT_ATTEMPTS` (5) such rejections, that row is replaced by a + placeholder span (`convert.invalid_row_placeholder_span`) which exports normally and + marks the row sent, so it leaves the pending set instead of being retried forever and + occupying part of the `LIMIT` window. The placeholder keeps the row's real `trace_id` + where the `session_id` allows it, and masks out-of-range ids into valid ones. It is + named `canyonos.invalid_span`, carries `canyonos.export.invalid` and the real + `future_id` as attributes, and deliberately has no exception event: it records that + telemetry could not be represented, **not** that the agent failed, and a dashboard must + be able to tell those apart. The counter is in-memory only — the placeholder is what + makes the outcome durable, so no schema change and no attempt column are needed. + Only per-row rejections count toward it; an export failure is shared by the whole + batch, and counting those would replace the entire queue with placeholders after a + spell of receiver downtime. +- The resulting spans are exported as one batch per destination via + `exporter.export(spans)`, which is **synchronous** and returns a `SpanExportResult`. + Rows are marked sent only when every destination returned SUCCESS *and* reported no + partial rejection. +- Retry needs no machinery: a failed batch simply leaves those rows at `sent = 0`, and + the next poll picks them up. `waiting` is the durable queue. The SDK already retries + transient failures internally with exponential backoff, bounded by the destination's + `timeout`, so a returned FAILURE means it genuinely gave up. +- Marking is all-or-nothing across destinations, so one destination failing re-delivers + to destinations that already accepted the batch. Span ids are deterministic, so those + duplicates collapse at the backend. +- **`partial_success` is checked, for HTTP destinations only.** A receiver may return + 200/OK while rejecting individual spans; the SDK exporters discard the response body + and report `SUCCESS` regardless, so `sent` would otherwise mean "the receiver accepted + the request", not "every span was stored". HTTP exporters are therefore built with a + `requests.Session` carrying a response hook (`_PartialSuccessRecorder`) that reads + `partial_success.rejected_spans`, and a non-zero count fails the batch. gRPC exposes no + equivalent public seam, so gRPC destinations cannot detect partial rejection — the + exporter logs a warning once when one is configured. +- `protocol` is validated against `SUPPORTED_PROTOCOLS` (`grpc`, `http`, + `http/protobuf`). It was previously read but never checked, so any other value — + including a typo or an absent field — silently selected the HTTP exporter. +- A queue that never yields a row is reported. `sqlite3.connect()` creates a missing + file instead of refusing, so a misdirected `DB_PATH` reads as a permanently idle + queue rather than an error. After `EMPTY_QUEUE_WARNING_POLLS` consecutive empty + polls the exporter checks the table's total row count and, if it is still zero, + warns once naming the path. A queue whose rows are all already exported is a + normal idle state and stays silent. +- No `TracerProvider` and no `BatchSpanProcessor` are used at all; spans are hand-built + and handed straight to the exporters. +- Every exporter is shut down on exit. Nothing is buffered in an OTLP exporter (its own + `force_flush()` is a documented no-op), so there is nothing to lose on shutdown — + anything unacknowledged is still `sent = 0` and resumes on the next start. + +**Why not `BatchSpanProcessor`** (the original design, removed): `on_end()` only enqueues +onto an in-memory queue and returns `None`, so the real send happened later on an SDK +background thread with no way to report back. Rows were marked sent immediately and a +failed export was lost silently and permanently, with the HTTP error stranded in the GC +container log. The in-memory queue also silently dropped spans when full, and lost its +entire contents on SIGKILL — while those rows already read `sent = 1`. ### 3. Future row → OTel span conversion (`canyonos/OTLP_Exporter/convert.py`) `future_id` maps to OTel `span_id`, not `trace_id` — `session_id` (== `request_id`) is @@ -174,23 +230,26 @@ config work, since `protocol: http` now needs that package importable). ## Known gaps (not yet built) - `WriteResult()` passes an undefined `error_message` variable to its fan-out callback, which can interrupt remote consumer propagation after the callback hash is persisted. -- Rows are marked `sent` immediately after `BatchSpanProcessor.on_end()` accepts them, +- ~~Rows are marked `sent` immediately after `BatchSpanProcessor.on_end()` accepts them, before the asynchronous OTLP export is confirmed; a later delivery failure can lose a - span while leaving `sent = 1`. + span while leaving `sent = 1`.~~ **Fixed** — export is now synchronous and `sent` is + written only on a SUCCESS from every destination. - Spans carry no explicit `resource`/`instrumentation_scope` — would show as `service.name=unknown_service` at a real backend. -- Destination-specific delivery acknowledgement/retry state is not tracked yet: - `sent` only records that the span was queued to all configured processors, so an - asynchronous export failure can still lose a span until a later delivery-state design - is added. +- Per-destination delivery state is still not tracked: `sent` is one boolean across all + destinations, so if one of several destinations fails the whole batch is retried + everywhere and the healthy destinations receive duplicates. Harmless for tracing + backends (ids are deterministic), but a per-destination table would avoid it. - `waiting` grows unboundedly: sent rows are never pruned, and futures that never finish (`finished_at` never arrives) also stay forever, invisible and un-expiring. - `error_name` is always `NULL` — CanyonOS's own Redis writer never records a distinct exception-type field, only a message string. - Test coverage is still limited; the waiting-field migration/normalization/conversion path is covered, but the exporter process and live OTLP delivery are not. -- Never verified against a live OTLP receiver — only against a refused connection - (confirmed the SDK's real retry/error-handling path is exercised correctly). +- Live-receiver coverage is now basic but real: the synchronous-export change was + verified against a local HTTP receiver (span accepted, real OTLP protobuf received, + row marked sent) as well as a refused connection (row left unsent and retried). Still + never verified against a production-grade OTLP backend. - No retry-limit/quarantine for a permanently malformed row — it logs an error every poll forever rather than being given up on. diff --git a/canyonos_core/OTLP_Exporter/convert.py b/canyonos_core/OTLP_Exporter/convert.py index fc749c8..c0c97c2 100644 --- a/canyonos_core/OTLP_Exporter/convert.py +++ b/canyonos_core/OTLP_Exporter/convert.py @@ -50,13 +50,15 @@ def waiting_row_to_span(row): events = [] status = Status(StatusCode.UNSET) if row["failed"]: + # An absent error_name means the producer never recorded one; naming a + # type here would invent an exception class the code never raised. + exception_attributes = {EXCEPTION_MESSAGE: row.get("error_message") or ""} + if row.get("error_name"): + exception_attributes[EXCEPTION_TYPE] = row["error_name"] events.append( Event( name="exception", - attributes={ - EXCEPTION_TYPE: row.get("error_name") or "RuntimeError", - EXCEPTION_MESSAGE: row.get("error_message") or "", - }, + attributes=exception_attributes, timestamp=to_epoch_nanos(row.get("finished_at")), ) ) @@ -103,3 +105,35 @@ def waiting_row_to_span(row): start_time=to_epoch_nanos(row.get("started_at")), end_time=to_epoch_nanos(row.get("finished_at")), ) + + +def invalid_row_placeholder_span(row, reason): + """Stand-in span for a future whose own telemetry cannot be encoded. + + Deliberately not named after the agent and not carrying an exception event: + this records that telemetry could not be represented, not that the agent + failed, and the two must stay distinguishable in a dashboard. + """ + row = dict(row) + context = SpanContext( + trace_id=(int(row["session_id"], 16) & (2**128 - 1)) or 1, + span_id=(int(row["future_id"], 16) & (2**64 - 1)) or 1, + is_remote=False, + trace_flags=_SAMPLED, + ) + return ReadableSpan( + name="canyonos.invalid_span", + context=context, + parent=None, + attributes={ + "canyonos.export.invalid": True, + "canyonos.export.error": reason, + "canyonos.future_id": row["future_id"], + }, + status=Status( + StatusCode.ERROR, description=f"span could not be exported: {reason}" + ), + kind=SpanKind.INTERNAL, + start_time=to_epoch_nanos(row.get("started_at")), + end_time=to_epoch_nanos(row.get("finished_at")), + ) diff --git a/canyonos_core/OTLP_Exporter/db.py b/canyonos_core/OTLP_Exporter/db.py index 16c02ae..c612190 100644 --- a/canyonos_core/OTLP_Exporter/db.py +++ b/canyonos_core/OTLP_Exporter/db.py @@ -1,14 +1,7 @@ -"""SQLite schema and writes for the OTel export pipeline's waiting table. - -`waiting` holds future rows as GlobalController observes them (including still-running -ones). There's no separate queue table -- OTel's own BatchSpanProcessor already queues -and batches spans in memory, so the only thing we need to track durably is which rows -have already been sent, which the `sent` column on this same table provides. (An earlier -version of this pipeline had a second `queue` table for that; collapsed away since it -wasn't doing anything BatchSpanProcessor doesn't already do -- see DESIGN.md.) -""" +"""SQLite schema and writes for the OTel export pipeline's waiting table.""" import json +import logging import os import sqlite3 @@ -17,8 +10,28 @@ # It is currently stored here for backcompat with the old telemetry collecting +logger = logging.getLogger(__name__) + DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "otel_queue.db") +# Cost lookups can fail on every row of every poll, so each kind is reported once. +_cost_failures_logged = set() + + +def _log_cost_failure(kind, exc): + """Report the first failure of each cost-lookup kind; suppress the rest.""" + if kind in _cost_failures_logged: + return + _cost_failures_logged.add(kind) + logger.warning( + "%s lookup failed; affected rows are recorded with a cost of 0. Further " + "%s failures are suppressed for the life of this process: %s", + kind, + kind, + exc, + exc_info=True, + ) + # Demo-only multipliers for scaling displayed costs, DELETE FOR MORE ACCURATE METRICS _TOKEN_COST_MULTIPLIER = 10000 _SERVER_COST_MULTIPLIER = 100000 @@ -88,7 +101,7 @@ def init_db(db_path=DB_PATH): def _normalize_json_text(value): """Return JSON text, encoding legacy scalar strings that are not valid JSON.""" - if value is None or value == "": + if value is None: return None try: json.loads(value) @@ -111,6 +124,18 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH fid = raw.get("future_id") session_id = raw.get("request_id") if not fid or not session_id: + missing = ", ".join( + field + for field, present in (("future_id", fid), ("request_id", session_id)) + if not present + ) + logger.warning( + "Dropping future row missing %s; it will never be exported " + "(future_id=%r, request_id=%r)", + missing, + fid, + session_id, + ) continue agent_id = raw.get("agent") started_at = float(raw.get("created_at") or 0) @@ -144,7 +169,8 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH ) * _TOKEN_COST_MULTIPLIER ) - except Exception: + except Exception as e: + _log_cost_failure("Token cost", e) token_cost = 0.0 try: server_cost = ( @@ -156,7 +182,8 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH ) * _SERVER_COST_MULTIPLIER ) - except Exception: + except Exception as e: + _log_cost_failure("Server cost", e) server_cost = 0.0 else: token_cost = 0.0 @@ -211,3 +238,30 @@ def mark_sent(future_id, db_path=DB_PATH): conn.commit() finally: conn.close() + + +def mark_sent_many(future_ids, db_path=DB_PATH): + """Mark every listed waiting row sent in one transaction.""" + if not future_ids: + return + try: + conn = sqlite3.connect(db_path) + except Exception as e: + # Re-raised, not swallowed: the caller reports these rows as delivered + # but unmarked, which is what tells an operator to expect duplicates. + logger.error( + "Failed to open %s to mark %d row(s) sent: %s", + db_path, + len(future_ids), + e, + exc_info=True, + ) + raise + try: + conn.executemany( + "UPDATE waiting SET sent = 1 WHERE future_id = ?", + [(future_id,) for future_id in future_ids], + ) + conn.commit() + finally: + conn.close() diff --git a/canyonos_core/OTLP_Exporter/otel_exporter.py b/canyonos_core/OTLP_Exporter/otel_exporter.py index 2d1cab1..2903cbc 100644 --- a/canyonos_core/OTLP_Exporter/otel_exporter.py +++ b/canyonos_core/OTLP_Exporter/otel_exporter.py @@ -1,14 +1,4 @@ -"""Entrypoint for the OTLP exporter process. - -Each poll tick reads finished, not-yet-sent rows from ``waiting``, converts each to a -span, hands it to every configured BatchSpanProcessor, and marks it sent only after -all processors accept it. Batching, OTLP serialization, and sending remain the SDK's -responsibility (see DESIGN.md). - -Destinations come from the ``otel:destinations`` Redis key (GlobalController writes it), -not env -- every poll tick re-reads it and rebuilds processors if it changed, so a config -reload (SIGHUP) reaches this process without a restart. -""" +"""Entrypoint for the OTLP exporter process.""" import json import logging @@ -28,7 +18,12 @@ from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter as HttpOTLPSpanExporter, ) -from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceResponse, +) +from opentelemetry.sdk.trace.export import SpanExportResult +import requests import convert import db @@ -37,9 +32,25 @@ logger = logging.getLogger(__name__) _running = True -_processors = [] +_exporters = [] _last_destinations_raw = None POLL_INTERVAL_SECONDS = 5 +# Bounds one export request, since a backlog is drained by repeated polls. +MAX_SPANS_PER_POLL = 512 +MAX_ROW_EXPORT_ATTEMPTS = 5 +# Counted in memory only: a placeholder marks the row sent, so it leaves the +# pending set for good and the count never needs to survive a restart. +_row_export_failures = {} +EMPTY_QUEUE_WARNING_POLLS = 12 +EMPTY_QUEUE_REWARN_POLLS = 720 +SUPPORTED_PROTOCOLS = ("grpc", "http", "http/protobuf") +_consecutive_empty_polls = 0 +_empty_queue_warned_at = None +# Keyed by destination name; only HTTP destinations can report partial success. +_partial_success_recorders = {} +_grpc_partial_success_warned = False +# Destination name -> whether its last export succeeded, so recovery is reported. +_destination_healthy = {} DESTINATIONS_KEY = "otel:destinations" # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY _redis = None @@ -52,7 +63,13 @@ def _validate_destination(destination, index): if not isinstance(name, str) or not name.strip(): raise ValueError(f"destination {index} name must be a non-empty string") - protocol = destination.get("protocol") # must be exactly "grpc" or "http" + protocol = destination.get("protocol") + protocol = protocol.lower() if isinstance(protocol, str) else protocol + if protocol not in SUPPORTED_PROTOCOLS: + raise ValueError( + f"destination {name!r} protocol must be one of " + f"{list(SUPPORTED_PROTOCOLS)}; got {protocol!r}" + ) endpoint = destination.get("endpoint") if not isinstance(endpoint, str) or not endpoint.strip(): raise ValueError(f"destination {name!r} endpoint must be a non-empty string") @@ -116,8 +133,41 @@ def _configured_destinations(raw): return validated +class _PartialSuccessRecorder: + """Records rejected_spans from OTLP responses, which the SDK exporters discard.""" + + def __init__(self, destination_name): + self.destination_name = destination_name + self.rejected_spans = 0 + self.error_message = "" + self._unparseable_logged = False + + def reset(self): + self.rejected_spans = 0 + self.error_message = "" + + def __call__(self, response, *args, **kwargs): + if not response.ok or not response.content: + return + try: + parsed = ExportTraceServiceResponse.FromString(response.content) + except Exception as e: + if not self._unparseable_logged: + self._unparseable_logged = True + logger.warning( + "Destination %s returned a body that is not an " + "ExportTraceServiceResponse, so partial rejections cannot be " + "detected there: %s", + self.destination_name, + e, + ) + return + self.rejected_spans = parsed.partial_success.rejected_spans + self.error_message = parsed.partial_success.error_message + + def _build_exporter(destination): - """Construct one OTLP exporter.""" + """Construct one OTLP exporter, and its partial-success recorder when supported.""" kwargs = { "endpoint": destination["endpoint"], } @@ -126,7 +176,15 @@ def _build_exporter(destination): if destination["protocol"] == "grpc": if destination["insecure"] is not None: kwargs["insecure"] = destination["insecure"] # fmt: skip - return GrpcOTLPSpanExporter(**kwargs) + global _grpc_partial_success_warned + if not _grpc_partial_success_warned: + _grpc_partial_success_warned = True + logger.warning( + "gRPC destinations cannot report partial rejections: the SDK " + "discards the response body, so spans this receiver rejects " + "individually will still be marked sent." + ) + return GrpcOTLPSpanExporter(**kwargs), None if destination["insecure"] is not None: logger.warning( @@ -134,35 +192,82 @@ def _build_exporter(destination): destination["name"], destination["insecure"], ) - return HttpOTLPSpanExporter(**kwargs) + recorder = _PartialSuccessRecorder(destination["name"]) + session = requests.Session() + session.hooks["response"].append(recorder) + kwargs["session"] = session + return HttpOTLPSpanExporter(**kwargs), recorder + + +def _probe_destination(destination_name, exporter): + """Report whether a destination actually answers, without failing startup. + + An empty export is a real OTLP request, so this exercises the endpoint, + path, TLS and auth rather than just proving a port is open. A destination + that is merely down yet is not fatal -- rows stay queued until it returns. + """ + try: + reachable = exporter.export([]) is SpanExportResult.SUCCESS + detail = "" + except Exception as e: + reachable = False + detail = f": {e}" + if reachable: + logger.info("OTel destination %s answered a connectivity check.", destination_name) + else: + logger.warning( + "OTel destination %s did not answer a connectivity check, so nothing " + "will reach it until that is fixed; spans stay queued meanwhile%s", + destination_name, + detail, + ) -def _build_processors(raw): - """Build one exporter/BatchSpanProcessor pair per configured destination.""" +def _build_exporters(raw): + """Build one OTLP exporter per configured destination.""" destinations = _configured_destinations(raw) if destinations is None: raise RuntimeError(f"{DESTINATIONS_KEY} is not set; otel.destinations is required") - processors = [] + exporters = [] + recorders = {} try: for destination in destinations: - exporter = _build_exporter(destination) - processors.append( - ( - destination["name"], - BatchSpanProcessor(exporter, schedule_delay_millis=1000), - ) - ) + exporter, recorder = _build_exporter(destination) + exporters.append((destination["name"], exporter)) + if recorder is not None: + recorders[destination["name"]] = recorder logger.info( "Configured OTel destination %s (%s).", destination["name"], destination["protocol"], ) + _probe_destination(destination["name"], exporter) except Exception: - for _, processor in processors: - processor.shutdown() + _shutdown_exporters( + exporters, + f"discarding destinations already built before {destination['name']!r} " + f"failed to build", + ) raise - return processors + _partial_success_recorders.clear() + _partial_success_recorders.update(recorders) + return exporters + + +def _shutdown_exporters(exporters, reason): + """Shut down each exporter, logging rather than propagating individual failures.""" + for destination_name, exporter in exporters: + try: + exporter.shutdown() + except Exception as e: + logger.error( + "Failed to shut down OTel destination %s while %s: %s", + destination_name, + reason, + e, + exc_info=True, + ) def _handle_shutdown(signum, frame): @@ -171,85 +276,292 @@ def _handle_shutdown(signum, frame): def _reload_destinations_if_changed(): - # Invalid Redis values are logged and ignored -- keep the previous processors + # Invalid Redis values are logged and ignored -- keep the previous exporters # running rather than tearing down a working config over a bad update. - global _processors, _last_destinations_raw - raw = _redis.get(DESTINATIONS_KEY) + global _exporters, _last_destinations_raw + try: + raw = _redis.get(DESTINATIONS_KEY) + except Exception as e: + logger.error( + "Failed to read %s from Redis; keeping the current %d destination(s): %s", + DESTINATIONS_KEY, + len(_exporters), + e, + exc_info=True, + ) + return if raw == _last_destinations_raw: return try: - new_processors = _build_processors(raw) + new_exporters = _build_exporters(raw) except Exception as e: logger.warning("Ignoring invalid %s update: %s", DESTINATIONS_KEY, e) return - for _, processor in _processors: - processor.shutdown() - _processors = new_processors + _shutdown_exporters(_exporters, "replacing it after a config reload") + _exporters = new_exporters _last_destinations_raw = raw - logger.info("Reloaded %d OTel destination(s) from Redis.", len(_processors)) - + logger.info("Reloaded %d OTel destination(s) from Redis.", len(_exporters)) -def _send_pending(): - """Convert and send each finished, not-yet-sent waiting row.""" - processors = _processors - if not processors: - raise RuntimeError("OTel exporter has no configured processors") - conn = sqlite3.connect(db.DB_PATH) +def _read_pending_rows(): + """Read one poll's worth of finished, not-yet-sent waiting rows.""" + try: + conn = sqlite3.connect(db.DB_PATH) + except Exception as e: + logger.error( + "Failed to open the waiting database at %s; no spans exported this poll: %s", + db.DB_PATH, + e, + exc_info=True, + ) + return [] conn.row_factory = sqlite3.Row try: - rows = conn.execute( + return conn.execute( "SELECT * FROM waiting WHERE finished_at IS NOT NULL " - "AND (sent IS NULL OR sent = 0)" + "AND (sent IS NULL OR sent = 0) LIMIT ?", + (MAX_SPANS_PER_POLL,), ).fetchall() + except Exception as e: + logger.error( + "Failed to read pending rows from %s; no spans exported this poll: %s", + db.DB_PATH, + e, + exc_info=True, + ) + return [] finally: conn.close() + + +def _reject_unexportable(span): + """Raise if the OTLP encoder cannot serialize this span. + + Encoding is what actually rejects a bad row (an out-of-range id, an + unencodable attribute), and it happens inside the batched export() call -- + where one row's failure discards every other span in the batch. Doing it per + row here keeps a single bad future from blocking everything behind it. + """ + encode_spans([span]).SerializePartialToString() + + +def _placeholder_after_repeated_failure(row, error): + """Return a placeholder span once a row has failed too often, else None. + + Only per-row conversion/encoding failures count here. An export failure is + shared by the whole batch, so counting those would replace every row in the + queue with a placeholder after a spell of receiver downtime. + """ + future_id = row["future_id"] + attempts = _row_export_failures.get(future_id, 0) + 1 + _row_export_failures[future_id] = attempts + if attempts < MAX_ROW_EXPORT_ATTEMPTS: + logger.error( + "Skipping waiting row %s -- cannot be exported (attempt %d of %d): %s", + future_id, + attempts, + MAX_ROW_EXPORT_ATTEMPTS, + error, + ) + return None + try: + placeholder = convert.invalid_row_placeholder_span(row, str(error)) + _reject_unexportable(placeholder) + except Exception as e: + logger.error( + "Waiting row %s cannot be exported and no placeholder could be built " + "for it either, so it stays in the queue: %s", + future_id, + e, + exc_info=True, + ) + return None + logger.warning( + "Waiting row %s failed %d export attempts; sending a placeholder span in " + "its place so the future is not lost silently: %s", + future_id, + attempts, + error, + ) + _row_export_failures.pop(future_id, None) + return placeholder + + +def _waiting_row_count(): + """Total rows in waiting, or None when the table cannot be counted.""" + try: + conn = sqlite3.connect(db.DB_PATH) + try: + return conn.execute("SELECT COUNT(*) FROM waiting").fetchone()[0] + finally: + conn.close() + except Exception as e: + logger.error( + "Failed to count rows in %s: %s", db.DB_PATH, e, exc_info=True + ) + return None + + +def _note_queue_state(found_pending): + """Warn once if no row ever appears, which sqlite cannot report as an error. + + A missing database file is created rather than refused, so a misdirected + DB_PATH looks exactly like an idle queue until someone compares the two paths. + """ + global _consecutive_empty_polls, _empty_queue_warned_at + if found_pending: + _consecutive_empty_polls = 0 + return + _consecutive_empty_polls += 1 + if _consecutive_empty_polls < EMPTY_QUEUE_WARNING_POLLS: + return + if ( + _empty_queue_warned_at is not None + and _consecutive_empty_polls - _empty_queue_warned_at < EMPTY_QUEUE_REWARN_POLLS + ): + return + # Only latch once the condition is confirmed, so a failed count re-checks + # next poll instead of silencing the warning for the life of the process. + if _waiting_row_count() != 0: + return + _empty_queue_warned_at = _consecutive_empty_polls + logger.warning( + "No rows have ever appeared in %s after %d consecutive polls. Spans are " + "only exported from this file, so GlobalController may be writing futures " + "to a different otel_queue.db than this process is reading.", + db.DB_PATH, + _consecutive_empty_polls, + ) + + +def _send_pending(): + """Export finished, not-yet-sent waiting rows and mark them only once delivered.""" + exporters = _exporters + if not exporters: + raise RuntimeError("OTel exporter has no configured destinations") + + rows = _read_pending_rows() + _note_queue_state(bool(rows)) if not rows: return - sent_count = 0 + + spans = [] + future_ids = [] for row in rows: try: span = convert.waiting_row_to_span(row) + _reject_unexportable(span) + except Exception as e: + span = _placeholder_after_repeated_failure(row, e) + if span is None: + continue + else: + _row_export_failures.pop(row["future_id"], None) + spans.append(span) + future_ids.append(row["future_id"]) + if not spans: + return + + failed_destinations = [] + for destination_name, exporter in exporters: + recorder = _partial_success_recorders.get(destination_name) + if recorder is not None: + recorder.reset() + delivered = False + try: + result = exporter.export(spans) except Exception as e: logger.error( - "Skipping waiting row %s -- failed to convert: %s", row["future_id"], e + "Destination %s raised while exporting %d span(s): %s", + destination_name, + len(spans), + e, ) - continue - - failed_destinations = [] - for destination_name, processor in processors: - try: - processor.on_end(span) - except Exception as e: - # Still offer the span to the remaining processors. The row is only - # acknowledged when every destination accepted it, so a failed - # destination will be retried by the next poll. - failed_destinations.append(destination_name) + else: + if result is not SpanExportResult.SUCCESS: logger.error( - "Destination %s rejected waiting row %s: %s", + "Destination %s failed to export %d span(s).", destination_name, - row["future_id"], - e, + len(spans), + ) + elif recorder is not None and recorder.rejected_spans: + logger.error( + "Destination %s accepted the request but rejected %d of %d " + "span(s), so the batch is not acknowledged: %s", + destination_name, + recorder.rejected_spans, + len(spans), + recorder.error_message or "no reason given", ) - if failed_destinations: - continue - db.mark_sent(row["future_id"]) - sent_count += 1 - logger.info("Queued %d span(s) for all configured OTel destinations.", sent_count) + else: + delivered = True + if not delivered: + failed_destinations.append(destination_name) + elif _destination_healthy.get(destination_name) is False: + logger.info( + "OTel destination %s is accepting spans again.", destination_name + ) + _destination_healthy[destination_name] = delivered + + # A partial success still leaves every row unsent, so the retry re-delivers to + # destinations that already accepted the batch; span ids are deterministic, so + # the duplicates collapse at the backend. + if failed_destinations: + logger.warning( + "Leaving %d span(s) unsent for retry; failed destination(s): %s", + len(spans), + ", ".join(failed_destinations), + ) + return + + try: + db.mark_sent_many(future_ids, db.DB_PATH) + except Exception as e: + logger.error( + "Exported %d span(s) but failed to mark them sent in %s -- they will be " + "re-exported and duplicated on the next poll: %s", + len(future_ids), + db.DB_PATH, + e, + exc_info=True, + ) + return + logger.info("Exported %d span(s) to all configured OTel destinations.", len(spans)) def main(): - global _processors, _redis, _last_destinations_raw + global _exporters, _redis, _last_destinations_raw signal.signal(signal.SIGTERM, _handle_shutdown) signal.signal(signal.SIGINT, _handle_shutdown) - db.init_db() - # GC reaches its own Redis via host.docker.internal (a sibling container, - # not the same network namespace, since GC runs on bridge networking) -- - # match that instead of plain localhost. - _redis = RedisClient(host="host.docker.internal") - _last_destinations_raw = _redis.get(DESTINATIONS_KEY) - _processors = _build_processors(_last_destinations_raw) - logger.info("OTel exporter process started with %d destination(s).", len(_processors)) + try: + db.init_db() + except Exception as e: + logger.error( + "Fatal: cannot initialize the waiting database at %s: %s", + db.DB_PATH, + e, + exc_info=True, + ) + raise + try: + _redis = RedisClient(host="host.docker.internal") + _last_destinations_raw = _redis.get(DESTINATIONS_KEY) + except Exception as e: + logger.error( + "Fatal: cannot reach Redis to read %s: %s", DESTINATIONS_KEY, e, exc_info=True + ) + raise + try: + _exporters = _build_exporters(_last_destinations_raw) + except Exception as e: + logger.error( + "Fatal: cannot build OTel destinations from %s: %s", + DESTINATIONS_KEY, + e, + exc_info=True, + ) + raise + logger.info("OTel exporter process started with %d destination(s).", len(_exporters)) try: last_poll = 0 while _running: @@ -258,17 +570,16 @@ def main(): _reload_destinations_if_changed() _send_pending() except Exception as e: - logger.warning("Poll cycle failed (non-fatal): %s", e) + logger.error( + "Unexpected error in OTel export poll cycle (non-fatal, " + "retrying next tick): %s", + e, + exc_info=True, + ) last_poll = time.time() time.sleep(1) finally: - for destination_name, processor in _processors: - try: - processor.shutdown() - except Exception as e: - logger.error( - "Failed to shut down OTel destination %s: %s", destination_name, e - ) + _shutdown_exporters(_exporters, "shutting the exporter process down") logger.info("OTel exporter process exiting.") diff --git a/canyonos_core/controller/global_controller.py b/canyonos_core/controller/global_controller.py index 5d0d0c1..934a30f 100644 --- a/canyonos_core/controller/global_controller.py +++ b/canyonos_core/controller/global_controller.py @@ -220,7 +220,7 @@ def _load_config(config_path): @staticmethod def _assign_new_project_id(config_path): """Generate a project_id and append it to the config file so it stays stable across reloads/restarts.""" - project_id = uuid.uuid4().hex + project_id = str(uuid.uuid4()) with open(config_path, "a") as f: f.write(f'project_id: "{project_id}"\n') return project_id diff --git a/canyonos_core/controller/local_controller.py b/canyonos_core/controller/local_controller.py index f2d85dd..3ee9329 100644 --- a/canyonos_core/controller/local_controller.py +++ b/canyonos_core/controller/local_controller.py @@ -472,7 +472,7 @@ def _process_request(self, data): for key, value in args.items(): if ( isinstance(value, str) - and len(value) == 32 + and len(value) == 16 and all(c in "0123456789abcdefABCDEF" for c in value) ): future_key = f"future:{value}" @@ -536,7 +536,7 @@ def _resolve_future_args(self, args, poll_interval=0.01, timeout=300): # Check if this arg value is a UUID hex string identifying a future if ( isinstance(value, str) - and len(value) == 32 + and len(value) == 16 and all(c in "0123456789abcdefABCDEF" for c in value) ): future_key = f"future:{value}" diff --git a/canyonos_core/stub_generator.py b/canyonos_core/stub_generator.py index a830735..ef3f4dc 100644 --- a/canyonos_core/stub_generator.py +++ b/canyonos_core/stub_generator.py @@ -411,11 +411,16 @@ def generate_docker( ), ] - # Copy provided agent stubs, overwriting the swept real file at the same path + # Copy provided agent stubs both flat (for `from price_agent import ...` style + # peer imports) and at their entrypoint-mirrored path (overwriting the swept + # real agent file there, as before), so both import styles resolve. if stub_files: for stub_file in stub_files: - destination = _stub_destination(stub_file, stub_entrypoints or {}) - files_to_copy.append((os.path.abspath(stub_file), destination)) + flat_dest = os.path.basename(stub_file) + entrypoint_dest = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), flat_dest)) + if entrypoint_dest != flat_dest: + files_to_copy.append((os.path.abspath(stub_file), entrypoint_dest)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -540,10 +545,15 @@ def generate_workflow_docker( ], ] - # Copy stub files, overwriting the swept real file at the same path + # Copy stub files both flat (for `from price_agent import ...` style imports + # in the workflow) and at their entrypoint-mirrored path (overwriting the + # swept real agent file there, as before), so both import styles resolve. for stub_file in stub_files: - destination = _stub_destination(stub_file, stub_entrypoints or {}) - files_to_copy.append((os.path.abspath(stub_file), destination)) + flat_dest = os.path.basename(stub_file) + entrypoint_dest = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), flat_dest)) + if entrypoint_dest != flat_dest: + files_to_copy.append((os.path.abspath(stub_file), entrypoint_dest)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): diff --git a/cli/canyonos/constants.py b/cli/canyonos/constants.py index 05ac30f..f7c9993 100644 --- a/cli/canyonos/constants.py +++ b/cli/canyonos/constants.py @@ -5,6 +5,8 @@ import ast import os import socket +import urllib.error +import urllib.request import yaml from ruamel.yaml import YAML @@ -12,6 +14,12 @@ DEFAULT_API_PORT = 8080 DEFAULT_DASHBOARD_PORT = 8081 +_EC2_TOKEN_URL = "http://169.254.169.254/latest/api/token" +_EC2_PUBLIC_IP_URL = "http://169.254.169.254/latest/meta-data/public-ipv4" + +_public_ip_cache = None +_public_ip_checked = False + # Fallback when the real function name/params can't be determined statically # (see workflow_entrypoint) -- canyonos_core's own examples all follow this shape. WORKFLOW_ROUTE = "main" @@ -24,6 +32,37 @@ def default_config_path(): return car if os.path.isfile(car) else os.path.join("config", "global_controller.yaml") +def public_ip(timeout=0.3): + """This machine's public IP via the EC2 IMDSv2 metadata service, or None. + + Only meaningful on an EC2 instance -- a laptop or any other host simply + fails to reach the link-local metadata address and gets None back. Kept to + a short timeout and cached for the life of the process so a non-EC2 host + doesn't pay a network-timeout tax on every call site that wants a display + host (deploy summary, `status`, etc). + """ + global _public_ip_cache, _public_ip_checked + if _public_ip_checked: + return _public_ip_cache + _public_ip_checked = True + try: + token_req = urllib.request.Request( + _EC2_TOKEN_URL, + method="PUT", + headers={"X-aws-ec2-metadata-token-ttl-seconds": "21600"}, + ) + token = urllib.request.urlopen(token_req, timeout=timeout).read().decode() + ip_req = urllib.request.Request( + _EC2_PUBLIC_IP_URL, headers={"X-aws-ec2-metadata-token": token} + ) + _public_ip_cache = ( + urllib.request.urlopen(ip_req, timeout=timeout).read().decode().strip() or None + ) + except (OSError, urllib.error.URLError): + _public_ip_cache = None + return _public_ip_cache + + def workflow_api_port(config_path): """Host port the workflow answers on, or None if there isn't one to read.""" try: diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py index 8e61e27..dd87b01 100644 --- a/cli/canyonos/deploy.py +++ b/cli/canyonos/deploy.py @@ -32,6 +32,7 @@ WORKFLOW_ROUTE, default_config_path, port_in_use, + public_ip, workflow_api_port, workflow_entrypoint, workspace_relative, @@ -57,6 +58,14 @@ "process did not complete successfully", ) +# Loggers whose own ERROR: lines are expected, self-recovering noise -- not a +# reason to abort the deploy. Checked before _ERROR_MARKERS so they never +# match: the OTel exporter logs at ERROR: when a destination (the dashboard's +# ingest) isn't reachable yet, which is normal on every cold deploy since +# `canyonos serve` hasn't been started at that point -- it retries and +# recovers on its own once the dashboard comes up. +_BENIGN_ERROR_PREFIXES = ("ERROR:opentelemetry.",) + # (substring, spinner message, completed message). A None spinner message keeps # whatever the spinner already shows; a None completed message prints nothing. # Matched by substring against the raw line, so a phase that never runs is simply @@ -93,6 +102,8 @@ def _agent_progress(self): return "Starting agents..." def feed(self, line): + if any(prefix in line for prefix in _BENIGN_ERROR_PREFIXES): + return None, None, False if any(marker in line for marker in _ERROR_MARKERS): return None, None, True @@ -186,6 +197,21 @@ def run_deploy(config_path=None, serve=True, verbose=False, quiet=False, extra_e return state +def _display_host(host): + """Substitute this machine's own public IP for a loopback host, when discoverable. + + A workflow placed on *this* machine reports `127.0.0.1`/`localhost` -- correct + for curling from the box itself, but useless from anywhere else (e.g. an EC2 + deploy meant to be queried from a laptop). Falls back to `127.0.0.1` off EC2 + (or if the metadata lookup fails), same as before. A workflow placed on a + *different* machine already reports its own real host and passes through + unchanged. + """ + if host not in ("127.0.0.1", "localhost"): + return host + return public_ip() or "127.0.0.1" + + def workflow_targets(gc_port, api_port): """(name, host, port) for each deployed workflow. @@ -196,7 +222,7 @@ def workflow_targets(gc_port, api_port): targets = [ ( endpoint.get("name"), - "127.0.0.1" if endpoint["host"] in ("127.0.0.1", "localhost") else endpoint["host"], + _display_host(endpoint["host"]), endpoint["port"], ) for endpoint in workflow_endpoints(gc_port) @@ -204,7 +230,7 @@ def workflow_targets(gc_port, api_port): ] if targets: return targets - return [(None, "127.0.0.1", api_port)] if api_port else [] + return [(None, _display_host("127.0.0.1"), api_port)] if api_port else [] def _example_route_and_body(config_path): @@ -221,10 +247,15 @@ def _example_route_and_body(config_path): def _curl_example(url, body): - """A copy-pasteable `curl -X POST ...` block, indented to sit under the summary's other rows.""" - json_lines = json.dumps(body, indent=2).splitlines() - indented_body = "\n".join(line if i == 0 else f" {line}" for i, line in enumerate(json_lines)) - return f'curl -X POST {url} \\\n -H "Content-Type: application/json" \\\n -d \'{indented_body}\'' + """A single-line, directly copy-pasteable `curl -X POST ...` command. + + Deliberately not split across `\\`-continued lines or pretty-printed JSON -- + a multi-line block is easy to mangle depending on what actually receives the + paste (some terminals/chat boxes drop the backslashes or the newlines), and + a single line always works no matter where it lands. + """ + compact_body = json.dumps(body) + return f'curl -X POST {url} -H "Content-Type: application/json" -d \'{compact_body}\'' def _summary_body(dashboard_url, targets, config_path): diff --git a/cli/canyonos/theme.py b/cli/canyonos/theme.py index cd2a44e..31fc0c4 100644 --- a/cli/canyonos/theme.py +++ b/cli/canyonos/theme.py @@ -29,8 +29,18 @@ def _is_light_background(): tty.setraw(fd) sys.stdout.write("\x1b]11;?\x07") sys.stdout.flush() - reply = os.read(fd, 32).decode(errors="ignore") if select.select([fd], [], [], 0.1)[0] else "" + # 100ms is fine locally but can be exceeded by a real SSH round-trip; + # 400ms gives a laggy remote session a real chance to answer before + # we give up and assume dark. + reply = os.read(fd, 32).decode(errors="ignore") if select.select([fd], [], [], 0.4)[0] else "" finally: + # A reply that arrives just after our timeout (or a second stray one) + # would otherwise sit in the tty buffer and get echoed as literal text + # ahead of the next command once we restore cooked/echo mode below -- + # drain anything still pending, non-blockingly, before that happens. + while select.select([fd], [], [], 0)[0]: + if not os.read(fd, 1024): + break termios.tcsetattr(fd, termios.TCSADRAIN, old) m = re.search(r"rgb:([0-9a-f]{2})\S*/([0-9a-f]{2})\S*/([0-9a-f]{2})", reply, re.I) return bool(m) and (0.299 * int(m[1], 16) + 0.587 * int(m[2], 16) + 0.114 * int(m[3], 16)) > 128 diff --git a/examples/portfolio/workflow/portfolio_workflow.py b/examples/portfolio/workflow/portfolio_workflow.py index 772f935..a55139d 100644 --- a/examples/portfolio/workflow/portfolio_workflow.py +++ b/examples/portfolio/workflow/portfolio_workflow.py @@ -46,7 +46,9 @@ def main( advisor = AdvisorAgent() # Stage 0: parse the free-text request into structured holdings + window. - intent = intent_agent.parse(query=query) + # parse() returns a Future in deployment -- .value() blocks for the result, + # which comes back as a JSON string like the other stage calls below. + intent = json.loads(intent_agent.parse(query=query).value()) holdings = intent["holdings"] lookback_days = intent["lookback_days"] diff --git a/tests/test_global_controller_project_id.py b/tests/test_global_controller_project_id.py index cf5f43a..88ab9dd 100644 --- a/tests/test_global_controller_project_id.py +++ b/tests/test_global_controller_project_id.py @@ -15,7 +15,9 @@ from canyonos_core.controller.global_controller import GlobalController -UUID_HEX_RE = re.compile(r"^[0-9a-f]{32}$") +UUID_HEX_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +) def _write_config(body): diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index 38d2cd4..8e47ad8 100644 --- a/tests/test_otel_exporter_fanout.py +++ b/tests/test_otel_exporter_fanout.py @@ -15,6 +15,12 @@ # therefore imports ``convert`` and ``db`` as top-level modules. sys.path.insert(0, os.path.join(ROOT, "canyonos_core", "OTLP_Exporter")) +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( # noqa: E402 + ExportTraceServiceResponse, +) +from opentelemetry.sdk.trace.export import SpanExportResult # noqa: E402 + +import convert # noqa: E402 import db # noqa: E402 import otel_exporter # noqa: E402 @@ -62,11 +68,9 @@ def _destination_config(): }, ] - def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): + def test_build_exporters_constructs_mixed_exporters_with_explicit_args(self): grpc_exporter = object() http_exporter = object() - grpc_processor = MagicMock(name="grpc_processor") - http_processor = MagicMock(name="http_processor") destinations = self._destination_config() with patch.object( @@ -77,15 +81,11 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): otel_exporter, "HttpOTLPSpanExporter", return_value=http_exporter, - ) as http_constructor, patch.object( - otel_exporter, - "BatchSpanProcessor", - side_effect=[grpc_processor, http_processor], - ) as processor_constructor: - processors = otel_exporter._build_processors(json.dumps(destinations)) + ) as http_constructor: + exporters = otel_exporter._build_exporters(json.dumps(destinations)) self.assertEqual( - processors, [("railway", grpc_processor), ("langfuse", http_processor)] + exporters, [("railway", grpc_exporter), ("langfuse", http_exporter)] ) grpc_constructor.assert_called_once_with( endpoint="receiver.example:4317", @@ -97,18 +97,12 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): endpoint="https://langfuse.example/api/public/otel", headers={"authorization": "Basic secret"}, timeout=7, - ) - self.assertEqual( - processor_constructor.call_args_list, - [ - unittest.mock.call(grpc_exporter, schedule_delay_millis=1000), - unittest.mock.call(http_exporter, schedule_delay_millis=1000), - ], + session=unittest.mock.ANY, ) - def test_build_processors_raises_when_destinations_raw_is_none(self): + def test_build_exporters_raises_when_destinations_raw_is_none(self): with self.assertRaisesRegex(RuntimeError, "otel.destinations is required"): - otel_exporter._build_processors(None) + otel_exporter._build_exporters(None) def test_configured_destinations_rejects_malformed_empty_and_duplicate_values(self): invalid_values = [ @@ -179,7 +173,7 @@ def _insert_pending_row(self): ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0) """, ( - "00112233445566778899aabbccddeeff", + "0011223344556677", "ffeeddccbbaa99887766554433221100", 1.0, 2.0, @@ -193,72 +187,86 @@ def _insert_pending_row(self): finally: conn.close() - def test_send_pending_delivers_the_same_span_to_every_processor(self): + def _assert_row_unsent(self): + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 0) + finally: + conn.close() + + def test_send_pending_delivers_the_same_spans_to_every_destination(self): self._insert_pending_row() first = MagicMock(name="first") + first.export.return_value = SpanExportResult.SUCCESS second = MagicMock(name="second") + second.export.return_value = SpanExportResult.SUCCESS with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( - otel_exporter.db, "mark_sent" - ) as mark_sent: - otel_exporter._processors = [("railway", first), ("langfuse", second)] + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._exporters = [("railway", first), ("langfuse", second)] otel_exporter._send_pending() - first.on_end.assert_called_once() - second.on_end.assert_called_once() - self.assertIs(first.on_end.call_args.args[0], second.on_end.call_args.args[0]) - mark_sent.assert_called_once_with("00112233445566778899aabbccddeeff") + first.export.assert_called_once() + second.export.assert_called_once() + self.assertEqual( + first.export.call_args.args[0], second.export.call_args.args[0] + ) + mark_sent_many.assert_called_once_with(["0011223344556677"], self.db_path) - def test_send_pending_attempts_remaining_processors_and_leaves_row_unsent_on_failure(self): + def test_send_pending_leaves_rows_unsent_when_a_destination_returns_failure(self): + self._insert_pending_row() + rejecting = MagicMock(name="rejecting") + rejecting.export.return_value = SpanExportResult.FAILURE + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._exporters = [("railway", rejecting)] + otel_exporter._send_pending() + + rejecting.export.assert_called_once() + self._assert_row_unsent() + + def test_send_pending_tries_every_destination_and_leaves_rows_unsent_on_error(self): self._insert_pending_row() failed = MagicMock(name="failed") - failed.on_end.side_effect = RuntimeError("destination unavailable") + failed.export.side_effect = RuntimeError("destination unavailable") remaining = MagicMock(name="remaining") + remaining.export.return_value = SpanExportResult.SUCCESS with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( - otel_exporter.db, "mark_sent" - ) as mark_sent: - otel_exporter._processors = [("railway", failed), ("langfuse", remaining)] + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._exporters = [("railway", failed), ("langfuse", remaining)] otel_exporter._send_pending() - failed.on_end.assert_called_once() - remaining.on_end.assert_called_once() - mark_sent.assert_not_called() + failed.export.assert_called_once() + remaining.export.assert_called_once() + mark_sent_many.assert_not_called() + self._assert_row_unsent() - conn = sqlite3.connect(self.db_path) - try: - self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 0) - finally: - conn.close() - - def test_processor_construction_failure_shuts_down_already_built_processors(self): - first_processor = MagicMock(name="first_processor") + def test_exporter_construction_failure_shuts_down_already_built_exporters(self): + first_exporter = MagicMock(name="first_exporter") destinations = self._destination_config() with patch.object( otel_exporter, "GrpcOTLPSpanExporter", - return_value=object(), + return_value=first_exporter, ), patch.object( otel_exporter, "HttpOTLPSpanExporter", side_effect=RuntimeError("bad HTTP exporter"), - ), patch.object( - otel_exporter, - "BatchSpanProcessor", - return_value=first_processor, ): with self.assertRaisesRegex(RuntimeError, "bad HTTP exporter"): - otel_exporter._build_processors(json.dumps(destinations)) + otel_exporter._build_exporters(json.dumps(destinations)) - first_processor.shutdown.assert_called_once_with() + first_exporter.shutdown.assert_called_once_with() class OTelExporterReloadTests(unittest.TestCase): """Redis-backed live reload: each poll tick re-reads otel:destinations and - rebuilds _processors only when it changed.""" + rebuilds _exporters only when it changed.""" def setUp(self): self._orig_redis = otel_exporter._redis self._orig_raw = otel_exporter._last_destinations_raw - self._orig_processors = otel_exporter._processors + self._orig_exporters = otel_exporter._exporters self.store = {} class FakeRedis: @@ -267,68 +275,531 @@ def get(_self, key): otel_exporter._redis = FakeRedis() otel_exporter._last_destinations_raw = None - otel_exporter._processors = [] + otel_exporter._exporters = [] def tearDown(self): otel_exporter._redis = self._orig_redis otel_exporter._last_destinations_raw = self._orig_raw - otel_exporter._processors = self._orig_processors + otel_exporter._exporters = self._orig_exporters - def test_reload_builds_processors_from_redis_on_first_read(self): + def test_reload_builds_exporters_from_redis_on_first_read(self): destinations = self._config_for("a", "grpc") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p") + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=MagicMock(name="e") ): otel_exporter._reload_destinations_if_changed() - self.assertEqual([name for name, _ in otel_exporter._processors], ["a"]) + self.assertEqual([name for name, _ in otel_exporter._exporters], ["a"]) def test_reload_is_a_noop_when_redis_value_is_unchanged(self): destinations = self._config_for("a", "grpc") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p") - ) as processor_ctor: + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=MagicMock(name="e") + ) as exporter_ctor: otel_exporter._reload_destinations_if_changed() otel_exporter._reload_destinations_if_changed() - processor_ctor.assert_called_once() + exporter_ctor.assert_called_once() - def test_reload_rebuilds_and_shuts_down_old_processors_when_redis_value_changes(self): - old_processor = MagicMock(name="old") - new_processor = MagicMock(name="new") + def test_reload_rebuilds_and_shuts_down_old_exporters_when_redis_value_changes(self): + old_exporter = MagicMock(name="old") + new_exporter = MagicMock(name="new") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc")) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=old_processor + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=old_exporter ): otel_exporter._reload_destinations_if_changed() self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("b", "http")) - with patch.object(otel_exporter, "HttpOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=new_processor + with patch.object( + otel_exporter, "HttpOTLPSpanExporter", return_value=new_exporter ): otel_exporter._reload_destinations_if_changed() - old_processor.shutdown.assert_called_once_with() - self.assertEqual([name for name, _ in otel_exporter._processors], ["b"]) + old_exporter.shutdown.assert_called_once_with() + self.assertEqual([name for name, _ in otel_exporter._exporters], ["b"]) - def test_reload_keeps_previous_processors_when_new_redis_value_is_invalid(self): + def test_reload_keeps_previous_exporters_when_new_redis_value_is_invalid(self): good = MagicMock(name="good") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc")) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=good - ): + with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=good): otel_exporter._reload_destinations_if_changed() self.store[otel_exporter.DESTINATIONS_KEY] = "not json" otel_exporter._reload_destinations_if_changed() good.shutdown.assert_not_called() - self.assertEqual([name for name, _ in otel_exporter._processors], ["a"]) + self.assertEqual([name for name, _ in otel_exporter._exporters], ["a"]) @staticmethod def _config_for(name, protocol): return [{"name": name, "protocol": protocol, "endpoint": "host:1"}] +class OTelExporterQuietFailureTests(unittest.TestCase): + """Failures sqlite and the pricing lookups cannot surface on their own.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + otel_exporter._consecutive_empty_polls = 0 + otel_exporter._empty_queue_warned = False + db._cost_failures_logged.clear() + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._consecutive_empty_polls = 0 + otel_exporter._empty_queue_warned = False + db._cost_failures_logged.clear() + os.unlink(self.db_path) + + def _poll(self, times): + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + for _ in range(times): + otel_exporter._send_pending() + + def _insert(self, future_id, sent): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', ?)", + (future_id, "ffeeddccbbaa99887766554433221100", sent), + ) + conn.commit() + finally: + conn.close() + + def test_queue_that_never_receives_rows_warns_once_and_names_the_path(self): + # sqlite creates a missing file rather than failing, so a misdirected + # DB_PATH is indistinguishable from an idle queue without this warning. + with self.assertLogs(otel_exporter.logger, level="WARNING") as captured: + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS + 4) + + warnings = [m for m in captured.output if "No rows have ever appeared" in m] + self.assertEqual(len(warnings), 1) + self.assertIn(self.db_path, warnings[0]) + + def test_no_warning_before_the_threshold(self): + with self.assertNoLogs(otel_exporter.logger, level="WARNING"): + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS - 1) + + def test_fully_exported_queue_does_not_warn(self): + # An empty pending set means "everything is delivered" here, not "wrong + # database" -- the row count is what separates the two. + self._insert("0011223344556677", sent=1) + with self.assertNoLogs(otel_exporter.logger, level="WARNING"): + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS + 4) + + def test_cost_lookup_failure_is_logged_once_per_kind_and_keeps_the_row(self): + rows = [ + { + "future_id": "0011223344556677", + "request_id": "r1", + "agent": "a1", + "created_at": 1.0, + "finished_at": 2.0, + "model": "m", + "input_token_count": 1, + "output_token_count": 1, + "token_count": 2, + } + ] + boom = RuntimeError("no aws_instance_pricing table") + with patch.object(db.pricing, "compute_token_cost", side_effect=boom), patch.object( + db.pricing, "compute_server_cost", side_effect=boom + ), self.assertLogs(db.logger, level="WARNING") as captured: + for index in range(3): + rows[0]["future_id"] = f"001122334455667{index}" + db.write_waiting_rows(rows, None, "proj", db_path=self.db_path) + + self.assertEqual( + len([m for m in captured.output if "Token cost lookup failed" in m]), 1 + ) + self.assertEqual( + len([m for m in captured.output if "Server cost lookup failed" in m]), 1 + ) + + conn = sqlite3.connect(self.db_path) + try: + written, cost = conn.execute( + "SELECT COUNT(*), COALESCE(SUM(total_cost), 0) FROM waiting" + ).fetchone() + finally: + conn.close() + self.assertEqual(written, 3) + self.assertEqual(cost, 0) + + +class OTelExporterDeliveryIntegrityTests(unittest.TestCase): + """Rows that cannot be delivered must not block or be reported as delivered.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + self._orig_recorders = dict(otel_exporter._partial_success_recorders) + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._partial_success_recorders.clear() + otel_exporter._partial_success_recorders.update(self._orig_recorders) + os.unlink(self.db_path) + + def _insert(self, future_id): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (future_id, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + def _sent(self, future_id): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute( + "SELECT sent FROM waiting WHERE future_id = ?", (future_id,) + ).fetchone()[0] + finally: + conn.close() + + def test_one_unexportable_row_does_not_block_the_rest_of_the_batch(self): + good = ["0011223344556677", "1122334455667788"] + poison = "00112233445566778899aabbccddeeff" # 128-bit, overflows span_id + for future_id in good: + self._insert(future_id) + self._insert(poison) + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._send_pending() + + self.assertEqual(len(exporter.export.call_args.args[0]), len(good)) + mark_sent_many.assert_called_once_with(good, self.db_path) + + def test_unexportable_row_is_named_in_the_log(self): + poison = "00112233445566778899aabbccddeeff" + self._insert(poison) + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + + with patch.object(otel_exporter.db, "DB_PATH", self.db_path), self.assertLogs( + otel_exporter.logger, level="ERROR" + ) as captured: + otel_exporter._send_pending() + + self.assertTrue(any(poison in line for line in captured.output)) + exporter.export.assert_not_called() + + def test_partial_success_rejection_leaves_the_batch_unsent(self): + # The SDK returns SUCCESS for a 200 that rejected spans, so delivery is + # only known by reading partial_success off the response. + future_id = "0011223344556677" + self._insert(future_id) + recorder = otel_exporter._PartialSuccessRecorder("live") + + def export_with_partial_rejection(spans): + # The real hook fires during the HTTP call, after _send_pending has + # reset the recorder, so the mock has to populate it the same way. + recorder.rejected_spans = 1 + recorder.error_message = "span rejected" + return SpanExportResult.SUCCESS + + exporter = MagicMock(name="exporter") + exporter.export.side_effect = export_with_partial_rejection + otel_exporter._exporters = [("live", exporter)] + otel_exporter._partial_success_recorders.clear() + otel_exporter._partial_success_recorders["live"] = recorder + + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._send_pending() + + self.assertEqual(self._sent(future_id), 0) + + def test_recorder_reads_rejected_spans_off_an_ok_response(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 2 + response.partial_success.error_message = "two bad spans" + http_response = MagicMock(ok=True, content=response.SerializeToString()) + + recorder = otel_exporter._PartialSuccessRecorder("live") + recorder(http_response) + + self.assertEqual(recorder.rejected_spans, 2) + self.assertEqual(recorder.error_message, "two bad spans") + + +class OTelExporterConfigValidationTests(unittest.TestCase): + def test_unsupported_protocol_is_rejected_instead_of_defaulting_to_http(self): + for protocol in ("grcp", None, "", 7): + with self.subTest(protocol=protocol): + with self.assertRaisesRegex(ValueError, "protocol must be one of"): + otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": protocol, "endpoint": "h:1"}]) + ) + + def test_protocol_case_is_normalized(self): + for given, expected in (("HTTP", "http"), ("GRPC", "grpc"), ("Http/Protobuf", "http/protobuf")): + with self.subTest(protocol=given): + parsed = otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": given, "endpoint": "h:1"}]) + ) + self.assertEqual(parsed[0]["protocol"], expected) + + def test_supported_protocols_are_accepted(self): + for protocol in otel_exporter.SUPPORTED_PROTOCOLS: + with self.subTest(protocol=protocol): + parsed = otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": protocol, "endpoint": "h:1"}]) + ) + self.assertEqual(parsed[0]["protocol"], protocol) + + +class OTelExporterRowFidelityTests(unittest.TestCase): + def test_missing_identifiers_are_logged_not_silently_dropped(self): + with self.assertLogs(db.logger, level="WARNING") as captured: + db.write_waiting_rows( + [{"future_id": "0011223344556677", "request_id": ""}], + None, + "proj", + db_path=":memory:", + ) + self.assertTrue(any("request_id" in line for line in captured.output)) + + def test_absent_error_name_does_not_fabricate_an_exception_type(self): + span = convert.waiting_row_to_span( + { + "future_id": "0011223344556677", + "session_id": "ffeeddccbbaa99887766554433221100", + "parent_id": None, + "failed": 1, + "error_name": None, + "error_message": "timed out", + "name": "A.b", + "started_at": 1.0, + "finished_at": 2.0, + } + ) + self.assertNotIn("exception.type", span.events[0].attributes) + self.assertEqual(span.events[0].attributes["exception.message"], "timed out") + + def test_empty_output_is_preserved_rather_than_treated_as_absent(self): + self.assertEqual(db._normalize_json_text(""), '""') + self.assertIsNone(db._normalize_json_text(None)) + + +class OTelExporterPlaceholderTests(unittest.TestCase): + """A row that can never be encoded is retired, not retried forever.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + otel_exporter._row_export_failures.clear() + self.poison = "00112233445566778899aabbccddeeff" + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (self.poison, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._row_export_failures.clear() + os.unlink(self.db_path) + + def _poll_once(self, exporter): + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._send_pending() + + def test_placeholder_is_sent_only_after_the_attempt_limit(self): + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + + for _ in range(otel_exporter.MAX_ROW_EXPORT_ATTEMPTS - 1): + self._poll_once(exporter) + exporter.export.assert_not_called() + + self._poll_once(exporter) + sent_spans = exporter.export.call_args.args[0] + self.assertEqual(len(sent_spans), 1) + self.assertEqual(sent_spans[0].name, "canyonos.invalid_span") + + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 1) + finally: + conn.close() + + def test_export_failures_do_not_count_toward_the_row_limit(self): + # Otherwise a spell of receiver downtime would replace every queued row + # with a placeholder. + good = "1122334455667788" + conn = sqlite3.connect(self.db_path) + try: + conn.execute("DELETE FROM waiting") + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (good, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + down = MagicMock(name="down") + down.export.return_value = SpanExportResult.FAILURE + for _ in range(otel_exporter.MAX_ROW_EXPORT_ATTEMPTS + 3): + self._poll_once(down) + + self.assertNotIn(good, otel_exporter._row_export_failures) + recovered = MagicMock(name="recovered") + recovered.export.return_value = SpanExportResult.SUCCESS + self._poll_once(recovered) + self.assertEqual( + recovered.export.call_args.args[0][0].name, "A.b" + ) + + def test_placeholder_is_not_disguised_as_an_agent_failure(self): + session_id = "ffeeddccbbaa99887766554433221100" + placeholder = convert.invalid_row_placeholder_span( + { + "future_id": self.poison, + "session_id": session_id, + "started_at": 1.0, + "finished_at": 2.0, + }, + "int too big to convert", + ) + + self.assertEqual(placeholder.name, "canyonos.invalid_span") + self.assertIs(placeholder.attributes["canyonos.export.invalid"], True) + self.assertEqual(placeholder.attributes["canyonos.future_id"], self.poison) + self.assertEqual(placeholder.events, ()) + self.assertEqual(placeholder.context.trace_id, int(session_id, 16)) + self.assertTrue(0 < placeholder.context.span_id <= 2**64 - 1) + + +class OTelExporterLifecycleLoggingTests(unittest.TestCase): + """Connectivity and recovery must be visible without waiting for traffic.""" + + def setUp(self): + self._orig_exporters = otel_exporter._exporters + otel_exporter._destination_healthy.clear() + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._destination_healthy.clear() + + def test_reachable_destination_is_reported_at_build_time(self): + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + with self.assertLogs(otel_exporter.logger, level="INFO") as captured: + otel_exporter._probe_destination("live", exporter) + exporter.export.assert_called_once_with([]) + self.assertTrue( + any("answered a connectivity check" in line for line in captured.output) + ) + + def test_unreachable_destination_is_reported_at_build_time(self): + exporter = MagicMock(name="exporter") + exporter.export.side_effect = RuntimeError("connection refused") + with self.assertLogs(otel_exporter.logger, level="WARNING") as captured: + otel_exporter._probe_destination("dead", exporter) + self.assertTrue( + any("did not answer a connectivity check" in line for line in captured.output) + ) + self.assertTrue(any("connection refused" in line for line in captured.output)) + + def test_recovery_is_logged_only_after_a_failure(self): + db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + db_path = db_file.name + db_file.close() + self.addCleanup(os.unlink, db_path) + db.init_db(db_path) + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES ('0011223344556677'," + " 'ffeeddccbbaa99887766554433221100', 1.0, 2.0, 0, 'A.b', 0)" + ) + conn.commit() + finally: + conn.close() + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.FAILURE + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", db_path): + otel_exporter._send_pending() + self.assertIs(otel_exporter._destination_healthy["live"], False) + + exporter.export.return_value = SpanExportResult.SUCCESS + with self.assertLogs(otel_exporter.logger, level="INFO") as captured: + otel_exporter._send_pending() + + self.assertTrue( + any("is accepting spans again" in line for line in captured.output) + ) + + def test_steady_success_does_not_log_recovery(self): + db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + db_path = db_file.name + db_file.close() + self.addCleanup(os.unlink, db_path) + db.init_db(db_path) + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES ('0011223344556677'," + " 'ffeeddccbbaa99887766554433221100', 1.0, 2.0, 0, 'A.b', 0)" + ) + conn.commit() + finally: + conn.close() + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._destination_healthy["live"] = True + otel_exporter._exporters = [("live", exporter)] + + with patch.object(otel_exporter.db, "DB_PATH", db_path), self.assertLogs( + otel_exporter.logger, level="INFO" + ) as captured: + otel_exporter._send_pending() + + self.assertTrue(any("Exported 1 span(s)" in line for line in captured.output)) + self.assertFalse( + any("is accepting spans again" in line for line in captured.output) + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_stub_generator.py b/tests/test_stub_generator.py index d3af7f2..2dd013c 100644 --- a/tests/test_stub_generator.py +++ b/tests/test_stub_generator.py @@ -116,7 +116,7 @@ def test_unsafe_entrypoint_falls_back_to_flat(self): class GenerateWorkflowDockerStubPlacementTests(unittest.TestCase): - def test_stub_lands_only_at_its_entrypoint_path(self): + def test_stub_lands_both_flat_and_at_its_entrypoint_path(self): with tempfile.TemporaryDirectory() as tmpdir: workflow_file = Path(tmpdir) / "workflow.py" workflow_file.write_text("from agents.split_agent import SplitAgent\n") @@ -136,7 +136,7 @@ def test_stub_lands_only_at_its_entrypoint_path(self): nested_path = Path(output_dir) / "agents" / "split_agent.py" flat_path = Path(output_dir) / "split_agent.py" self.assertIn("class SplitAgent", nested_path.read_text()) - self.assertFalse(flat_path.exists(), "stub must not be duplicated flat") + self.assertIn("class SplitAgent", flat_path.read_text()) if __name__ == "__main__":