From c98e5c58a2fe80c2c591b4bf0b178d4c7073f89f Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Thu, 5 Nov 2026 15:30:18 -0500 Subject: [PATCH 1/2] ax-server: set up OpenTelemetry tracing, metrics, events, and JSON logs ax-server had no telemetry: plain-text logs and nothing to scrape or collect. Set up the OpenTelemetry SDK in a new internal/telemetry package and wire it into ax-server. - Traces export over OTLP/gRPC. New traces are sampled at 10% (ParentBased(TraceIDRatio)), a caller's decision is followed, and only W3C trace context is propagated, not baggage. - A logger provider exports structured events over OTLP/gRPC. ax-server's own log lines stay on stdout. - Metrics are served for Prometheus on :9090/metrics (--metrics-addr), with Go and process metrics and a cap of 100 attribute sets per instrument. Probes and the API stay on :8080. - Logs are JSON, with trace_id and span_id when a span is active. - SDK errors go to slog instead of stderr. Exporters and sampling follow the standard OTEL_* environment variables. internal/axattr names the attributes AX emits, reusing Substrate's ate.atespace and ate.actor.uid so records join on them. The OTel versions match Agent Substrate's (otel v1.45.0, semconv v1.43.0). Nothing records spans or emits events yet; RPC instrumentation and task state events build on this. --- AGENTS.md | 4 +- README.md | 1 + cmd/ax-server/main.go | 36 ++- deploy/ax-server.yaml | 5 + docs/observability.md | 54 +++++ go.mod | 28 ++- go.sum | 48 +++- internal/axattr/axattr.go | 53 +++++ internal/telemetry/log.go | 50 +++++ internal/telemetry/telemetry.go | 237 +++++++++++++++++++ internal/telemetry/telemetry_test.go | 325 +++++++++++++++++++++++++++ test/e2e/manifests/ax-server.yaml | 5 + 12 files changed, 839 insertions(+), 7 deletions(-) create mode 100644 docs/observability.md create mode 100644 internal/axattr/axattr.go create mode 100644 internal/telemetry/log.go create mode 100644 internal/telemetry/telemetry.go create mode 100644 internal/telemetry/telemetry_test.go diff --git a/AGENTS.md b/AGENTS.md index d9363c20..3d3150b9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -27,6 +27,8 @@ internal/store/ memory + postgres backends (schema.sql and schema.go chan internal/substrate The only place that talks to the Substrate Control API internal/workspace git clone / inline files / skills dir (MCP is spec-only; agents read it via the metadata server) internal/{tunnel,guest,metadata} k8s port-forward, in-sandbox guest client, metadata server +internal/telemetry ax-server's OpenTelemetry setup: OTLP traces, metrics, and events, Prometheus /metrics (:9090), JSON slog with trace IDs +internal/axattr OpenTelemetry attribute names (ax.task.*, and Substrate's ate.* for actor identity) pkg/apis/v1alpha1 *.proto (source of truth) -> *.pb.go (generated), types.go, resources.go test/e2e Kind-based; compiled by `go test ./...` but self-skips without AX_E2E=true hack/ E2E provisioning scripts @@ -40,7 +42,7 @@ skills/ Developer skills (SKILL.md format). .agents/skills, .clau export GOTOOLCHAIN=auto # go.mod wants 1.27.1; older local Go auto-downloads it go mod tidy && git diff --exit-code go.mod go.sum go build ./cmd/ax ./cmd/ax-server ./cmd/ax-task-runner -go test ./... # ~20 s, 16 packages with tests; Postgres + E2E tiers self-skip +go test ./... # ~20 s, 17 packages with tests; Postgres + E2E tiers self-skip go vet ./... && gofmt -l . # not run by CI; both are clean at HEAD ``` diff --git a/README.md b/README.md index bd6ac673..475d4e73 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,7 @@ Want to see the whole lifecycle end to end? Run [`./demo.sh`](demo.sh). It appli | [Sandbox](docs/sandbox.md) | See what the runner does on boot and what your command can rely on: metadata server, guest services, environment. | | [Runners](docs/runner.md) | Understand the contract between the control plane and the task container, and build your own runner image to replace the default. | | [Networking](docs/networking.md) | Reach a running task through the atenet router from the cluster, your laptop, or a gRPC client. | +| [Observability](docs/observability.md) | Collect ax-server's logs, metrics, traces, and events, and configure where they go. | | [Architecture](DESIGN.md) | Understand how the control plane fits together, plus the [API reference](DESIGN.md#api-reference). | | [Development](docs/development.md) | Build, test, and ship changes to AX itself. | | [Roadmap](docs/roadmap.md) | See planned milestones across core specs, actor architecture, agentic environments, and governance. | diff --git a/cmd/ax-server/main.go b/cmd/ax-server/main.go index a834615f..759ebcab 100644 --- a/cmd/ax-server/main.go +++ b/cmd/ax-server/main.go @@ -31,6 +31,7 @@ import ( "github.com/google/ax/internal/server" "github.com/google/ax/internal/store/postgres" "github.com/google/ax/internal/substrate" + "github.com/google/ax/internal/telemetry" "github.com/google/ax/pkg/apis/v1alpha1" "github.com/jackc/pgx/v5/pgconn" ) @@ -38,6 +39,7 @@ import ( func main() { var ( listenAddr string + metricsAddr string storeURL string substrateEndpoint string substrateAuthority string @@ -51,6 +53,7 @@ func main() { ) flag.StringVar(&listenAddr, "addr", ":8080", "HTTP listen address") + flag.StringVar(&metricsAddr, "metrics-addr", ":9090", "Listen address for Prometheus /metrics; empty disables it") flag.StringVar(&storeURL, "store", "postgres://ax@localhost:5432/ax?sslmode=disable", "PostgreSQL connection URL (env: AX_STORE). libpq variables such as PGPASSWORD fill in what it leaves out") flag.StringVar(&substrateEndpoint, "substrate-endpoint", "api.ate-system.svc.cluster.local:443", "Agent Substrate Control API endpoint") flag.StringVar(&substrateAuthority, "substrate-authority", "api.ate-system.svc", "Authority / TLS ServerName for Substrate endpoint") @@ -76,11 +79,25 @@ func main() { defaultTaskImage = envImage } - logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + logger := slog.New(telemetry.NewLogHandler(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))) slog.SetDefault(logger) + tel, err := telemetry.Init(context.Background(), "ax-server") + if err != nil { + slog.Error("initializing telemetry", "error", err) + os.Exit(1) + } + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := tel.Shutdown(ctx); err != nil { + slog.Warn("flushing telemetry", "error", err) + } + }() + slog.Info("starting ax-server", "listenAddr", listenAddr, + "metricsAddr", metricsAddr, "store", describeStore(storeURL), "substrateEndpoint", substrateEndpoint, "template", defaultTemplate, @@ -133,12 +150,29 @@ func main() { } }() + // Metrics get their own listener so the probe and API port stays as it is. + var metricsServer *http.Server + if metricsAddr != "" { + mux := http.NewServeMux() + mux.Handle("GET /metrics", tel.MetricsHandler()) + metricsServer = &http.Server{Addr: metricsAddr, Handler: mux, ReadHeaderTimeout: 10 * time.Second} + go func() { + if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + slog.Error("metrics server failed", "error", err) + os.Exit(1) + } + }() + } + <-ctx.Done() slog.Info("shutting down ax-server") shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) defer shutdownCancel() _ = httpServer.Shutdown(shutdownCtx) + if metricsServer != nil { + _ = metricsServer.Shutdown(shutdownCtx) + } } // describeStore summarizes a store connection string for logs. It parses the diff --git a/deploy/ax-server.yaml b/deploy/ax-server.yaml index f0049e0c..7fd9bc4a 100644 --- a/deploy/ax-server.yaml +++ b/deploy/ax-server.yaml @@ -92,6 +92,8 @@ spec: ports: - containerPort: 8080 name: http + - containerPort: 9090 + name: metrics resources: requests: cpu: 100m @@ -162,6 +164,9 @@ spec: - port: 8080 targetPort: 8080 name: http + - port: 9090 + targetPort: 9090 + name: metrics selector: app.kubernetes.io/name: ax-server diff --git a/docs/observability.md b/docs/observability.md new file mode 100644 index 00000000..2241c97e --- /dev/null +++ b/docs/observability.md @@ -0,0 +1,54 @@ +# Observability + +ax-server sets up OpenTelemetry itself. It needs no sidecar, but it does need +a collector to send traces and events to. + +| Signal | Where it goes | Default | +|---|---|---| +| Logs | stdout, one JSON object per line | Always on | +| Metrics | Prometheus `/metrics` on port 9090 (`--metrics-addr`) | On; `--metrics-addr=` turns it off | +| Traces | OTLP/gRPC to a collector | Exported; 10% of new traces sampled | +| Events | OTLP/gRPC log records to a collector | Exported | + +Health probes and the AX API stay on port 8080. The bundled manifests expose +port 9090 as `metrics` on the `ax-server` Service, so a Prometheus scrape +config or a GKE `PodMonitoring` can target it. + +## Configuration + +Telemetry is configured with the standard +[OpenTelemetry environment variables](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/) +on the ax-server container: + +| Variable | Effect | +|---|---| +| `OTEL_EXPORTER_OTLP_ENDPOINT` | Collector address, for example `http://otel-collector.observability.svc:4317`. The SDK default is `localhost:4317`. An `http://` endpoint is reached without TLS. | +| `OTEL_TRACES_EXPORTER`, `OTEL_LOGS_EXPORTER` | `otlp` (default) or `none`. With `none`, traces are still sampled and propagated, only not sent. | +| `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | Replace the default sampler, `parentbased_traceidratio` at `0.1`. A caller's sampling decision is always followed. | +| `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES` | Override the resource. The defaults are `service.name=ax-server` and `service.instance.id` set to the pod name. | + +Nothing is sent until ax-server records a span or emits an event, so a +deployment without a collector logs no export errors until then. + +## Logs + +ax-server writes structured JSON logs to stdout. A line logged while a trace +is active carries `trace_id` and `span_id`, so a log backend can link it to +the trace. ax-server's log lines are not sent over OTLP; collect them from +stdout as you would for any pod. OTLP log records carry structured events +only. + +## Metrics + +Each instrument keeps at most 100 attribute sets per collection. Further sets +are folded into one series labelled `otel_metric_overflow="true"`, so an +unbounded label cannot exhaust memory. `/metrics` also serves the Go runtime +and process metrics (`go_*`, `process_*`). + +## Attribute names + +Spans, events, and log attributes use the names in +[`internal/axattr`](../internal/axattr/axattr.go). Attributes that identify an +actor reuse Agent Substrate's names, `ate.atespace` and `ate.actor.uid`, so +ax-server and Substrate records join on them. Identity attributes such as +`ax.task.id` and `ate.actor.uid` never appear on metrics. diff --git a/go.mod b/go.mod index 968855e5..5d6905c1 100644 --- a/go.mod +++ b/go.mod @@ -6,21 +6,45 @@ require ( github.com/agent-substrate/env v0.0.11-0.20260912052224-4468a200b170 github.com/cenkalti/backoff/v5 v5.0.3 github.com/jackc/pgx/v5 v5.11.0 + github.com/prometheus/client_golang v1.24.1 go.ate.dev/substrate v1.0.0 + go.opentelemetry.io/otel v1.45.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0 + go.opentelemetry.io/otel/exporters/prometheus v0.65.0 + go.opentelemetry.io/otel/log v0.21.0 + go.opentelemetry.io/otel/metric v1.45.0 + go.opentelemetry.io/otel/sdk v1.45.0 + go.opentelemetry.io/otel/sdk/log v0.21.0 + go.opentelemetry.io/otel/sdk/metric v1.45.0 + go.opentelemetry.io/otel/trace v1.45.0 + go.opentelemetry.io/proto/otlp v1.11.0 google.golang.org/grpc v1.83.2 google.golang.org/protobuf v1.36.12 gopkg.in/yaml.v3 v3.0.1 ) require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/go-logr/logr v1.4.4 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - github.com/kr/pretty v0.3.1 // indirect - github.com/rogpeppe/go-internal v1.14.1 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/otlptranslator v1.0.0 // indirect + github.com/prometheus/procfs v0.21.1 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 // indirect golang.org/x/net v0.60.0 // indirect golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect golang.org/x/text v0.42.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect ) diff --git a/go.sum b/go.sum index 9c33c634..83854ddf 100644 --- a/go.sum +++ b/go.sum @@ -1,11 +1,13 @@ github.com/agent-substrate/env v0.0.11-0.20260912052224-4468a200b170 h1:8AxhEx3BL7J8c+zqnYq5S3aH0J/4z85/JYWoo1dr6xY= github.com/agent-substrate/env v0.0.11-0.20260912052224-4468a200b170/go.mod h1:RZ4ACmDqWQx7fgX9b1EaFJskeYTZh1Y7uZinDtp7U2o= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= 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/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -16,6 +18,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.30.0 h1:/Tnpcb2E0Pz/tN9s3bfEY2Q8ePCEX9iuS+cneUwncnw= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0/go.mod h1:zOBXOsUaBSjKgmH4OGzV1esUpR3oUSCPYVd2cUBjKYY= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= @@ -24,13 +28,27 @@ github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/otlptranslator v1.0.0 h1:s0LJW/iN9dkIH+EnhiD3BlkkP5QVIUVEoIwkU+A6qos= +github.com/prometheus/otlptranslator v1.0.0/go.mod h1:vRYWnXvI6aWGpsdY/mOT/cbeVRBlPWtBNDb7kGR3uKM= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -44,14 +62,36 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0 h1:WseeVYf5dJZTsyPiyW5L14k5qsSibqXAMTSiFEDiWr0= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0/go.mod h1:SiLZnQS6Qk2eCpvr2CH/XMAOa64TWGXxEZJZCpD2Lmc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 h1:QRefszxJmfPdjXUUm3j6iDzY03mTPXMjqErFqQ67vUg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0/go.mod h1:Tiz03lTBVBrm7eWZBOidzEaYaJa8tjwGUGv6d8mlTyk= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0 h1:fG5MCxGz8+2VtrN/WgqSpJFctVz24gpxj8CxkKmc8Ww= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0/go.mod h1:BmAYTn+3ysbRe+IU2msxmf5Rx3g6DHvex+tWI3LdhYI= +go.opentelemetry.io/otel/exporters/prometheus v0.65.0 h1:jOveH/b4lU9HT7y+Gfamf18BqlOuz2PWEvs8yM7Q6XE= +go.opentelemetry.io/otel/exporters/prometheus v0.65.0/go.mod h1:i1P8pcumauPtUI4YNopea1dhzEMuEqWP1xoUZDylLHo= +go.opentelemetry.io/otel/log v0.21.0 h1:SLsVDGmtyBrdw8/a2Z0bOIxou/+bN4z56GebH7T0LvA= +go.opentelemetry.io/otel/log v0.21.0/go.mod h1:iReetQrZL9Wyg84cCkOoCmqDHS5RCFfyxC7J+r8fn8g= go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypRm3br/M= go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= +go.opentelemetry.io/otel/metric/x v0.67.0 h1:PcicCNZFkZ4bXfSooXdo3WN7RBOVOtjVdo1wD358Uns= +go.opentelemetry.io/otel/metric/x v0.67.0/go.mod h1:FBjCWZe6wgcqxcMtjdGiClDKXb2YxxXii0CXftE4QtI= go.opentelemetry.io/otel/sdk v1.45.0 h1:4VVSMgQ83dUgW2aoX5f6JgLvHwIvzcuLnF9lUdCSpCw= go.opentelemetry.io/otel/sdk v1.45.0/go.mod h1:Sr40LgXV7DsKMMJMKOhUWOgMWTfAaqvm2kF0g7ilwuA= +go.opentelemetry.io/otel/sdk/log v0.21.0 h1:QsE7XSR0ktQdKmRKGnR+f1ObGF32WG+7MER/P9KgmYc= +go.opentelemetry.io/otel/sdk/log v0.21.0/go.mod h1:m9mApjCoD2/1QuKCAptjv+BrG9WKOvQLVdNx+iBldTo= +go.opentelemetry.io/otel/sdk/log/logtest v0.21.0 h1:X+JBBgKlswCGYsmgL0CnoUUtlE//VB345c84jYAYkdQ= +go.opentelemetry.io/otel/sdk/log/logtest v0.21.0/go.mod h1:HD1575K8e6sIFBBDd5tZB3t9DlMytWXq9FuR+Y4rfjE= go.opentelemetry.io/otel/sdk/metric v1.45.0 h1:oVFszMfyj1Am6s24Vtc7wBb8BKLcwepJjNEYILuiE3o= go.opentelemetry.io/otel/sdk/metric v1.45.0/go.mod h1:vUWUxDZvu1WVRj8JA8S0AdhsPrZoDpA2DdZauIh4mDA= go.opentelemetry.io/otel/trace v1.45.0 h1:l/mP6Uv7oNO7/TblbhpbgMidxhq1uO/rPsikOyVhxag= go.opentelemetry.io/otel/trace v1.45.0/go.mod h1:qoJJA2xNMnxRrdISU/kLtfUH2wNeQbiv+jhs/CxI8bc= +go.opentelemetry.io/proto/otlp v1.11.0 h1:5rrYs0Ykyj50sdU/JU0x8etU+LubXWb+gED6TbEdMIk= +go.opentelemetry.io/proto/otlp v1.11.0/go.mod h1:SmVizdCOAm3XBtG1g1NnOdhW6jtddT72hLMhv8VwA8E= +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/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU= @@ -64,6 +104,8 @@ golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d h1:FarXi840EJWSHYTN3ERkADbPWjl307+FGrA22KAVjjc= +google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d/go.mod h1:K/+WGbmBY7aNW1HDw1fJnKYo10i0DkAX6pows00dLig= google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d h1:IL4hdHzcUv2l/gcg98/Rj3FbtE6axwqslOW8SW0C+S0= google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= diff --git a/internal/axattr/axattr.go b/internal/axattr/axattr.go new file mode 100644 index 00000000..93e5b123 --- /dev/null +++ b/internal/axattr/axattr.go @@ -0,0 +1,53 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package axattr names the OpenTelemetry attributes AX puts on its spans, logs, +// and events, so every emitter spells them the same way. Attributes that +// identify an actor reuse Agent Substrate's names, so AX and Substrate records +// join on them. +// +// The identity attributes (task ID and name, actor UID, exit code, signal) are +// unbounded. Put them on spans, logs, and events only, never on metrics. +package axattr + +import "go.opentelemetry.io/otel/attribute" + +const ( + // TaskID is the task's status.id, unique across the task's lifetime. + TaskID = attribute.Key("ax.task.id") + // TaskName is the task's metadata.name. + TaskName = attribute.Key("ax.task.name") + // TaskPhase is the task's status.phase, for example Running. + TaskPhase = attribute.Key("ax.task.phase") + // TaskResidency is where the task's actor is in its lifecycle: queued, + // resuming, running, suspending, suspended, reverting, crashed, deleting, + // or deleted. + TaskResidency = attribute.Key("ax.task.residency") + // TaskOutcome is how the task's command ended: succeeded, failed, or + // cancelled. + TaskOutcome = attribute.Key("ax.task.outcome") + // TaskOutcomeSource is what decided the outcome: sandbox, orchestrator, or + // external. + TaskOutcomeSource = attribute.Key("ax.task.outcome_source") + // TaskExitCode is the exit code of the task's command. + TaskExitCode = attribute.Key("ax.task.exit_code") + // TaskSignal is the signal that ended the task's command. + TaskSignal = attribute.Key("ax.task.signal") + + // Atespace is the atespace of the task and its actor, named as Substrate + // names it. + Atespace = attribute.Key("ate.atespace") + // ActorUID is the UID of the task's actor, named as Substrate names it. + ActorUID = attribute.Key("ate.actor.uid") +) diff --git a/internal/telemetry/log.go b/internal/telemetry/log.go new file mode 100644 index 00000000..57492acd --- /dev/null +++ b/internal/telemetry/log.go @@ -0,0 +1,50 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package telemetry + +import ( + "context" + "log/slog" + + "go.opentelemetry.io/otel/trace" +) + +// NewLogHandler wraps h so that a record logged with a context carrying a +// valid span gets trace_id and span_id attributes, joining log lines to +// traces. Records logged without such a context pass through unchanged. +func NewLogHandler(h slog.Handler) slog.Handler { + return traceHandler{h} +} + +type traceHandler struct { + slog.Handler +} + +func (h traceHandler) Handle(ctx context.Context, r slog.Record) error { + if sc := trace.SpanContextFromContext(ctx); sc.IsValid() { + r = r.Clone() + r.AddAttrs(slog.String("trace_id", sc.TraceID().String()), slog.String("span_id", sc.SpanID().String())) + } + return h.Handler.Handle(ctx, r) +} + +func (h traceHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + return traceHandler{h.Handler.WithAttrs(attrs)} +} + +// WithGroup nests later attributes, trace_id and span_id included, under name. +func (h traceHandler) WithGroup(name string) slog.Handler { + return traceHandler{h.Handler.WithGroup(name)} +} diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go new file mode 100644 index 00000000..e71c0c8d --- /dev/null +++ b/internal/telemetry/telemetry.go @@ -0,0 +1,237 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package telemetry sets up OpenTelemetry for ax-server: a tracer provider and +// a logger provider that export over OTLP/gRPC, a meter provider read by a +// Prometheus exporter, and a slog handler that stamps log lines with the +// current trace and span IDs. +// +// Configuration follows the OpenTelemetry environment variables: +// OTEL_EXPORTER_OTLP_ENDPOINT (and the per-signal variants) choose the +// collector, OTEL_TRACES_EXPORTER and OTEL_LOGS_EXPORTER choose otlp (the +// default) or none, OTEL_TRACES_SAMPLER overrides the default sampler, and +// OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES override the resource. +package telemetry + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net/http" + "os" + "strings" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" + "github.com/prometheus/client_golang/prometheus/promhttp" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + promexporter "go.opentelemetry.io/otel/exporters/prometheus" + "go.opentelemetry.io/otel/log/global" + "go.opentelemetry.io/otel/propagation" + sdklog "go.opentelemetry.io/otel/sdk/log" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" +) + +const ( + // CardinalityLimit caps the attribute sets an instrument keeps per + // collection. Sets beyond it are folded into one overflow series, so a + // label that turns out to be unbounded cannot exhaust ax-server's memory or + // the metrics backend. + CardinalityLimit = 100 + // DefaultSampleRatio is the share of new traces sampled when + // OTEL_TRACES_SAMPLER is unset. A caller's sampling decision, carried in + // traceparent, is always followed. + DefaultSampleRatio = 0.1 + + exporterOTLP = "otlp" + exporterNone = "none" +) + +// Providers holds the providers Init installed as the OpenTelemetry globals. +type Providers struct { + Tracer *sdktrace.TracerProvider + Meter *sdkmetric.MeterProvider + // Logger is nil when OTEL_LOGS_EXPORTER turns OTLP logs off. + Logger *sdklog.LoggerProvider + + registry *prometheus.Registry +} + +// Init builds the providers for serviceName, installs them as the +// OpenTelemetry globals with the W3C trace context propagator, and routes SDK +// errors to slog. Call Shutdown on the result before the process exits. +// +// Nothing is exported until something records a span or emits a log record, +// so a deployment without a collector only sees errors once it does. +func Init(ctx context.Context, serviceName string) (*Providers, error) { + // The SDK's default handler writes to stderr, outside the JSON logs. + otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) { + slog.Warn("OpenTelemetry SDK error", "error", err) + })) + + res, err := newResource(ctx, serviceName) + if err != nil { + return nil, err + } + p := &Providers{} + if p.Tracer, err = newTracerProvider(ctx, res); err != nil { + return nil, err + } + if p.Meter, p.registry, err = newMeterProvider(res); err != nil { + return nil, errors.Join(err, p.Shutdown(ctx)) + } + if p.Logger, err = newLoggerProvider(ctx, res); err != nil { + return nil, errors.Join(err, p.Shutdown(ctx)) + } + + otel.SetTracerProvider(p.Tracer) + otel.SetMeterProvider(p.Meter) + if p.Logger != nil { + global.SetLoggerProvider(p.Logger) + } + // Only trace context crosses process boundaries. Baggage from a caller is + // not trusted, so it is not propagated. + otel.SetTextMapPropagator(propagation.TraceContext{}) + return p, nil +} + +// MetricsHandler serves the meter provider's metrics, together with Go +// runtime and process metrics, in the Prometheus exposition format. +func (p *Providers) MetricsHandler() http.Handler { + return promhttp.HandlerFor(p.registry, promhttp.HandlerOpts{}) +} + +// Shutdown flushes and stops the providers. It is safe to call on providers +// that Init only partly built. +func (p *Providers) Shutdown(ctx context.Context) error { + var errs []error + if p.Tracer != nil { + errs = append(errs, p.Tracer.Shutdown(ctx)) + } + if p.Meter != nil { + errs = append(errs, p.Meter.Shutdown(ctx)) + } + if p.Logger != nil { + errs = append(errs, p.Logger.Shutdown(ctx)) + } + return errors.Join(errs...) +} + +// newResource describes the process. OTEL_SERVICE_NAME and +// OTEL_RESOURCE_ATTRIBUTES override the defaults. +func newResource(ctx context.Context, serviceName string) (*resource.Resource, error) { + attrs := resource.WithAttributes(semconv.ServiceName(serviceName)) + opts := []resource.Option{resource.WithTelemetrySDK(), resource.WithSchemaURL(semconv.SchemaURL), attrs} + // In Kubernetes the hostname is the pod name, which tells replicas apart. + if host, err := os.Hostname(); err == nil { + opts = append(opts, resource.WithAttributes(semconv.ServiceInstanceID(host))) + } + opts = append(opts, resource.WithFromEnv()) + res, err := resource.New(ctx, opts...) + if errors.Is(err, resource.ErrPartialResource) || errors.Is(err, resource.ErrSchemaURLConflict) { + slog.WarnContext(ctx, "partial telemetry resource", "error", err) + return res, nil + } + if err != nil { + return nil, fmt.Errorf("building telemetry resource: %w", err) + } + return res, nil +} + +func newTracerProvider(ctx context.Context, res *resource.Resource) (*sdktrace.TracerProvider, error) { + opts := []sdktrace.TracerProviderOption{sdktrace.WithResource(res)} + // Without WithSampler, the SDK reads OTEL_TRACES_SAMPLER itself. + if strings.TrimSpace(os.Getenv("OTEL_TRACES_SAMPLER")) == "" { + opts = append(opts, sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(DefaultSampleRatio)))) + } + export := exporterSelected(ctx, "OTEL_TRACES_EXPORTER") + if export { + exp, err := otlptracegrpc.New(ctx) + if err != nil { + return nil, fmt.Errorf("creating OTLP trace exporter: %w", err) + } + opts = append(opts, sdktrace.WithBatcher(exp)) + } + // Without an exporter the provider still samples and propagates context, + // so traces through ax-server stay whole. + tp := sdktrace.NewTracerProvider(opts...) + slog.InfoContext(ctx, "tracing initialized", "export", export) + return tp, nil +} + +func newMeterProvider(res *resource.Resource) (*sdkmetric.MeterProvider, *prometheus.Registry, error) { + reg := prometheus.NewRegistry() + if err := reg.Register(collectors.NewGoCollector()); err != nil { + return nil, nil, fmt.Errorf("registering Go metrics: %w", err) + } + if err := reg.Register(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})); err != nil { + return nil, nil, fmt.Errorf("registering process metrics: %w", err) + } + reader, err := promexporter.New(promexporter.WithRegisterer(reg)) + if err != nil { + return nil, nil, fmt.Errorf("creating Prometheus exporter: %w", err) + } + mp := sdkmetric.NewMeterProvider( + sdkmetric.WithResource(res), + sdkmetric.WithReader(reader), + sdkmetric.WithCardinalityLimit(CardinalityLimit), + ) + return mp, reg, nil +} + +// newLoggerProvider returns the provider for structured events, or nil when +// OTEL_LOGS_EXPORTER turns OTLP off. ax-server's own log lines go to stdout, +// not through it. +func newLoggerProvider(ctx context.Context, res *resource.Resource) (*sdklog.LoggerProvider, error) { + if !exporterSelected(ctx, "OTEL_LOGS_EXPORTER") { + slog.InfoContext(ctx, "OTLP logs export disabled") + return nil, nil + } + exp, err := otlploggrpc.New(ctx) + if err != nil { + return nil, fmt.Errorf("creating OTLP log exporter: %w", err) + } + // Batched, so an unreachable collector never blocks the caller emitting. + lp := sdklog.NewLoggerProvider(sdklog.WithResource(res), sdklog.WithProcessor(sdklog.NewBatchProcessor(exp))) + slog.InfoContext(ctx, "OTLP logs export initialized") + return lp, nil +} + +// exporterSelected reports whether the exporter list in the environment +// variable key selects otlp. Unset selects otlp. ax-server knows only otlp and +// none; other names are logged and ignored. +func exporterSelected(ctx context.Context, key string) bool { + val := strings.TrimSpace(os.Getenv(key)) + if val == "" { + return true + } + selected := false + for _, name := range strings.Split(val, ",") { + switch name = strings.ToLower(strings.TrimSpace(name)); name { + case exporterOTLP: + selected = true + case exporterNone: + return false + default: + slog.WarnContext(ctx, "ignoring unsupported OpenTelemetry exporter", "variable", key, "exporter", name) + } + } + return selected +} diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go new file mode 100644 index 00000000..cd99e13c --- /dev/null +++ b/internal/telemetry/telemetry_test.go @@ -0,0 +1,325 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package telemetry + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + otellog "go.opentelemetry.io/otel/log" + "go.opentelemetry.io/otel/log/global" + "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/trace" + collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + "google.golang.org/grpc" +) + +// fakeCollector is an in-process OTLP/gRPC collector that keeps what it +// receives. +type fakeCollector struct { + coltracepb.UnimplementedTraceServiceServer + collogspb.UnimplementedLogsServiceServer + + mu sync.Mutex + spans []*coltracepb.ExportTraceServiceRequest + logs []*collogspb.ExportLogsServiceRequest +} + +func (c *fakeCollector) Export(_ context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.spans = append(c.spans, req) + return &coltracepb.ExportTraceServiceResponse{}, nil +} + +// logsService adapts fakeCollector to the logs service, whose Export method +// has the same name as the trace service's. +type logsService struct { + collogspb.UnimplementedLogsServiceServer + c *fakeCollector +} + +func (s logsService) Export(_ context.Context, req *collogspb.ExportLogsServiceRequest) (*collogspb.ExportLogsServiceResponse, error) { + s.c.mu.Lock() + defer s.c.mu.Unlock() + s.c.logs = append(s.c.logs, req) + return &collogspb.ExportLogsServiceResponse{}, nil +} + +// startCollector starts a fakeCollector and points the OTLP exporters at it. +func startCollector(t *testing.T) *fakeCollector { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + c := &fakeCollector{} + srv := grpc.NewServer() + coltracepb.RegisterTraceServiceServer(srv, c) + collogspb.RegisterLogsServiceServer(srv, logsService{c: c}) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://"+lis.Addr().String()) + return c +} + +// initForTest runs Init and shuts the providers down when the test ends. +func initForTest(t *testing.T) *Providers { + t.Helper() + p, err := Init(context.Background(), "ax-server") + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = p.Shutdown(context.Background()) }) + return p +} + +func resourceAttr(attrs []*commonpb.KeyValue, key string) string { + for _, kv := range attrs { + if kv.GetKey() == key { + return kv.GetValue().GetStringValue() + } + } + return "" +} + +func TestInitExportsSpansAndLogsOverOTLP(t *testing.T) { + c := startCollector(t) + t.Setenv("OTEL_TRACES_SAMPLER", "always_on") + p := initForTest(t) + + _, span := otel.Tracer("test").Start(context.Background(), "test-span") + span.End() + var rec otellog.Record + rec.SetEventName("ax.test.event") + rec.SetBody(attribute.StringValue("hello")) + global.GetLoggerProvider().Logger("test").Emit(context.Background(), rec) + + // Shutdown flushes both batchers. + if err := p.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + + c.mu.Lock() + defer c.mu.Unlock() + if len(c.spans) == 0 || len(c.spans[0].GetResourceSpans()) == 0 { + t.Fatal("collector received no spans") + } + rs := c.spans[0].GetResourceSpans()[0] + if got := resourceAttr(rs.GetResource().GetAttributes(), "service.name"); got != "ax-server" { + t.Errorf("span resource service.name = %q, want ax-server", got) + } + if name := rs.GetScopeSpans()[0].GetSpans()[0].GetName(); name != "test-span" { + t.Errorf("span name = %q, want test-span", name) + } + if len(c.logs) == 0 || len(c.logs[0].GetResourceLogs()) == 0 { + t.Fatal("collector received no log records") + } + lr := c.logs[0].GetResourceLogs()[0].GetScopeLogs()[0].GetLogRecords()[0] + if lr.GetEventName() != "ax.test.event" || lr.GetBody().GetStringValue() != "hello" { + t.Errorf("log record = %v, want event ax.test.event with body hello", lr) + } +} + +func TestInitHonorsExporterNone(t *testing.T) { + c := startCollector(t) + t.Setenv("OTEL_TRACES_SAMPLER", "always_on") + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_LOGS_EXPORTER", "none") + p := initForTest(t) + if p.Logger != nil { + t.Error("Logger provider built although OTEL_LOGS_EXPORTER=none") + } + + // Spans are still sampled, so trace context keeps flowing, but not sent. + _, span := otel.Tracer("test").Start(context.Background(), "unexported") + if !span.SpanContext().IsSampled() { + t.Error("span not sampled with OTEL_TRACES_EXPORTER=none") + } + span.End() + if err := p.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.spans) != 0 || len(c.logs) != 0 { + t.Errorf("collector received %d span and %d log exports, want none", len(c.spans), len(c.logs)) + } +} + +func TestDefaultSamplerFollowsParentAndSamplesTenthOfRoots(t *testing.T) { + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_LOGS_EXPORTER", "none") + initForTest(t) + tracer := otel.Tracer("test") + + const roots = 4000 + sampled := 0 + for range roots { + _, span := tracer.Start(context.Background(), "root") + if span.SpanContext().IsSampled() { + sampled++ + } + span.End() + } + // 10% of 4000 is 400, with a standard deviation of about 19. + if sampled < 300 || sampled > 500 { + t.Errorf("%d of %d root spans sampled, want about %d", sampled, roots, int(roots*DefaultSampleRatio)) + } + + for _, parentSampled := range []bool{true, false} { + flags := trace.TraceFlags(0) + if parentSampled { + flags = trace.FlagsSampled + } + parent := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: trace.TraceID{1}, SpanID: trace.SpanID{1}, TraceFlags: flags, Remote: true, + }) + _, span := tracer.Start(trace.ContextWithRemoteSpanContext(context.Background(), parent), "child") + if got := span.SpanContext().IsSampled(); got != parentSampled { + t.Errorf("child of a parent with sampled=%v has sampled=%v", parentSampled, got) + } + span.End() + } +} + +func TestPropagatorCarriesTraceContextOnly(t *testing.T) { + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_LOGS_EXPORTER", "none") + initForTest(t) + got := strings.Join(otel.GetTextMapPropagator().Fields(), ",") + if got != "traceparent,tracestate" { + t.Errorf("propagated fields = %s, want traceparent,tracestate (no baggage)", got) + } +} + +func scrape(t *testing.T, h http.Handler) string { + t.Helper() + srv := httptest.NewServer(h) + defer srv.Close() + resp, err := http.Get(srv.URL) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return string(body) +} + +func TestMetricsHandlerServesInstrumentsWithCardinalityCap(t *testing.T) { + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_LOGS_EXPORTER", "none") + p := initForTest(t) + + counter, err := otel.Meter("test").Int64Counter("ax.test.requests") + if err != nil { + t.Fatal(err) + } + for i := range CardinalityLimit * 3 { + counter.Add(context.Background(), 1, metric.WithAttributes(attribute.String("key", fmt.Sprint(i)))) + } + + body := scrape(t, p.MetricsHandler()) + series := 0 + for _, line := range strings.Split(body, "\n") { + if strings.HasPrefix(line, "ax_test_requests_total{") { + series++ + } + } + // The limit counts the overflow series, so the 300 sets become 100 series. + if series != CardinalityLimit { + t.Errorf("ax_test_requests_total has %d series, want %d", series, CardinalityLimit) + } + for _, want := range []string{`otel_metric_overflow="true"`, `service_name="ax-server"`, "go_goroutines", "process_cpu_seconds_total"} { + if !strings.Contains(body, want) { + t.Errorf("/metrics output has no %s", want) + } + } +} + +func TestExporterSelected(t *testing.T) { + tests := []struct { + val string + want bool + }{ + {"", true}, + {"otlp", true}, + {" OTLP ", true}, + {"none", false}, + {"otlp,none", false}, + {"console", false}, + {"console,otlp", true}, + } + for _, tt := range tests { + t.Setenv("OTEL_TEST_EXPORTER", tt.val) + if got := exporterSelected(context.Background(), "OTEL_TEST_EXPORTER"); got != tt.want { + t.Errorf("exporterSelected(%q) = %v, want %v", tt.val, got, tt.want) + } + } +} + +func TestLogHandlerAddsTraceAndSpanIDs(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(NewLogHandler(slog.NewJSONHandler(&buf, nil))) + sc := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: trace.TraceID{0x0a, 0xf7}, + SpanID: trace.SpanID{0x00, 0xf0, 0x67}, + TraceFlags: trace.FlagsSampled, + }) + + logger.InfoContext(trace.ContextWithSpanContext(context.Background(), sc), "with span", "k", "v") + logger.With("component", "test").InfoContext(trace.ContextWithSpanContext(context.Background(), sc), "with attrs") + logger.InfoContext(context.Background(), "without span") + + lines := strings.Split(strings.TrimSpace(buf.String()), "\n") + if len(lines) != 3 { + t.Fatalf("got %d log lines, want 3:\n%s", len(lines), buf.String()) + } + for i, line := range lines { + var rec map[string]any + if err := json.Unmarshal([]byte(line), &rec); err != nil { + t.Fatalf("line %d is not JSON: %v\n%s", i, err, line) + } + if i == 2 { + if _, ok := rec["trace_id"]; ok { + t.Errorf("line without a span has trace_id: %s", line) + } + continue + } + if rec["trace_id"] != sc.TraceID().String() || rec["span_id"] != sc.SpanID().String() { + t.Errorf("line %d trace_id=%v span_id=%v, want %s %s", i, rec["trace_id"], rec["span_id"], sc.TraceID(), sc.SpanID()) + } + } + if !strings.Contains(lines[1], `"component":"test"`) { + t.Errorf("WithAttrs attribute lost: %s", lines[1]) + } +} diff --git a/test/e2e/manifests/ax-server.yaml b/test/e2e/manifests/ax-server.yaml index 63a896f0..b247106c 100644 --- a/test/e2e/manifests/ax-server.yaml +++ b/test/e2e/manifests/ax-server.yaml @@ -71,6 +71,8 @@ spec: ports: - containerPort: 8080 name: http + - containerPort: 9090 + name: metrics resources: requests: cpu: 100m @@ -141,5 +143,8 @@ spec: - port: 8080 targetPort: 8080 name: http + - port: 9090 + targetPort: 9090 + name: metrics selector: app.kubernetes.io/name: ax-server From 345d8d5a0e1d06cb01b29f255ecb6e992e129751 Mon Sep 17 00:00:00 2001 From: Yu Yi Date: Fri, 6 Nov 2026 16:17:46 -0500 Subject: [PATCH 2/2] telemetry: push metrics over OTLP, keep defaults on bad settings, send events to stdout Address review on the OpenTelemetry setup: - Resolve OTEL_TRACES_SAMPLER and OTEL_*_EXPORTER in ax-server, as Agent Substrate does. An unknown or invalid value is logged and keeps the default, so a typo no longer samples every trace or turns export off. - Add an OTLP periodic metric reader. OTEL_METRICS_EXPORTER controls only the push; --metrics-addr still controls /metrics. - Default OTEL_LOGS_EXPORTER to none and accept console. Add telemetry.Emit, which writes an event to stdout when OTLP logs are off, or beside OTLP with console, so events are not dropped. - Keep the SDK's default cardinality limit (2000), so OTEL_GO_X_CARDINALITY_LIMIT applies again. - Add trace_flags to log lines. - Read OTEL_* settings from an ax-otel-config ConfigMap, and set the k8s.* resource attributes and service.instance.id from the downward API. The production manifest ships with OTLP push off; the E2E manifest points at the collector from Substrate's kind overlay. - List ate.atespace among the attributes kept off metrics. --- AGENTS.md | 2 +- deploy/ax-server.yaml | 38 +++ docs/observability.md | 102 ++++++-- go.mod | 1 + go.sum | 2 + internal/axattr/axattr.go | 10 +- internal/telemetry/config.go | 167 +++++++++++++ internal/telemetry/event.go | 137 +++++++++++ internal/telemetry/log.go | 13 +- internal/telemetry/telemetry.go | 137 +++++------ internal/telemetry/telemetry_test.go | 355 +++++++++++++++++++++++---- skills/substrate/SKILL.md | 1 + test/e2e/manifests/ax-server.yaml | 35 +++ 13 files changed, 845 insertions(+), 155 deletions(-) create mode 100644 internal/telemetry/config.go create mode 100644 internal/telemetry/event.go diff --git a/AGENTS.md b/AGENTS.md index 3d3150b9..c3b38a0c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -27,7 +27,7 @@ internal/store/ memory + postgres backends (schema.sql and schema.go chan internal/substrate The only place that talks to the Substrate Control API internal/workspace git clone / inline files / skills dir (MCP is spec-only; agents read it via the metadata server) internal/{tunnel,guest,metadata} k8s port-forward, in-sandbox guest client, metadata server -internal/telemetry ax-server's OpenTelemetry setup: OTLP traces, metrics, and events, Prometheus /metrics (:9090), JSON slog with trace IDs +internal/telemetry ax-server's OpenTelemetry setup: OTLP traces, metrics, and events (Emit, stdout fallback), Prometheus /metrics (:9090), JSON slog with trace IDs internal/axattr OpenTelemetry attribute names (ax.task.*, and Substrate's ate.* for actor identity) pkg/apis/v1alpha1 *.proto (source of truth) -> *.pb.go (generated), types.go, resources.go test/e2e Kind-based; compiled by `go test ./...` but self-skips without AX_E2E=true diff --git a/deploy/ax-server.yaml b/deploy/ax-server.yaml index 7fd9bc4a..0c8931db 100644 --- a/deploy/ax-server.yaml +++ b/deploy/ax-server.yaml @@ -47,6 +47,25 @@ roleRef: kind: ClusterRole name: ax-server --- +# OpenTelemetry settings for ax-server, read with envFrom. Point ax-server at a +# collector here; see docs/observability.md. On a cluster that runs Agent +# Substrate, the endpoint can be the one in ate-system's ate-otel-config. +# Editing this ConfigMap does not restart ax-server; follow a change with +# `kubectl -n ax-system rollout restart deployment/ax-server`. +apiVersion: v1 +kind: ConfigMap +metadata: + name: ax-otel-config + namespace: ax-system +data: + # OTLP export is off until you set a collector. Metrics stay on /metrics + # (port 9090) and events go to stdout. To export, set the endpoint and + # remove the two exporter lines; add OTEL_LOGS_EXPORTER: otlp to send + # events too. + # OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector.observability.svc:4317 + OTEL_TRACES_EXPORTER: none + OTEL_METRICS_EXPORTER: prometheus +--- apiVersion: apps/v1 kind: Deployment metadata: @@ -79,7 +98,26 @@ spec: - "--substrate-ca-file=/run/servicedns-ca/trust-bundle.pem" - "--template=default-template" - "--template-atespace=ax-system" + envFrom: + - configMapRef: + name: ax-otel-config env: + # The resource attributes come from the downward API. Kubernetes + # expands $(VAR) only for a variable listed earlier. + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid + - name: OTEL_RESOURCE_ATTRIBUTES + value: "k8s.namespace.name=$(POD_NAMESPACE),k8s.pod.name=$(POD_NAME),k8s.pod.uid=$(POD_UID),service.instance.id=$(POD_UID)" - name: PGPASSWORD valueFrom: secretKeyRef: diff --git a/docs/observability.md b/docs/observability.md index 2241c97e..55dde0ad 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -1,49 +1,101 @@ # Observability -ax-server sets up OpenTelemetry itself. It needs no sidecar, but it does need -a collector to send traces and events to. +ax-server sets up OpenTelemetry itself and needs no sidecar. It exports +traces, metrics, and events over OTLP/gRPC to a collector you choose, and +serves metrics for Prometheus as well. -| Signal | Where it goes | Default | -|---|---|---| -| Logs | stdout, one JSON object per line | Always on | -| Metrics | Prometheus `/metrics` on port 9090 (`--metrics-addr`) | On; `--metrics-addr=` turns it off | -| Traces | OTLP/gRPC to a collector | Exported; 10% of new traces sampled | -| Events | OTLP/gRPC log records to a collector | Exported | +| Signal | Where it goes | Default in the code | Default in `deploy/ax-server.yaml` | +|---|---|---|---| +| Logs | stdout, one JSON object per line | Always on | Always on | +| Metrics | OTLP push, and Prometheus `/metrics` on port 9090 (`--metrics-addr`) | Both on | `/metrics` only | +| Traces | OTLP push; 10% of new traces sampled | On | Off | +| Events | stdout as JSON log lines, or OTLP log records | stdout | stdout | Health probes and the AX API stay on port 8080. The bundled manifests expose port 9090 as `metrics` on the `ax-server` Service, so a Prometheus scrape -config or a GKE `PodMonitoring` can target it. +config or a GKE `PodMonitoring` can target it. `--metrics-addr=` turns +`/metrics` off. + +## Connect a collector + +The manifests read every `OTEL_*` setting from the `ax-otel-config` ConfigMap +in `ax-system`. `deploy/ax-server.yaml` ships with OTLP export off, so an +installation without a collector logs no export errors and loses nothing: +metrics stay on `/metrics` and events go to stdout. To export, edit the +ConfigMap and restart ax-server: + +```yaml +data: + OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector.observability.svc:4317 + OTEL_LOGS_EXPORTER: otlp # also send events; add ,console to keep them on stdout + # delete OTEL_TRACES_EXPORTER: none and OTEL_METRICS_EXPORTER: prometheus +``` + +```bash +kubectl -n ax-system rollout restart deployment/ax-server +``` + +On a cluster that runs Agent Substrate, the endpoint can be the one in +`ate-otel-config` in `ate-system`, so ax-server and Substrate telemetry reach +the same backend. The E2E manifests point at the collector that Substrate's +kind overlay deploys. + +The manifests also set `OTEL_RESOURCE_ATTRIBUTES` from the downward API: +`k8s.namespace.name`, `k8s.pod.name`, `k8s.pod.uid`, and `service.instance.id` +set to the pod UID, as Substrate does. This variable is in the Deployment's +`env`, which overrides `envFrom`, so the same key in `ax-otel-config` has no +effect. Add other resource attributes to the value in the Deployment. ## Configuration -Telemetry is configured with the standard -[OpenTelemetry environment variables](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/) -on the ax-server container: +Telemetry follows the standard +[OpenTelemetry environment variables](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/). +As the specification asks, ax-server warns about a value it does not +recognize and ignores it, which keeps the default in the table below. | Variable | Effect | |---|---| | `OTEL_EXPORTER_OTLP_ENDPOINT` | Collector address, for example `http://otel-collector.observability.svc:4317`. The SDK default is `localhost:4317`. An `http://` endpoint is reached without TLS. | -| `OTEL_TRACES_EXPORTER`, `OTEL_LOGS_EXPORTER` | `otlp` (default) or `none`. With `none`, traces are still sampled and propagated, only not sent. | -| `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | Replace the default sampler, `parentbased_traceidratio` at `0.1`. A caller's sampling decision is always followed. | -| `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES` | Override the resource. The defaults are `service.name=ax-server` and `service.instance.id` set to the pod name. | +| `OTEL_TRACES_EXPORTER` | `otlp` (default) or `none`. With `none`, traces are still sampled and propagated, only not sent. | +| `OTEL_METRICS_EXPORTER` | Controls only the OTLP push: `otlp` (default) turns it on; `prometheus` or `none` turns it off. `/metrics` is controlled by `--metrics-addr`. | +| `OTEL_LOGS_EXPORTER` | Where events go: `none` (default) writes them to stdout; `otlp` sends them to the collector instead; `otlp,console` does both. | +| `OTEL_METRIC_EXPORT_INTERVAL` | Milliseconds between metric pushes; the SDK default is 60000. | +| `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | Replace the default sampler, `parentbased_traceidratio` at `0.1`. With a `parentbased_*` sampler, ax-server always follows the sampling decision of the caller. An unknown sampler, or a ratio sampler whose argument is missing or outside [0, 1], keeps the default. | +| `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES` | Override the resource. The code defaults are `service.name=ax-server` and `service.instance.id` set to the hostname. | +| `OTEL_GO_X_CARDINALITY_LIMIT` | Attribute sets each metric instrument keeps per collection; the SDK default is 2000. | -Nothing is sent until ax-server records a span or emits an event, so a -deployment without a collector logs no export errors until then. +The OTLP metric push runs every interval whether or not anything recorded a +metric, so with the push on and no reachable collector the SDK logs an export +error each interval. Traces and events are exported only once ax-server +records them. ## Logs ax-server writes structured JSON logs to stdout. A line logged while a trace -is active carries `trace_id` and `span_id`, so a log backend can link it to -the trace. ax-server's log lines are not sent over OTLP; collect them from -stdout as you would for any pod. OTLP log records carry structured events -only. +is active carries `trace_id`, `span_id`, and `trace_flags`, the fields the +OpenTelemetry specification defines for trace context in non-OTLP logs, so a +log backend can link it to the trace. ax-server's log lines are not sent over +OTLP; collect them from stdout as you would for any pod. + +## Events + +Structured events, such as task state changes, use the OTLP logs signal: +each is a log record with an event name. With `OTEL_LOGS_EXPORTER` unset or +`none`, an event is written to stdout as a JSON log line instead, with the +event name in `event`, so it is never dropped because OTLP is off. This +follows Agent Substrate, whose actor events default to stdout the same way. + +OTLP event records are batched, so a slow collector never delays the task +operation that emits them. Records still queued when ax-server exits +ungracefully are lost, unless `console` also wrote them to stdout. ## Metrics -Each instrument keeps at most 100 attribute sets per collection. Further sets -are folded into one series labelled `otel_metric_overflow="true"`, so an +Each instrument keeps at most 2000 attribute sets per collection (the SDK +default, which `OTEL_GO_X_CARDINALITY_LIMIT` overrides). Further sets are +folded into one series labelled `otel_metric_overflow="true"`, so an unbounded label cannot exhaust memory. `/metrics` also serves the Go runtime -and process metrics (`go_*`, `process_*`). +and process metrics (`go_*`, `process_*`); they are not pushed over OTLP. ## Attribute names @@ -51,4 +103,4 @@ Spans, events, and log attributes use the names in [`internal/axattr`](../internal/axattr/axattr.go). Attributes that identify an actor reuse Agent Substrate's names, `ate.atespace` and `ate.actor.uid`, so ax-server and Substrate records join on them. Identity attributes such as -`ax.task.id` and `ate.actor.uid` never appear on metrics. +`ax.task.id`, `ate.atespace`, and `ate.actor.uid` never appear on metrics. diff --git a/go.mod b/go.mod index 5d6905c1..38d369ca 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( go.ate.dev/substrate v1.0.0 go.opentelemetry.io/otel v1.45.0 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.45.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0 go.opentelemetry.io/otel/exporters/prometheus v0.65.0 go.opentelemetry.io/otel/log v0.21.0 diff --git a/go.sum b/go.sum index 83854ddf..308c4e13 100644 --- a/go.sum +++ b/go.sum @@ -64,6 +64,8 @@ go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0 h1:WseeVYf5dJZTsyPiyW5L14k5qsSibqXAMTSiFEDiWr0= go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.21.0/go.mod h1:SiLZnQS6Qk2eCpvr2CH/XMAOa64TWGXxEZJZCpD2Lmc= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.45.0 h1:klTViGcsvLCd1xN3rZzfZ12NslC/OimbmR+k+A006RI= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.45.0/go.mod h1:jRsK04CWmXuY8A0O+wMpSf+t90RHZ53o5Qmxn2PQPfk= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 h1:QRefszxJmfPdjXUUm3j6iDzY03mTPXMjqErFqQ67vUg= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0/go.mod h1:Tiz03lTBVBrm7eWZBOidzEaYaJa8tjwGUGv6d8mlTyk= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0 h1:fG5MCxGz8+2VtrN/WgqSpJFctVz24gpxj8CxkKmc8Ww= diff --git a/internal/axattr/axattr.go b/internal/axattr/axattr.go index 93e5b123..7db6c5fe 100644 --- a/internal/axattr/axattr.go +++ b/internal/axattr/axattr.go @@ -17,8 +17,14 @@ // identify an actor reuse Agent Substrate's names, so AX and Substrate records // join on them. // -// The identity attributes (task ID and name, actor UID, exit code, signal) are -// unbounded. Put them on spans, logs, and events only, never on metrics. +// Keep these attributes off metrics: +// - The identity attributes (task ID and name, atespace, actor UID) have one +// value per task or per user-chosen atespace, so their set of values is +// unbounded. Agent Substrate's no-actor-identity rule also keeps ate.atespace +// and ate.actor.uid off its metrics. +// - The exit code and the signal have a bounded set of values, but they +// describe one task's command, so ax-server records them only beside the +// task identity, on spans, logs, and events. package axattr import "go.opentelemetry.io/otel/attribute" diff --git a/internal/telemetry/config.go b/internal/telemetry/config.go new file mode 100644 index 00000000..8c6f57cc --- /dev/null +++ b/internal/telemetry/config.go @@ -0,0 +1,167 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package telemetry + +import ( + "context" + "fmt" + "log/slog" + "maps" + "os" + "slices" + "strconv" + "strings" + + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +// The environment variables ax-server reads itself rather than leaving to the +// SDK, so that a value it does not understand keeps ax-server's default. +const ( + tracesExporterEnv = "OTEL_TRACES_EXPORTER" + metricsExporterEnv = "OTEL_METRICS_EXPORTER" + logsExporterEnv = "OTEL_LOGS_EXPORTER" + tracesSamplerEnv = "OTEL_TRACES_SAMPLER" + tracesSamplerArgEnv = "OTEL_TRACES_SAMPLER_ARG" +) + +// The names the OTEL_*_EXPORTER variables use. none selects the empty set; it +// is not an exporter. +const ( + exporterOTLP = "otlp" + exporterConsole = "console" + exporterPrometheus = "prometheus" + exporterNone = "none" +) + +// exporters is the set of exporters an OTEL_*_EXPORTER variable selects. The +// empty set is none. +type exporters map[string]bool + +func (e exporters) has(name string) bool { return e[name] } + +// String lists the names in order, or none. +func (e exporters) String() string { + if len(e) == 0 { + return exporterNone + } + return strings.Join(slices.Sorted(maps.Keys(e)), ",") +} + +// resolveExportersEnv reads env and applies resolveExporters. A problem with +// the value is logged, not returned: a telemetry setting never stops startup. +func resolveExportersEnv(ctx context.Context, env string, known []string, def exporters) exporters { + value := os.Getenv(env) + got, err := resolveExporters(env, value, known, def) + if err != nil { + slog.WarnContext(ctx, "invalid OpenTelemetry exporter setting", "variable", env, "value", value, "using", got.String(), "error", err) + } + return got +} + +// resolveExporters reads value as a comma-separated exporter list, as the +// OpenTelemetry specification allows. Names are trimmed and lowercased; empty +// and unknown names are skipped with an error, and the rest apply. A value +// that is empty, names no known exporter, or lists none with an exporter +// keeps def, as the specification asks for a setting it cannot use. The +// returned set is the one to use even when err is not nil. +func resolveExporters(env, value string, known []string, def exporters) (exporters, error) { + // Unset and empty read the same, because templated manifests can render + // an empty variable. + if strings.TrimSpace(value) == "" { + return maps.Clone(def), nil + } + got := exporters{} + none := false + var skipped []string + for item := range strings.SplitSeq(value, ",") { + switch name := strings.ToLower(strings.TrimSpace(item)); { + case name == exporterNone: + none = true + case slices.Contains(known, name): + got[name] = true + default: + skipped = append(skipped, name) + } + } + var skipErr error + if len(skipped) > 0 { + skipErr = fmt.Errorf("%s: skipped empty or unknown exporters %q", env, skipped) + } + switch { + case none && len(got) > 0: + return maps.Clone(def), fmt.Errorf("%s %q lists none with an exporter", env, value) + case none: + return exporters{}, skipErr + case len(got) == 0: + return maps.Clone(def), fmt.Errorf("%s %q names no known exporter", env, value) + } + return got, skipErr +} + +// resolveSampler applies OTEL_TRACES_SAMPLER and OTEL_TRACES_SAMPLER_ARG on top +// of def. ax-server reads them itself because the SDK falls back to sampling +// every trace on a value it does not know, and reads a missing or invalid +// ratio as 1.0. Either would multiply the trace volume by ten, so ax-server +// keeps def instead and logs a warning. +func resolveSampler(ctx context.Context, def sdktrace.Sampler) sdktrace.Sampler { + name, nameSet := os.LookupEnv(tracesSamplerEnv) + arg, argSet := os.LookupEnv(tracesSamplerArgEnv) + s, err := parseSampler(name, nameSet, arg, argSet, def) + if err != nil { + slog.WarnContext(ctx, "invalid trace sampler setting, keeping the default", "sampler", name, "arg", arg, "default", def.Description(), "error", err) + } + return s +} + +// parseSampler accepts the sampler names the OpenTelemetry Go SDK knows. Any +// error means def was kept. +func parseSampler(name string, nameSet bool, arg string, argSet bool, def sdktrace.Sampler) (sdktrace.Sampler, error) { + name = strings.ToLower(strings.TrimSpace(name)) + // Unlike the SDK, an empty value reads as unset. + if !nameSet || name == "" { + return def, nil + } + var ratio float64 + if name == "traceidratio" || name == "parentbased_traceidratio" { + trimmed := strings.TrimSpace(arg) + if !argSet || trimmed == "" { + return def, fmt.Errorf("%s %q requires %s to be a ratio in [0, 1]", tracesSamplerEnv, name, tracesSamplerArgEnv) + } + parsed, err := strconv.ParseFloat(trimmed, 64) + if err != nil { + return def, fmt.Errorf("parsing %s %q: %w", tracesSamplerArgEnv, arg, err) + } + if parsed < 0 || parsed > 1 { + return def, fmt.Errorf("%s %q is outside [0, 1]", tracesSamplerArgEnv, arg) + } + ratio = parsed + } + switch name { + case "always_on": + return sdktrace.AlwaysSample(), nil + case "always_off": + return sdktrace.NeverSample(), nil + case "traceidratio": + return sdktrace.TraceIDRatioBased(ratio), nil + case "parentbased_always_on": + return sdktrace.ParentBased(sdktrace.AlwaysSample()), nil + case "parentbased_always_off": + return sdktrace.ParentBased(sdktrace.NeverSample()), nil + case "parentbased_traceidratio": + return sdktrace.ParentBased(sdktrace.TraceIDRatioBased(ratio)), nil + } + return def, fmt.Errorf("unsupported %s %q", tracesSamplerEnv, name) +} diff --git a/internal/telemetry/event.go b/internal/telemetry/event.go new file mode 100644 index 00000000..841c9e2d --- /dev/null +++ b/internal/telemetry/event.go @@ -0,0 +1,137 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package telemetry + +import ( + "context" + "log/slog" + "math" + "sync/atomic" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/log" + "go.opentelemetry.io/otel/log/global" +) + +// EventScope is the instrumentation scope of every event ax-server emits. +const EventScope = "github.com/google/ax/internal/telemetry" + +// Event is a kind of structured event, such as a task state change. Name is +// the OTLP log record's event name. Body and Severity belong to the kind, not +// the call site, so the OTLP and stdout forms of an event cannot differ. +type Event struct { + Name string + Body string + Severity log.Severity +} + +// level maps the event's severity onto slog's four levels. +func (ev Event) level() slog.Level { + switch { + case ev.Severity >= log.SeverityError: + return slog.LevelError + case ev.Severity >= log.SeverityWarn: + return slog.LevelWarn + case ev.Severity >= log.SeverityInfo: + return slog.LevelInfo + default: + return slog.LevelDebug + } +} + +// console makes Emit also write the stdout form of an event that went over +// OTLP. Init sets it from OTEL_LOGS_EXPORTER. +var console atomic.Bool + +// Emit records ev with attrs. It sends an OTLP log record when +// OTEL_LOGS_EXPORTER selects otlp, and writes the event as a JSON log line +// through slog.Default() when it does not, or also when console is selected. +// An event is therefore never dropped because OTLP is off. +// +// The OTLP record is batched, so a slow or unreachable collector never blocks +// the caller. A record still queued when the process dies is lost, unless +// console also wrote it to stdout. +func Emit(ctx context.Context, ev Event, attrs ...slog.Attr) { + t := time.Now() + if emitOTLP(ctx, ev, t, attrs) && !console.Load() { + return + } + h := slog.Default().Handler() + if !h.Enabled(ctx, ev.level()) { + return + } + rec := slog.NewRecord(t, ev.level(), ev.Body, 0) + rec.AddAttrs(slog.String("event", ev.Name)) + rec.AddAttrs(attrs...) + _ = h.Handle(ctx, rec) +} + +// emitOTLP sends the OTLP record and reports whether it did. Init installs a +// no-op provider when OTLP logs are off, so the check is cheap. +func emitOTLP(ctx context.Context, ev Event, t time.Time, attrs []slog.Attr) bool { + logger := global.GetLoggerProvider().Logger(EventScope) + if !logger.Enabled(ctx, log.EnabledParameters{Severity: ev.Severity, EventName: ev.Name}) { + return false + } + var rec log.Record + rec.SetEventName(ev.Name) + rec.SetTimestamp(t) + rec.SetSeverity(ev.Severity) + rec.SetBody(attribute.StringValue(ev.Body)) + kvs := make([]attribute.KeyValue, 0, len(attrs)) + for _, a := range attrs { + kvs = append(kvs, attribute.KeyValue{Key: attribute.Key(a.Key), Value: attrValue(a.Value)}) + } + rec.AddAttributes(kvs...) + // The SDK takes the trace and span IDs from ctx. + logger.Emit(ctx, rec) + return true +} + +// attrValue converts an slog value to the attribute kind slog's JSON handler +// writes, so the two forms of an event match. +func attrValue(v slog.Value) attribute.Value { + v = v.Resolve() + switch v.Kind() { + case slog.KindString: + return attribute.StringValue(v.String()) + case slog.KindInt64: + return attribute.Int64Value(v.Int64()) + case slog.KindUint64: + // OTel has no unsigned kind; clamp rather than wrap. + if u := v.Uint64(); u <= math.MaxInt64 { + return attribute.Int64Value(int64(u)) + } + return attribute.Int64Value(math.MaxInt64) + case slog.KindFloat64: + return attribute.Float64Value(v.Float64()) + case slog.KindBool: + return attribute.BoolValue(v.Bool()) + case slog.KindDuration: + // Nanoseconds, as slog's JSON handler writes them. + return attribute.Int64Value(int64(v.Duration())) + case slog.KindTime: + return attribute.StringValue(v.Time().Format(time.RFC3339Nano)) + case slog.KindGroup: + kvs := make([]attribute.KeyValue, 0, len(v.Group())) + for _, a := range v.Group() { + kvs = append(kvs, attribute.KeyValue{Key: attribute.Key(a.Key), Value: attrValue(a.Value)}) + } + return attribute.MapValue(kvs...) + default: + return attribute.StringValue(v.String()) + } +} diff --git a/internal/telemetry/log.go b/internal/telemetry/log.go index 57492acd..eab2c0cc 100644 --- a/internal/telemetry/log.go +++ b/internal/telemetry/log.go @@ -22,8 +22,9 @@ import ( ) // NewLogHandler wraps h so that a record logged with a context carrying a -// valid span gets trace_id and span_id attributes, joining log lines to -// traces. Records logged without such a context pass through unchanged. +// valid span gets trace_id, span_id, and trace_flags attributes, the fields +// the OpenTelemetry specification defines for trace context in non-OTLP logs. +// Records logged without such a context pass through unchanged. func NewLogHandler(h slog.Handler) slog.Handler { return traceHandler{h} } @@ -35,7 +36,11 @@ type traceHandler struct { func (h traceHandler) Handle(ctx context.Context, r slog.Record) error { if sc := trace.SpanContextFromContext(ctx); sc.IsValid() { r = r.Clone() - r.AddAttrs(slog.String("trace_id", sc.TraceID().String()), slog.String("span_id", sc.SpanID().String())) + r.AddAttrs( + slog.String("trace_id", sc.TraceID().String()), + slog.String("span_id", sc.SpanID().String()), + slog.String("trace_flags", sc.TraceFlags().String()), + ) } return h.Handler.Handle(ctx, r) } @@ -44,7 +49,7 @@ func (h traceHandler) WithAttrs(attrs []slog.Attr) slog.Handler { return traceHandler{h.Handler.WithAttrs(attrs)} } -// WithGroup nests later attributes, trace_id and span_id included, under name. +// WithGroup nests later attributes, the trace context included, under name. func (h traceHandler) WithGroup(name string) slog.Handler { return traceHandler{h.Handler.WithGroup(name)} } diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index e71c0c8d..ccb57bd7 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -12,16 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package telemetry sets up OpenTelemetry for ax-server: a tracer provider and -// a logger provider that export over OTLP/gRPC, a meter provider read by a -// Prometheus exporter, and a slog handler that stamps log lines with the -// current trace and span IDs. +// Package telemetry sets up OpenTelemetry for ax-server: a tracer provider, a +// meter provider, and a logger provider for structured events, all exporting +// over OTLP/gRPC; a Prometheus handler for the same metrics; and a slog +// handler that stamps log lines with the current trace context. // // Configuration follows the OpenTelemetry environment variables: // OTEL_EXPORTER_OTLP_ENDPOINT (and the per-signal variants) choose the -// collector, OTEL_TRACES_EXPORTER and OTEL_LOGS_EXPORTER choose otlp (the -// default) or none, OTEL_TRACES_SAMPLER overrides the default sampler, and -// OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES override the resource. +// collector; OTEL_TRACES_EXPORTER, OTEL_METRICS_EXPORTER, and +// OTEL_LOGS_EXPORTER choose the exporters; OTEL_TRACES_SAMPLER and +// OTEL_TRACES_SAMPLER_ARG override the default sampler; and OTEL_SERVICE_NAME +// and OTEL_RESOURCE_ATTRIBUTES override the resource. A value ax-server does +// not understand is logged and ignored, which keeps ax-server's default. package telemetry import ( @@ -31,16 +33,17 @@ import ( "log/slog" "net/http" "os" - "strings" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/collectors" "github.com/prometheus/client_golang/prometheus/promhttp" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" promexporter "go.opentelemetry.io/otel/exporters/prometheus" "go.opentelemetry.io/otel/log/global" + lognoop "go.opentelemetry.io/otel/log/noop" "go.opentelemetry.io/otel/propagation" sdklog "go.opentelemetry.io/otel/sdk/log" sdkmetric "go.opentelemetry.io/otel/sdk/metric" @@ -49,26 +52,17 @@ import ( semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) -const ( - // CardinalityLimit caps the attribute sets an instrument keeps per - // collection. Sets beyond it are folded into one overflow series, so a - // label that turns out to be unbounded cannot exhaust ax-server's memory or - // the metrics backend. - CardinalityLimit = 100 - // DefaultSampleRatio is the share of new traces sampled when - // OTEL_TRACES_SAMPLER is unset. A caller's sampling decision, carried in - // traceparent, is always followed. - DefaultSampleRatio = 0.1 - - exporterOTLP = "otlp" - exporterNone = "none" -) +// DefaultSampleRatio is the share of new traces sampled unless +// OTEL_TRACES_SAMPLER validly says otherwise. The default sampler is parent +// based, so a caller's sampling decision, carried in traceparent, is followed. +const DefaultSampleRatio = 0.1 // Providers holds the providers Init installed as the OpenTelemetry globals. type Providers struct { Tracer *sdktrace.TracerProvider Meter *sdkmetric.MeterProvider - // Logger is nil when OTEL_LOGS_EXPORTER turns OTLP logs off. + // Logger is nil when OTEL_LOGS_EXPORTER does not select otlp. Emit then + // writes events to stdout. Logger *sdklog.LoggerProvider registry *prometheus.Registry @@ -78,8 +72,10 @@ type Providers struct { // OpenTelemetry globals with the W3C trace context propagator, and routes SDK // errors to slog. Call Shutdown on the result before the process exits. // -// Nothing is exported until something records a span or emits a log record, -// so a deployment without a collector only sees errors once it does. +// Metrics are pushed every OTEL_METRIC_EXPORT_INTERVAL (60 s by default) while +// OTEL_METRICS_EXPORTER selects otlp, so without a reachable collector the SDK +// logs an export error at that interval. Traces and events are exported only +// once something records them. func Init(ctx context.Context, serviceName string) (*Providers, error) { // The SDK's default handler writes to stderr, outside the JSON logs. otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) { @@ -94,7 +90,7 @@ func Init(ctx context.Context, serviceName string) (*Providers, error) { if p.Tracer, err = newTracerProvider(ctx, res); err != nil { return nil, err } - if p.Meter, p.registry, err = newMeterProvider(res); err != nil { + if p.Meter, p.registry, err = newMeterProvider(ctx, res); err != nil { return nil, errors.Join(err, p.Shutdown(ctx)) } if p.Logger, err = newLoggerProvider(ctx, res); err != nil { @@ -105,6 +101,9 @@ func Init(ctx context.Context, serviceName string) (*Providers, error) { otel.SetMeterProvider(p.Meter) if p.Logger != nil { global.SetLoggerProvider(p.Logger) + } else { + // Emit checks the global provider; a no-op one sends events to stdout. + global.SetLoggerProvider(lognoop.NewLoggerProvider()) } // Only trace context crosses process boundaries. Baggage from a caller is // not trusted, so it is not propagated. @@ -113,7 +112,8 @@ func Init(ctx context.Context, serviceName string) (*Providers, error) { } // MetricsHandler serves the meter provider's metrics, together with Go -// runtime and process metrics, in the Prometheus exposition format. +// runtime and process metrics, in the Prometheus exposition format. The Go +// and process metrics are served here only, not pushed over OTLP. func (p *Providers) MetricsHandler() http.Handler { return promhttp.HandlerFor(p.registry, promhttp.HandlerOpts{}) } @@ -135,11 +135,12 @@ func (p *Providers) Shutdown(ctx context.Context) error { } // newResource describes the process. OTEL_SERVICE_NAME and -// OTEL_RESOURCE_ATTRIBUTES override the defaults. +// OTEL_RESOURCE_ATTRIBUTES override the defaults; the bundled manifests set +// service.instance.id to the pod UID and add the k8s.* attributes there. func newResource(ctx context.Context, serviceName string) (*resource.Resource, error) { attrs := resource.WithAttributes(semconv.ServiceName(serviceName)) opts := []resource.Option{resource.WithTelemetrySDK(), resource.WithSchemaURL(semconv.SchemaURL), attrs} - // In Kubernetes the hostname is the pod name, which tells replicas apart. + // Without OTEL_RESOURCE_ATTRIBUTES, the hostname tells replicas apart. if host, err := os.Hostname(); err == nil { opts = append(opts, resource.WithAttributes(semconv.ServiceInstanceID(host))) } @@ -156,12 +157,9 @@ func newResource(ctx context.Context, serviceName string) (*resource.Resource, e } func newTracerProvider(ctx context.Context, res *resource.Resource) (*sdktrace.TracerProvider, error) { - opts := []sdktrace.TracerProviderOption{sdktrace.WithResource(res)} - // Without WithSampler, the SDK reads OTEL_TRACES_SAMPLER itself. - if strings.TrimSpace(os.Getenv("OTEL_TRACES_SAMPLER")) == "" { - opts = append(opts, sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(DefaultSampleRatio)))) - } - export := exporterSelected(ctx, "OTEL_TRACES_EXPORTER") + def := sdktrace.ParentBased(sdktrace.TraceIDRatioBased(DefaultSampleRatio)) + opts := []sdktrace.TracerProviderOption{sdktrace.WithResource(res), sdktrace.WithSampler(resolveSampler(ctx, def))} + export := resolveExportersEnv(ctx, tracesExporterEnv, []string{exporterOTLP}, exporters{exporterOTLP: true}).has(exporterOTLP) if export { exp, err := otlptracegrpc.New(ctx) if err != nil { @@ -176,7 +174,14 @@ func newTracerProvider(ctx context.Context, res *resource.Resource) (*sdktrace.T return tp, nil } -func newMeterProvider(res *resource.Resource) (*sdkmetric.MeterProvider, *prometheus.Registry, error) { +// newMeterProvider builds a meter provider with a Prometheus reader, which +// MetricsHandler serves, and an OTLP periodic reader while +// OTEL_METRICS_EXPORTER selects otlp. The variable controls only the push: +// prometheus or none turns it off, and --metrics-addr controls /metrics. +// +// Each instrument keeps the SDK's default of 2000 attribute sets per +// collection, which OTEL_GO_X_CARDINALITY_LIMIT overrides. +func newMeterProvider(ctx context.Context, res *resource.Resource) (*sdkmetric.MeterProvider, *prometheus.Registry, error) { reg := prometheus.NewRegistry() if err := reg.Register(collectors.NewGoCollector()); err != nil { return nil, nil, fmt.Errorf("registering Go metrics: %w", err) @@ -184,24 +189,40 @@ func newMeterProvider(res *resource.Resource) (*sdkmetric.MeterProvider, *promet if err := reg.Register(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})); err != nil { return nil, nil, fmt.Errorf("registering process metrics: %w", err) } - reader, err := promexporter.New(promexporter.WithRegisterer(reg)) + promReader, err := promexporter.New(promexporter.WithRegisterer(reg)) if err != nil { return nil, nil, fmt.Errorf("creating Prometheus exporter: %w", err) } - mp := sdkmetric.NewMeterProvider( - sdkmetric.WithResource(res), - sdkmetric.WithReader(reader), - sdkmetric.WithCardinalityLimit(CardinalityLimit), - ) + opts := []sdkmetric.Option{sdkmetric.WithResource(res), sdkmetric.WithReader(promReader)} + + selected := resolveExportersEnv(ctx, metricsExporterEnv, []string{exporterOTLP, exporterPrometheus}, + exporters{exporterOTLP: true, exporterPrometheus: true}) + push := selected.has(exporterOTLP) + if push { + exp, err := otlpmetricgrpc.New(ctx) + if err != nil { + return nil, nil, fmt.Errorf("creating OTLP metric exporter: %w", err) + } + opts = append(opts, sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exp))) + } + mp := sdkmetric.NewMeterProvider(opts...) + slog.InfoContext(ctx, "metrics initialized", "export", push) return mp, reg, nil } // newLoggerProvider returns the provider for structured events, or nil when -// OTEL_LOGS_EXPORTER turns OTLP off. ax-server's own log lines go to stdout, -// not through it. +// OTEL_LOGS_EXPORTER does not select otlp. It also sets whether Emit writes +// events to stdout beside OTLP (console). ax-server's own log lines go to +// stdout, not through this provider. +// +// The default is none rather than the specification's otlp, as in Agent +// Substrate: events stay on stdout, where they are never lost, until an +// operator points them at a collector. func newLoggerProvider(ctx context.Context, res *resource.Resource) (*sdklog.LoggerProvider, error) { - if !exporterSelected(ctx, "OTEL_LOGS_EXPORTER") { - slog.InfoContext(ctx, "OTLP logs export disabled") + selected := resolveExportersEnv(ctx, logsExporterEnv, []string{exporterOTLP, exporterConsole}, exporters{}) + console.Store(selected.has(exporterConsole)) + if !selected.has(exporterOTLP) { + slog.InfoContext(ctx, "OTLP logs export disabled", "exporter", selected.String()) return nil, nil } exp, err := otlploggrpc.New(ctx) @@ -210,28 +231,6 @@ func newLoggerProvider(ctx context.Context, res *resource.Resource) (*sdklog.Log } // Batched, so an unreachable collector never blocks the caller emitting. lp := sdklog.NewLoggerProvider(sdklog.WithResource(res), sdklog.WithProcessor(sdklog.NewBatchProcessor(exp))) - slog.InfoContext(ctx, "OTLP logs export initialized") + slog.InfoContext(ctx, "OTLP logs export initialized", "exporter", selected.String()) return lp, nil } - -// exporterSelected reports whether the exporter list in the environment -// variable key selects otlp. Unset selects otlp. ax-server knows only otlp and -// none; other names are logged and ignored. -func exporterSelected(ctx context.Context, key string) bool { - val := strings.TrimSpace(os.Getenv(key)) - if val == "" { - return true - } - selected := false - for _, name := range strings.Split(val, ",") { - switch name = strings.ToLower(strings.TrimSpace(name)); name { - case exporterOTLP: - selected = true - case exporterNone: - return false - default: - slog.WarnContext(ctx, "ignoring unsupported OpenTelemetry exporter", "variable", key, "exporter", name) - } - } - return selected -} diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go index cd99e13c..1d5be9e2 100644 --- a/internal/telemetry/telemetry_test.go +++ b/internal/telemetry/telemetry_test.go @@ -21,34 +21,39 @@ import ( "fmt" "io" "log/slog" + "math" "net" "net/http" "net/http/httptest" + "slices" "strings" "sync" "testing" + "time" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" otellog "go.opentelemetry.io/otel/log" - "go.opentelemetry.io/otel/log/global" "go.opentelemetry.io/otel/metric" + sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/trace" collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" commonpb "go.opentelemetry.io/proto/otlp/common/v1" "google.golang.org/grpc" ) // fakeCollector is an in-process OTLP/gRPC collector that keeps what it -// receives. +// receives. It serves the trace service itself; logsService and +// metricsService serve the others, whose Export methods share the name. type fakeCollector struct { coltracepb.UnimplementedTraceServiceServer - collogspb.UnimplementedLogsServiceServer - mu sync.Mutex - spans []*coltracepb.ExportTraceServiceRequest - logs []*collogspb.ExportLogsServiceRequest + mu sync.Mutex + spans []*coltracepb.ExportTraceServiceRequest + logs []*collogspb.ExportLogsServiceRequest + metrics []*colmetricspb.ExportMetricsServiceRequest } func (c *fakeCollector) Export(_ context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) { @@ -58,8 +63,6 @@ func (c *fakeCollector) Export(_ context.Context, req *coltracepb.ExportTraceSer return &coltracepb.ExportTraceServiceResponse{}, nil } -// logsService adapts fakeCollector to the logs service, whose Export method -// has the same name as the trace service's. type logsService struct { collogspb.UnimplementedLogsServiceServer c *fakeCollector @@ -72,6 +75,18 @@ func (s logsService) Export(_ context.Context, req *collogspb.ExportLogsServiceR return &collogspb.ExportLogsServiceResponse{}, nil } +type metricsService struct { + colmetricspb.UnimplementedMetricsServiceServer + c *fakeCollector +} + +func (s metricsService) Export(_ context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) { + s.c.mu.Lock() + defer s.c.mu.Unlock() + s.c.metrics = append(s.c.metrics, req) + return &colmetricspb.ExportMetricsServiceResponse{}, nil +} + // startCollector starts a fakeCollector and points the OTLP exporters at it. func startCollector(t *testing.T) *fakeCollector { t.Helper() @@ -83,12 +98,21 @@ func startCollector(t *testing.T) *fakeCollector { srv := grpc.NewServer() coltracepb.RegisterTraceServiceServer(srv, c) collogspb.RegisterLogsServiceServer(srv, logsService{c: c}) + colmetricspb.RegisterMetricsServiceServer(srv, metricsService{c: c}) go func() { _ = srv.Serve(lis) }() t.Cleanup(srv.Stop) t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://"+lis.Addr().String()) return c } +// noExport turns off every OTLP exporter, for tests without a collector. +func noExport(t *testing.T) { + t.Helper() + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_METRICS_EXPORTER", "prometheus") + t.Setenv("OTEL_LOGS_EXPORTER", "none") +} + // initForTest runs Init and shuts the providers down when the test ends. func initForTest(t *testing.T) *Providers { t.Helper() @@ -100,6 +124,34 @@ func initForTest(t *testing.T) *Providers { return p } +// captureLogs points slog.Default() at a JSON buffer until the test ends. +func captureLogs(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return &buf +} + +// stdoutEvents returns the log lines in buf that carry an event attribute. +func stdoutEvents(t *testing.T, buf *bytes.Buffer) []map[string]any { + t.Helper() + var events []map[string]any + for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") { + var rec map[string]any + if err := json.Unmarshal([]byte(line), &rec); err != nil { + t.Fatalf("log line is not JSON: %v\n%s", err, line) + } + if _, ok := rec["event"]; ok { + events = append(events, rec) + } + } + return events +} + +var testEvent = Event{Name: "ax.test.event", Body: "Test event", Severity: otellog.SeverityInfo} + func resourceAttr(attrs []*commonpb.KeyValue, key string) string { for _, kv := range attrs { if kv.GetKey() == key { @@ -109,19 +161,23 @@ func resourceAttr(attrs []*commonpb.KeyValue, key string) string { return "" } -func TestInitExportsSpansAndLogsOverOTLP(t *testing.T) { +func TestInitExportsAllSignalsOverOTLP(t *testing.T) { c := startCollector(t) t.Setenv("OTEL_TRACES_SAMPLER", "always_on") + t.Setenv("OTEL_LOGS_EXPORTER", "otlp") + buf := captureLogs(t) p := initForTest(t) _, span := otel.Tracer("test").Start(context.Background(), "test-span") span.End() - var rec otellog.Record - rec.SetEventName("ax.test.event") - rec.SetBody(attribute.StringValue("hello")) - global.GetLoggerProvider().Logger("test").Emit(context.Background(), rec) + counter, err := otel.Meter("test").Int64Counter("ax.test.requests") + if err != nil { + t.Fatal(err) + } + counter.Add(context.Background(), 1) + Emit(context.Background(), testEvent, slog.String("ax.task.id", "t-1"), slog.Int("count", 3)) - // Shutdown flushes both batchers. + // Shutdown flushes the batchers and runs a last metrics export. if err := p.Shutdown(context.Background()); err != nil { t.Fatalf("Shutdown: %v", err) } @@ -138,12 +194,35 @@ func TestInitExportsSpansAndLogsOverOTLP(t *testing.T) { if name := rs.GetScopeSpans()[0].GetSpans()[0].GetName(); name != "test-span" { t.Errorf("span name = %q, want test-span", name) } + + if len(c.metrics) == 0 || len(c.metrics[0].GetResourceMetrics()) == 0 { + t.Fatal("collector received no metrics") + } + var names []string + for _, rm := range c.metrics[0].GetResourceMetrics() { + for _, sm := range rm.GetScopeMetrics() { + for _, m := range sm.GetMetrics() { + names = append(names, m.GetName()) + } + } + } + if !slices.Contains(names, "ax.test.requests") { + t.Errorf("pushed metrics = %v, want ax.test.requests", names) + } + if len(c.logs) == 0 || len(c.logs[0].GetResourceLogs()) == 0 { t.Fatal("collector received no log records") } lr := c.logs[0].GetResourceLogs()[0].GetScopeLogs()[0].GetLogRecords()[0] - if lr.GetEventName() != "ax.test.event" || lr.GetBody().GetStringValue() != "hello" { - t.Errorf("log record = %v, want event ax.test.event with body hello", lr) + if lr.GetEventName() != testEvent.Name || lr.GetBody().GetStringValue() != testEvent.Body { + t.Errorf("log record = %v, want event %s with body %q", lr, testEvent.Name, testEvent.Body) + } + if got := resourceAttr(lr.GetAttributes(), "ax.task.id"); got != "t-1" { + t.Errorf("log record ax.task.id = %q, want t-1", got) + } + // Without console, an event sent over OTLP is not also written to stdout. + if events := stdoutEvents(t, buf); len(events) != 0 { + t.Errorf("event also written to stdout without console: %v", events) } } @@ -151,6 +230,7 @@ func TestInitHonorsExporterNone(t *testing.T) { c := startCollector(t) t.Setenv("OTEL_TRACES_SAMPLER", "always_on") t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_METRICS_EXPORTER", "none") t.Setenv("OTEL_LOGS_EXPORTER", "none") p := initForTest(t) if p.Logger != nil { @@ -163,36 +243,101 @@ func TestInitHonorsExporterNone(t *testing.T) { t.Error("span not sampled with OTEL_TRACES_EXPORTER=none") } span.End() + counter, err := otel.Meter("test").Int64Counter("ax.test.requests") + if err != nil { + t.Fatal(err) + } + counter.Add(context.Background(), 1) + // /metrics still serves with the push off. + if body := scrape(t, p.MetricsHandler()); !strings.Contains(body, "ax_test_requests_total") { + t.Error("/metrics has no ax_test_requests_total with OTEL_METRICS_EXPORTER=none") + } if err := p.Shutdown(context.Background()); err != nil { t.Fatalf("Shutdown: %v", err) } c.mu.Lock() defer c.mu.Unlock() - if len(c.spans) != 0 || len(c.logs) != 0 { - t.Errorf("collector received %d span and %d log exports, want none", len(c.spans), len(c.logs)) + if len(c.spans) != 0 || len(c.metrics) != 0 || len(c.logs) != 0 { + t.Errorf("collector received %d span, %d metric, and %d log exports, want none", len(c.spans), len(c.metrics), len(c.logs)) } } -func TestDefaultSamplerFollowsParentAndSamplesTenthOfRoots(t *testing.T) { +func TestEmitWritesEventsToStdoutByDefault(t *testing.T) { + c := startCollector(t) t.Setenv("OTEL_TRACES_EXPORTER", "none") - t.Setenv("OTEL_LOGS_EXPORTER", "none") - initForTest(t) - tracer := otel.Tracer("test") + t.Setenv("OTEL_METRICS_EXPORTER", "none") + buf := captureLogs(t) + p := initForTest(t) + if p.Logger != nil { + t.Error("Logger provider built with OTEL_LOGS_EXPORTER unset; the default is none") + } + + Emit(context.Background(), testEvent, slog.String("ax.task.id", "t-1")) + if err := p.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + + events := stdoutEvents(t, buf) + if len(events) != 1 { + t.Fatalf("got %d stdout events, want 1:\n%s", len(events), buf.String()) + } + if ev := events[0]; ev["event"] != testEvent.Name || ev["msg"] != testEvent.Body || ev["level"] != "INFO" || ev["ax.task.id"] != "t-1" { + t.Errorf("stdout event = %v", ev) + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.logs) != 0 { + t.Errorf("collector received %d log exports, want none", len(c.logs)) + } +} - const roots = 4000 +func TestEmitWithConsoleWritesBoth(t *testing.T) { + c := startCollector(t) + t.Setenv("OTEL_TRACES_EXPORTER", "none") + t.Setenv("OTEL_METRICS_EXPORTER", "none") + t.Setenv("OTEL_LOGS_EXPORTER", "otlp,console") + buf := captureLogs(t) + p := initForTest(t) + + Emit(context.Background(), testEvent) + if err := p.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown: %v", err) + } + + if events := stdoutEvents(t, buf); len(events) != 1 { + t.Errorf("got %d stdout events, want 1:\n%s", len(events), buf.String()) + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.logs) != 1 { + t.Errorf("collector received %d log exports, want 1", len(c.logs)) + } +} + +// sampledShare starts n root spans and returns the share sampled. +func sampledShare(n int) float64 { + tracer := otel.Tracer("test") sampled := 0 - for range roots { + for range n { _, span := tracer.Start(context.Background(), "root") if span.SpanContext().IsSampled() { sampled++ } span.End() } + return float64(sampled) / float64(n) +} + +func TestDefaultSamplerFollowsParentAndSamplesTenthOfRoots(t *testing.T) { + noExport(t) + initForTest(t) + // 10% of 4000 is 400, with a standard deviation of about 19. - if sampled < 300 || sampled > 500 { - t.Errorf("%d of %d root spans sampled, want about %d", sampled, roots, int(roots*DefaultSampleRatio)) + if got := sampledShare(4000); got < 0.075 || got > 0.125 { + t.Errorf("%.3f of root spans sampled, want about %.1f", got, DefaultSampleRatio) } + tracer := otel.Tracer("test") for _, parentSampled := range []bool{true, false} { flags := trace.TraceFlags(0) if parentSampled { @@ -209,9 +354,57 @@ func TestDefaultSamplerFollowsParentAndSamplesTenthOfRoots(t *testing.T) { } } +func TestInvalidSamplerSettingKeepsDefault(t *testing.T) { + for _, tt := range []struct{ sampler, arg string }{ + {"parentbased_traceidratoi", "0.1"}, // typo + {"parentbased_traceidratio", ""}, + {"parentbased_traceidratio", "10"}, + } { + t.Run(tt.sampler+"="+tt.arg, func(t *testing.T) { + noExport(t) + t.Setenv("OTEL_TRACES_SAMPLER", tt.sampler) + t.Setenv("OTEL_TRACES_SAMPLER_ARG", tt.arg) + initForTest(t) + if got := sampledShare(2000); got < 0.065 || got > 0.135 { + t.Errorf("%.3f of root spans sampled, want about %.1f", got, DefaultSampleRatio) + } + }) + } +} + +func TestParseSampler(t *testing.T) { + def := sdktrace.ParentBased(sdktrace.TraceIDRatioBased(DefaultSampleRatio)) + tests := []struct { + name, arg string + nameSet bool + argSet bool + want string // the sampler's Description + wantErr bool + }{ + {name: "", want: def.Description()}, + {name: " ", nameSet: true, want: def.Description()}, + {name: "always_on", nameSet: true, want: "AlwaysOnSampler"}, + {name: "ALWAYS_OFF", nameSet: true, want: "AlwaysOffSampler"}, + {name: "traceidratio", nameSet: true, arg: "0.5", argSet: true, want: "TraceIDRatioBased{0.5}"}, + {name: "parentbased_traceidratio", nameSet: true, arg: " 0.25 ", argSet: true, want: sdktrace.ParentBased(sdktrace.TraceIDRatioBased(0.25)).Description()}, + {name: "parentbased_always_on", nameSet: true, want: sdktrace.ParentBased(sdktrace.AlwaysSample()).Description()}, + {name: "parentbased_always_off", nameSet: true, want: sdktrace.ParentBased(sdktrace.NeverSample()).Description()}, + {name: "parentbased_traceidratoi", nameSet: true, arg: "0.1", argSet: true, want: def.Description(), wantErr: true}, + {name: "jaeger_remote", nameSet: true, want: def.Description(), wantErr: true}, + {name: "traceidratio", nameSet: true, want: def.Description(), wantErr: true}, + {name: "traceidratio", nameSet: true, arg: "half", argSet: true, want: def.Description(), wantErr: true}, + {name: "traceidratio", nameSet: true, arg: "-0.1", argSet: true, want: def.Description(), wantErr: true}, + } + for _, tt := range tests { + got, err := parseSampler(tt.name, tt.nameSet, tt.arg, tt.argSet, def) + if got.Description() != tt.want || (err != nil) != tt.wantErr { + t.Errorf("parseSampler(%q, %q) = %s, %v; want %s, error %v", tt.name, tt.arg, got.Description(), err, tt.want, tt.wantErr) + } + } +} + func TestPropagatorCarriesTraceContextOnly(t *testing.T) { - t.Setenv("OTEL_TRACES_EXPORTER", "none") - t.Setenv("OTEL_LOGS_EXPORTER", "none") + noExport(t) initForTest(t) got := strings.Join(otel.GetTextMapPropagator().Fields(), ",") if got != "traceparent,tracestate" { @@ -235,19 +428,17 @@ func scrape(t *testing.T, h http.Handler) string { return string(body) } -func TestMetricsHandlerServesInstrumentsWithCardinalityCap(t *testing.T) { - t.Setenv("OTEL_TRACES_EXPORTER", "none") - t.Setenv("OTEL_LOGS_EXPORTER", "none") - p := initForTest(t) - +// countSeries adds n attribute sets to a counter and returns the number of +// series /metrics serves for it, and the body. +func countSeries(t *testing.T, p *Providers, n int) (int, string) { + t.Helper() counter, err := otel.Meter("test").Int64Counter("ax.test.requests") if err != nil { t.Fatal(err) } - for i := range CardinalityLimit * 3 { + for i := range n { counter.Add(context.Background(), 1, metric.WithAttributes(attribute.String("key", fmt.Sprint(i)))) } - body := scrape(t, p.MetricsHandler()) series := 0 for _, line := range strings.Split(body, "\n") { @@ -255,34 +446,90 @@ func TestMetricsHandlerServesInstrumentsWithCardinalityCap(t *testing.T) { series++ } } - // The limit counts the overflow series, so the 300 sets become 100 series. - if series != CardinalityLimit { - t.Errorf("ax_test_requests_total has %d series, want %d", series, CardinalityLimit) + return series, body +} + +func TestMetricsHandlerServesInstrumentsAndRuntimeMetrics(t *testing.T) { + noExport(t) + p := initForTest(t) + + // 300 sets stay under the SDK's default limit of 2000. + series, body := countSeries(t, p, 300) + if series != 300 { + t.Errorf("ax_test_requests_total has %d series, want 300", series) } - for _, want := range []string{`otel_metric_overflow="true"`, `service_name="ax-server"`, "go_goroutines", "process_cpu_seconds_total"} { + for _, want := range []string{`service_name="ax-server"`, "go_goroutines", "process_cpu_seconds_total"} { if !strings.Contains(body, want) { t.Errorf("/metrics output has no %s", want) } } + if strings.Contains(body, "otel_metric_overflow") { + t.Error("/metrics has an overflow series under the default cardinality limit") + } +} + +func TestCardinalityLimitFollowsEnvironment(t *testing.T) { + noExport(t) + t.Setenv("OTEL_GO_X_CARDINALITY_LIMIT", "100") + p := initForTest(t) + + // The limit counts the overflow series, so the 300 sets become 100 series. + series, body := countSeries(t, p, 300) + if series != 100 { + t.Errorf("ax_test_requests_total has %d series, want 100", series) + } + if !strings.Contains(body, `otel_metric_overflow="true"`) { + t.Error("/metrics has no overflow series") + } +} + +func TestResolveExporters(t *testing.T) { + known := []string{exporterOTLP, exporterConsole} + def := exporters{exporterOTLP: true} + tests := []struct { + val string + want string + wantErr bool + }{ + {"", "otlp", false}, + {" ", "otlp", false}, + {"otlp", "otlp", false}, + {" OTLP ", "otlp", false}, + {"console", "console", false}, + {"otlp,console", "console,otlp", false}, + {"none", "none", false}, + {"none,", "none", true}, + {"otlp,none", "otlp", true}, // contradictory: keep the default + {"otpl", "otlp", true}, // typo: keep the default + {"zipkin", "otlp", true}, + {"zipkin,console", "console", true}, + } + for _, tt := range tests { + got, err := resolveExporters("OTEL_TEST_EXPORTER", tt.val, known, def) + if got.String() != tt.want || (err != nil) != tt.wantErr { + t.Errorf("resolveExporters(%q) = %s, %v; want %s, error %v", tt.val, got, err, tt.want, tt.wantErr) + } + } } -func TestExporterSelected(t *testing.T) { +func TestAttrValueMatchesJSONKinds(t *testing.T) { + ts := time.Date(2026, 11, 6, 21, 0, 0, 123, time.UTC) tests := []struct { - val string - want bool + in slog.Value + want attribute.Value }{ - {"", true}, - {"otlp", true}, - {" OTLP ", true}, - {"none", false}, - {"otlp,none", false}, - {"console", false}, - {"console,otlp", true}, + {slog.StringValue("s"), attribute.StringValue("s")}, + {slog.IntValue(-3), attribute.Int64Value(-3)}, + {slog.Uint64Value(math.MaxUint64), attribute.Int64Value(math.MaxInt64)}, + {slog.Float64Value(1.5), attribute.Float64Value(1.5)}, + {slog.BoolValue(true), attribute.BoolValue(true)}, + {slog.DurationValue(time.Second), attribute.Int64Value(int64(time.Second))}, + {slog.TimeValue(ts), attribute.StringValue("2026-11-06T21:00:00.000000123Z")}, + {slog.GroupValue(slog.String("k", "v")), attribute.MapValue(attribute.String("k", "v"))}, } for _, tt := range tests { - t.Setenv("OTEL_TEST_EXPORTER", tt.val) - if got := exporterSelected(context.Background(), "OTEL_TEST_EXPORTER"); got != tt.want { - t.Errorf("exporterSelected(%q) = %v, want %v", tt.val, got, tt.want) + if got := attrValue(tt.in); got != tt.want { + t.Errorf("attrValue(%v) = %v, want %v", tt.in, got.Emit(), tt.want.Emit()) } } } @@ -315,8 +562,8 @@ func TestLogHandlerAddsTraceAndSpanIDs(t *testing.T) { } continue } - if rec["trace_id"] != sc.TraceID().String() || rec["span_id"] != sc.SpanID().String() { - t.Errorf("line %d trace_id=%v span_id=%v, want %s %s", i, rec["trace_id"], rec["span_id"], sc.TraceID(), sc.SpanID()) + if rec["trace_id"] != sc.TraceID().String() || rec["span_id"] != sc.SpanID().String() || rec["trace_flags"] != "01" { + t.Errorf("line %d trace_id=%v span_id=%v trace_flags=%v, want %s %s 01", i, rec["trace_id"], rec["span_id"], rec["trace_flags"], sc.TraceID(), sc.SpanID()) } } if !strings.Contains(lines[1], `"component":"test"`) { diff --git a/skills/substrate/SKILL.md b/skills/substrate/SKILL.md index d236e54d..9f177eea 100644 --- a/skills/substrate/SKILL.md +++ b/skills/substrate/SKILL.md @@ -56,6 +56,7 @@ All Control API traffic goes through this package. It encapsulates: | `api.ate-system.svc.cluster.local:443` + authority `api.ate-system.svc` | `cmd/ax-server/main.go` (`--substrate-endpoint` / `--substrate-authority` flag defaults), `internal/substrate/client.go` (3×), `deploy/ax-server.yaml`, `test/e2e/manifests/ax-server.yaml`, `test/e2e/harness_test.go` | | `atenet-router.ate-system.svc[.cluster.local]` | `internal/controller/reconciler.go` (`defaultRouterServerName`), `internal/tunnel/router_tls.go` (`routerServerName`), `cmd/ax/main.go` (ssh port-forward path), `internal/guest/client.go` (doc comment), `deploy/ax-server.yaml`, `test/e2e/manifests/ax-server.yaml`, `docs/networking.md` | | namespace `ate-system` | all of the above + `test/e2e/harness_test.go` (`ateSystemNamespace`) | +| `opentelemetry-collector.otel-system.svc:4317` (the collector Substrate's kind overlay deploys, `manifests/ate-install/kind/ate-otel-config.yaml` in Substrate) | `test/e2e/manifests/ax-server.yaml` (`ax-otel-config` ConfigMap). An unreachable address does not fail E2E; ax-server only logs OTLP export errors | Why it matters: tasks never get a Service/Ingress. Every request to a task is proxied by the **atenet router**, which picks the target from the diff --git a/test/e2e/manifests/ax-server.yaml b/test/e2e/manifests/ax-server.yaml index b247106c..79656939 100644 --- a/test/e2e/manifests/ax-server.yaml +++ b/test/e2e/manifests/ax-server.yaml @@ -26,6 +26,22 @@ roleRef: kind: ClusterRole name: ax-server --- +# OpenTelemetry settings for ax-server, read with envFrom. Point ax-server at a +# collector here; see docs/observability.md. On a cluster that runs Agent +# Substrate, the endpoint can be the one in ate-system's ate-otel-config. +# Editing this ConfigMap does not restart ax-server; follow a change with +# `kubectl -n ${AX_SYSTEM_NAMESPACE} rollout restart deployment/ax-server`. +apiVersion: v1 +kind: ConfigMap +metadata: + name: ax-otel-config + namespace: ${AX_SYSTEM_NAMESPACE} +data: + # The collector Substrate's kind overlay deploys (hack/setup-e2e-kind.sh + # installs Substrate with --kind). console keeps events in kubectl logs too. + OTEL_EXPORTER_OTLP_ENDPOINT: http://opentelemetry-collector.otel-system.svc:4317 + OTEL_LOGS_EXPORTER: otlp,console +--- apiVersion: apps/v1 kind: Deployment metadata: @@ -58,7 +74,26 @@ spec: - "--substrate-ca-file=/run/servicedns-ca/trust-bundle.pem" - "--template=default-template" - "--template-atespace=${AX_SYSTEM_NAMESPACE}" + envFrom: + - configMapRef: + name: ax-otel-config env: + # The resource attributes come from the downward API. Kubernetes + # expands $(VAR) only for a variable listed earlier. + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid + - name: OTEL_RESOURCE_ATTRIBUTES + value: "k8s.namespace.name=$(POD_NAMESPACE),k8s.pod.name=$(POD_NAME),k8s.pod.uid=$(POD_UID),service.instance.id=$(POD_UID)" - name: PGPASSWORD valueFrom: secretKeyRef: