OTEL Collector

otel

OTLP HTTP/JSON receiver and token/cost summary.

Lightweight collector that accepts OTLP exports for metrics and logs, tracks token usage over a sliding window, and prints a summary.

UsageSummary

Bases: NamedTuple

Usage totals and models with Codex completion events from one OTEL log.

start_collector(run_dir, bind_addr='127.0.0.1')

Start the OTEL collector as a subprocess. Returns (proc, port).

Source code in src/agentic_ci/otel.py

def start_collector(run_dir, bind_addr="127.0.0.1"):
    """Start the OTEL collector as a subprocess. Returns (proc, port)."""
    otel_log = os.path.join(run_dir, "claude-otel.jsonl")
    otel_rate = os.path.join(run_dir, "claude-otel-rate.json")
    port_file = os.path.join(run_dir, "otel-port")

    for f in [otel_log, port_file]:
        try:
            os.unlink(f)
        except FileNotFoundError:
            pass

    env = {
        **os.environ,
        "OTEL_LOG_FILE": otel_log,
        "OTEL_RATE_FILE": otel_rate,
        "OTEL_COLLECTOR_PORT": "0",
        "OTEL_PORT_FILE": port_file,
        "OTEL_BIND_ADDR": bind_addr,
    }
    proc = subprocess.Popen(
        [sys.executable, "-m", "agentic_ci.otel"],
        env=env,
        stderr=subprocess.DEVNULL,
    )

    for _ in range(50):
        if os.path.exists(port_file):
            break
        time.sleep(0.1)
    else:
        proc.kill()
        raise RuntimeError("OTEL collector did not write port file")

    with open(port_file) as f:
        port = int(f.read().strip())

    return proc, port, otel_log, otel_rate

stop_collector(proc)

Stop the OTEL collector subprocess.

Source code in src/agentic_ci/otel.py

def stop_collector(proc):
    """Stop the OTEL collector subprocess."""
    proc.terminate()
    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait()

parse_metrics(records)

Parse OTLP JSONL records into structured token/cost data.

Source code in src/agentic_ci/otel.py

def parse_metrics(records):
    """Parse OTLP JSONL records into structured token/cost data."""
    token_totals, cost_totals, api_requests, active_time, _ = _parse_metrics(records)
    return token_totals, cost_totals, api_requests, active_time

read_usage_summary(log_file)

Read curated usage totals without printing or modifying the log.

Raises OSError for unreadable logs and ValueError for a line that is not JSON. Valid JSON with an unexpected shape is skipped, not raised.

Source code in src/agentic_ci/otel.py

def read_usage_summary(log_file):
    """Read curated usage totals without printing or modifying the log.

    Raises OSError for unreadable logs and ValueError for a line that is not
    JSON. Valid JSON with an unexpected shape is skipped, not raised.
    """
    tokens, costs, _, _, models = _read_metrics(log_file)
    return UsageSummary(frozenset(models), dict(tokens), dict(costs))

print_summary(log_file)

Print a human-readable token/cost summary from an OTEL JSONL log.

Returns a :class:UsageSummary. Missing, unreadable, empty, or invalid logs return empty model names, token totals, and cost totals.

Source code in src/agentic_ci/otel.py

def print_summary(log_file):
    """Print a human-readable token/cost summary from an OTEL JSONL log.

    Returns a :class:`UsageSummary`. Missing, unreadable, empty, or invalid
    logs return empty model names, token totals, and cost totals.
    """
    try:
        token_totals, cost_totals, api_requests, active_time, codex_models = _read_metrics(log_file)
    except FileNotFoundError:
        print("No OTEL data collected (log file not found).")
        return UsageSummary(frozenset(), {}, {})
    except OSError as e:
        print(f"Error reading OTEL log: {e}")
        return UsageSummary(frozenset(), {}, {})
    except ValueError as e:
        print(f"Error parsing OTEL log: {e}")
        return UsageSummary(frozenset(), {}, {})

    if not any((token_totals, cost_totals, api_requests, active_time)):
        print("No OTEL data collected.")

    if token_totals:
        models = sorted(set(m for m, _ in token_totals.keys()))
        for model in models:
            model_tokens = {}
            for (m, token_type), count in token_totals.items():
                if m != model:
                    continue
                number = _finite_number(count)
                if number is not None:
                    model_tokens[token_type] = number
            if not model_tokens:
                continue
            print(f"\n  Model: {model}")
            print(f"  {'Token Type':<20} {'Count':>12}")
            print(f"  {'-' * 20} {'-' * 12}")
            for token_type in ["input", "cacheRead", "cacheCreation", "output"]:
                if token_type in model_tokens:
                    print(f"  {token_type:<20} {model_tokens[token_type]:>12,.0f}")
            total = sum(model_tokens.values())
            print(f"  {'TOTAL':<20} {total:>12,.0f}")

    priced = []
    for model in sorted(cost_totals):
        cost = _finite_number(cost_totals[model])
        if cost is not None:
            priced.append((model, cost))
    if priced:
        print(f"\n  {'Model':<30} {'Cost (USD)':>12}")
        print(f"  {'-' * 30} {'-' * 12}")
        grand_total = 0.0
        for model, cost in priced:
            grand_total += cost
            print(f"  {model:<30} ${cost:>11.4f}")
        if len(priced) > 1:
            print(f"  {'TOTAL':<30} ${grand_total:>11.4f}")

    if active_time:
        print("\n  Active Time:")
        for time_type, seconds in sorted(active_time.items()):
            mins, secs = divmod(int(seconds), 60)
            print(f"    {time_type}: {mins}m {secs}s")

    if api_requests:
        print(f"\n  API Requests: {len(api_requests)}")
        total_duration = sum(_non_negative_count(r.get("duration_ms")) or 0.0 for r in api_requests)
        if total_duration:
            print(f"  Total API time: {total_duration / 1000:.1f}s")

    return UsageSummary(frozenset(codex_models), dict(token_totals), dict(cost_totals))

append_codex_usage(log_file, summary, *, start_ns=None)

Append agentic-ci-owned Codex usage totals to an OTEL JSONL log.

Only models with completion events in summary are written. Unknown prices omit the cost point but retain tokens. Returns True when written. start_ns is the run start time for the cumulative sums; when unknown, the collection time is used as the start of the series.

Source code in src/agentic_ci/otel.py

def append_codex_usage(log_file, summary, *, start_ns=None):
    """Append agentic-ci-owned Codex usage totals to an OTEL JSONL log.

    Only models with completion events in ``summary`` are written. Unknown
    prices omit the cost point but retain tokens. Returns True when written.
    ``start_ns`` is the run start time for the cumulative sums; when unknown,
    the collection time is used as the start of the series.
    """
    if not summary.codex_models:
        return False

    end_ns = time.time_ns()
    timestamps = {"startTimeUnixNano": str(start_ns or end_ns), "timeUnixNano": str(end_ns)}
    token_points = []
    cost_points = []
    for model in sorted(summary.codex_models):
        for token_type in _CODEX_TOKEN_TYPES:
            count = summary.token_totals.get((model, token_type), 0)
            if not _reportable_token_count(count):
                continue
            token_points.append(
                {
                    **_metric_number(count),
                    **timestamps,
                    "attributes": [
                        _string_attribute("model", model),
                        _string_attribute("type", token_type),
                    ],
                }
            )
        cost = _non_negative_count(summary.cost_totals.get(model))
        if cost is not None:
            cost_points.append(
                {
                    "asDouble": cost,
                    **timestamps,
                    "attributes": [_string_attribute("model", model)],
                }
            )

    if not token_points and not cost_points:
        return False

    metrics = []
    if cost_points:
        metrics.append(
            {
                "name": _CODEX_COST_METRIC,
                "unit": "USD",
                "sum": {
                    "aggregationTemporality": 2,
                    "isMonotonic": True,
                    "dataPoints": cost_points,
                },
            }
        )
    if token_points:
        metrics.append(
            {
                "name": _CODEX_TOKEN_METRIC,
                "unit": "{token}",
                "sum": {
                    "aggregationTemporality": 2,
                    "isMonotonic": True,
                    "dataPoints": token_points,
                },
            }
        )

    record = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "path": "/v1/metrics",
        "payload": {
            "resourceMetrics": [
                {"scopeMetrics": [{"scope": {"name": "agentic-ci"}, "metrics": metrics}]}
            ]
        },
    }
    with open(log_file, "a") as f:
        f.write(json.dumps(record, allow_nan=False) + "\n")
    return True

generate_trace_context()

Generate W3C Trace Context components for orchestrator-owned root spans.

Returns (trace_id, span_id, traceparent) where traceparent is a valid W3C traceparent header value. If the agent's OTEL SDK picks up the TRACEPARENT env var, its spans become children of this root span.

Source code in src/agentic_ci/otel.py

def generate_trace_context():
    """Generate W3C Trace Context components for orchestrator-owned root spans.

    Returns (trace_id, span_id, traceparent) where traceparent is a valid
    W3C traceparent header value. If the agent's OTEL SDK picks up the
    TRACEPARENT env var, its spans become children of this root span.
    """
    trace_id = uuid.uuid4().hex
    span_id = uuid.uuid4().hex[:16]
    traceparent = f"00-{trace_id}-{span_id}-01"
    return trace_id, span_id, traceparent

root_span_attributes(backend, harness, model, effort=None, subagent_effort=None)

Return the synthetic root span attributes for one agent run.

Always sets agent.backend, agent.harness and agent.model; adds agent.reasoning_effort and agent.subagent_reasoning_effort when those efforts are in effect, so MLflow and JSONL consumers can see the effective effort next to the model.

Source code in src/agentic_ci/otel.py

def root_span_attributes(
    backend: str,
    harness: str,
    model: str,
    effort: str | None = None,
    subagent_effort: str | None = None,
) -> dict[str, str]:
    """Return the synthetic root span attributes for one agent run.

    Always sets ``agent.backend``, ``agent.harness`` and ``agent.model``;
    adds ``agent.reasoning_effort`` and ``agent.subagent_reasoning_effort``
    when those efforts are in effect, so MLflow and JSONL consumers can see
    the effective effort next to the model.
    """
    attributes = {"agent.backend": backend, "agent.harness": harness, "agent.model": model}
    if effort is not None:
        attributes["agent.reasoning_effort"] = effort
    if subagent_effort is not None:
        attributes["agent.subagent_reasoning_effort"] = subagent_effort
    return attributes

inject_root_spans(log_file, start_ns, end_ns, exit_code, fallback_trace_id=None, fallback_span_id=None, attributes=None)

Stitch the run's traces together and append a synthetic root.

All traces in the JSONL belong to one agent run, so they are first merged into a single trace (see :func:_stitch_run_traces). Then, for each trace that has child spans but no root span (agent ignored TRACEPARENT, container crash, OOM, timeout), a synthetic root span is appended to the JSONL file. If no spans exist at all and fallback_trace_id is provided, a standalone root span is created so MLflow always has at least one complete trace. When fallback_span_id is set, the fallback root reuses that span ID so children that were parented via TRACEPARENT stay connected.

Returns the number of synthetic root spans injected.

Source code in src/agentic_ci/otel.py

def inject_root_spans(
    log_file,
    start_ns,
    end_ns,
    exit_code,
    fallback_trace_id=None,
    fallback_span_id=None,
    attributes=None,
):
    """Stitch the run's traces together and append a synthetic root.

    All traces in the JSONL belong to one agent run, so they are first merged
    into a single trace (see :func:`_stitch_run_traces`). Then, for each trace
    that has child spans but no root span (agent ignored TRACEPARENT,
    container crash, OOM, timeout), a synthetic root span is appended to the
    JSONL file. If no spans exist at all and fallback_trace_id is provided, a
    standalone root span is created so MLflow always has at least one
    complete trace. When fallback_span_id is set, the fallback root reuses
    that span ID so children that were parented via TRACEPARENT stay
    connected.

    Returns the number of synthetic root spans injected.
    """
    records = []
    try:
        with open(log_file) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
    except FileNotFoundError:
        pass

    primary, _ = _stitch_run_traces(records, fallback_trace_id, fallback_span_id)
    if primary:
        _write_records(log_file, records)

    orphans = _find_orphan_traces(records)

    injected = 0
    root_ids = {}

    for trace_id, (child_start, child_end, dangling_parent) in orphans.items():
        if dangling_parent:
            span_id = dangling_parent
        elif fallback_span_id:
            span_id = fallback_span_id
        else:
            span_id = uuid.uuid4().hex[:16]
        root_ids[trace_id] = span_id

    # An agent can drop more than one parent span (Codex loses its exec root
    # and a session span). The synthetic root reuses the most common missing
    # parent; every other span whose parent was never exported is re-homed
    # under it so nothing renders detached. Traces that already have a real
    # root get their dangling spans re-homed under that root instead.
    for trace_id, root_span_id in _real_root_ids(records).items():
        root_ids.setdefault(trace_id, root_span_id)
    if _rehome_dangling_spans(records, root_ids):
        _write_records(log_file, records)

    for trace_id, (child_start, child_end, _dangling_parent) in orphans.items():
        span_id = root_ids[trace_id]
        span_start = min(start_ns, child_start) if child_start else start_ns
        span_end = max(end_ns, child_end) if child_end else end_ns
        record = _build_root_span_record(
            trace_id, span_id, span_start, span_end, exit_code, attributes
        )
        with open(log_file, "a") as f:
            f.write(json.dumps(record) + "\n")
        injected += 1

    if not orphans and not _has_any_traces(records) and fallback_trace_id:
        span_id = fallback_span_id if fallback_span_id else uuid.uuid4().hex[:16]
        record = _build_root_span_record(
            fallback_trace_id, span_id, start_ns, end_ns, exit_code, attributes
        )
        with open(log_file, "a") as f:
            f.write(json.dumps(record) + "\n")
        injected += 1

    return injected

main()

Run the OTEL collector server.

Source code in src/agentic_ci/otel.py

def main():
    """Run the OTEL collector server."""
    port = int(os.environ.get("OTEL_COLLECTOR_PORT", "4318"))
    bind_addr = os.environ.get("OTEL_BIND_ADDR", "127.0.0.1")
    server = HTTPServer((bind_addr, port), OTLPHandler)
    actual_port = server.server_address[1]
    port_file = os.environ.get("OTEL_PORT_FILE")
    if port_file:
        with open(port_file, "w") as f:
            f.write(str(actual_port))
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    log_file = os.environ.get("OTEL_LOG_FILE", "/tmp/claude-otel.jsonl")
    print(
        f"OTLP collector listening on {bind_addr}:{actual_port}, writing to {log_file}",
        file=sys.stderr,
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    server.server_close()