From b331587e4667cfaf7d0e1b745c24a851f020fd63 Mon Sep 17 00:00:00 2001 From: Hector Martinez Date: Fri, 24 Jul 2026 10:51:01 +0200 Subject: [PATCH] fix(telemetry)!: respect OTEL variables default behaviour Fixed a bug in which OTEL_EXPORTER_OTLP_ENDPOINT was not appending /v1/traces always, as intended. Fixed a bug in which the OTEL exporter had no time to flush all spans in 5 seconds. Now spans are retried starting at 250ms instead of ~5s, changing the default behaviour of the SDK. Fixed a bug in docs which stated that OTEL_SDK_DISABLED disabled some systems. The Golang SDK does not honor OTEL_SDK_DISABLED and to disable OTLP you can unset the ENDPOINT variable. Reworded the documentation accordingly. Added `OTEL_EXPORTER_OTLP_ENDPOINT` to the workflows, so it can be used instead of `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`. For developers: added new tests using a local testing server to inspect traces sent. BREAKING CHANGE: OTEL_TRACES_EXPORTER and OTEL_EXPORTER_OTLP_PROTOCOL are no longer supported; see https://fullsend.sh for migration. Signed-off-by: Hector Martinez --- .github/workflows/reusable-code.yml | 1 + .github/workflows/reusable-dispatch.yml | 7 + .github/workflows/reusable-fix.yml | 1 + .github/workflows/reusable-prioritize.yml | 1 + .github/workflows/reusable-retro.yml | 1 + .github/workflows/reusable-review.yml | 1 + .github/workflows/reusable-triage.yml | 1 + .../infrastructure/distributed-tracing.md | 54 +- go.mod | 5 +- internal/cli/run.go | 2 +- internal/telemetry/main_test.go | 11 + internal/telemetry/otlpsink_test.go | 83 ++ internal/telemetry/telemetry.go | 104 +-- internal/telemetry/telemetry_test.go | 719 +++++++++++++++--- 14 files changed, 829 insertions(+), 162 deletions(-) create mode 100644 internal/telemetry/main_test.go create mode 100644 internal/telemetry/otlpsink_test.go diff --git a/.github/workflows/reusable-code.yml b/.github/workflows/reusable-code.yml index 3bd91ec4b3..3ce6ccd071 100644 --- a/.github/workflows/reusable-code.yml +++ b/.github/workflows/reusable-code.yml @@ -202,6 +202,7 @@ jobs: REPO_FULL_NAME: ${{ inputs.source_repo }} CODE_ALLOWED_TARGET_BRANCHES: '' TARGET_BRANCH: main + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-dispatch.yml b/.github/workflows/reusable-dispatch.yml index f309f8f882..f5f0238fd4 100644 --- a/.github/workflows/reusable-dispatch.yml +++ b/.github/workflows/reusable-dispatch.yml @@ -616,6 +616,7 @@ jobs: env: GITHUB_ISSUE_URL: ${{ fromJSON(needs.route.outputs.event_payload).issue.html_url }} REPO_FULL_NAME: ${{ github.repository }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -738,6 +739,7 @@ jobs: REPO_FULL_NAME: ${{ github.repository }} CODE_ALLOWED_TARGET_BRANCHES: '' TARGET_BRANCH: main + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -850,6 +852,7 @@ jobs: PRIOR_REVIEW_SHA: ${{ steps.prior-review.outputs.prior_sha }} PRIOR_REVIEW_FILE: ${{ steps.prior-review.outputs.prior_review_file }} PRIOR_REVIEW_PROVENANCE: ${{ steps.prior-review.outputs.prior_review_provenance }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -1133,6 +1136,7 @@ jobs: FIX_ITERATION: ${{ steps.context.outputs.iteration }} REVIEW_BODY_FILE: ${{ steps.review-body.outputs.review_file }} PRE_AGENT_HEAD: ${{ steps.pre-agent.outputs.head }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -1228,6 +1232,7 @@ jobs: RETRO_COMMENT: ${{ fromJSON(needs.route.outputs.event_payload).comment.body || '' }} REPO_FULL_NAME: ${{ github.repository }} MINT_REPOS: ${{ steps.workspace.outputs.repo_name != '' && (inputs.install_mode == 'per-repo' && steps.workspace.outputs.repo_name || format('{0},.fullsend', steps.workspace.outputs.repo_name)) || '' }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -1308,6 +1313,7 @@ jobs: env: GITHUB_ISSUE_URL: ${{ fromJSON(needs.route.outputs.event_payload).issue.html_url }} REPO_FULL_NAME: ${{ github.repository }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} @@ -1615,6 +1621,7 @@ jobs: env: GITHUB_ISSUE_URL: ${{ steps.dispatch-env.outputs.issue_url }} REPO_FULL_NAME: ${{ matrix.source_repo }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-fix.yml b/.github/workflows/reusable-fix.yml index 181f75fef5..b11e575245 100644 --- a/.github/workflows/reusable-fix.yml +++ b/.github/workflows/reusable-fix.yml @@ -399,6 +399,7 @@ jobs: FIX_ITERATION: ${{ steps.context.outputs.iteration }} REVIEW_BODY_FILE: ${{ steps.review-body.outputs.review_file }} PRE_AGENT_HEAD: ${{ steps.pre-agent.outputs.head }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-prioritize.yml b/.github/workflows/reusable-prioritize.yml index d631229359..b69dcf3ab9 100644 --- a/.github/workflows/reusable-prioritize.yml +++ b/.github/workflows/reusable-prioritize.yml @@ -163,6 +163,7 @@ jobs: env: GITHUB_ISSUE_URL: ${{ fromJSON(inputs.event_payload).issue.html_url }} REPO_FULL_NAME: ${{ inputs.source_repo }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-retro.yml b/.github/workflows/reusable-retro.yml index 07268daaa6..608a7050be 100644 --- a/.github/workflows/reusable-retro.yml +++ b/.github/workflows/reusable-retro.yml @@ -174,6 +174,7 @@ jobs: RETRO_COMMENT: ${{ fromJSON(inputs.event_payload).comment.body || '' }} REPO_FULL_NAME: ${{ inputs.source_repo }} MINT_REPOS: ${{ steps.repo-parts.outputs.name != '' && (inputs.install_mode == 'per-repo' && steps.repo-parts.outputs.name || format('{0},.fullsend', steps.repo-parts.outputs.name)) || '' }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-review.yml b/.github/workflows/reusable-review.yml index 5ce210c16f..9d1cbd9ddb 100644 --- a/.github/workflows/reusable-review.yml +++ b/.github/workflows/reusable-review.yml @@ -190,6 +190,7 @@ jobs: PRIOR_REVIEW_SHA: ${{ steps.prior-review.outputs.prior_sha }} PRIOR_REVIEW_FILE: ${{ steps.prior-review.outputs.prior_review_file }} PRIOR_REVIEW_PROVENANCE: ${{ steps.prior-review.outputs.prior_review_provenance }} + OTEL_EXPORTER_OTLP_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/.github/workflows/reusable-triage.yml b/.github/workflows/reusable-triage.yml index 6e3681c973..57f0a046c4 100644 --- a/.github/workflows/reusable-triage.yml +++ b/.github/workflows/reusable-triage.yml @@ -180,6 +180,7 @@ jobs: # 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_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }} 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 }} diff --git a/docs/guides/infrastructure/distributed-tracing.md b/docs/guides/infrastructure/distributed-tracing.md index 7ecf820c1e..5facdd7186 100644 --- a/docs/guides/infrastructure/distributed-tracing.md +++ b/docs/guides/infrastructure/distributed-tracing.md @@ -12,9 +12,8 @@ configuration required: (sandbox creation, agent iterations, validation) with timestamps, durations, trace IDs, and token/cost attributes. -This file is written on every run unless `OTEL_SDK_DISABLED=true`, which -suppresses all telemetry output including the local file. It contains -metadata only — no prompts, completions, or source code content. +This file is written on every run. It contains metadata only — no prompts, +completions, or source code content. ## Prerequisites @@ -31,16 +30,31 @@ Level 1 requires nothing. To enable OTLP export (Level 2 and Level 3) you need: and bring-your-own-workflow runs only — the managed workflows do not yet pass a CA bundle through. +## Disabling telemetry + +To disable all telemetry, including the local file exporter: + +```bash +export OTEL_SDK_DISABLED=true # case-insensitive +``` + +To disable only the OTLP exporter: + +```bash +unset OTEL_EXPORTER_OTLP_ENDPOINT +unset OTEL_EXPORTER_OTLP_TRACES_ENDPOINT +``` + ## Enabling OTLP export (Level 2) To send metadata spans to an OpenTelemetry-compatible backend, set one of the standard OTEL environment variables: ```bash -# Signal-specific (takes precedence, used as-is — no /v1/traces appended) +# Signal-specific (used as-is, no path appended) export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="https://your-backend:4318/v1/traces" -# Base URL (SDK appends /v1/traces automatically) +# Base URL (SDK appends /v1/traces) export OTEL_EXPORTER_OTLP_ENDPOINT="https://your-backend:4318" ``` @@ -48,13 +62,13 @@ export OTEL_EXPORTER_OTLP_ENDPOINT="https://your-backend:4318" Headers follow the same pattern: `OTEL_EXPORTER_OTLP_TRACES_HEADERS` > `OTEL_EXPORTER_OTLP_HEADERS`. The local file (`run-telemetry.jsonl`) is produced with no configuration -needed (Level 1), unless `OTEL_SDK_DISABLED=true`. +needed (Level 1). When an endpoint is configured, spans are exported via OTLP/HTTP. Any backend that speaks OTLP works: Jaeger, Grafana Tempo, MLflow, Arize Phoenix, Langfuse, SigNoz, Honeycomb, Datadog, etc. -If the endpoint is unreachable, the CLI continues normally — local files are +If the endpoint is unreachable, the CLI continues normally; local files are still produced and the run is not affected. Operational details: @@ -62,6 +76,11 @@ Operational details: - **Export timing:** spans are exported live via the OTel SDK's batch processor as they complete. On shutdown, the provider flushes remaining spans within a 5-second budget. A dead endpoint does not block the run. +- **Retry:** the exporter retries on transient failures (HTTP 503, etc.) + with an initial interval of 250 ms and a max interval of 2 s. The + 5-second context deadline passed to `tp.Shutdown` bounds both retries + and in-flight requests, so a persistently failing or hanging endpoint + does not extend shutdown. - **Crashed runs:** completed spans that were already flushed mid-run reach the backend; spans still in the batch buffer are lost. The local `run-telemetry.jsonl` (written synchronously per span) remains the @@ -69,15 +88,15 @@ Operational details: - **Sampling:** when the run continues an inbound `TRACEPARENT` whose W3C sampled flag is unset (`-00`), the upstream sampling decision is respected: nothing is exported. The local file is still written. -- **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` disables all telemetry output - (OTLP export *and* the local file). `OTEL_TRACES_EXPORTER=none` disables - only the OTLP export; the local file is still written. +- **Endpoint validation:** before creating the OTLP exporter, the CLI + validates whichever endpoint the SDK will actually use + (`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` if set, else + `OTEL_EXPORTER_OTLP_ENDPOINT`). An endpoint is invalid if it cannot be + parsed as a URL, has no scheme, uses a scheme other than `http` or `https`, + or has no host (e.g. `localhost:4318` instead of + `http://localhost:4318`). When invalid, the CLI prints a warning to stderr + and skips OTLP export; the local file exporter is unaffected. A valid + signal-specific endpoint is not blocked by an invalid generic endpoint. - **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. @@ -205,6 +224,8 @@ 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`). + Alternatively, set `OTEL_EXPORTER_OTLP_ENDPOINT` (the base URL without a + signal path); managed workflows forward both variants. 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`). @@ -223,6 +244,7 @@ Add the environment variables to any job that runs `fullsend run`: ```yaml env: + OTEL_EXPORTER_OTLP_ENDPOINT: "${{ vars.OTEL_EXPORTER_OTLP_ENDPOINT }}" OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "${{ vars.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT }}" OTEL_EXPORTER_OTLP_TRACES_HEADERS: "${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }}" ``` diff --git a/go.mod b/go.mod index 8007300a99..af54577fc3 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,8 @@ require ( 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 + go.uber.org/goleak v1.3.0 golang.org/x/crypto v0.54.0 golang.org/x/oauth2 v0.36.0 golang.org/x/term v0.45.0 @@ -38,7 +40,6 @@ require ( 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 - go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/net v0.56.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect @@ -83,7 +84,7 @@ require ( golang.org/x/image v0.44.0 // indirect golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.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/internal/cli/run.go b/internal/cli/run.go index ec8f391f78..b5534c0f87 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -910,7 +910,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep } rootSpan.End() - flushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + flushCtx, cancel := context.WithTimeout(context.Background(), telemetry.FlushTimeout) defer cancel() tracingCleanup(flushCtx) }() diff --git a/internal/telemetry/main_test.go b/internal/telemetry/main_test.go new file mode 100644 index 0000000000..557159db5e --- /dev/null +++ b/internal/telemetry/main_test.go @@ -0,0 +1,11 @@ +package telemetry + +import ( + "testing" + + "go.uber.org/goleak" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} diff --git a/internal/telemetry/otlpsink_test.go b/internal/telemetry/otlpsink_test.go new file mode 100644 index 0000000000..d406692b4e --- /dev/null +++ b/internal/telemetry/otlpsink_test.go @@ -0,0 +1,83 @@ +package telemetry + +import ( + "bytes" + "compress/gzip" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + "google.golang.org/protobuf/proto" +) + +type otlpSink struct { + mu sync.Mutex + reqs []*coltracepb.ExportTraceServiceRequest + headers []http.Header + paths []string + srv *httptest.Server +} + +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) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if r.Header.Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(raw)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + raw, err = io.ReadAll(zr) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } + var req coltracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(raw, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + 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) spanNames() []string { + s.mu.Lock() + defer s.mu.Unlock() + var names []string + for _, req := range s.reqs { + for _, rs := range req.GetResourceSpans() { + for _, ss := range rs.GetScopeSpans() { + for _, sp := range ss.GetSpans() { + names = append(names, sp.GetName()) + } + } + } + } + return names +} + +func (s *otlpSink) requestCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.reqs) +} diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index 87cce46940..838583f5ff 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -18,6 +18,7 @@ import ( "path/filepath" "strings" "sync" + "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" @@ -30,11 +31,51 @@ import ( // TelemetryFile is the artifact name written to the output dir. const TelemetryFile = "run-telemetry.jsonl" +// FlushTimeout is the budget for tp.Shutdown to flush pending spans at CLI exit. +const FlushTimeout = 5 * time.Second + const scopeName = "github.com/fullsend-ai/fullsend/internal/telemetry" // newOTLPExporter is a seam over exporter construction for tests. -var newOTLPExporter = func(ctx context.Context, endpoint string) (sdktrace.SpanExporter, error) { - return otlptracehttp.New(ctx, otlptracehttp.WithEndpointURL(endpoint)) +// The SDK reads OTEL_EXPORTER_OTLP_*ENDPOINT from the environment. +var newOTLPExporter = func(ctx context.Context) (sdktrace.SpanExporter, error) { + retryOption := otlptracehttp.WithRetry(otlptracehttp.RetryConfig{ + Enabled: true, + InitialInterval: 250 * time.Millisecond, + MaxInterval: 2 * time.Second, + }) + return otlptracehttp.New(ctx, retryOption) +} + +func validateEndpoints(endpoint, tracesEndpoint string) error { + // The SDK uses TRACES_ENDPOINT when set, falling back to ENDPOINT. + // Validate only the value that will actually be used. + ep := tracesEndpoint + if ep == "" { + ep = endpoint + } + if ep == "" { + return nil + } + + u, err := url.Parse(ep) + if err != nil { + return err + } + + if u.Scheme == "" { + return fmt.Errorf("endpoint %q has no scheme, it is required", ep) + } + + if u.Scheme != "https" && u.Scheme != "http" { + return fmt.Errorf("endpoint %q uses the %q scheme which is not supported", ep, u.Scheme) + } + + if u.Host == "" { + return fmt.Errorf("endpoint %q has no host, it is required", ep) + } + + return nil } // Setup creates a TracerProvider with file and (optionally) OTLP exporters. @@ -46,7 +87,7 @@ var newOTLPExporter = func(ctx context.Context, endpoint string) (sdktrace.SpanE func Setup(dir string, serviceVersion string) (trace.Tracer, func(context.Context)) { noop := func(context.Context) {} - if isSDKDisabled() { + if sdkDisable := os.Getenv("OTEL_SDK_DISABLED"); strings.EqualFold(strings.TrimSpace(sdkDisable), "true") { return tracenoop.NewTracerProvider().Tracer(""), noop } @@ -56,22 +97,24 @@ func Setup(dir string, serviceVersion string) (trace.Tracer, func(context.Contex } res := buildResource(serviceVersion) - opts := []sdktrace.TracerProviderOption{ sdktrace.WithResource(res), sdktrace.WithSampler(sdktrace.AlwaysSample()), sdktrace.WithSpanProcessor(sdktrace.NewSimpleSpanProcessor(newFileExporter(f))), } - if endpoint := endpointFromEnv(); endpoint != "" && !isExporterNone() { - if err := validateEndpoint(endpoint); err != nil { - fmt.Fprintf(os.Stderr, "fullsend: OTLP export skipped: %v\n", err) - } else if exp, err := newOTLPExporter(context.Background(), endpoint); err != nil { - fmt.Fprintf(os.Stderr, "fullsend: OTLP exporter failed: %v\n", err) + endpoint := strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")) + tracesEndpoint := strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")) + if endpoint != "" || tracesEndpoint != "" { + if err := validateEndpoints(endpoint, tracesEndpoint); err != nil { + fmt.Fprintf(os.Stderr, "fullsend: OTLP endpoints validation failed: %v\n", err) } else { - opts = append(opts, sdktrace.WithSpanProcessor( - &parentSampledProcessor{base: sdktrace.NewBatchSpanProcessor(exp)}, - )) + exp, err := newOTLPExporter(context.Background()) + if err != nil { + fmt.Fprintf(os.Stderr, "fullsend: OTLP export setup failed: %v\n", err) + } else { + opts = append(opts, sdktrace.WithSpanProcessor(&parentSampledProcessor{base: sdktrace.NewBatchSpanProcessor(exp)})) + } } } @@ -79,46 +122,15 @@ func Setup(dir string, serviceVersion string) (trace.Tracer, func(context.Contex tracer := tp.Tracer(scopeName, trace.WithInstrumentationVersion(serviceVersion)) cleanup := func(ctx context.Context) { - _ = tp.Shutdown(ctx) + if err := tp.Shutdown(ctx); err != nil { + fmt.Fprintf(os.Stderr, "fullsend: telemetry flush incomplete: %v\n", err) + } _ = f.Close() } return tracer, cleanup } -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")) -} - -func isSDKDisabled() bool { - return strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_SDK_DISABLED")), "true") -} - -func isExporterNone() bool { - return strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_TRACES_EXPORTER")), "none") -} - -func validateEndpoint(endpoint string) error { - u, err := url.Parse(endpoint) - if 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", endpoint) - } - if p := protocolFromEnv(); p != "" && p != "http/protobuf" { - return fmt.Errorf("OTEL_EXPORTER_OTLP_(TRACES_)PROTOCOL %q is not supported (only http/protobuf)", p) - } - return nil -} - -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")) -} - // parentSampledProcessor wraps a SpanProcessor and only forwards spans whose // trace was not explicitly unsampled by a remote parent. When a root span // arrives with a remote unsampled parent, the entire trace is suppressed from diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go index 1e1c38292e..b5751c6768 100644 --- a/internal/telemetry/telemetry_test.go +++ b/internal/telemetry/telemetry_test.go @@ -1,30 +1,61 @@ package telemetry import ( + "bytes" + "compress/gzip" "context" "encoding/json" "fmt" + "io" + "net" + "net/http" + "net/http/httptest" "os" "path/filepath" + "strings" + "sync" "testing" + "time" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/trace" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + "google.golang.org/protobuf/proto" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// pinOTELEnv clears ambient OTEL variables so tests are hermetic in CI +// (where OTEL_EXPORTER_OTLP_TRACES_ENDPOINT may be set by org vars). +func pinOTELEnv(t *testing.T) { + t.Helper() + t.Setenv("OTEL_SDK_DISABLED", "") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") +} + func TestSetup_FileExporter(t *testing.T) { + pinOTELEnv(t) + orig := newOTLPExporter + t.Cleanup(func() { newOTLPExporter = orig }) + var exporterCreated bool + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { + exporterCreated = true + return orig(context.Background()) + } + dir := t.TempDir() tracer, cleanup := Setup(dir, "1.0.0-test") - defer cleanup(context.Background()) _, span := tracer.Start(context.Background(), "test-span") span.End() cleanup(context.Background()) + assert.False(t, exporterCreated, + "OTLP exporter must not be created when no endpoint is configured") + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) require.NoError(t, err) require.NotEmpty(t, data, "file exporter must have written span data") @@ -37,6 +68,7 @@ func TestSetup_FileExporter(t *testing.T) { } func TestSetup_NoopOnBadDir(t *testing.T) { + pinOTELEnv(t) tracer, cleanup := Setup("/nonexistent/path/that/should/fail", "1.0.0") defer cleanup(context.Background()) @@ -46,140 +78,654 @@ func TestSetup_NoopOnBadDir(t *testing.T) { } func TestSetup_SDKDisabled(t *testing.T) { - t.Setenv("OTEL_SDK_DISABLED", "true") - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318") + for _, tt := range []struct { + name string + disabledValue string + }{ + { + name: "is disabled lowercase", + disabledValue: "true", + }, + { + name: "is disabled uppercase", + disabledValue: "TRUE", + }, + { + name: "is disabled title case", + disabledValue: "True", + }, + { + name: "is disabled mixed case", + disabledValue: "truE", + }, + } { + t.Run(tt.name, func(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_SDK_DISABLED", tt.disabledValue) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0") + + _, span := tracer.Start(context.Background(), "disabled-span") + span.End() + cleanup(context.Background()) + + assert.False(t, span.SpanContext().IsValid(), "noop tracer when SDK disabled") + + _, err := os.Stat(filepath.Join(dir, TelemetryFile)) + assert.True(t, os.IsNotExist(err), "no telemetry file when SDK disabled") + + assert.Equal(t, 0, sink.requestCount(), "no OTLP export when SDK disabled") + }) + } +} + +func TestSetup_NoEndpoint_FileOnly(t *testing.T) { + pinOTELEnv(t) + orig := newOTLPExporter + t.Cleanup(func() { newOTLPExporter = orig }) + var exporterCreated bool + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { + exporterCreated = true + return orig(context.Background()) + } dir := t.TempDir() tracer, cleanup := Setup(dir, "1.0.0") - defer cleanup(context.Background()) - _, span := tracer.Start(context.Background(), "test") - assert.False(t, span.SpanContext().IsValid(), "SDK disabled returns noop tracer") + _, span := tracer.Start(context.Background(), "file-only-span") + assert.True(t, span.SpanContext().IsValid(), "tracer is active without OTLP endpoint") + span.End() + cleanup(context.Background()) + + assert.False(t, exporterCreated, + "OTLP exporter must not be created when no endpoint is configured") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter writes spans when no OTLP endpoint is set") +} + +func TestSetup_WhitespaceOnlyEndpoint_NoOTLP(t *testing.T) { + pinOTELEnv(t) + orig := newOTLPExporter + t.Cleanup(func() { newOTLPExporter = orig }) + var exporterCreated bool + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { + exporterCreated = true + return orig(context.Background()) + } + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", " \t ") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", " ") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0") + + _, span := tracer.Start(context.Background(), "whitespace-span") + assert.True(t, span.SpanContext().IsValid(), "tracer is active with file exporter") span.End() + cleanup(context.Background()) - _, err := os.Stat(filepath.Join(dir, TelemetryFile)) - assert.True(t, os.IsNotExist(err), "no telemetry file when SDK disabled") + assert.False(t, exporterCreated, + "OTLP exporter must not be created when endpoints are whitespace-only") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter writes spans") } -func TestSetup_OTLPExporterNone(t *testing.T) { - t.Setenv("OTEL_TRACES_EXPORTER", "none") - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318") +func TestSetup_TracesEndpointAlone(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/v1/traces") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0") + + _, span := tracer.Start(context.Background(), "traces-only-span") + span.End() + cleanup(context.Background()) + + assert.Contains(t, sink.spanNames(), "traces-only-span", + "OTLP exporter activates on OTEL_EXPORTER_OTLP_TRACES_ENDPOINT alone") + + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "/v1/traces", sink.paths[0], + "signal-specific endpoint must be used verbatim, no path appended") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter still writes") +} + +func TestSetup_TracesEndpointUsedVerbatim(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/otlp") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0") + + _, span := tracer.Start(context.Background(), "verbatim-span") + span.End() + cleanup(context.Background()) + + assert.Contains(t, sink.spanNames(), "verbatim-span") + + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "/otlp", sink.paths[0], + "signal-specific endpoint path must be used verbatim, not have /v1/traces appended") +} + +func TestSetup_GeneralEndpointAlone(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0") + + _, span := tracer.Start(context.Background(), "traces-only-span") + span.End() + cleanup(context.Background()) + + assert.Contains(t, sink.spanNames(), "traces-only-span", + "OTLP exporter activates on OTEL_EXPORTER_OTLP_ENDPOINT alone") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter still writes") +} + +func TestSetup_ExporterCreationFails(t *testing.T) { + pinOTELEnv(t) + orig := newOTLPExporter + t.Cleanup(func() { newOTLPExporter = orig }) + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { + return nil, fmt.Errorf("bad endpoint") + } + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://bad-host:4318") dir := t.TempDir() tracer, cleanup := Setup(dir, "1.0.0") defer cleanup(context.Background()) - _, span := tracer.Start(context.Background(), "test") - assert.True(t, span.SpanContext().IsValid()) + _, span := tracer.Start(context.Background(), "span") + assert.True(t, span.SpanContext().IsValid(), "if the OTLP fails file still has traces") span.End() + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file spans written when OTLP exporter creation fails") } -func TestSetup_OTLPExporterSeam(t *testing.T) { +func TestSetup_OTLPWirePath(t *testing.T) { + t.Run("delivery_and_path", func(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "wire-span") + span.End() + cleanup(context.Background()) + + // File exporter wrote the span. + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + require.NotEmpty(t, data) + + // OTLP exporter delivered the span as valid protobuf. + require.NotEmpty(t, sink.spanNames(), "span must arrive at the OTLP collector") + assert.Contains(t, sink.spanNames(), "wire-span") + + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "/v1/traces", sink.paths[0]) + }) + + t.Run("gzip_compression", func(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_EXPORTER_OTLP_COMPRESSION", "gzip") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "gzip-span") + span.End() + cleanup(context.Background()) + + require.NotEmpty(t, sink.spanNames()) + assert.Contains(t, sink.spanNames(), "gzip-span") + + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "gzip", sink.headers[0].Get("Content-Encoding")) + }) + + t.Run("base_endpoint_appends_v1_traces", func(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL+"/otlp") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "path-span") + span.End() + cleanup(context.Background()) + + require.NotEmpty(t, sink.spanNames()) + assert.Contains(t, sink.spanNames(), "path-span") + + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "/otlp/v1/traces", sink.paths[0], + "base endpoint must have /v1/traces appended per OTLP spec") + }) + + t.Run("traces_endpoint_precedence", func(t *testing.T) { + pinOTELEnv(t) + primary := newOTLPSink(t) + decoy := newOTLPSink(t) + + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", primary.srv.URL) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", decoy.srv.URL) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "precedence-span") + span.End() + cleanup(context.Background()) + + assert.Contains(t, primary.spanNames(), "precedence-span") + assert.Equal(t, 0, decoy.requestCount(), "generic endpoint must not receive spans when traces endpoint is set") + + primary.mu.Lock() + defer primary.mu.Unlock() + assert.Equal(t, "/", primary.paths[0], + "signal-specific endpoint must be used verbatim, no path appended") + }) + + t.Run("custom_headers", func(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", sink.srv.URL) + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "x-test-key=test-value") + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "header-span") + span.End() + cleanup(context.Background()) + + require.NotEmpty(t, sink.spanNames()) + sink.mu.Lock() + defer sink.mu.Unlock() + assert.Equal(t, "test-value", sink.headers[0].Get("X-Test-Key")) + }) + + t.Run("retry_delivers_within_cli_flush_budget", func(t *testing.T) { + pinOTELEnv(t) + var ( + mu sync.Mutex + attempts int + delivered bool + ) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + attempts++ + n := attempts + mu.Unlock() + + if n == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + + raw, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if r.Header.Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(raw)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + raw, err = io.ReadAll(zr) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } + var req coltracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(raw, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + mu.Lock() + delivered = true + mu.Unlock() + resp, _ := proto.Marshal(&coltracepb.ExportTraceServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + w.Write(resp) + })) + defer func() { srv.CloseClientConnections(); srv.Close() }() + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", srv.URL) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "flush-budget-span") + span.End() + + ctx, cancel := context.WithTimeout(context.Background(), FlushTimeout) + defer cancel() + cleanup(ctx) + + mu.Lock() + defer mu.Unlock() + assert.True(t, delivered, + "retry after 503 must complete within the CLI flush budget") + }) + + t.Run("persistent_503_emits_stderr_warning", func(t *testing.T) { + pinOTELEnv(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer func() { srv.CloseClientConnections(); srv.Close() }() + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", srv.URL) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "doomed-span") + span.End() + + pr, pw, err := os.Pipe() + require.NoError(t, err) + oldStderr := os.Stderr + os.Stderr = pw + defer func() { os.Stderr = oldStderr }() + + // Drain the pipe concurrently so cleanup can't deadlock + // filling the pipe buffer. + var captured []byte + done := make(chan struct{}) + go func() { + captured, _ = io.ReadAll(pr) + close(done) + }() + + ctx, cancel := context.WithTimeout(context.Background(), FlushTimeout) + defer cancel() + cleanup(ctx) + + os.Stderr = oldStderr + pw.Close() + <-done + + assert.Contains(t, string(captured), "fullsend: telemetry flush incomplete:", + "cleanup must warn on stderr when OTLP export fails persistently") + }) + + t.Run("bare_ip_port_rejected_by_validation", func(t *testing.T) { + pinOTELEnv(t) + // url.Parse("127.0.0.1:PORT") returns a parse error ("first path + // segment in URL cannot contain colon"), so validateEndpoints + // rejects it before the SDK is ever invoked. + sink := newOTLPSink(t) + + addr := strings.TrimPrefix(sink.srv.URL, "http://") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", addr) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "schemeless-span") + span.End() + + ctx, cancel := context.WithTimeout(context.Background(), FlushTimeout) + defer cancel() + cleanup(ctx) + + assert.Equal(t, 0, sink.requestCount(), + "bare IP:port fails url.Parse and must be rejected by validation") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter still writes when OTLP export fails") + }) + + t.Run("shutdown_does_not_hang_on_hanging_endpoint", func(t *testing.T) { + pinOTELEnv(t) + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer ln.Close() + + // Accept connections but never respond; close each in its own + // goroutine when the listener shuts down (via defer above). + var conns []net.Conn + var connsMu sync.Mutex + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + connsMu.Lock() + conns = append(conns, conn) + connsMu.Unlock() + } + }() + t.Cleanup(func() { + connsMu.Lock() + defer connsMu.Unlock() + for _, c := range conns { + c.Close() + } + }) + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://"+ln.Addr().String()) + + dir := t.TempDir() + tracer, cleanup := Setup(dir, "1.0.0-wire") + + _, span := tracer.Start(context.Background(), "blackhole-span") + span.End() + + ctx, cancel := context.WithTimeout(context.Background(), FlushTimeout) + defer cancel() + + start := time.Now() + cleanup(ctx) + elapsed := time.Since(start) + + assert.Less(t, elapsed, FlushTimeout+time.Second, + "cleanup must return within the flush budget even when the endpoint hangs") + + data, err := os.ReadFile(filepath.Join(dir, TelemetryFile)) + require.NoError(t, err) + assert.NotEmpty(t, data, "file exporter must still write when OTLP endpoint hangs") + }) +} + +func TestValidateEndpoints(t *testing.T) { + tests := []struct { + name string + endpoint string + tracesEndpoint string + wantErr string // substring; empty means no error + }{ + { + name: "both empty", + endpoint: "", + tracesEndpoint: "", + }, + { + name: "valid endpoint alone", + endpoint: "http://localhost:4318", + tracesEndpoint: "", + }, + { + name: "valid traces endpoint alone", + endpoint: "", + tracesEndpoint: "https://backend:4318/v1/traces", + }, + { + name: "traces endpoint takes precedence over endpoint", + endpoint: "not-a-url", + tracesEndpoint: "https://backend:4318/v1/traces", + }, + { + name: "endpoint used when traces endpoint empty", + endpoint: "not-a-url", + tracesEndpoint: "", + wantErr: "no scheme", + }, + { + name: "parse error on bare ip:port", + endpoint: "127.0.0.1:4318", + tracesEndpoint: "", + wantErr: "cannot contain colon", + }, + { + name: "no scheme rejected", + endpoint: "localhost", + tracesEndpoint: "", + wantErr: "no scheme", + }, + { + name: "unsupported scheme rejected", + endpoint: "ftp://localhost:4318", + tracesEndpoint: "", + wantErr: "not supported", + }, + { + name: "no host rejected", + endpoint: "http://", + tracesEndpoint: "", + wantErr: "no host", + }, + { + name: "http scheme accepted", + endpoint: "http://collector:4318", + tracesEndpoint: "", + }, + { + name: "https scheme accepted", + endpoint: "https://collector:4318", + tracesEndpoint: "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateEndpoints(tt.endpoint, tt.tracesEndpoint) + if tt.wantErr == "" { + assert.NoError(t, err) + } else { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + } + }) + } +} + +func TestSetup_SchemelessEndpointFailed(t *testing.T) { + pinOTELEnv(t) orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() + t.Cleanup(func() { newOTLPExporter = orig }) var called bool - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { called = true - return orig(context.Background(), "http://localhost:4318") + return nil, fmt.Errorf("expected: SDK will reject this") } - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "not-a-url") dir := t.TempDir() _, cleanup := Setup(dir, "1.0.0") cleanup(context.Background()) - assert.True(t, called, "OTLP exporter must be created when endpoint is set") + assert.False(t, called, "schemeless string fails validation") } -func TestSetup_TracesEndpointPreferred(t *testing.T) { +func TestSetup_SchemelessTracesEndpointFailed(t *testing.T) { + pinOTELEnv(t) orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() + t.Cleanup(func() { newOTLPExporter = orig }) var called bool - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { called = true - return orig(context.Background(), "http://localhost:4318") + return nil, fmt.Errorf("expected: SDK will reject this") } - t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://traces.local:4318") - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://generic.local:4318") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "yes-a-url") dir := t.TempDir() _, cleanup := Setup(dir, "1.0.0") cleanup(context.Background()) - assert.True(t, called, "OTLP exporter created when traces-specific endpoint set") + assert.False(t, called, "schemeless string fails validation") } -func TestSetup_InvalidEndpointSkipsOTLP(t *testing.T) { +func TestSetup_SchemelessEndpointsFailed(t *testing.T) { + pinOTELEnv(t) orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() + t.Cleanup(func() { newOTLPExporter = orig }) - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { - t.Fatal("should not be called for invalid endpoint") - return nil, nil + var called bool + newOTLPExporter = func(_ context.Context) (sdktrace.SpanExporter, error) { + called = true + return nil, fmt.Errorf("expected: SDK will reject this") } t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "not-a-url") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "yes-a-url") dir := t.TempDir() - tracer, cleanup := Setup(dir, "1.0.0") - defer cleanup(context.Background()) + _, cleanup := Setup(dir, "1.0.0") + cleanup(context.Background()) - _, span := tracer.Start(context.Background(), "test") - assert.True(t, span.SpanContext().IsValid(), "file exporter still works") - span.End() + assert.False(t, called, "schemeless string fails validation") } -func TestSetup_UnsupportedProtocolSkipsOTLP(t *testing.T) { - orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() - - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { - t.Fatal("should not be called for unsupported protocol") - return nil, nil - } +func TestSetup_InvalidEndpointValidTracesEndpoint(t *testing.T) { + pinOTELEnv(t) + sink := newOTLPSink(t) - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317") - t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "not-a-url") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", sink.srv.URL+"/v1/traces") dir := t.TempDir() tracer, cleanup := Setup(dir, "1.0.0") - defer cleanup(context.Background()) - _, span := tracer.Start(context.Background(), "test") - assert.True(t, span.SpanContext().IsValid(), "file exporter still works") + _, span := tracer.Start(context.Background(), "precedence-bypass-span") span.End() -} - -func TestSetup_TracesProtocolPreferred(t *testing.T) { - orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() - - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { - t.Fatal("traces protocol=grpc should block exporter creation") - return nil, nil - } - - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317") - t.Setenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "grpc") - t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") - - dir := t.TempDir() - _, cleanup := Setup(dir, "1.0.0") cleanup(context.Background()) + + assert.Contains(t, sink.spanNames(), "precedence-bypass-span", + "valid TRACES_ENDPOINT must not be blocked by an invalid generic ENDPOINT") } // spyProcessor records span names forwarded to OnEnd. @@ -246,24 +792,3 @@ func TestParentSampledProcessor_AllowsSampledTrace(t *testing.T) { assert.ElementsMatch(t, []string{"root", "child"}, spy.ended) } - -func TestSetup_OTLPExporterError(t *testing.T) { - orig := newOTLPExporter - defer func() { newOTLPExporter = orig }() - - newOTLPExporter = func(_ context.Context, _ string) (sdktrace.SpanExporter, error) { - return nil, fmt.Errorf("connection refused") - } - - t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318") - t.Setenv("OTEL_SDK_DISABLED", "") - t.Setenv("OTEL_TRACES_EXPORTER", "") - - dir := t.TempDir() - tracer, cleanup := Setup(dir, "1.0.0") - defer cleanup(context.Background()) - - _, span := tracer.Start(context.Background(), "test") - assert.True(t, span.SpanContext().IsValid(), "file exporter works even when OTLP fails") - span.End() -}