diff --git a/AGENTS.md b/AGENTS.md index d9363c20..c3b38a0c 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 (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 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..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: @@ -92,6 +130,8 @@ spec: ports: - containerPort: 8080 name: http + - containerPort: 9090 + name: metrics resources: requests: cpu: 100m @@ -162,6 +202,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..55dde0ad --- /dev/null +++ b/docs/observability.md @@ -0,0 +1,106 @@ +# Observability + +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 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. `--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 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` | `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. | + +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`, `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 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_*`); they are not pushed over OTLP. + +## 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`, `ate.atespace`, and `ate.actor.uid` never appear on metrics. diff --git a/go.mod b/go.mod index 968855e5..38d369ca 100644 --- a/go.mod +++ b/go.mod @@ -6,21 +6,46 @@ 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/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 + 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..308c4e13 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,38 @@ 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/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= +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 +106,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..7db6c5fe --- /dev/null +++ b/internal/axattr/axattr.go @@ -0,0 +1,59 @@ +// 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. +// +// 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" + +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/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 new file mode 100644 index 00000000..eab2c0cc --- /dev/null +++ b/internal/telemetry/log.go @@ -0,0 +1,55 @@ +// 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, 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} +} + +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()), + slog.String("trace_flags", sc.TraceFlags().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, 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 new file mode 100644 index 00000000..ccb57bd7 --- /dev/null +++ b/internal/telemetry/telemetry.go @@ -0,0 +1,236 @@ +// 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, 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, 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 ( + "context" + "errors" + "fmt" + "log/slog" + "net/http" + "os" + + "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" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" +) + +// 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 does not select otlp. Emit then + // writes events to stdout. + 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. +// +// 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) { + 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(ctx, 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) + } 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. + 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. 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{}) +} + +// 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; 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} + // Without OTEL_RESOURCE_ATTRIBUTES, the hostname 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) { + 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 { + 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 +} + +// 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) + } + if err := reg.Register(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})); err != nil { + return nil, nil, fmt.Errorf("registering process metrics: %w", err) + } + promReader, err := promexporter.New(promexporter.WithRegisterer(reg)) + if err != nil { + return nil, nil, fmt.Errorf("creating Prometheus exporter: %w", err) + } + 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 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) { + 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) + 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", "exporter", selected.String()) + return lp, nil +} diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go new file mode 100644 index 00000000..1d5be9e2 --- /dev/null +++ b/internal/telemetry/telemetry_test.go @@ -0,0 +1,572 @@ +// 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" + "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/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. It serves the trace service itself; logsService and +// metricsService serve the others, whose Export methods share the name. +type fakeCollector struct { + coltracepb.UnimplementedTraceServiceServer + + mu sync.Mutex + spans []*coltracepb.ExportTraceServiceRequest + logs []*collogspb.ExportLogsServiceRequest + metrics []*colmetricspb.ExportMetricsServiceRequest +} + +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 +} + +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 +} + +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() + 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}) + 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() + p, err := Init(context.Background(), "ax-server") + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = p.Shutdown(context.Background()) }) + 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 { + return kv.GetValue().GetStringValue() + } + } + return "" +} + +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() + 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 the batchers and runs a last metrics export. + 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.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() != 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) + } +} + +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 { + 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() + 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.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 TestEmitWritesEventsToStdoutByDefault(t *testing.T) { + c := startCollector(t) + t.Setenv("OTEL_TRACES_EXPORTER", "none") + 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)) + } +} + +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 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 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 { + 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 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) { + noExport(t) + 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) +} + +// 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 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") { + if strings.HasPrefix(line, "ax_test_requests_total{") { + series++ + } + } + 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{`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 TestAttrValueMatchesJSONKinds(t *testing.T) { + ts := time.Date(2026, 11, 6, 21, 0, 0, 123, time.UTC) + tests := []struct { + in slog.Value + want attribute.Value + }{ + {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 { + if got := attrValue(tt.in); got != tt.want { + t.Errorf("attrValue(%v) = %v, want %v", tt.in, got.Emit(), tt.want.Emit()) + } + } +} + +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() || 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"`) { + t.Errorf("WithAttrs attribute lost: %s", lines[1]) + } +} 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 63a896f0..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: @@ -71,6 +106,8 @@ spec: ports: - containerPort: 8080 name: http + - containerPort: 9090 + name: metrics resources: requests: cpu: 100m @@ -141,5 +178,8 @@ spec: - port: 8080 targetPort: 8080 name: http + - port: 9090 + targetPort: 9090 + name: metrics selector: app.kubernetes.io/name: ax-server