diff --git a/.github/workflows/reusable-dispatch.yml b/.github/workflows/reusable-dispatch.yml index 81ef1fb552..cba6b359d9 100644 --- a/.github/workflows/reusable-dispatch.yml +++ b/.github/workflows/reusable-dispatch.yml @@ -65,6 +65,13 @@ on: required: false FULLSEND_GCP_PROJECT_ID: required: true + OTEL_EXPORTER_OTLP_TRACES_HEADERS: + description: >- + OTLP headers for ADR 0050 Level 2 trace export, forwarded to the + triage stage (baggage-style k=v,k=v; may carry auth, hence a + secret). Optional — unset leaves export disabled unless the + endpoint variable alone suffices. + required: false jobs: route: @@ -453,6 +460,7 @@ jobs: secrets: FULLSEND_GCP_WIF_PROVIDER: ${{ secrets.FULLSEND_GCP_WIF_PROVIDER }} FULLSEND_GCP_PROJECT_ID: ${{ secrets.FULLSEND_GCP_PROJECT_ID }} + OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }} code: name: Code diff --git a/.github/workflows/reusable-triage.yml b/.github/workflows/reusable-triage.yml index 2261191a71..de00903124 100644 --- a/.github/workflows/reusable-triage.yml +++ b/.github/workflows/reusable-triage.yml @@ -47,6 +47,12 @@ on: required: true FULLSEND_GCP_PROJECT_ID: required: true + OTEL_EXPORTER_OTLP_TRACES_HEADERS: + description: >- + OTLP headers for ADR 0050 Level 2 trace export (baggage-style + k=v,k=v; may carry auth, hence a secret). Optional — unset leaves + export disabled unless the endpoint variable alone suffices. + required: false concurrency: group: fullsend-triage-agent-${{ inputs.source_repo }}-${{ fromJSON(inputs.event_payload).issue.number || fromJSON(inputs.event_payload).pull_request.number }} @@ -171,6 +177,12 @@ jobs: env: GITHUB_ISSUE_URL: ${{ fromJSON(inputs.event_payload).issue.html_url }} REPO_FULL_NAME: ${{ inputs.source_repo }} + # ADR 0050 Level 2: orgs opt into OTLP trace export by defining the + # endpoint as an Actions variable (and, when the backend needs auth + # or routing headers, the headers secret). Unset = export inert. + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT }} + OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }} + OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} with: agent: triage version: ${{ inputs.fullsend_version }} diff --git a/docs/guides/infrastructure/distributed-tracing.md b/docs/guides/infrastructure/distributed-tracing.md index 34f47bab78..ff31c3c985 100644 --- a/docs/guides/infrastructure/distributed-tracing.md +++ b/docs/guides/infrastructure/distributed-tracing.md @@ -19,6 +19,21 @@ configuration required: These files are always written, even when no OTLP backend is configured. They contain metadata only — no prompts, completions, or source code content. +## Prerequisites + +Level 1 requires nothing. To enable OTLP export (Level 2 and Level 3) you need: + +- An **OTLP/HTTP-capable backend** and its endpoint URL — e.g. Jaeger, Tempo, + Grafana, MLflow ≥ 3.6, or any OpenTelemetry Collector. +- Any **backend authentication** (bearer token or basic auth) for the + `OTEL_EXPORTER_OTLP_TRACES_HEADERS` variable. +- **Network reachability** from where runs execute (your machine or CI runners) + to the backend endpoint. +- For a backend behind a **private CA** (e.g. an internal MLflow): the CA + certificate bundle, pointed to by `OTEL_EXPORTER_OTLP_CERTIFICATE`. Local + and bring-your-own-workflow runs only — the managed workflows do not yet + pass a CA bundle through. + ## Enabling OTLP export (Level 2) To send metadata spans to an OpenTelemetry-compatible backend, set one of the @@ -45,8 +60,59 @@ Langfuse, SigNoz, Honeycomb, Datadog, etc. If the endpoint is unreachable, the CLI continues normally — local files are still produced and the run is not affected. +Operational details: + +- **Export timing:** spans are exported once, when the run closes, inside a + hard wall-clock budget (5 seconds). There is no mid-run network traffic; a + dead endpoint costs at most the budget and one warning line. +- **Crashed runs are not exported:** export replays the finalized artifacts, + so a run that never finalizes (crash, OOM, SIGKILL) writes no + `run-summary.json` and exports nothing. Its `run-telemetry.jsonl` remains + the local forensic record. +- **Sampling:** when the run continues an inbound `TRACEPARENT` whose W3C + sampled flag is unset (`-00`), the upstream sampling decision is respected: + nothing is exported. Local files are always written regardless. +- **Protocol:** OTLP over `http/protobuf` only. Setting + `OTEL_EXPORTER_OTLP_PROTOCOL` (or the traces-specific variant) to anything + else — e.g. `grpc` — skips export with a warning rather than posting + protobuf at a gRPC endpoint. +- **Validation:** a malformed endpoint value skips export with a warning; it + is never silently replaced with the SDK's `localhost:4318` default. +- **Kill switches:** `OTEL_SDK_DISABLED=true` and `OTEL_TRACES_EXPORTER=none` + are honored. +- **Private CAs:** point `OTEL_EXPORTER_OTLP_CERTIFICATE` at a PEM bundle for + backends with certificates outside the system trust store. There is no + skip-verify option. + +### MLflow example + +MLflow ≥ 3.6 ingests OTLP/HTTP natively at `{server}/v1/traces` and routes +traces to an experiment via a required header: + +```bash +export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="https://mlflow.example.com/v1/traces" +export OTEL_EXPORTER_OTLP_TRACES_HEADERS="x-mlflow-experiment-id=42" +``` + +Header values are URL-decoded, so spaces are percent-encoded — for a +Basic-auth-fronted instance: + +```bash +export OTEL_EXPORTER_OTLP_TRACES_HEADERS="authorization=Basic%20${CREDS_B64},x-mlflow-experiment-id=42" +``` + +> **Cost columns:** MLflow's per-trace cost is its own estimate — extracted +> input/output token counts priced against MLflow's internal model table. It +> excludes cache-creation/cache-read tokens, which dominate agent-run cost. +> The authoritative figure is the runtime-reported `fullsend.cost_usd` on +> `agent` spans (also in `run-summary.json`). + ## Enabling content capture (Level 3) +> **Planned:** Level 3 content capture is not yet implemented. This section +> documents the contract decided in +> [ADR 0050](../../ADRs/0050-distributed-tracing-instrumentation.md). + By default, spans contain metadata only (timing, token counts, tool names, errors). To include full prompt/completion content in spans: @@ -92,83 +158,132 @@ consumers (scripts, other agents) can continue the trace chain. ## Span structure -A typical agent run produces this span hierarchy: +A run produces this span hierarchy (span names match the `name` field in +`run-telemetry.jsonl` — the exported spans and the local file are two views +of the same trace, with identical span ids): ``` -fullsend-run (root, SpanKind=Consumer if dispatched) -├── load-harness -├── setup-sandbox -│ └── create-sandbox (gen_ai.operation.name=create_agent) -├── agent-execution.iteration-0 -│ └── (gen_ai.operation.name=invoke_agent) -├── agent-execution.iteration-1 -├── collect-artifacts -├── security-scan -└── validation +run (root; Consumer when dispatched with TRACEPARENT, else Internal) +├── sandbox_create (gen_ai.operation.name=create_agent) +└── agent (one per iteration; gen_ai.operation.name=invoke_agent) ``` ### GenAI semantic conventions -Root and iteration spans carry [OTEL GenAI semantic convention](https://opentelemetry.io/docs/specs/semconv/gen-ai/) attributes: +Spans carry [OTEL GenAI semantic convention](https://opentelemetry.io/docs/specs/semconv/gen-ai/) attributes: -| Attribute | Example | Description | -|-----------|---------|-------------| -| `gen_ai.operation.name` | `invoke_agent` | The GenAI operation type | -| `gen_ai.agent.name` | `triage` | The agent being executed | -| `gen_ai.request.model` | `claude-sonnet-4-20250514` | The model configured in the harness | -| `gen_ai.system` | `anthropic` | The LLM provider | +| Attribute | Example | On | +|-----------|---------|-----| +| `gen_ai.operation.name` | `invoke_agent` | `run` and `agent` spans (`create_agent` on `sandbox_create`) | +| `gen_ai.agent.name` | `triage` | `run` and `agent` spans | +| `gen_ai.request.model` | `claude-opus-4-6` | `agent` spans (resolved model) | +| `gen_ai.system` | `anthropic` | `agent` spans (the model vendor, from the runtime) | +| `gen_ai.usage.input_tokens` / `output_tokens` / `cache_*_input_tokens` | `109938` | `agent` spans | These attributes enable LLM-aware backends to recognize fullsend spans as agent operations and surface them in GenAI-specific dashboards. ### SpanKind -- **Consumer**: The root span when `TRACEPARENT` is set (the run was - dispatched by an external system). -- **Internal**: The root span for local/manual invocations. +- **Consumer**: The root span when a valid inbound `TRACEPARENT` was adopted + (the run was dispatched by an instrumented system). +- **Internal**: The root span for local/manual invocations, and all child + spans. ## Custom attributes -Every span also carries fullsend-specific attributes: +Fullsend-specific attributes: + +| Attribute | On | Description | +|-----------|-----|-------------| +| `fullsend.work_item_id` | every span | Work item identity (e.g. `owner/repo#123`) — the primary cross-run correlation key | +| `fullsend.cost_usd` | `agent` spans | Iteration cost in USD, rounded to cents | +| `fullsend.tool_calls` | `agent` spans | Tool invocations in the iteration | +| `agent` | `run` span | Agent name (predates `gen_ai.agent.name`; kept for Level 1 consumers) | + +## GHA workflow configuration -| Attribute | Description | -|-----------|-------------| -| `fullsend.agent` | Agent name from the harness | -| `fullsend.harness` | Path to the harness YAML | -| `fullsend.model` | Model identifier | -| `fullsend.image` | Container image used | -| `fullsend.work_item_id` | Issue/PR number being addressed | +### Managed workflows -## GHA workflow configuration example +Only the **triage** stage forwards OTEL configuration in this release; the +other agents (code, fix, review, retro, prioritize) do not export yet. -Add these environment variables to workflow jobs that run `fullsend run`: +To enable export for triage runs, set on the org (or repo) that hosts the +fullsend caller workflows: + +1. Actions **variable** `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` — the backend's + full traces URL (e.g. `https://mlflow.example.com/v1/traces`). +2. Actions **secret** `OTEL_EXPORTER_OTLP_TRACES_HEADERS` — the complete + header string, auth and routing included (e.g. + `Authorization=Bearer%20,x-mlflow-experiment-id=42`). +3. Optional: Actions **variable** `OTEL_RESOURCE_ATTRIBUTES` — static + `k=v,k=v` trace tags. The value is used verbatim: `${{ github.* }}` + expressions evaluate only in workflow YAML, not in variables. + +Installations scaffolded before this release must also forward the secret +(add `OTEL_EXPORTER_OTLP_TRACES_HEADERS` under `secrets:`) until the scaffold +is re-synced: in the `.fullsend` repo's `triage.yml` (per-org), or in the +fullsend shim workflow's dispatch job (per-repo). + +### Bring your own workflow + +Add the environment variables to any job that runs `fullsend run`: ```yaml env: - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "${{ secrets.OTLP_ENDPOINT }}" - OTEL_EXPORTER_OTLP_TRACES_HEADERS: "Authorization=Bearer ${{ secrets.OTLP_TOKEN }}" + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "${{ vars.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT }}" + OTEL_EXPORTER_OTLP_TRACES_HEADERS: "${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }}" ``` -The secret names and values depend on your chosen backend. Consult your -backend's documentation for the endpoint URL and authentication mechanism. +Any variable and secret names work here — the values reach the exporter +as-is. Consult your backend's documentation for the endpoint URL and +authentication mechanism. + +### Organizing traces for an org + +Two conventions keep a shared backend navigable as repos onboard: + +1. **One backend bucket per org.** On MLflow, create one experiment per org + (e.g. `fullsend-`) and point the org's header secret at its id. The + backend's per-bucket access controls then align with org boundaries. +2. **Slice inside the bucket with resource attributes.** Standard OTel + resource env is honored, so workflows can tag every trace with repo, + agent, and environment: + + ```yaml + env: + OTEL_RESOURCE_ATTRIBUTES: "fullsend.repo=${{ github.repository }},fullsend.agent=triage,deployment.environment=prod" + ``` + + The example is inline workflow `env:`, where `${{ github.* }}` evaluates. + On the managed path, set the `OTEL_RESOURCE_ATTRIBUTES` Actions variable + to a static value instead — variables are not expression-expanded. + + These become filterable trace tags (enable them as columns in MLflow's + Traces table). `fullsend.work_item_id` is already on every span, so runs + for the same issue correlate without configuration. ## Local development Run an agent locally with traces going to a local backend: -```bash -# Start a local Jaeger instance (OTLP-compatible) -podman run -d --name jaeger \ - -p 16686:16686 \ - -p 4318:4318 \ - jaegertracing/jaeger +1. Start a local Jaeger instance (OTLP-compatible): -# Run an agent with tracing enabled -export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4318" -fullsend run triage --issue 42 + ```bash + podman run -d --name jaeger \ + -p 16686:16686 \ + -p 4318:4318 \ + jaegertracing/jaeger + ``` -# View traces at http://localhost:16686 -``` +2. Point the exporter at it and run an agent: + + ```bash + export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4318" + fullsend run triage --issue 42 + ``` + +3. View the traces at . Other lightweight local backends: @@ -176,7 +291,7 @@ Other lightweight local backends: |---------|---------|-----| | Jaeger | `podman run -p 16686:16686 -p 4318:4318 jaegertracing/jaeger` | `localhost:16686` | | Arize Phoenix | `podman run -p 6006:6006 -p 4318:4318 arizephoenix/phoenix` | `localhost:6006` | -| MLflow | `uvx mlflow server` (with OTLP plugin) | `localhost:5000` | +| MLflow ≥ 3.6 | `uvx "mlflow>=3.6" server --backend-store-uri sqlite:///mlflow.db` (native OTLP at `/v1/traces`; requires the `x-mlflow-experiment-id` header — see the MLflow example above) | `localhost:5000` | ## Other backends diff --git a/go.mod b/go.mod index 79ef5a60e8..b1b8eb351e 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,11 @@ require ( github.com/knights-analytics/hugot v0.7.5 github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 + go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/trace v1.44.0 + go.opentelemetry.io/proto/otlp v1.10.0 golang.org/x/crypto v0.52.0 golang.org/x/oauth2 v0.36.0 golang.org/x/term v0.43.0 @@ -17,16 +22,27 @@ require ( ) require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cucumber/gherkin/go/v26 v26.2.0 // indirect github.com/cucumber/messages/go/v21 v21.0.1 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/gofrs/uuid v4.4.0+incompatible // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/hashicorp/go-immutable-radix v1.3.1 // indirect github.com/hashicorp/go-memdb v1.3.4 // indirect github.com/hashicorp/golang-lru v0.5.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + golang.org/x/net v0.55.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.81.1 // indirect ) require ( - cloud.google.com/go/compute/metadata v0.3.0 // indirect + cloud.google.com/go/compute/metadata v0.9.0 // indirect github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect github.com/charmbracelet/colorprofile v0.4.3 // indirect github.com/charmbracelet/x/ansi v0.11.6 // indirect @@ -64,7 +80,7 @@ require ( golang.org/x/image v0.41.0 // indirect golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.45.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect + google.golang.org/protobuf v1.36.11 k8s.io/klog/v2 v2.140.0 // indirect ) diff --git a/go.sum b/go.sum index 31f3b0081e..44c840a9a5 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -cloud.google.com/go/compute/metadata v0.3.0 h1:Tz+eQXMEqDIKRsmY3cHTL6FVaynIjX2QxYC4trgAKZc= -cloud.google.com/go/compute/metadata v0.3.0/go.mod h1:zFmK7XCadkQkj6TtorcaGlCW1hT1fIilQDwofLpJ20k= +cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= +cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= codeberg.org/go-fonts/liberation v0.5.0 h1:SsKoMO1v1OZmzkG2DY+7ZkCL9U+rrWI09niOLfQ5Bo0= codeberg.org/go-fonts/liberation v0.5.0/go.mod h1:zS/2e1354/mJ4pGzIIaEtm/59VFCFnYC7YV6YdGl5GU= codeberg.org/go-latex/latex v0.1.0 h1:hoGO86rIbWVyjtlDLzCqZPjNykpWQ9YuTZqAzPcfL3c= @@ -14,6 +14,10 @@ github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiE github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= github.com/campoy/embedmd v1.0.0 h1:V4kI2qTJJLf4J29RzI/MAt2c3Bl4dQSYPuflzwFH2hY= github.com/campoy/embedmd v1.0.0/go.mod h1:oxyr9RCiSXg0M3VJ3ks0UGfp98BpSSGr0kpiX3MzVl8= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/charmbracelet/colorprofile v0.4.3 h1:QPa1IWkYI+AOB+fE+mg/5/4HRMZcaXex9t5KX76i20Q= github.com/charmbracelet/colorprofile v0.4.3/go.mod h1:/zT4BhpD5aGFpqQQqw7a+VtHCzu+zrQtt1zhMt9mR4Q= github.com/charmbracelet/lipgloss v1.1.0 h1:vYXsiLHVkK7fp74RkV7b2kq9+zDLoEU4MZoFqR/noCY= @@ -47,8 +51,11 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/go-errors/errors v1.5.1 h1:ZwEMSLRCapFLflTpT7NKaAc7ukJ8ZPEjzlxt8rPN8bk= github.com/go-errors/errors v1.5.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/gofrs/flock v0.13.0 h1:95JolYOvGMqeH31+FC7D2+uULf6mG61mEZ/A8dRYMzw= github.com/gofrs/flock v0.13.0/go.mod h1:jxeyy9R1auM5S6JYDBhDt+E2TCo7DkratH4Pgi8P+Z0= github.com/gofrs/uuid v4.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= @@ -57,6 +64,8 @@ github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1 github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 h1:DACJavvAHhabrF08vX0COfcOBJRhZ8lUbR+ZWIs0Y5g= github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/gomlx/exceptions v0.0.3 h1:HKnTgEjj4jlmhr8zVFkTP9qmV1ey7ypYYosQ8GzXWuM= github.com/gomlx/exceptions v0.0.3/go.mod h1:uHL0TQwJ0xaV2/snJOJV6hSE4yRmhhfymuYgNredGxU= github.com/gomlx/go-huggingface v0.3.5 h1:eZz1huOvfr0TW30e11TkGAUZY4Jj5Oh/g0Thz4cvu0I= @@ -71,6 +80,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/hashicorp/go-immutable-radix v1.3.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc= github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= @@ -146,6 +157,26 @@ github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavM github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM= github.com/yalue/onnxruntime_go v1.31.0 h1:1ln4YW1SFOFfGJZXe3jNOb2JUSt+l2pEneZfV8HdtFA= github.com/yalue/onnxruntime_go v1.31.0/go.mod h1:b4X26A8pekNb1ACJ58wAXgNKeUCGEAQ9dmACut9Sm/4= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 h1:lgh3PiVrRUWMLOVSkQicxzZll5NjF1r+AtsX1XRIHw0= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0/go.mod h1:5Cnhth3m/AgOeTgE3ex12pPmiu/gGtZit03kSzx9X7s= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= +go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= @@ -153,6 +184,8 @@ golang.org/x/exp v0.0.0-20260603202125-055de637280b h1:v1uXiEBHo8QA0LiGCo7UgHMzH golang.org/x/exp v0.0.0-20260603202125-055de637280b/go.mod h1:d2fgXJLVs4dYDHUk5lwMIfzRzSrWCfGZb0ZqeLa/Vcw= golang.org/x/image v0.41.0 h1:8wS72eGJMJaBxK6okTzd4WaXumUlTVlb753MlsSvTCo= golang.org/x/image v0.41.0/go.mod h1:uIc348UZMSvS5Z65CVZ7iDPaNobNFEPeJ4kbqTOszmA= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= @@ -164,8 +197,16 @@ golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= gonum.org/v1/plot v0.15.2 h1:Tlfh/jBk2tqjLZ4/P8ZIwGrLEWQSPDLRm/SNWKNXiGI= gonum.org/v1/plot v0.15.2/go.mod h1:DX+x+DWso3LTha+AdkJEv5Txvi+Tql3KAGkehP0/Ubg= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/hack/telemetry-replay/main.go b/hack/telemetry-replay/main.go new file mode 100644 index 0000000000..b0fe9aeabc --- /dev/null +++ b/hack/telemetry-replay/main.go @@ -0,0 +1,56 @@ +// Command telemetry-replay replays a run directory's Level 1 telemetry +// artifacts (run-telemetry.jsonl + run-summary.json) through the production +// OTLP export path (internal/telemetry/otlp). It exists for validating the +// Level 2 export against a real backend without running an agent: point the +// standard OTEL_EXPORTER_OTLP_* env vars at the backend and replay a +// captured artifact directory. +// +// OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 \ +// go run ./hack/telemetry-replay --input /path/to/run-dir +// +// For MLflow (>= 3.6): +// +// OTEL_EXPORTER_OTLP_TRACES_ENDPOINT={server}/v1/traces \ +// OTEL_EXPORTER_OTLP_TRACES_HEADERS="x-mlflow-experiment-id={id}" \ +// go run ./hack/telemetry-replay --input /path/to/run-dir +package main + +import ( + "encoding/json" + "flag" + "fmt" + "os" + "path/filepath" + + "github.com/fullsend-ai/fullsend/internal/telemetry" + "github.com/fullsend-ai/fullsend/internal/telemetry/otlp" +) + +func main() { + dir := flag.String("input", "", "run directory containing "+telemetry.TelemetryFile+" and "+telemetry.SummaryFile) + version := flag.String("service-version", "telemetry-replay-dev", "service.version resource attribute") + flag.Parse() + + if *dir == "" { + fmt.Fprintln(os.Stderr, "usage: telemetry-replay --input ") + os.Exit(2) + } + if !otlp.Enabled() { + fmt.Fprintln(os.Stderr, "no OTLP endpoint configured: set OTEL_EXPORTER_OTLP_ENDPOINT or OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") + os.Exit(2) + } + + if err := otlp.ExportRunDir(*dir, *version); err != nil { + fmt.Fprintln(os.Stderr, "export failed:", err) + os.Exit(1) + } + + // Echo the trace id so the operator can find the trace in the backend. + var s struct { + TraceID string `json:"trace_id"` + } + if data, err := os.ReadFile(filepath.Join(*dir, telemetry.SummaryFile)); err == nil { + _ = json.Unmarshal(data, &s) + } + fmt.Printf("exported %s (trace_id %s)\n", *dir, s.TraceID) +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 126c058457..6629a00ca4 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -41,6 +41,7 @@ import ( "github.com/fullsend-ai/fullsend/internal/security" "github.com/fullsend-ai/fullsend/internal/statuscomment" "github.com/fullsend-ai/fullsend/internal/telemetry" + "github.com/fullsend-ai/fullsend/internal/telemetry/otlp" "github.com/fullsend-ai/fullsend/internal/ui" ) @@ -788,11 +789,20 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep var lastExitCode int var transcriptErrorOverride bool rec := telemetry.New(runDir, traceCtx, agentName, workItemID, runStart) - defer func() { rec.Finalize(telemetryExitCode(lastExitCode, runErr)) }() + defer func() { + rec.Finalize(telemetryExitCode(lastExitCode, runErr)) + // ADR 0050 Level 2: best-effort OTLP export of the finalized + // artifacts. Inert without an endpoint configured; fail-open — a + // dead endpoint costs at most the package's bounded budget and + // never affects the run's outcome. + if err := otlp.ExportRunDir(runDir, Version()); err != nil { + printer.StepWarn("OTLP export failed (run unaffected): " + err.Error()) + } + }() createStart := time.Now() printer.StepStart("Creating sandbox: " + sandboxName) - sandboxSpan := rec.StartSpan("sandbox_create", "", nil) + sandboxSpan := rec.StartSpan("sandbox_create", "", map[string]any{"gen_ai.operation.name": "create_agent"}) readyTimeout := time.Duration(h.SandboxTimeoutSeconds) * time.Second if err := sandbox.CreateWithRetry(sandboxName, h.Providers, h.Image, h.Policy, sandbox.DefaultMaxCreateAttempts, readyTimeout); err != nil { @@ -1160,7 +1170,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep heartbeatDone := make(chan struct{}) go runHeartbeat(printer, agentStart, timeout, heartbeatDone) - agentSpan := rec.StartSpan("agent", "", map[string]any{"iteration": iteration}) + agentSpan := rec.StartSpan("agent", "", agentSpanStartAttrs(iteration, agentName)) var metrics agentruntime.RunMetrics exitCode, runErr := rt.Run(ctx, agentruntime.RunParams{ SandboxName: sandboxName, @@ -1775,6 +1785,17 @@ func validationFailMessage(output []byte, execErr error) string { // metrics.json keeps full precision. func roundUSD(c float64) float64 { return math.Round(c*100) / 100 } +// agentSpanStartAttrs builds the span_start attributes for one agent +// iteration: the iteration counter plus the OTEL GenAI semconv identity +// (gen_ai.operation.name, gen_ai.agent.name) named by ADR 0050. +func agentSpanStartAttrs(iteration int, agentName string) map[string]any { + return map[string]any{ + "iteration": iteration, + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.name": agentName, + } +} + // agentSpanEndAttrs builds the span_end attributes for one agent iteration, // using OTEL GenAI semconv names (gen_ai.*) so the later L2 OTLP transform is // ~1:1. system is the runtime's gen_ai.system vendor (kept runtime-agnostic, not diff --git a/internal/cli/telemetry_run_test.go b/internal/cli/telemetry_run_test.go index 1869e28b94..c9336cd259 100644 --- a/internal/cli/telemetry_run_test.go +++ b/internal/cli/telemetry_run_test.go @@ -252,6 +252,13 @@ func TestResolveTraceIdentity(t *testing.T) { } } +func TestAgentSpanStartAttrs(t *testing.T) { + attrs := agentSpanStartAttrs(3, "code") + assert.Equal(t, 3, attrs["iteration"]) + assert.Equal(t, "invoke_agent", attrs["gen_ai.operation.name"]) + assert.Equal(t, "code", attrs["gen_ai.agent.name"]) +} + func TestAgentSpanEndAttrs(t *testing.T) { var m agentruntime.RunMetrics m.Model = "claude-opus-4-6" diff --git a/internal/scaffold/fullsend-repo/.github/workflows/triage.yml b/internal/scaffold/fullsend-repo/.github/workflows/triage.yml index 2c102f6af0..327b8b373d 100644 --- a/internal/scaffold/fullsend-repo/.github/workflows/triage.yml +++ b/internal/scaffold/fullsend-repo/.github/workflows/triage.yml @@ -39,3 +39,4 @@ jobs: secrets: FULLSEND_GCP_WIF_PROVIDER: ${{ secrets.FULLSEND_GCP_WIF_PROVIDER }} FULLSEND_GCP_PROJECT_ID: ${{ secrets.FULLSEND_GCP_PROJECT_ID }} + OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }} diff --git a/internal/scaffold/fullsend-repo/templates/shim-per-repo.yaml b/internal/scaffold/fullsend-repo/templates/shim-per-repo.yaml index 3e5f9035c9..16a9f01ae8 100644 --- a/internal/scaffold/fullsend-repo/templates/shim-per-repo.yaml +++ b/internal/scaffold/fullsend-repo/templates/shim-per-repo.yaml @@ -52,6 +52,7 @@ jobs: secrets: FULLSEND_GCP_WIF_PROVIDER: ${{ secrets.FULLSEND_GCP_WIF_PROVIDER }} FULLSEND_GCP_PROJECT_ID: ${{ secrets.FULLSEND_GCP_PROJECT_ID }} + OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }} stop-fix: if: >- diff --git a/internal/scaffold/workflow_call_alignment_test.go b/internal/scaffold/workflow_call_alignment_test.go index 560b28c25d..df07da1b62 100644 --- a/internal/scaffold/workflow_call_alignment_test.go +++ b/internal/scaffold/workflow_call_alignment_test.go @@ -326,6 +326,34 @@ func TestReusableDispatchProjectNumberInput(t *testing.T) { "prioritize job should thread project_number from dispatch inputs") } +// TestOTELHeadersSecretThreading validates that the optional OTLP headers +// secret (#2862) is forwarded along both installation-mode chains to +// reusable-triage.yml. TestWorkflowCallInputAlignment only enforces required +// secrets — an omitted optional forward silently arrives empty, which turns +// into a 401 at authenticated backends instead of failing loudly. +func TestOTELHeadersSecretThreading(t *testing.T) { + const forward = "OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }}" + + cases := []struct { + name string + content func(t *testing.T) []byte + }{ + // consumer: env injection on the agent step + {"reusable-triage.yml", loadRepoFile(".github/workflows/reusable-triage.yml")}, + // per-repo chain: shim → reusable-dispatch → reusable-triage + {"scaffold/templates/shim-per-repo.yaml", loadScaffoldFile("templates/shim-per-repo.yaml")}, + {"reusable-dispatch.yml", loadRepoFile(".github/workflows/reusable-dispatch.yml")}, + // per-org chain: thin caller → reusable-triage + {"scaffold/triage.yml", loadScaffoldFile(".github/workflows/triage.yml")}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert.Contains(t, string(tc.content(t)), forward, + "%s must forward/inject the OTLP headers secret", tc.name) + }) + } +} + // TestReusableDispatchStageConcurrency validates per-role cancel-in-progress groups // on all stage jobs in reusable-dispatch.yml (#981, #982, ADR 0033). func TestReusableDispatchStageConcurrency(t *testing.T) { diff --git a/internal/telemetry/otlp/otlp.go b/internal/telemetry/otlp/otlp.go new file mode 100644 index 0000000000..1d4c24cd24 --- /dev/null +++ b/internal/telemetry/otlp/otlp.go @@ -0,0 +1,129 @@ +// Package otlp implements ADR 0050 Level 2: best-effort OTLP/HTTP export of +// the Level 1 telemetry artifacts. Export is a pure function of the run +// directory — run-telemetry.jsonl and run-summary.json are replayed into +// OpenTelemetry span snapshots and sent through the OTel Go SDK's OTLP/HTTP +// exporter — so the exported trace is, by construction, the same trace the +// local files record. +// +// The package never affects the run: it is inert unless a standard +// OTEL_EXPORTER_OTLP_(TRACES_)ENDPOINT is configured, every network +// operation is bounded by a hard wall-clock deadline, and all failures are +// returned as an error the caller may surface as a warning — the exit code +// and the Level 1 artifacts are never touched. "Non-blocking flush" is +// implemented as a bounded flush: a fire-and-forget goroutine would be +// killed at process exit and silently lose every span. +// +// All endpoint, header, TLS, timeout, and compression configuration is +// delegated to the exporter's standard env-var handling, which matches the +// contract published in docs/guides/infrastructure/distributed-tracing.md. +package otlp + +import ( + "context" + "errors" + "fmt" + "net/url" + "os" + "strings" + "time" + + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +// exportTimeout is the hard wall-clock budget for the entire export +// (exporter construction, send incl. retries, shutdown). It intentionally +// derives from context.Background(), not the run context: at Finalize time +// after a cancellation the run context is already dead, and the traces of +// failed runs are the most valuable ones. Package variable as a test seam. +var exportTimeout = 5 * time.Second + +// newExporter is a seam over exporter construction. Retries are capped well +// below exportTimeout — the SDK defaults (5s initial backoff, 1min max +// elapsed) are tuned for long-lived services, not a CLI at exit. +var newExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return otlptracehttp.New(ctx, otlptracehttp.WithRetry(otlptracehttp.RetryConfig{ + Enabled: true, + InitialInterval: 1 * time.Second, + MaxInterval: 2 * time.Second, + MaxElapsedTime: 4 * time.Second, + })) +} + +// endpointFromEnv returns the configured OTLP traces endpoint, honoring the +// standard precedence: the signal-specific variable wins over the generic +// one. Whitespace-only values count as unset per the OTel spec. +func endpointFromEnv() string { + if v := strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")); v != "" { + return v + } + return strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")) +} + +// Enabled reports whether an OTLP traces endpoint is configured. +func Enabled() bool { + return endpointFromEnv() != "" +} + +// ExportRunDir exports the completed spans recorded in dir's Level 1 +// artifacts. It is a no-op (nil) when no endpoint is configured, when the +// standard kill switches (OTEL_SDK_DISABLED, OTEL_TRACES_EXPORTER=none) are +// set, when the run's trace is unsampled (W3C sampled bit unset — an +// upstream sampling decision must be respected), or when the artifacts are +// missing or contain no completed spans. It never blocks longer than +// exportTimeout and never modifies anything in dir. +func ExportRunDir(dir, serviceVersion string) error { + endpoint := endpointFromEnv() + if endpoint == "" { + return nil + } + if strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_SDK_DISABLED")), "true") { + return nil + } + if strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_TRACES_EXPORTER")), "none") { + return nil + } + // Pre-validate: on a malformed endpoint value the SDK reports to its + // global error handler and silently falls back to localhost:4318 — a + // typo would spray spans at localhost. Refuse instead. + if u, err := url.Parse(endpoint); err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { + return fmt.Errorf("OTLP endpoint %q is not an absolute http(s) URL with a host; export skipped", endpoint) + } + // Only OTLP over HTTP/protobuf is implemented (matches MLflow and the + // published guide). Silently posting protobuf at a gRPC-only endpoint + // fails cryptically, so refuse loudly. + if p := protocolFromEnv(); p != "" && p != "http/protobuf" { + return fmt.Errorf("OTEL_EXPORTER_OTLP_(TRACES_)PROTOCOL %q is not supported (only http/protobuf); export skipped", p) + } + + spans, sampled, err := readRun(dir, serviceVersion) + if err != nil { + return err + } + if !sampled || len(spans) == 0 { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), exportTimeout) + defer cancel() + + exp, err := newExporter(ctx) + if err != nil { + return fmt.Errorf("creating OTLP exporter: %w", err) + } + expErr := exp.ExportSpans(ctx, spans) + shutErr := exp.Shutdown(ctx) + if err := errors.Join(expErr, shutErr); err != nil { + return fmt.Errorf("exporting %d spans: %w", len(spans), err) + } + return nil +} + +// protocolFromEnv returns the configured OTLP protocol (signal-specific +// variable wins), or "" when unset. +func protocolFromEnv() string { + if v := strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL")); v != "" { + return v + } + return strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_PROTOCOL")) +} diff --git a/internal/telemetry/otlp/otlp_test.go b/internal/telemetry/otlp/otlp_test.go new file mode 100644 index 0000000000..a78fd8839c --- /dev/null +++ b/internal/telemetry/otlp/otlp_test.go @@ -0,0 +1,698 @@ +package otlp + +import ( + "bytes" + "compress/gzip" + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + tracepb "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/protobuf/proto" + + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +const ( + testTraceID = "4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d" + testRootID = "a1b2c3d4e5f60718" + testVersion = "test-version" +) + +// otlpSink is an in-process OTLP/HTTP trace collector for tests. It decodes +// each POST body into an ExportTraceServiceRequest and records the request +// headers and paths for assertions. +type otlpSink struct { + srv *httptest.Server + mu sync.Mutex + reqs []*coltracepb.ExportTraceServiceRequest + headers []http.Header + paths []string +} + +func newOTLPSink(t *testing.T) *otlpSink { + t.Helper() + s := &otlpSink{} + s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, err := io.ReadAll(r.Body) + require.NoError(t, err) + if r.Header.Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(raw)) + require.NoError(t, err) + raw, err = io.ReadAll(zr) + require.NoError(t, err) + } + var req coltracepb.ExportTraceServiceRequest + require.NoError(t, proto.Unmarshal(raw, &req)) + s.mu.Lock() + s.reqs = append(s.reqs, &req) + s.headers = append(s.headers, r.Header.Clone()) + s.paths = append(s.paths, r.URL.Path) + s.mu.Unlock() + resp, _ := proto.Marshal(&coltracepb.ExportTraceServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(resp) + })) + t.Cleanup(s.srv.Close) + return s +} + +func (s *otlpSink) requestCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.reqs) +} + +// spans flattens all received resourceSpans/scopeSpans into a single list. +func (s *otlpSink) spans() []*tracepb.Span { + s.mu.Lock() + defer s.mu.Unlock() + var out []*tracepb.Span + for _, req := range s.reqs { + for _, rs := range req.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + out = append(out, ss.GetSpans()...) + } + } + } + return out +} + +func (s *otlpSink) resourceAttrs() map[string]string { + s.mu.Lock() + defer s.mu.Unlock() + out := map[string]string{} + for _, req := range s.reqs { + for _, rs := range req.GetResourceSpans() { + for _, kv := range rs.GetResource().GetAttributes() { + out[kv.GetKey()] = kv.GetValue().GetStringValue() + } + } + } + return out +} + +// spanAttrs returns the attribute map of the first received span with name. +func (s *otlpSink) spanAttrs(t *testing.T, name string) map[string]*commonpb.AnyValue { + t.Helper() + for _, sp := range s.spans() { + if sp.GetName() == name { + out := map[string]*commonpb.AnyValue{} + for _, kv := range sp.GetAttributes() { + out[kv.GetKey()] = kv.GetValue() + } + return out + } + } + t.Fatalf("no span named %q received", name) + return nil +} + +// writeRunFixture drives a real Level 1 Recorder to produce genuine +// run-telemetry.jsonl and run-summary.json artifacts in dir. +func writeRunFixture(t *testing.T, dir string, tc telemetry.TraceContext, exitCode int) { + t.Helper() + r := telemetry.New(dir, tc, "code", "octo/repo#2862", time.Now().Add(-2*time.Second)) + sb := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sb, "ok", nil) + ag := r.StartSpan("agent", "", map[string]any{"iteration": 1}) + r.EndSpan(ag, "ok", map[string]any{ + "iteration": 1, + "exit_code": 0, + "gen_ai.request.model": "claude-opus-4-6", + "gen_ai.usage.input_tokens": 1234, + "fullsend.cost_usd": 0.34, + "cache_hit": true, + }) + r.Finalize(exitCode) +} + +func defaultTC() telemetry.TraceContext { + return telemetry.TraceContext{TraceID: testTraceID, RootSpanID: testRootID} +} + +// --- Gating --- + +func TestEnabled(t *testing.T) { + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + assert.False(t, Enabled(), "no endpoint => disabled") + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318") + assert.True(t, Enabled(), "generic endpoint => enabled") + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://localhost:4318/v1/traces") + assert.True(t, Enabled(), "signal-specific endpoint => enabled") + + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", " \t ") + assert.False(t, Enabled(), "whitespace-only endpoint is unset per the OTel spec") +} + +func TestExportRunDir_NoEndpoint_NoOpAndNoNetwork(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount(), "no endpoint => zero network activity") +} + +func TestExportRunDir_BaseEndpointAppendsV1Traces(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "/v1/traces", sink.paths[0], "generic endpoint gets /v1/traces appended (published contract)") +} + +func TestExportRunDir_TracesEndpointUsedAsIs(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:1/ignored") // must not be used + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/custom/ingest/path") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "/custom/ingest/path", sink.paths[0], + "signal-specific endpoint wins and is used as-is, no path appended (published contract)") +} + +func TestExportRunDir_MalformedEndpointFailsOpen(t *testing.T) { + sink := newOTLPSink(t) + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + for _, bad := range []string{"://bad", "not a url", "ftp://host:4318", "http://"} { + t.Run(bad, func(t *testing.T) { + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", bad) + err := ExportRunDir(dir, testVersion) + assert.Error(t, err, "malformed endpoint must surface an error for the caller's warning") + assert.Equal(t, 0, sink.requestCount(), + "must not fall back to the SDK default localhost endpoint") + assertL1Intact(t, dir) + }) + } +} + +func TestExportRunDir_HeadersReachWire(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + // Percent-encoded space + base64 padding: the exact form the MLflow + // runbook uses (headers are parsed baggage-style and URL-decoded). + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "authorization=Basic%20dXNlcjpwYXNz,x-mlflow-experiment-id=42") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "Basic dXNlcjpwYXNz", sink.headers[0].Get("Authorization")) + assert.Equal(t, "42", sink.headers[0].Get("x-mlflow-experiment-id")) +} + +func TestExportRunDir_TracesHeadersPrecedence(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "x-mlflow-experiment-id=1") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "x-mlflow-experiment-id=2") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "2", sink.headers[0].Get("x-mlflow-experiment-id"), + "signal-specific headers win (published contract)") +} + +func TestExportRunDir_GzipCompression(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_COMPRESSION", "gzip") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "gzip", sink.headers[0].Get("Content-Encoding")) + assert.Len(t, sink.spans(), 3, "compressed payload still decodes to all spans") +} + +func TestExportRunDir_SDKDisabled(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_SDK_DISABLED", "true") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount(), "OTEL_SDK_DISABLED=true must be honored") +} + +func TestExportRunDir_TracesExporterNone(t *testing.T) { + // The OTel spec recommends case-insensitive comparison of env var values. + for _, v := range []string{"none", "NONE", "None"} { + t.Run(v, func(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_TRACES_EXPORTER", v) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount(), "OTEL_TRACES_EXPORTER=%s must be honored", v) + }) + } +} + +func TestExportRunDir_UnsupportedProtocolRejected(t *testing.T) { + // Only OTLP over http/protobuf is implemented. Posting protobuf at a + // gRPC-only endpoint fails cryptically, so the mismatch is refused + // loudly instead — for either protocol env var. + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc") + assert.Error(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount()) + + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "grpc") + assert.Error(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount()) + + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "http/protobuf") + assert.NoError(t, ExportRunDir(dir, testVersion), "explicit http/protobuf proceeds") + assert.Equal(t, 1, sink.requestCount()) +} + +func TestExportRunDir_ExporterConstructionErrorSurfaced(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + orig := newExporter + newExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + return nil, errors.New("boom") + } + defer func() { newExporter = orig }() + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + err := ExportRunDir(dir, testVersion) + assert.ErrorContains(t, err, "boom") + assert.Equal(t, 0, sink.requestCount()) + assertL1Intact(t, dir) +} + +func TestExportRunDir_MalformedResourceEnvStillExports(t *testing.T) { + // A garbage OTEL_RESOURCE_ATTRIBUTES must not break export — the + // resource degrades to fullsend's defaults. + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "%%%%not=valid=pairs%%%%,,,=") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + require.Equal(t, 1, sink.requestCount()) + assert.Equal(t, "fullsend", sink.resourceAttrs()["service.name"]) +} + +func TestExportRunDir_UnsampledRunSkipsExport(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + tc := defaultTC() + tc.Flags = "00" // upstream said: do not sample + writeRunFixture(t, dir, tc, 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount(), + "an upstream-unsampled trace must not be exported (ParentBased semantics)") + assertL1Intact(t, dir) +} + +// --- Fidelity --- + +func TestExportRunDir_SpanIdentityPreserved(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + spans := sink.spans() + require.Len(t, spans, 3, "run + sandbox_create + agent") + + events := readEvents(t, dir) + starts := map[string]map[string]any{} // span_id (hex) -> span_start line + ends := map[string]map[string]any{} + for _, e := range events { + id, _ := e["span_id"].(string) + if e["event"] == "span_start" { + starts[id] = e + } else { + ends[id] = e + } + } + + byName := map[string]*tracepb.Span{} + for _, sp := range spans { + byName[sp.GetName()] = sp + // Identity: ids on the wire are byte-equal to the L1 file's hex ids. + assert.Equal(t, testTraceID, hexid(sp.GetTraceId()), "trace id preserved on %s", sp.GetName()) + st, ok := starts[hexid(sp.GetSpanId())] + require.True(t, ok, "wire span %s must exist in the L1 file", sp.GetName()) + // Timestamps: nanosecond-exact against the file's RFC3339Nano values. + startTS, err := time.Parse(time.RFC3339Nano, st["ts"].(string)) + require.NoError(t, err) + assert.Equal(t, uint64(startTS.UnixNano()), sp.GetStartTimeUnixNano(), "start time exact on %s", sp.GetName()) + endTS, err := time.Parse(time.RFC3339Nano, ends[hexid(sp.GetSpanId())]["ts"].(string)) + require.NoError(t, err) + assert.Equal(t, uint64(endTS.UnixNano()), sp.GetEndTimeUnixNano(), "end time exact on %s", sp.GetName()) + } + + root := byName["run"] + require.NotNil(t, root) + assert.Equal(t, testRootID, hexid(root.GetSpanId())) + assert.Empty(t, root.GetParentSpanId(), "local trace root has no parent") + assert.Equal(t, tracepb.Span_SPAN_KIND_INTERNAL, root.GetKind()) + for _, name := range []string{"sandbox_create", "agent"} { + child := byName[name] + require.NotNil(t, child, "%s span exported", name) + assert.Equal(t, testRootID, hexid(child.GetParentSpanId()), "%s parents to root", name) + assert.Equal(t, tracepb.Span_SPAN_KIND_INTERNAL, child.GetKind()) + } +} + +func TestExportRunDir_RemoteParentContinuesTrace(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + tc := defaultTC() + tc.ParentSpanID = "beefbeefbeefbeef" // inbound TRACEPARENT parent (issue #2779) + writeRunFixture(t, dir, tc, 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + for _, sp := range sink.spans() { + if sp.GetName() != "run" { + continue + } + assert.Equal(t, "beefbeefbeefbeef", hexid(sp.GetParentSpanId()), + "root span must join the inbound parent trace on the wire") + assert.Equal(t, tracepb.Span_SPAN_KIND_CONSUMER, sp.GetKind(), + "dispatched run roots are CONSUMER spans (published contract)") + return + } + t.Fatal("run span not exported") +} + +func TestExportRunDir_AttributeTypesPreserved(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + attrs := sink.spanAttrs(t, "agent") + + assert.Equal(t, int64(1234), attrs["gen_ai.usage.input_tokens"].GetIntValue(), "integral numbers stay Int64") + assert.Equal(t, int64(1), attrs["iteration"].GetIntValue()) + assert.InDelta(t, 0.34, attrs["fullsend.cost_usd"].GetDoubleValue(), 1e-9, "fractional numbers stay Double") + assert.Equal(t, "claude-opus-4-6", attrs["gen_ai.request.model"].GetStringValue()) + assert.True(t, attrs["cache_hit"].GetBoolValue()) +} + +func TestExportRunDir_WorkItemIDOnEverySpan(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + spans := sink.spans() + require.NotEmpty(t, spans) + for _, sp := range spans { + found := false + for _, kv := range sp.GetAttributes() { + if kv.GetKey() == "fullsend.work_item_id" { + found = true + assert.Equal(t, "octo/repo#2862", kv.GetValue().GetStringValue()) + } + } + assert.True(t, found, "fullsend.work_item_id on span %s (primary correlation key, ADR 0050)", sp.GetName()) + } +} + +func TestExportRunDir_StatusMapping(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 2) // nonzero exit => root span status "error" + + require.NoError(t, ExportRunDir(dir, testVersion)) + for _, sp := range sink.spans() { + switch sp.GetName() { + case "run": + assert.Equal(t, tracepb.Status_STATUS_CODE_ERROR, sp.GetStatus().GetCode(), "error status maps to ERROR") + default: + assert.Equal(t, tracepb.Status_STATUS_CODE_OK, sp.GetStatus().GetCode(), "ok status maps to OK on %s", sp.GetName()) + } + } +} + +func TestExportRunDir_ResourceServiceNameAndVersion(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + res := sink.resourceAttrs() + assert.Equal(t, "fullsend", res["service.name"]) + assert.Equal(t, testVersion, res["service.version"]) +} + +func TestExportRunDir_ServiceNameEnvOverride(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_SERVICE_NAME", "my-deployment") + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, "my-deployment", sink.resourceAttrs()["service.name"], + "OTEL_SERVICE_NAME overrides the default (standard resource env)") +} + +// --- Fail-open --- + +// assertL1Intact asserts both Level 1 artifacts exist and parse. +func assertL1Intact(t *testing.T, dir string) { + t.Helper() + events := readEvents(t, dir) + assert.NotEmpty(t, events, "run-telemetry.jsonl intact") + _, err := os.Stat(filepath.Join(dir, telemetry.SummaryFile)) + assert.NoError(t, err, "run-summary.json intact") +} + +// exportWithBudget runs ExportRunDir against endpoint with a pinned export +// budget and asserts the fail-open invariants: an error is returned (for the +// caller's single warning line), the call is time-bounded, and the Level 1 +// artifacts are untouched. The 15s bound is ~10x the pinned budget so loaded +// CI runners cannot flake it, while a regression to the SDK's default +// unbounded shutdown blows through it reliably. +func exportWithBudget(t *testing.T, dir, endpoint string) { + t.Helper() + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", endpoint) + t.Setenv("OTEL_EXPORTER_OTLP_TIMEOUT", "1000") // 1s per-request budget + + orig := exportTimeout + exportTimeout = 1500 * time.Millisecond + defer func() { exportTimeout = orig }() + + start := time.Now() + err := ExportRunDir(dir, testVersion) + elapsed := time.Since(start) + + assert.Error(t, err, "endpoint pathology must surface an error") + assert.Less(t, elapsed, 15*time.Second, "export must be hard-bounded, got %v", elapsed) + assertL1Intact(t, dir) +} + +func TestExportFailOpen_TCPBlackHole(t *testing.T) { + // Accepts TCP connections and never reads or responds — the nastiest + // endpoint pathology (issue #2862's mandated test). + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + var conns []net.Conn + var mu sync.Mutex + done := make(chan struct{}) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + mu.Lock() + conns = append(conns, c) + mu.Unlock() + select { + case <-done: + return + default: + } + } + }() + t.Cleanup(func() { + close(done) + _ = ln.Close() + mu.Lock() + for _, c := range conns { + _ = c.Close() + } + mu.Unlock() + }) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + exportWithBudget(t, dir, "http://"+ln.Addr().String()) +} + +func TestExportFailOpen_HangingHTTPServer(t *testing.T) { + // Accepts the HTTP request but never responds. + block := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-block + })) + t.Cleanup(func() { + close(block) + srv.Close() + }) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + exportWithBudget(t, dir, srv.URL) +} + +func TestExportFailOpen_ConnectionRefused(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := ln.Addr().String() + require.NoError(t, ln.Close()) // dead address => RST on connect + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + exportWithBudget(t, dir, "http://"+addr) +} + +func TestExportFailOpen_DNSFailure(t *testing.T) { + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + // .invalid is RFC 2606-reserved: never resolves, deterministically. + exportWithBudget(t, dir, "http://fullsend-l2.invalid:4318") +} + +func TestExportFailOpen_HTTP4xx(t *testing.T) { + // 401 is exactly what an unauthenticated MLflow returns; 400 is a + // missing x-mlflow-experiment-id. Neither is retryable per the OTLP + // spec, so the export fails fast without a retry storm. + for _, code := range []int{401, 400} { + t.Run(http.StatusText(code), func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(code) + })) + t.Cleanup(srv.Close) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + exportWithBudget(t, dir, srv.URL) + }) + } +} + +func TestExportFailOpen_Retry503ThenDelivered(t *testing.T) { + // A transient 503 with Retry-After must be retried (within the bounded + // budget) and the spans eventually delivered. + var mu sync.Mutex + calls := 0 + var got *coltracepb.ExportTraceServiceRequest + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + calls++ + if calls == 1 { + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusServiceUnavailable) + return + } + raw, _ := io.ReadAll(r.Body) + var req coltracepb.ExportTraceServiceRequest + _ = proto.Unmarshal(raw, &req) + got = &req + resp, _ := proto.Marshal(&coltracepb.ExportTraceServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(resp) + })) + t.Cleanup(srv.Close) + + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", srv.URL) + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + start := time.Now() + err := ExportRunDir(dir, testVersion) + require.NoError(t, err, "transient 503 must be retried to success") + assert.Less(t, time.Since(start), 15*time.Second) + + mu.Lock() + defer mu.Unlock() + assert.GreaterOrEqual(t, calls, 2, "must have retried") + require.NotNil(t, got) + spans := 0 + for _, rs := range got.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + spans += len(ss.GetSpans()) + } + } + assert.Equal(t, 3, spans, "all spans delivered after retry") +} diff --git a/internal/telemetry/otlp/replay.go b/internal/telemetry/otlp/replay.go new file mode 100644 index 0000000000..e874486e8d --- /dev/null +++ b/internal/telemetry/otlp/replay.go @@ -0,0 +1,289 @@ +package otlp + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/sdk/instrumentation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" + + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +// scopeName identifies the instrumentation scope on exported spans. +const scopeName = "github.com/fullsend-ai/fullsend/internal/telemetry" + +// event mirrors the Level 1 eventRecord NDJSON line shape. It is declared +// locally so internal/telemetry keeps its internals unexported and stays +// dependency-free. +type event struct { + V int `json:"v"` + Event string `json:"event"` + TraceID string `json:"trace_id"` + SpanID string `json:"span_id"` + Parent string `json:"parent"` + Name string `json:"name"` + TS string `json:"ts"` + WorkItemID string `json:"fullsend.work_item_id"` + Status string `json:"status"` + Attrs map[string]any `json:"attrs"` +} + +// readRun parses dir's Level 1 artifacts into exportable span snapshots and +// reports whether the run's trace is sampled (from the summary traceparent's +// W3C flags). Missing artifacts mean "nothing to export" (nil spans, no +// error): a missing summary means the run never finalized — the NDJSON file +// on disk is the crash-forensics record and stays local. Malformed lines, +// invalid ids, and span_starts without a span_end are skipped: OTLP spans +// require a complete identity and an end time, and a partial artifact must +// never block export of the well-formed remainder. +func readRun(dir, serviceVersion string) ([]sdktrace.ReadOnlySpan, bool, error) { + summary, err := os.ReadFile(filepath.Join(dir, telemetry.SummaryFile)) + if err != nil { + return nil, false, nil // never finalized (or no telemetry at all) + } + if !summarySampled(summary) { + return nil, false, nil // upstream said: do not sample + } + + f, err := os.Open(filepath.Join(dir, telemetry.TelemetryFile)) + if err != nil { + return nil, false, nil + } + defer f.Close() + + starts := map[string]event{} + ends := map[string]event{} + var order []string // span ids in file (start-line) order, for determinism + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var e event + dec := json.NewDecoder(strings.NewReader(line)) + dec.UseNumber() // keep numbers exact: 2^53+1 must not round through float64 + if dec.Decode(&e) != nil || e.SpanID == "" { + continue + } + switch e.Event { + case "span_start": + if _, dup := starts[e.SpanID]; !dup { + starts[e.SpanID] = e + order = append(order, e.SpanID) + } + case "span_end": + ends[e.SpanID] = e + } + } + if err := sc.Err(); err != nil { + return nil, false, fmt.Errorf("reading %s: %w", telemetry.TelemetryFile, err) + } + + res := buildResource(serviceVersion) + scope := instrumentation.Scope{Name: scopeName, Version: serviceVersion} + + var spans []sdktrace.ReadOnlySpan + for _, id := range order { + start := starts[id] + end, finished := ends[id] + if !finished { + continue // in-flight at crash: no end time, stays file-only + } + stub, ok := buildStub(start, end, starts, res, scope) + if !ok { + continue + } + spans = append(spans, stub.Snapshot()) + } + return spans, true, nil +} + +// summarySampled extracts the W3C sampled bit from the summary's traceparent. +// Artifacts without a parseable traceparent count as sampled — Level 1 always +// writes one, so this only affects hand-edited files. +func summarySampled(summary []byte) bool { + var s struct { + Traceparent string `json:"traceparent"` + } + if json.Unmarshal(summary, &s) != nil { + return true + } + _, _, flags, ok := telemetry.ParseTraceParent(s.Traceparent) + if !ok { + return true + } + bits, err := strconv.ParseUint(flags, 16, 8) + if err != nil { + return true + } + return bits&0x01 != 0 +} + +// buildStub converts a span_start/span_end pair into a span snapshot stub. +// tracetest.SpanStub is the SDK's only public ReadOnlySpan constructor +// (ReadOnlySpan has an unexported method), and the replay design needs exact +// control of ids and timestamps so the exported span is identical to the one +// in the Level 1 file. +func buildStub(start, end event, starts map[string]event, res *resource.Resource, scope instrumentation.Scope) (tracetest.SpanStub, bool) { + tid, err := trace.TraceIDFromHex(start.TraceID) + if err != nil { + return tracetest.SpanStub{}, false + } + sid, err := trace.SpanIDFromHex(start.SpanID) + if err != nil { + return tracetest.SpanStub{}, false + } + startTime, err := time.Parse(time.RFC3339Nano, start.TS) + if err != nil { + return tracetest.SpanStub{}, false + } + endTime, err := time.Parse(time.RFC3339Nano, end.TS) + if err != nil { + return tracetest.SpanStub{}, false + } + + // Parent: a span id recorded in this file is a local parent; anything + // else is the remote parent adopted from an inbound TRACEPARENT (#2779). + // Only export happens when sampled, so flags are always FlagsSampled. + parent := trace.SpanContext{} + kind := trace.SpanKindInternal + if start.Parent != "" { + psid, err := trace.SpanIDFromHex(start.Parent) + if err != nil { + return tracetest.SpanStub{}, false + } + _, local := starts[start.Parent] + parent = trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: tid, SpanID: psid, TraceFlags: trace.FlagsSampled, Remote: !local, + }) + if !local { + // A dispatched run's root span consumes work from the parent + // pipeline (matches the published SpanKind contract). + kind = trace.SpanKindConsumer + } + } + + status := sdktrace.Status{Code: codes.Unset} + switch end.Status { + case "ok": + status.Code = codes.Ok + case "error": + status.Code = codes.Error + } + + return tracetest.SpanStub{ + Name: start.Name, + SpanContext: trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: tid, SpanID: sid, TraceFlags: trace.FlagsSampled, + }), + Parent: parent, + SpanKind: kind, + StartTime: startTime, + EndTime: endTime, + Attributes: mergedAttrs(start, end), + Status: status, + Resource: res, + InstrumentationScope: scope, + }, true +} + +// mergedAttrs merges span_start and span_end attributes (end wins on key +// conflicts) plus the per-line work-item id, in sorted key order for +// deterministic output. +func mergedAttrs(start, end event) []attribute.KeyValue { + merged := map[string]any{} + for k, v := range start.Attrs { + merged[k] = v + } + for k, v := range end.Attrs { + merged[k] = v + } + if wi := firstNonEmpty(end.WorkItemID, start.WorkItemID); wi != "" { + merged["fullsend.work_item_id"] = wi + } + + keys := make([]string, 0, len(merged)) + for k := range merged { + keys = append(keys, k) + } + sort.Strings(keys) + + attrs := make([]attribute.KeyValue, 0, len(keys)) + for _, k := range keys { + if kv, ok := attrKV(k, merged[k]); ok { + attrs = append(attrs, kv) + } + } + return attrs +} + +// attrKV maps a decoded JSON attribute value onto the matching OTel +// attribute type. Nothing is dropped and nothing panics — an attribute of an +// unexpected shape degrades to its string form rather than disappearing. +func attrKV(k string, v any) (attribute.KeyValue, bool) { + switch val := v.(type) { + case nil: + return attribute.KeyValue{}, false + case string: + return attribute.String(k, val), true + case bool: + return attribute.Bool(k, val), true + case json.Number: + if i, err := val.Int64(); err == nil { + return attribute.Int64(k, i), true + } + if f, err := val.Float64(); err == nil { + return attribute.Float64(k, f), true + } + return attribute.String(k, val.String()), true + default: + return attribute.String(k, fmt.Sprint(val)), true + } +} + +// buildResource assembles the export Resource: fullsend's identity plus any +// standard env overrides (OTEL_SERVICE_NAME, OTEL_RESOURCE_ATTRIBUTES) — +// env detectors run last so operator configuration wins. Plain attribute +// keys are used instead of semconv constants: newer semconv versions renamed +// keys this package's contract (ADR 0050) pins, and service.name/version are +// stable strings. +func buildResource(serviceVersion string) *resource.Resource { + res, err := resource.New(context.Background(), + resource.WithAttributes( + attribute.String("service.name", "fullsend"), + attribute.String("service.version", serviceVersion), + ), + resource.WithFromEnv(), + ) + if err != nil || res == nil { + return resource.NewSchemaless( + attribute.String("service.name", "fullsend"), + attribute.String("service.version", serviceVersion), + ) + } + return res +} + +// firstNonEmpty returns the first non-empty string. +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} diff --git a/internal/telemetry/otlp/replay_test.go b/internal/telemetry/otlp/replay_test.go new file mode 100644 index 0000000000..69f5eeea72 --- /dev/null +++ b/internal/telemetry/otlp/replay_test.go @@ -0,0 +1,295 @@ +package otlp + +import ( + "bufio" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +// readEvents reads run-telemetry.jsonl from dir into decoded maps. +func readEvents(t *testing.T, dir string) []map[string]any { + t.Helper() + f, err := os.Open(filepath.Join(dir, telemetry.TelemetryFile)) + require.NoError(t, err) + defer f.Close() + var out []map[string]any + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.TrimSpace(sc.Text()) == "" { + continue + } + var m map[string]any + require.NoError(t, json.Unmarshal(sc.Bytes(), &m)) + out = append(out, m) + } + require.NoError(t, sc.Err()) + return out +} + +// hexid renders an OTLP id byte slice as lowercase hex ("" for empty/zero). +func hexid(b []byte) string { + if len(b) == 0 { + return "" + } + return hex.EncodeToString(b) +} + +func TestReadRun_MissingTelemetryFileIsNoOp(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // Endpoint configured but the run dir has no L1 artifacts at all (e.g. a + // disabled recorder): nothing to export, no error, no network. + require.NoError(t, ExportRunDir(t.TempDir(), testVersion)) + assert.Equal(t, 0, sink.requestCount()) +} + +func TestReadRun_MissingSummarySkipsExport(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // A telemetry file without a summary means the run never finalized + // (crash). The L1 file is the forensic record; export skips quietly. + dir := t.TempDir() + r := telemetry.New(dir, defaultTC(), "code", "wi", time.Now()) + sp := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sp, "ok", nil) + // no Finalize + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount()) +} + +func TestReadRun_SummaryWithoutTelemetryFileIsNoOp(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // The converse of a missing summary: a summary with no telemetry file + // (hand-pruned artifacts). Nothing to export, no error, no network. + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), + []byte(`{"traceparent":"00-`+testTraceID+`-a1b2c3d4e5f60718-01"}`), 0o644)) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount()) +} + +func TestReadRun_UnpairedStartSkipped(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // A span_start without a span_end (in-flight at crash, but the run still + // finalized) must be skipped: OTLP spans require an end time. The L1 + // line on disk remains the forensic record. + dir := t.TempDir() + r := telemetry.New(dir, defaultTC(), "code", "wi", time.Now()) + _ = r.StartSpan("agent", "", nil) // never ended + sb := r.StartSpan("sandbox_create", "", nil) + r.EndSpan(sb, "ok", nil) + r.Finalize(0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + names := []string{} + for _, sp := range sink.spans() { + names = append(names, sp.GetName()) + } + assert.ElementsMatch(t, []string{"run", "sandbox_create"}, names, + "unpaired agent span_start skipped; paired spans exported") +} + +func TestReadRun_MalformedLinesSkipped(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + // Corrupt the file: append blank lines, garbage, and a torn line (crash + // artifacts). + f, err := os.OpenFile(filepath.Join(dir, telemetry.TelemetryFile), os.O_APPEND|os.O_WRONLY, 0o644) + require.NoError(t, err) + _, err = f.WriteString("\n \nnot json at all\n" + `{"v":1,"event":"span_start","trace_id":"x","spa`) + require.NoError(t, err) + require.NoError(t, f.Close()) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Len(t, sink.spans(), 3, "well-formed spans exported; garbage skipped") +} + +func TestReadRun_OversizedLineReportsError(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + + // A line beyond the scanner's 1 MiB cap is an I/O-level failure, not a + // skippable bad record: the error is reported (the runner warns and moves + // on) and nothing is exported. + f, err := os.OpenFile(filepath.Join(dir, telemetry.TelemetryFile), os.O_APPEND|os.O_WRONLY, 0o644) + require.NoError(t, err) + _, err = f.WriteString(strings.Repeat("a", 2*1024*1024) + "\n") + require.NoError(t, err) + require.NoError(t, f.Close()) + + require.Error(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount()) +} + +func TestReadRun_InvalidIDsSkipped(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // Hand-written artifacts with a non-hex trace id: the affected span is + // dropped rather than exported with a garbage identity. + dir := t.TempDir() + lines := []string{ + `{"v":1,"event":"span_start","trace_id":"ZZZ","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:00Z","fullsend.work_item_id":"wi"}`, + `{"v":1,"event":"span_end","trace_id":"ZZZ","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:01Z","fullsend.work_item_id":"wi","status":"ok"}`, + } + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.TelemetryFile), + []byte(strings.Join(lines, "\n")+"\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), + []byte(`{"v":1,"trace_id":"ZZZ","traceparent":"00-4f3a9c1b2d8e4a7c9f0b1e2d3c4a5b6d-a1b2c3d4e5f60718-01","exit_code":0}`), 0o644)) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 0, sink.requestCount(), "no valid spans => nothing to send") +} + +func TestReadRun_MalformedSummaryDefaultsToSampled(t *testing.T) { + // Level 1 always writes a parseable summary; a hand-mangled one must not + // silently suppress export — default is sampled. + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + writeRunFixture(t, dir, defaultTC(), 0) + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), []byte("{not json"), 0o644)) + + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 1, sink.requestCount()) + + // Same for a summary whose traceparent doesn't parse. + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), + []byte(`{"traceparent":"garbage"}`), 0o644)) + require.NoError(t, ExportRunDir(dir, testVersion)) + assert.Equal(t, 2, sink.requestCount()) +} + +func TestReadRun_OddSpanLinesSkipped(t *testing.T) { + // Hand-written artifact pathologies beyond invalid trace ids: bad span + // ids, bad parent ids, unparseable timestamps, duplicate span_starts. + // Each bad span is dropped; the good one survives. + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + const tid = testTraceID + lines := []string{ + // good root, duplicated span_start (second ignored) + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:00Z","fullsend.work_item_id":"wi"}`, + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"a1b2c3d4e5f60718","parent":"","name":"run-dup","ts":"2026-07-02T10:00:00Z"}`, + // bad span id + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"zz","parent":"","name":"bad-sid","ts":"2026-07-02T10:00:00Z"}`, + `{"v":1,"event":"span_end","trace_id":"` + tid + `","span_id":"zz","parent":"","name":"bad-sid","ts":"2026-07-02T10:00:01Z","status":"ok"}`, + // bad parent id + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"bbbbbbbbbbbbbbbb","parent":"nothex","name":"bad-parent","ts":"2026-07-02T10:00:00Z"}`, + `{"v":1,"event":"span_end","trace_id":"` + tid + `","span_id":"bbbbbbbbbbbbbbbb","parent":"nothex","name":"bad-parent","ts":"2026-07-02T10:00:01Z","status":"ok"}`, + // bad timestamps + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"cccccccccccccccc","parent":"","name":"bad-ts","ts":"yesterday"}`, + `{"v":1,"event":"span_end","trace_id":"` + tid + `","span_id":"cccccccccccccccc","parent":"","name":"bad-ts","ts":"2026-07-02T10:00:01Z","status":"ok"}`, + // bad end timestamp (start side fine) + `{"v":1,"event":"span_start","trace_id":"` + tid + `","span_id":"dddddddddddddddd","parent":"","name":"bad-end-ts","ts":"2026-07-02T10:00:00Z"}`, + `{"v":1,"event":"span_end","trace_id":"` + tid + `","span_id":"dddddddddddddddd","parent":"","name":"bad-end-ts","ts":"tomorrow","status":"ok"}`, + // good root end; span_end without work item id exercises the + // start-side fallback + `{"v":1,"event":"span_end","trace_id":"` + tid + `","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:02Z","status":"ok"}`, + } + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.TelemetryFile), + []byte(strings.Join(lines, "\n")+"\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), + []byte(`{"traceparent":"00-`+tid+`-a1b2c3d4e5f60718-01"}`), 0o644)) + + require.NoError(t, ExportRunDir(dir, testVersion)) + spans := sink.spans() + require.Len(t, spans, 1, "only the well-formed span survives") + assert.Equal(t, "run", spans[0].GetName(), "first span_start wins over the duplicate") + attrs := sink.spanAttrs(t, "run") + assert.Equal(t, "wi", attrs["fullsend.work_item_id"].GetStringValue(), + "work item id falls back to the span_start line") +} + +func TestReadRun_UnexpectedAttrShapesDegradeToStrings(t *testing.T) { + // Arrays and nested objects are not part of the L1 schema, but if they + // ever appear they must degrade to strings — never dropped, never a panic. + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + r := telemetry.New(dir, defaultTC(), "code", "wi", time.Now()) + sp := r.StartSpan("agent", "", nil) + r.EndSpan(sp, "ok", map[string]any{ + "arr": []string{"a", "b"}, + "nested": map[string]any{"k": 1}, + "nothing": nil, + }) + r.Finalize(0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + attrs := sink.spanAttrs(t, "agent") + assert.NotEmpty(t, attrs["arr"].GetStringValue(), "array degrades to its string form") + assert.NotEmpty(t, attrs["nested"].GetStringValue(), "object degrades to its string form") + _, present := attrs["nothing"] + assert.False(t, present, "null attributes carry no information and are omitted") +} + +func TestReadRun_NumberFidelity(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // Large integers must survive the JSON round-trip without float64 + // mangling: 9007199254740993 (2^53+1) is not representable as float64. + dir := t.TempDir() + r := telemetry.New(dir, defaultTC(), "code", "wi", time.Now()) + sp := r.StartSpan("agent", "", nil) + r.EndSpan(sp, "ok", map[string]any{"big": int64(9007199254740993), "neg": -7, "frac": 0.5}) + r.Finalize(0) + + require.NoError(t, ExportRunDir(dir, testVersion)) + attrs := sink.spanAttrs(t, "agent") + assert.Equal(t, int64(9007199254740993), attrs["big"].GetIntValue(), "2^53+1 preserved exactly") + assert.Equal(t, int64(-7), attrs["neg"].GetIntValue()) + assert.InDelta(t, 0.5, attrs["frac"].GetDoubleValue(), 1e-12) +} + +func TestReadRun_NumberBeyondFloat64DegradesToString(t *testing.T) { + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL) + + // A number that fits neither int64 nor float64 (hand-written artifact; L1 + // never emits one) degrades to its literal string — never dropped, never + // a panic. + dir := t.TempDir() + lines := []string{ + `{"v":1,"event":"span_start","trace_id":"` + testTraceID + `","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:00Z","fullsend.work_item_id":"wi"}`, + `{"v":1,"event":"span_end","trace_id":"` + testTraceID + `","span_id":"a1b2c3d4e5f60718","parent":"","name":"run","ts":"2026-07-02T10:00:01Z","status":"ok","attrs":{"huge":1e400}}`, + } + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.TelemetryFile), + []byte(strings.Join(lines, "\n")+"\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, telemetry.SummaryFile), + []byte(`{"traceparent":"00-`+testTraceID+`-a1b2c3d4e5f60718-01"}`), 0o644)) + + require.NoError(t, ExportRunDir(dir, testVersion)) + attrs := sink.spanAttrs(t, "run") + assert.Equal(t, "1e400", attrs["huge"].GetStringValue()) +} diff --git a/internal/telemetry/recorder.go b/internal/telemetry/recorder.go index 34499b0c00..a1acd41e4a 100644 --- a/internal/telemetry/recorder.go +++ b/internal/telemetry/recorder.go @@ -152,7 +152,13 @@ func New(dir string, tc TraceContext, agent, workItemID string, start time.Time) r.emit(eventRecord{ V: SchemaVersion, Event: "span_start", TraceID: r.traceID, SpanID: r.rootSpanID, Parent: r.parentSpanID, Name: "run", TS: start.UTC().Format(time.RFC3339Nano), - WorkItemID: workItemID, Attrs: map[string]any{"agent": agent}, + WorkItemID: workItemID, Attrs: map[string]any{ + // OTEL GenAI semconv identity per ADR 0050; the bare "agent" key + // predates it and stays for consumers of the Level 1 schema. + "agent": agent, + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.name": agent, + }, }) r.mu.Unlock() return r diff --git a/internal/telemetry/recorder_test.go b/internal/telemetry/recorder_test.go index 1af5775126..60b94f1225 100644 --- a/internal/telemetry/recorder_test.go +++ b/internal/telemetry/recorder_test.go @@ -411,6 +411,24 @@ func TestRecorder_SetMetricsNilAndDisabledSafe(t *testing.T) { assert.NotPanics(t, func() { disabled.SetMetrics(RunMetrics{InputTokens: 1}) }) } +func TestRecorder_RootSpanCarriesGenAIIdentity(t *testing.T) { + // ADR 0050 names gen_ai.operation.name and gen_ai.agent.name on run + // spans; they live at the source (Level 1) so the local file and any + // Level 2 export stay two views of one truth. + dir := t.TempDir() + r := New(dir, TraceContext{TraceID: testTraceID, RootSpanID: testRootID}, + "triage", "wi", time.Now()) + r.Finalize(0) + + lines := readLines(t, filepath.Join(dir, TelemetryFile)) + require.NotEmpty(t, lines) + attrs, ok := lines[0]["attrs"].(map[string]any) + require.True(t, ok, "root span_start has attrs") + assert.Equal(t, "invoke_agent", attrs["gen_ai.operation.name"]) + assert.Equal(t, "triage", attrs["gen_ai.agent.name"]) + assert.Equal(t, "triage", attrs["agent"], "existing key kept for consumers of the L1 schema") +} + func TestRecorder_RootSpanRecordsRemoteParent(t *testing.T) { // When the run adopts an inbound TRACEPARENT (issue #2779), the root span // must record the inbound span-id as its parent so the exported trace