diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 60cca4c3c..0ab4d8310 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -43,6 +43,7 @@ import ( "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/credbundle" "github.com/agent-substrate/substrate/internal/imagecache" + "github.com/agent-substrate/substrate/internal/otlprelay" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/resources" @@ -95,6 +96,8 @@ var ( showVersion = pflag.Bool("version", false, "Print version and exit.") logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") + otlpRelaySocket = pflag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), "Unix socket to serve the OTLP relay on, which forwards the node's ateom telemetry to OTEL_EXPORTER_OTLP_ENDPOINT so worker pods need no network path to the collector. Empty disables the relay.") + drainDelay = pflag.Duration("drain-delay", 0, "How long to keep accepting new RPCs after SIGTERM before starting the gRPC drain.") drainTimeout = pflag.Duration("drain-timeout", 5*time.Minute, "Deadline for the graceful gRPC drain on shutdown. In-flight RPCs still running past it are forcefully cancelled.") ) @@ -150,6 +153,27 @@ func main() { EnableHealthz: true, }) + // The OTLP relay lets the ateom pods on this node export telemetry over a + // unix socket instead of their own network (see internal/otlprelay). Started + // early: an ateom that finds no socket at startup falls back to exporting + // directly for its whole life, so the socket should exist before any worker + // pod on this node boots. + if relay, err := otlprelay.NewServer(ctx, *otlpRelaySocket); err != nil { + slog.ErrorContext(ctx, "Failed to create the OTLP relay; ateoms will export directly", slog.Any("err", err)) + } else if relay != nil { + // Deferred rather than tied to the drain: the relay carries other + // processes' telemetry, so it should outlive atelet's own RPC serving + // and stay up while the ateoms it serves are themselves shutting down. + defer relay.Stop() + go func() { + if err := relay.Serve(ctx); err != nil { + // Not fatal: atelet's actual job does not depend on the relay, + // and the ateoms fall back to exporting directly. + slog.ErrorContext(ctx, "OTLP relay stopped", slog.Any("err", err)) + } + }() + } + ateomDialer := &AteomDialer{ conns: lru.New(256), } diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 80cd2b793..d03208377 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -38,6 +38,7 @@ import ( "github.com/agent-substrate/substrate/internal/atunnel" "github.com/agent-substrate/substrate/internal/contextlogging" "github.com/agent-substrate/substrate/internal/imagecache" + "github.com/agent-substrate/substrate/internal/otlprelay" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/readyz" "github.com/agent-substrate/substrate/internal/resources" @@ -66,6 +67,9 @@ var ( showVersion = pflag.Bool("version", false, "Print version and exit.") logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") + otlpRelaySocket = pflag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), + "Unix socket of atelet's OTLP relay to export telemetry through, keeping it off the pod network. Empty, or absent at startup, exports directly to OTEL_EXPORTER_OTLP_ENDPOINT instead.") + reapLock sync.RWMutex ) @@ -102,16 +106,39 @@ func do(ctx context.Context) error { slog.InfoContext(ctx, "ateom booting") const serviceName = "ateom-gvisor" + // Export through atelet's node-local relay when it is there, so telemetry + // never touches the worker pod's network. A nil conn means it is not, and + // both providers fall back to dialing the collector directly. + // + // A relay that cannot be dialed is logged rather than fatal, matching both + // ends of the same decision: Dial already treats an absent socket as a + // fallback rather than an error, and atelet logs and keeps going when it + // cannot serve the relay at all. What is lost here is the node-local export + // path, not the ateom's ability to run actors, and failing the worker pod + // over its telemetry route would turn a misconfigured flag into an outage. + relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket) + if err != nil { + slog.ErrorContext(ctx, "Failed to connect to the OTLP relay; exporting telemetry directly over the pod network", + slog.String("socket", *otlpRelaySocket), slog.Any("err", err)) + } + if relayConn != nil { + defer relayConn.Close() + } + tp, err := serverboot.InitTracing(ctx, serverboot.TracingOptions{ - ServiceName: serviceName, - Sampling: serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)), + ServiceName: serviceName, + Sampling: serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)), + ExporterConn: relayConn, + // So the spans say which path they took, including when relayConn is nil + // because the dial above failed and this ateom is exporting directly. + RelayCapable: true, }) if err != nil { serverboot.Fatal(ctx, "Failed to initialize tracing", err) } defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown) - mp, err := serverboot.InitMetricsPushOnly(ctx, serviceName) + mp, err := serverboot.InitMetricsPushOnlyVia(ctx, serviceName, relayConn) if err != nil { serverboot.Fatal(ctx, "Failed to initialize metrics", err) } diff --git a/cmd/ateom-microvm/main.go b/cmd/ateom-microvm/main.go index d0fe13029..86b1d9345 100644 --- a/cmd/ateom-microvm/main.go +++ b/cmd/ateom-microvm/main.go @@ -42,6 +42,7 @@ import ( "github.com/agent-substrate/substrate/internal/ateomnet" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/atunnel" + "github.com/agent-substrate/substrate/internal/otlprelay" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/serverboot" "github.com/agent-substrate/substrate/internal/version" @@ -60,6 +61,9 @@ var ( showVersion = flag.Bool("version", false, "Print version and exit.") logLevelFlag = flag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.") + otlpRelaySocket = flag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), + "Unix socket of atelet's OTLP relay to export telemetry through, keeping it off the pod network. Empty, or absent at startup, exports directly to OTEL_EXPORTER_OTLP_ENDPOINT instead.") + atunnelListenAddress = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS") workerCredentialBundle = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS") podIdentityTrustBundle = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet") @@ -101,16 +105,39 @@ func do(ctx context.Context) error { slog.InfoContext(ctx, "ateom-microvm booting", slog.String("version", version.String())) const serviceName = "ateom-microvm" + // Export through atelet's node-local relay when it is there, so telemetry + // never touches the worker pod's network. A nil conn means it is not, and + // both providers fall back to dialing the collector directly. + // + // A relay that cannot be dialed is logged rather than fatal, matching both + // ends of the same decision: Dial already treats an absent socket as a + // fallback rather than an error, and atelet logs and keeps going when it + // cannot serve the relay at all. What is lost here is the node-local export + // path, not the ateom's ability to run actors, and failing the worker pod + // over its telemetry route would turn a misconfigured flag into an outage. + relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket) + if err != nil { + slog.ErrorContext(ctx, "Failed to connect to the OTLP relay; exporting telemetry directly over the pod network", + slog.String("socket", *otlpRelaySocket), slog.Any("err", err)) + } + if relayConn != nil { + defer relayConn.Close() + } + tp, err := serverboot.InitTracing(ctx, serverboot.TracingOptions{ - ServiceName: serviceName, - Sampling: serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)), + ServiceName: serviceName, + Sampling: serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)), + ExporterConn: relayConn, + // So the spans say which path they took, including when relayConn is nil + // because the dial above failed and this ateom is exporting directly. + RelayCapable: true, }) if err != nil { serverboot.Fatal(ctx, "Failed to initialize tracing", err) } defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown) - mp, err := serverboot.InitMetricsPushOnly(ctx, serviceName) + mp, err := serverboot.InitMetricsPushOnlyVia(ctx, serviceName, relayConn) if err != nil { serverboot.Fatal(ctx, "Failed to initialize metrics", err) } diff --git a/docs/observability.md b/docs/observability.md index 231c4c4d0..5d158c44b 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -198,7 +198,9 @@ To visualize traces locally: ``` The kind overlay pins `ateapi` to `parentbased_always_on`, so API calls show up even without `--trace`; the flag additionally prints the trace ID and forces sampling on every hop. -4. **Search and Inspect**: Copy the printed Trace ID from the CLI output and paste it into the Jaeger search box (top right), or select `ateapi` or `atelet` under the **Service** dropdown and click **Find Traces** to inspect detailed call stacks, DB transactions, state updates, and worker pod handoffs. +4. **Search and Inspect**: Copy the printed Trace ID from the CLI output and paste it into the Jaeger search box (top right), or select `ateapi`, `atelet`, or `ateom-gvisor` under the **Service** dropdown and click **Find Traces** to inspect detailed call stacks, DB transactions, state updates, and worker pod handoffs. + +> ateom carries no manual spans — its only instrumentation is the `otelgrpc` interceptor on the gRPC surface `atelet` calls. So it produces a span for an actor lifecycle operation (`suspend`, `resume`) and nothing at all for a read like `kubectl ate get actor`. Its sampler is parent based, so a lifecycle command is traced end to end into ateom whenever `ateapi` roots a sampled trace, which the kind overlay makes unconditional; the per-component ratio never enters into it. `ateom-gvisor` appearing here at all is also the sign its [OTLP relay](#the-ateom-otlp-relay) is working, since that is the only path its telemetry has out of the worker pod. > **Developer Guide:** For detailed instructions on configuring OpenTelemetry tracer providers, middleware, and exporters in your servers or clients, please refer to the [Tracing Best Practices](dev/best-practices/tracing.md) guide. @@ -221,6 +223,24 @@ Telemetry is emitted the same way everywhere; only the backend differs between a > > ateom workers don't read the ConfigMap at all — `ate-controller` copies the value into each worker pod at creation. A new endpoint reaches them only once the controller itself restarts, and that restart then rolls every WorkerPool Deployment, replacing the running workers along with the actors on them. +### The ateom OTLP relay + +ateom is the one component that does not talk to the collector directly. It exports over a unix socket at `/var/lib/ateom-gvisor/atelet-otlp.sock`, which `atelet` serves and forwards to the collector on the node's network ([`internal/otlprelay`](../internal/otlprelay)): + +``` +ateom ──OTLP/gRPC over unix socket──► atelet relay ──OTLP/gRPC──► collector +``` + +The socket sits in the `BasePath` hostPath already mounted into both, so nothing new is mounted. `atelet` is a DaemonSet, so every ateom on a node shares one relay, and the many per-pod collector connections collapse into one per node. Four things motivate it: the worker pod runs untrusted agent code and will not need egress to the collector once direct fallback is phased out; the connection count drops; ateom's own telemetry stays clear of the transparent egress redirect it installs for the actor; and `atelet` outlives the worker pod, so spans still queued at teardown are not lost with it. + +The relay is best-effort. If the socket is absent when ateom starts — `atelet` not up yet, `--otlp-relay-socket=""`, or no collector configured for the relay to forward to — ateom logs it and exports directly to `OTEL_EXPORTER_OTLP_ENDPOINT` as before. That fallback is decided once at startup, not per export, and is stamped on telemetry as the `substrate.otlp.relay` resource attribute (`relay` vs `direct`). + +> **Note on Network Egress Lockdown:** Complete network policy lockdown of worker pod egress to the collector is planned as a Phase 2 milestone once the relay path is fully proven and direct fallback is deprecated. While the fallback path remains active, worker pods retain network egress to the collector and `ate-controller` continues to inject `OTEL_EXPORTER_OTLP_ENDPOINT`. + +For verified ateom sources, the relay forwards each request verbatim rather than decoding and re-exporting, which is what keeps every ateom its own service in Jaeger/GCP Trace instead of being absorbed into `atelet`'s. `ate-controller` injects `k8s.pod.name`, `k8s.namespace.name`, `k8s.pod.uid`, and `service.instance.id` directly into `OTEL_RESOURCE_ATTRIBUTES` via the Kubernetes Downward API; because the relay preserves resources verbatim, Kubernetes attributes remain intact even though the TCP connection to the collector originates from `atelet` rather than the worker pod IP (bypassing reliance on collector-side IP-based `k8sattributes` enrichment). + +Verbatim forwarding is restricted to known ateom sources and refuses anything else with `PermissionDenied`. Actor telemetry is what that excludes: actors share a hostname (`runsc`) and an interior IP, so their series merge unless identity is injected from outside the actor ([#761](https://github.com/agent-substrate/substrate/issues/761)) — a rewrite, which will be implemented as an explicit rewriting path alongside this forwarder. + --- ## 5. Dashboards diff --git a/go.mod b/go.mod index 2d0191f3b..6349b2168 100644 --- a/go.mod +++ b/go.mod @@ -47,6 +47,7 @@ require ( go.opentelemetry.io/otel/sdk v1.43.0 go.opentelemetry.io/otel/sdk/metric v1.43.0 go.opentelemetry.io/otel/trace v1.43.0 + go.opentelemetry.io/proto/otlp v1.10.0 golang.org/x/oauth2 v0.36.0 golang.org/x/sync v0.21.0 golang.org/x/sys v0.46.0 @@ -168,7 +169,6 @@ require ( go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect - go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 680e329c1..38d14e709 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -53,6 +53,28 @@ func GVisorReleaseDir(sha256 string) string { return filepath.Join(StaticFilesDir, "gvisor-"+sha256) } +// AteletOTLPSocketPath is the node-scoped unix socket atelet serves the OTLP +// relay on (see internal/otlprelay). It is node-scoped rather than per-pod +// because every ateom on the node pushes into the same relay: atelet is a +// DaemonSet, so one socket collapses N per-pod collector connections into one +// per-node connection. +// +// It sits directly under BasePath, which is the host directory already mounted +// at the same path into atelet and into every ateom pod, so no new volume is +// needed for ateom to reach it. Note that BasePath is mounted writable +// (workerpool_apply.go) and shared with CredentialBrokerSocket and the image +// cache, so a worker pod can unlink or replace this socket. Confining +// atelet-owned sockets to a subdirectory mounted read-only would be an +// improvement, but it is a property of the whole BasePath mount rather than of +// this socket — a read-only subdir needs its own volume and mount, and the pod +// keeps CAP_SYS_ADMIN. Tracked separately rather than solved here. +func AteletOTLPSocketPath() string { + return filepath.Join( + BasePath, + "atelet-otlp.sock", + ) +} + func AteomPath(podUID string) string { return filepath.Join( BasePath, diff --git a/internal/ateompath/ateompath_test.go b/internal/ateompath/ateompath_test.go index edd934c93..d4b372ba0 100644 --- a/internal/ateompath/ateompath_test.go +++ b/internal/ateompath/ateompath_test.go @@ -47,6 +47,28 @@ func TestAteomSocketPathLimits(t *testing.T) { } } +func TestAteletOTLPSocketPath(t *testing.T) { + sockPath := AteletOTLPSocketPath() + + // Unix domain socket path limit is 107 bytes (108 with NUL terminator) + const maxUnixSocketLen = 107 + if len(sockPath) > maxUnixSocketLen { + t.Errorf("socket path length %d exceeds max allowed length %d: %q", len(sockPath), maxUnixSocketLen, sockPath) + } + + // It must sit under BasePath: that is the host directory already mounted at + // the same path into atelet and into every ateom pod, which is the whole + // reason the relay needs no new volume. + if !strings.HasPrefix(sockPath, BasePath+"/") { + t.Errorf("AteletOTLPSocketPath() = %q, want it under %q so ateom and atelet see the same file", sockPath, BasePath) + } + + // Node-scoped, so it must not collide with any per-pod ateom socket. + if other := AteomSocketPath("123e4567-e89b-12d3-a456-426614174000"); sockPath == other { + t.Errorf("AteletOTLPSocketPath() collides with AteomSocketPath: %q", sockPath) + } +} + func TestAteomPathUniqueness(t *testing.T) { uid1 := "123e4567-e89b-12d3-a456-426614174000" uid2 := "987f6543-e21b-32d1-b654-246614174111" diff --git a/internal/otlprelay/client.go b/internal/otlprelay/client.go new file mode 100644 index 000000000..fef990666 --- /dev/null +++ b/internal/otlprelay/client.go @@ -0,0 +1,68 @@ +// 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 otlprelay + +import ( + "context" + "errors" + "fmt" + "io/fs" + "log/slog" + "os" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// Dial opens the ateom half of the relay: a gRPC connection over atelet's unix +// socket, to be handed to the OTLP exporters via serverboot's ExporterConn. +// +// It returns (nil, nil) when sockPath is empty or absent, which the caller reads +// as "export directly instead". The existence check is what makes the fallback +// deterministic at startup: grpc.NewClient is lazy, so a connection to a missing +// socket would be created happily and only fail later, per export, with the +// telemetry already lost. Losing spans is not worth failing ateom over either, +// hence a fallback rather than an error. +// +// The connection is plaintext by design. A unix socket cannot leave the node, so +// there is no transport to protect; access is controlled by the socket's file +// permissions instead (see socketMode). +func Dial(ctx context.Context, sockPath string) (*grpc.ClientConn, error) { + if sockPath == "" { + return nil, nil + } + if err := validateSocketPath(sockPath); err != nil { + return nil, err + } + if _, err := os.Stat(sockPath); err != nil { + if errors.Is(err, fs.ErrNotExist) { + slog.WarnContext(ctx, "OTLP relay socket absent, exporting telemetry directly over the pod network", + slog.String("socket", sockPath)) + return nil, nil + } + return nil, fmt.Errorf("while checking the OTLP relay socket %q: %w", sockPath, err) + } + + // gRPC resolves a "unix://" target to a unix socket dialer natively, so the + // OTLP exporters above this connection are unchanged: OTLP is gRPC, and gRPC + // needs only a reliable byte stream. + conn, err := grpc.NewClient("unix://"+sockPath, + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, fmt.Errorf("while dialing the OTLP relay socket %q: %w", sockPath, err) + } + slog.InfoContext(ctx, "Exporting telemetry through the atelet OTLP relay", slog.String("socket", sockPath)) + return conn, nil +} diff --git a/internal/otlprelay/e2e_test.go b/internal/otlprelay/e2e_test.go new file mode 100644 index 000000000..8c64258cf --- /dev/null +++ b/internal/otlprelay/e2e_test.go @@ -0,0 +1,202 @@ +// 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 otlprelay + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/agent-substrate/substrate/internal/serverboot" +) + +// TestEndToEndThroughServerboot exercises the whole path an ateom span actually +// takes, rather than the relay in isolation: +// +// serverboot.InitTracing → OTLP exporter → unix socket → relay → collector +// +// The relay tests above speak the collector protocol directly, so they would +// still pass if TracingOptions.ExporterConn were wired up wrong and the exporter +// quietly kept dialing OTEL_EXPORTER_OTLP_ENDPOINT. This one would not: the +// endpoint variable points at the fake collector *through* the relay only, and +// the assertion is that the span arrived carrying ateom's own service.name. +// +// Run it on its own to watch the hop happen: +// +// go test ./internal/otlprelay/ -run TestEndToEndThroughServerboot -v +func TestEndToEndThroughServerboot(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + // Re-point the generic endpoint at an unroutable address so an exporter + // that ignored ExporterConn would fail deterministically instead of dialing + // the test collector directly. + t.Setenv(endpointEnv, "http://127.0.0.1:1") + t.Logf("fake collector on %s, relay socket %s", collector, sock) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + if conn == nil { + t.Fatal("Dial returned no connection; the exporter would have bypassed the relay") + } + defer conn.Close() + + const serviceName = "ateom-microvm" + tp, err := serverboot.InitTracing(context.Background(), serverboot.TracingOptions{ + ServiceName: serviceName, + // Ratio 1.0: this test asserts on delivery, not on sampling. + Sampling: serverboot.ParentRatioSampling(1.0), + ExporterConn: conn, + RelayCapable: true, + }) + if err != nil { + t.Fatalf("InitTracing: %v", err) + } + + _, span := tp.Tracer("relay-e2e").Start(context.Background(), "RunWorkload") + span.End() + + // Shutdown flushes the batch processor, which is what actually puts the + // span on the wire. + if err := tp.Shutdown(context.Background()); err != nil { + t.Fatalf("TracerProvider.Shutdown: %v", err) + } + + select { + case <-sink.got: + case <-time.After(10 * time.Second): + t.Fatal("collector never received a span through the relay") + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.traces) == 0 { + t.Fatal("collector recorded no trace exports") + } + + var gotService, gotSpan, gotRelay string + for _, req := range sink.traces { + for _, rs := range req.GetResourceSpans() { + for _, attr := range rs.GetResource().GetAttributes() { + if attr.GetKey() == "service.name" { + gotService = attr.GetValue().GetStringValue() + } + if attr.GetKey() == relayAttrKey { + gotRelay = attr.GetValue().GetStringValue() + } + } + for _, ss := range rs.GetScopeSpans() { + for _, s := range ss.GetSpans() { + gotSpan = s.GetName() + } + } + } + } + t.Logf("collector received span %q from service %q (relay=%q)", gotSpan, gotService, gotRelay) + + // The point of forwarding the request verbatim: the span is still ateom's, + // not atelet's. + if gotService != serviceName { + t.Errorf("span arrived with service.name %q, want %q; the relay must not re-attribute it", gotService, serviceName) + } + if gotSpan != "RunWorkload" { + t.Errorf("span arrived named %q, want %q", gotSpan, "RunWorkload") + } + if gotRelay != "relay" { + t.Errorf("span arrived with %s %q, want %q", relayAttrKey, gotRelay, "relay") + } +} + +// relayAttrKey duplicates serverboot's unexported constant. Keeping a literal +// here is the point: if serverboot renames the attribute, the dashboards and +// alerts keyed on it break too, and this test is where that shows up. +const relayAttrKey = "ate.otlp.relay" + +// TestEndToEndFallsBackToDirect is the other half of TestEndToEndThroughServerboot: +// the ateom asked for the relay, atelet was not serving one, and the exporter +// must fall back to the network path rather than dropping telemetry. +// +// This is the case the ateoms degrade into instead of exiting (see the Dial call +// in cmd/ateom-*/main.go), so it needs to be more than a nil check: the span has +// to reach the collector, and it has to be distinguishable from a relayed one at +// query time — hence the "direct" attribute. +func TestEndToEndFallsBackToDirect(t *testing.T) { + sink, collector := startFakeCollector(t) + // No relay: the socket path is inside a fresh temp dir nothing created. + sock := filepath.Join(t.TempDir(), "absent-atelet-otlp.sock") + // The direct path is the exporter dialing this itself, which is exactly what + // the relay test points at an unroutable address to rule out. + t.Setenv(endpointEnv, "http://"+collector) + t.Logf("fake collector on %s, absent relay socket %s", collector, sock) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial with no relay present must not fail: %v", err) + } + if conn != nil { + conn.Close() + t.Fatal("Dial returned a connection for a socket that does not exist") + } + + const serviceName = "ateom-microvm" + tp, err := serverboot.InitTracing(context.Background(), serverboot.TracingOptions{ + ServiceName: serviceName, + Sampling: serverboot.ParentRatioSampling(1.0), + ExporterConn: conn, // nil: the fallback + RelayCapable: true, + }) + if err != nil { + t.Fatalf("InitTracing: %v", err) + } + + _, span := tp.Tracer("relay-e2e").Start(context.Background(), "RunWorkload") + span.End() + + if err := tp.Shutdown(context.Background()); err != nil { + t.Fatalf("TracerProvider.Shutdown: %v", err) + } + + select { + case <-sink.got: + case <-time.After(10 * time.Second): + t.Fatal("collector never received a span over the direct path") + } + + sink.mu.Lock() + defer sink.mu.Unlock() + var gotService, gotRelay string + for _, req := range sink.traces { + for _, rs := range req.GetResourceSpans() { + for _, attr := range rs.GetResource().GetAttributes() { + switch attr.GetKey() { + case "service.name": + gotService = attr.GetValue().GetStringValue() + case relayAttrKey: + gotRelay = attr.GetValue().GetStringValue() + } + } + } + } + if gotService != serviceName { + t.Errorf("span arrived with service.name %q, want %q", gotService, serviceName) + } + // Without this, a node whose atelet never came up looks identical to a + // healthy one in the trace store. + if gotRelay != "direct" { + t.Errorf("span arrived with %s %q, want %q", relayAttrKey, gotRelay, "direct") + } +} diff --git a/internal/otlprelay/relay.go b/internal/otlprelay/relay.go new file mode 100644 index 000000000..b68b65ebc --- /dev/null +++ b/internal/otlprelay/relay.go @@ -0,0 +1,569 @@ +// 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 otlprelay carries ateom's OTLP telemetry to the collector over a unix +// socket served by atelet, so a worker pod needs no network path of its own to +// export spans and metrics. +// +// Motivation. ateom runs inside the worker pod that hosts the actor, and until +// now exported OTLP straight to the collector over the pod's network (the +// endpoint is injected by atecontroller, see workerpool_apply.go). That has four +// costs the relay removes: +// +// - Blast radius. The pod runs untrusted agent code. Exporting over the pod +// network means the pod must be allowed egress to the collector, which is +// reachable to anything that escapes the sandbox. A unix socket cannot leave +// the node, so the pod can be denied network egress entirely. +// - Connection count. Worker pods are heavily oversubscribed, so a node runs +// many ateoms, each holding its own gRPC connection to the collector. They +// collapse into atelet's single per-node connection. +// - Interference. ateom installs a transparent redirect of actor egress to its +// own atunnel listener; its own outbound traffic has to stay clear of the +// rules it installs. A unix socket is not IP traffic and cannot be caught. +// - Shutdown loss. Teardown frees the actor's network and then the pod goes +// away, which is exactly when the spans describing teardown are still queued +// in the batch processor. atelet outlives the worker pod. +// +// The relay forwards the OTLP request message verbatim rather than decoding it +// into SDK records and re-exporting. Verbatim pass-through keeps each ateom's +// own resource (service.name, service.instance.id, pod attributes) intact, so +// its spans stay attributed to ateom instead of being absorbed into atelet's. +// Restricting this pass-through to verified ateom sources ensures that future +// actor telemetry requiring identity rewrites (#761) will be added as an +// explicit rewriting path alongside this forwarder; see ateomServices. +// +// Verbatim applies to the payload, not to the call around it. The request's +// metadata is dropped and replaced with the headers atelet resolves from its own +// OTEL_EXPORTER_OTLP_HEADERS, since the upstream leg is atelet's connection and +// authenticating it is atelet's business; see upstreamContext. +package otlprelay + +import ( + "context" + "errors" + "fmt" + "io/fs" + "log/slog" + "net" + "net/url" + "os" + "path/filepath" + "sort" + "strings" + "sync" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/encoding/gzip" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + semconv "go.opentelemetry.io/otel/semconv/v1.40.0" + colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1" + resourcepb "go.opentelemetry.io/proto/otlp/resource/v1" +) + +const ( + // endpointEnv and its signal-specific overrides are the standard OTLP + // exporter variables. The relay resolves them itself because it dials the + // collector directly rather than through an OTel SDK exporter. + endpointEnv = "OTEL_EXPORTER_OTLP_ENDPOINT" + tracesEndpointEnv = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" + metricsEndpointEnv = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" + + // compressionEnv and its signal-specific overrides configure upstream + // gRPC compression (gzip or none). + compressionEnv = "OTEL_EXPORTER_OTLP_COMPRESSION" + tracesCompressionEnv = "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION" + metricsCompressionEnv = "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION" + + // headersEnv and its signal-specific overrides carry the headers the + // collector expects (an API key, a tenant id). Unlike the endpoint and the + // compression, these are per-call metadata rather than per-connection, so + // traces and metrics may legitimately differ and are resolved separately. + headersEnv = "OTEL_EXPORTER_OTLP_HEADERS" + tracesHeadersEnv = "OTEL_EXPORTER_OTLP_TRACES_HEADERS" + metricsHeadersEnv = "OTEL_EXPORTER_OTLP_METRICS_HEADERS" + + // otlpDefaultPort matches atenet's normalizeOtlpCollector. + otlpDefaultPort = "4317" + + // socketMode keeps the relay socket private to root: both atelet and the + // ateom worker pods run as root (runAsUser: 0). The socket lives inside + // BasePath, a root-owned host directory. + socketMode = 0o600 + + // maxRecvMsgSize bounds a single Export payload. One misbehaving ateom + // should not be able to make atelet allocate without limit; the OTel SDK's + // batch processor emits far smaller messages than this. + maxRecvMsgSize = 16 << 20 // 16 MiB +) + +// Server is the atelet half of the relay: an OTLP receiver on a unix socket +// that forwards to the real collector over the node's network. +type Server struct { + upstream *grpc.ClientConn + grpc *grpc.Server + sockPath string +} + +// ateomServices are the only sources this relay carries, keyed by the +// service.name their resource declares — which the OTEL_* environment can +// override out from under an ateom; see sourceGate for what that looks like. +// Mirrors the serviceName constants in +// cmd/ateom-gvisor and cmd/ateom-microvm, which are package main and cannot be +// imported; TestAteomServicesMatchTheAteomBinaries guards the duplication. +// +// This allowlist is a protocol contract rather than a security boundary: +// service.name is client-provided, so a compromised process could claim an +// ateom name. Its purpose is to prevent accidental misuse (e.g. an actor SDK +// pointed at the socket) and keep the pass-through contract explicit for #761. +// Peer authentication, if needed, would require per-pod sockets or UDS peer +// credentials (SO_PEERCRED) tied to #741. +var ateomServices = map[string]bool{ + "ateom-gvisor": true, + "ateom-microvm": true, +} + +// sourceGate applies the ateomServices allowlist and reports the first +// rejection of each service.name. +// +// The log matters because of how this failure presents. service.name is +// whatever the resource declares, and resource.WithFromEnv() runs last in +// serverboot.newResource, so OTEL_SERVICE_NAME or an OTEL_RESOURCE_ATTRIBUTES +// entry set on a worker pod overrides the ateom's own name. The relay then +// refuses every export with PermissionDenied, which the OTel SDK does not retry +// — it drops the batch and reports through the SDK error handler. Telemetry from +// that ateom simply stops, with nothing on the collector side to say why. One +// line per distinct name on the node makes it greppable; the ateom itself is +// unaffected, so this is a diagnosability problem rather than an outage. +// +// Dedup keyed by name, because a rejected exporter keeps retrying on its own +// schedule (seconds) for the pod's whole life: unbounded logging would be the +// bigger operational problem. The map is bounded by the number of distinct names +// a node's own ateoms declare. +type sourceGate struct { + logged sync.Map // service.name -> struct{} +} + +// check rejects a missing service.name along with an unrecognized one: an +// unidentified source is the one the relay cannot vouch for. +func (g *sourceGate) check(ctx context.Context, r *resourcepb.Resource) error { + name := resourceServiceName(r) + if ateomServices[name] { + return nil + } + if _, dup := g.logged.LoadOrStore(name, struct{}{}); !dup { + slog.WarnContext(ctx, "OTLP relay rejected telemetry from an unrecognized source; it is being dropped, not retried. If this is an ateom, check whether OTEL_SERVICE_NAME or OTEL_RESOURCE_ATTRIBUTES on the worker pod is overriding its service.name", + slog.String("service.name", name), + slog.Any("allowed", allowedServices()), + slog.String("note", "logged once per distinct service.name")) + } + return status.Errorf(codes.PermissionDenied, + "the OTLP relay carries ateom telemetry only, got service.name %q; a source whose identity has to be rewritten (#761) must not be forwarded verbatim. If this is an ateom, an OTEL_SERVICE_NAME or OTEL_RESOURCE_ATTRIBUTES override on the worker pod would produce exactly this", + name) +} + +func allowedServices() []string { + names := make([]string, 0, len(ateomServices)) + for name := range ateomServices { + names = append(names, name) + } + sort.Strings(names) + return names +} + +func resourceServiceName(r *resourcepb.Resource) string { + for _, attr := range r.GetAttributes() { + if attr.GetKey() == string(semconv.ServiceNameKey) { + return attr.GetValue().GetStringValue() + } + } + return "" +} + +// upstreamContext builds the metadata for the upstream call from atelet's own +// configuration, dropping whatever the ateom sent. +// +// The upstream leg is atelet's connection to the collector, so its credentials +// belong to atelet: the relay resolves OTEL_EXPORTER_OTLP_HEADERS from its own +// environment, exactly as the SDK exporter it replaces would have. Forwarding +// the client's headers instead would let anything that reached the socket choose +// what atelet presents to the collector — a header set is not telemetry to be +// passed through verbatim the way the resource is, and unlike service.name (see +// ateomServices) it is not merely claimed identity but an actual credential. +// +// Nothing is allow-listed through. atecontroller injects only +// OTEL_EXPORTER_OTLP_ENDPOINT into worker pods (workerpool_apply.go), so no +// ateom has a header to lose today; add an allowlist here, not a blanket +// forward, if one ever needs to reach the collector. +// +// The incoming metadata is dropped by simply not copying it: gRPC never +// propagates incoming metadata to an outgoing call on its own. +func upstreamContext(ctx context.Context, md metadata.MD) context.Context { + if len(md) == 0 { + return ctx + } + return metadata.NewOutgoingContext(ctx, md) +} + +// parseHeaders reads the W3C-Baggage-shaped list the OTLP headers variables +// carry ("key1=value1,key2=value2", values percent-encoded), as the OTel SDK +// exporters do. +// +// Keys are lower-cased because gRPC metadata keys are case-insensitive and +// metadata.MD is documented to hold them lower-cased; a mixed-case key set here +// would otherwise be invisible to metadata.Get. +func parseHeaders(raw string) (metadata.MD, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + md := metadata.MD{} + // Errors name the position rather than the offending text: any of these + // entries may be a credential, and this error reaches a log line. + for i, pair := range strings.Split(raw, ",") { + pair = strings.TrimSpace(pair) + if pair == "" { + continue + } + key, value, found := strings.Cut(pair, "=") + if !found { + return nil, fmt.Errorf("OTLP header %d is not in key=value form", i+1) + } + key = strings.ToLower(strings.TrimSpace(key)) + if key == "" { + return nil, fmt.Errorf("OTLP header %d has an empty name", i+1) + } + // Percent-decoding is what makes a value containing "," or "=" (a base64 + // token, say) expressible in this format at all. + decoded, err := url.QueryUnescape(strings.TrimSpace(value)) + if err != nil { + // The value is deliberately not in the message: these are credentials. + return nil, fmt.Errorf("OTLP header %q has a value that is not valid percent-encoding: %w", key, err) + } + md.Append(key, decoded) + } + return md, nil +} + +// upstreamHeaders resolves the headers for one signal. Per the OTLP spec the +// signal-specific variable replaces the generic one rather than merging with +// it, so a component that sets both gets exactly what the SDK would have sent. +func upstreamHeaders(signalEnv string) (metadata.MD, error) { + env, raw := signalEnv, strings.TrimSpace(os.Getenv(signalEnv)) + if raw == "" { + env, raw = headersEnv, os.Getenv(headersEnv) + } + md, err := parseHeaders(raw) + if err != nil { + return nil, fmt.Errorf("while reading %s: %w", env, err) + } + return md, nil +} + +// The two OTLP services both declare a method named Export, with different +// request types, so one type cannot implement both: the embedded Unimplemented +// structs would give Server an ambiguous promoted Export and satisfy neither +// interface. Each service gets its own tiny forwarder instead. + +type traceRelay struct { + coltracepb.UnimplementedTraceServiceServer + upstream coltracepb.TraceServiceClient + // headers atelet presents to the collector; see upstreamContext. Resolved + // once at construction: they come from atelet's environment, not the call. + headers metadata.MD + // gate is shared with metricRelay so a misnamed ateom is reported once, not + // once per signal. + gate *sourceGate +} + +// Export forwards a batch of spans to the collector unchanged. +// +// Deliberately not wrapped in a span of atelet's own: the relay must not inject +// itself into the trace it is carrying. +// +// A batch is refused whole rather than having the offending resource dropped: a +// partial success the sender reads as success loses telemetry silently. +func (t *traceRelay) Export(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) { + for _, rs := range req.GetResourceSpans() { + if err := t.gate.check(ctx, rs.GetResource()); err != nil { + return nil, err + } + } + return t.upstream.Export(upstreamContext(ctx, t.headers), req) +} + +type metricRelay struct { + colmetricspb.UnimplementedMetricsServiceServer + upstream colmetricspb.MetricsServiceClient + headers metadata.MD + gate *sourceGate +} + +// Export forwards a batch of metric datapoints to the collector unchanged. +func (m *metricRelay) Export(ctx context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) { + for _, rm := range req.GetResourceMetrics() { + if err := m.gate.check(ctx, rm.GetResource()); err != nil { + return nil, err + } + } + return m.upstream.Export(upstreamContext(ctx, m.headers), req) +} + +// validateSocketPath rejects a relative path, which gRPC does not resolve: +// "unix://foo/r.sock" parses as authority "foo", path "/r.sock". grpc.NewClient +// being lazy, that wrong target is accepted at startup and fails per export +// afterwards, which is why this errors rather than falling back. +func validateSocketPath(sockPath string) error { + if !filepath.IsAbs(sockPath) { + return fmt.Errorf("the OTLP relay socket path %q is relative; it must be absolute, since atelet and ateom would otherwise resolve it against different working directories", sockPath) + } + return nil +} + +// NewServer builds a relay that forwards to the collector named by the standard +// OTLP endpoint environment variables. It returns (nil, nil) when sockPath is +// empty (the relay is switched off) or when no endpoint is configured: a relay +// with nowhere to forward to would accept an ateom's spans and drop them, which +// is worse than ateom finding no socket and falling back to a direct export. +func NewServer(ctx context.Context, sockPath string) (*Server, error) { + if sockPath == "" { + return nil, nil + } + if err := validateSocketPath(sockPath); err != nil { + return nil, err + } + target, err := upstreamTarget() + if err != nil { + return nil, err + } + if target == "" { + slog.InfoContext(ctx, "OTLP relay disabled: no collector endpoint configured", + slog.String("env", endpointEnv)) + return nil, nil + } + + comp, err := upstreamCompression() + if err != nil { + return nil, err + } + + // Resolved before the socket exists: a header set atelet cannot parse would + // otherwise become a per-export failure against a collector that rejects the + // unauthenticated calls, which is harder to read than refusing to start the + // relay. ateom then finds no socket and exports directly. + traceHeaders, err := upstreamHeaders(tracesHeadersEnv) + if err != nil { + return nil, err + } + metricHeaders, err := upstreamHeaders(metricsHeadersEnv) + if err != nil { + return nil, err + } + + dialOpts := []grpc.DialOption{ + // Plaintext by design today; TLS support for the upstream leg will be added + // in tandem with #741. + grpc.WithTransportCredentials(insecure.NewCredentials()), + } + if comp == "gzip" { + dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name))) + } + + // Lazy by design: grpc.NewClient does not block on the collector being up, + // so atelet startup does not depend on the collector's readiness. + upstream, err := grpc.NewClient(target, dialOpts...) + if err != nil { + return nil, fmt.Errorf("while dialing OTLP collector %q: %w", target, err) + } + + s := &Server{ + upstream: upstream, + sockPath: sockPath, + grpc: grpc.NewServer(grpc.MaxRecvMsgSize(maxRecvMsgSize)), + } + gate := &sourceGate{} + coltracepb.RegisterTraceServiceServer(s.grpc, &traceRelay{ + upstream: coltracepb.NewTraceServiceClient(upstream), + headers: traceHeaders, + gate: gate, + }) + colmetricspb.RegisterMetricsServiceServer(s.grpc, &metricRelay{ + upstream: colmetricspb.NewMetricsServiceClient(upstream), + headers: metricHeaders, + gate: gate, + }) + // Header names only: the values are credentials. + slog.InfoContext(ctx, "OTLP relay forwarding to collector", + slog.String("collector", target), + slog.String("compression", comp), + slog.Any("traceHeaders", headerNames(traceHeaders)), + slog.Any("metricHeaders", headerNames(metricHeaders))) + return s, nil +} + +// Serve listens on the relay socket and blocks until the server stops. Designed +// to be `go`-launched; it returns an error only if the socket cannot be opened +// or serving fails. +func (s *Server) Serve(ctx context.Context) error { + if err := os.MkdirAll(filepath.Dir(s.sockPath), 0o755); err != nil { + return fmt.Errorf("while creating the OTLP relay socket directory: %w", err) + } + // A socket left behind by a previous atelet would make Listen fail with + // EADDRINUSE even though nothing holds it. + if err := os.Remove(s.sockPath); err != nil && !errors.Is(err, fs.ErrNotExist) { + return fmt.Errorf("while removing a stale OTLP relay socket %q: %w", s.sockPath, err) + } + lis, err := net.Listen("unix", s.sockPath) + if err != nil { + return fmt.Errorf("while opening the OTLP relay socket %q: %w", s.sockPath, err) + } + // net.Listen applies the umask, which on atelet would typically leave the + // socket group/other-unwritable and unreachable from an ateom running as a + // different uid. Widen it explicitly. + if err := os.Chmod(s.sockPath, socketMode); err != nil { + _ = lis.Close() + return fmt.Errorf("while setting the OTLP relay socket mode: %w", err) + } + + slog.InfoContext(ctx, "OTLP relay serving", slog.String("socket", s.sockPath)) + return s.grpc.Serve(lis) +} + +// Stop drains the relay, closes the upstream connection and removes the socket. +func (s *Server) Stop() { + s.grpc.GracefulStop() + _ = s.upstream.Close() + _ = os.Remove(s.sockPath) +} + +// headerNames lists the configured header names, sorted, for logging. It exists +// so an operator can confirm the relay picked up the headers without the values +// reaching the node's logs. +func headerNames(md metadata.MD) []string { + names := make([]string, 0, len(md)) + for k := range md { + names = append(names, k) + } + sort.Strings(names) + return names +} + +// upstreamCompression resolves the compression algorithm (gzip or none) to use +// for upstream export. +func upstreamCompression() (string, error) { + generic := strings.TrimSpace(os.Getenv(compressionEnv)) + traces := strings.TrimSpace(os.Getenv(tracesCompressionEnv)) + metrics := strings.TrimSpace(os.Getenv(metricsCompressionEnv)) + + traceComp := generic + if traces != "" { + traceComp = traces + } + metricComp := generic + if metrics != "" { + metricComp = metrics + } + + if traceComp != "" && metricComp != "" && traceComp != metricComp { + return "", fmt.Errorf("signal-specific compression settings conflict (%q for traces vs %q for metrics); the relay carries both signals over one connection", + traceComp, metricComp) + } + + resolved := traceComp + if resolved == "" { + resolved = metricComp + } + switch resolved { + case "", "none": + return "none", nil + case "gzip": + return "gzip", nil + default: + return "", fmt.Errorf("unsupported OTLP compression %q, want gzip or none", resolved) + } +} + +// upstreamTarget resolves the collector address the relay forwards to, from the +// standard OTLP endpoint variables, into the bare host:port grpc.NewClient wants. +// +// The signal-specific variables must agree: the relay carries traces and metrics +// over one connection, so it cannot honor two different collectors. Configuring +// both differently is a misconfiguration rather than something to silently pick +// a winner for. +func upstreamTarget() (string, error) { + generic := strings.TrimSpace(os.Getenv(endpointEnv)) + traces := strings.TrimSpace(os.Getenv(tracesEndpointEnv)) + metrics := strings.TrimSpace(os.Getenv(metricsEndpointEnv)) + + traceTarget := generic + if traces != "" { + traceTarget = traces + } + metricTarget := generic + if metrics != "" { + metricTarget = metrics + } + + if traceTarget != "" && metricTarget != "" && traceTarget != metricTarget { + return "", fmt.Errorf("signal-specific endpoints conflict (%q for traces vs %q for metrics); the relay carries both signals over one connection", + traceTarget, metricTarget) + } + + resolved := traceTarget + if resolved == "" { + resolved = metricTarget + } + if resolved == "" { + return "", nil + } + return normalizeEndpoint(resolved) +} + +// normalizeEndpoint accepts both a bare "host:port" and the URL form the OTLP +// environment variables carry, and returns the host:port grpc.NewClient dials. +// +// https is rejected rather than downgraded: the relay dials with insecure +// credentials, so honoring it would ship telemetry in plaintext to an endpoint +// that asked for TLS. +func normalizeEndpoint(addr string) (string, error) { + hostport := addr + if strings.Contains(addr, "://") { + u, err := url.Parse(addr) + if err != nil { + return "", fmt.Errorf("parse OTLP collector endpoint %q: %w", addr, err) + } + switch u.Scheme { + case "http": + case "https": + return "", fmt.Errorf("OTLP collector endpoint %q uses https, which the relay does not support: it forwards over an insecure gRPC connection. Point it at an http:// endpoint", addr) + default: + return "", fmt.Errorf("OTLP collector endpoint %q has unsupported scheme %q, want http", addr, u.Scheme) + } + hostport = u.Host + } + + host, port, err := net.SplitHostPort(hostport) + if err != nil { + host = strings.Trim(hostport, "[]") + port = otlpDefaultPort + } + if host == "" { + return "", fmt.Errorf("OTLP collector endpoint %q names no host", addr) + } + return net.JoinHostPort(host, port), nil +} diff --git a/internal/otlprelay/relay_test.go b/internal/otlprelay/relay_test.go new file mode 100644 index 000000000..5aaccf2f4 --- /dev/null +++ b/internal/otlprelay/relay_test.go @@ -0,0 +1,869 @@ +// 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 otlprelay + +import ( + "bytes" + "context" + "errors" + "io/fs" + "log/slog" + "net" + "os" + "path/filepath" + "regexp" + "strings" + "sync" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/testing/protocmp" + + 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" + metricspb "go.opentelemetry.io/proto/otlp/metrics/v1" + resourcepb "go.opentelemetry.io/proto/otlp/resource/v1" + tracepb "go.opentelemetry.io/proto/otlp/trace/v1" +) + +// fakeCollector is a stand-in for the real OTLP collector: it records what the +// relay forwards so the tests can assert the payload arrived unchanged. +type fakeCollector struct { + coltracepb.UnimplementedTraceServiceServer + + mu sync.Mutex + traces []*coltracepb.ExportTraceServiceRequest + metrics []*colmetricspb.ExportMetricsServiceRequest + traceMD []metadata.MD + metricMD []metadata.MD + got chan struct{} +} + +func (f *fakeCollector) Export(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) { + f.mu.Lock() + f.traces = append(f.traces, req) + if md, ok := metadata.FromIncomingContext(ctx); ok { + f.traceMD = append(f.traceMD, md.Copy()) + } + f.mu.Unlock() + f.got <- struct{}{} + return &coltracepb.ExportTraceServiceResponse{}, nil +} + +// metricsSink exists because the two OTLP services both declare Export with +// different request types, the same collision the relay itself works around. +type metricsSink struct { + colmetricspb.UnimplementedMetricsServiceServer + parent *fakeCollector +} + +func (m *metricsSink) Export(ctx context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) { + m.parent.mu.Lock() + m.parent.metrics = append(m.parent.metrics, req) + if md, ok := metadata.FromIncomingContext(ctx); ok { + m.parent.metricMD = append(m.parent.metricMD, md.Copy()) + } + m.parent.mu.Unlock() + m.parent.got <- struct{}{} + return &colmetricspb.ExportMetricsServiceResponse{}, nil +} + +// startFakeCollector serves the OTLP collector services on a loopback TCP port +// (the shape the relay forwards to) and returns the sink and its host:port. +func startFakeCollector(t *testing.T) (*fakeCollector, string) { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + sink := &fakeCollector{got: make(chan struct{}, 8)} + srv := grpc.NewServer() + coltracepb.RegisterTraceServiceServer(srv, sink) + colmetricspb.RegisterMetricsServiceServer(srv, &metricsSink{parent: sink}) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + return sink, lis.Addr().String() +} + +// startRelay brings up a relay on a socket in a temp dir, wired to collector. +func startRelay(t *testing.T, collector string) string { + t.Helper() + t.Setenv(endpointEnv, collector) + + // Short filename: a unix socket path is capped at ~104 bytes and the test + // temp dir already eats most of that on darwin. + sock := filepath.Join(t.TempDir(), "r.sock") + relay, err := NewServer(context.Background(), sock) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + if relay == nil { + t.Fatal("NewServer returned nil with a collector endpoint set") + } + serveErr := make(chan error, 1) + go func() { serveErr <- relay.Serve(context.Background()) }() + t.Cleanup(relay.Stop) + + // Serve creates the socket asynchronously; Dial's existence check needs it. + waitForSocket(t, sock, serveErr) + return sock +} + +func waitForSocket(t *testing.T, sock string, serveErr <-chan error) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-serveErr: + t.Fatalf("relay.Serve returned early: %v", err) + default: + } + // Closed immediately: a probe connection that never speaks HTTP/2 sits + // in the server's handshake path until its 120s timeout, and + // GracefulStop would wait the whole of it. + if c, err := net.Dial("unix", sock); err == nil { + _ = c.Close() + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("relay socket %q never became connectable", sock) +} + +// serviceResource builds the one resource attribute the relay's scoping looks +// at. Passing "" yields a resource that declares no service.name at all. +func serviceResource(name string) *resourcepb.Resource { + if name == "" { + return &resourcepb.Resource{} + } + return &resourcepb.Resource{ + Attributes: []*commonpb.KeyValue{{ + Key: "service.name", + Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: name}}, + }}, + } +} + +// TestRelayForwardsTracesVerbatim is the property the whole design rests on: +// what an ateom exports is what the collector sees, including the resource +// attributes that attribute the spans to that ateom rather than to atelet. +func TestRelayForwardsTracesVerbatim(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + if conn == nil { + t.Fatal("Dial returned no connection for an existing socket") + } + defer conn.Close() + + req := &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{ + Resource: serviceResource("ateom-microvm"), + ScopeSpans: []*tracepb.ScopeSpans{{ + Spans: []*tracepb.Span{{ + Name: "RunWorkload", + TraceId: []byte("0123456789abcdef"), + SpanId: []byte("01234567"), + }}, + }}, + }}, + } + if _, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), req); err != nil { + t.Fatalf("Export through the relay: %v", err) + } + + select { + case <-sink.got: + case <-time.After(5 * time.Second): + t.Fatal("collector never received the forwarded trace export") + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.traces) != 1 { + t.Fatalf("collector got %d trace exports, want 1", len(sink.traces)) + } + if diff := cmp.Diff(req, sink.traces[0], protocmp.Transform()); diff != "" { + t.Errorf("forwarded request differs from what was sent (-sent +received):\n%s", diff) + } +} + +func TestRelayForwardsMetrics(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + req := &colmetricspb.ExportMetricsServiceRequest{ + ResourceMetrics: []*metricspb.ResourceMetrics{{ + Resource: serviceResource("ateom-microvm"), + ScopeMetrics: []*metricspb.ScopeMetrics{{ + Metrics: []*metricspb.Metric{{Name: "ateom.workload.runs"}}, + }}, + }}, + } + if _, err := colmetricspb.NewMetricsServiceClient(conn).Export(context.Background(), req); err != nil { + t.Fatalf("Export through the relay: %v", err) + } + + select { + case <-sink.got: + case <-time.After(5 * time.Second): + t.Fatal("collector never received the forwarded metric export") + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.metrics) != 1 { + t.Fatalf("collector got %d metric exports, want 1", len(sink.metrics)) + } + if diff := cmp.Diff(req, sink.metrics[0], protocmp.Transform()); diff != "" { + t.Errorf("forwarded request differs from what was sent (-sent +received):\n%s", diff) + } +} + +// TestStopRemovesSocket matters for the restart path: a leftover socket makes +// the next atelet's Listen fail with EADDRINUSE, and in the meantime makes +// every ateom on the node believe a relay is there. +func TestStopRemovesSocket(t *testing.T) { + _, collector := startFakeCollector(t) + t.Setenv(endpointEnv, collector) + + sock := filepath.Join(t.TempDir(), "r.sock") + relay, err := NewServer(context.Background(), sock) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + serveErr := make(chan error, 1) + go func() { serveErr <- relay.Serve(context.Background()) }() + waitForSocket(t, sock, serveErr) + + relay.Stop() + if _, err := os.Stat(sock); !errors.Is(err, fs.ErrNotExist) { + t.Errorf("os.Stat(%q) after Stop = %v, want the socket to be gone", sock, err) + } +} + +// TestServeReplacesStaleSocket covers the atelet-crashed-and-restarted case: +// the socket file survives the process, and Listen would refuse to reuse it. +func TestServeReplacesStaleSocket(t *testing.T) { + _, collector := startFakeCollector(t) + t.Setenv(endpointEnv, collector) + + sock := filepath.Join(t.TempDir(), "r.sock") + if err := os.WriteFile(sock, nil, 0o600); err != nil { + t.Fatalf("planting a stale socket file: %v", err) + } + + relay, err := NewServer(context.Background(), sock) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + serveErr := make(chan error, 1) + go func() { serveErr <- relay.Serve(context.Background()) }() + t.Cleanup(relay.Stop) + waitForSocket(t, sock, serveErr) +} + +func TestNewServerDisabled(t *testing.T) { + t.Setenv(endpointEnv, "otel-collector:4317") + relay, err := NewServer(context.Background(), "") + if err != nil { + t.Fatalf("NewServer with an empty socket path: %v", err) + } + if relay != nil { + t.Error("NewServer with an empty socket path returned a server, want nil (relay disabled)") + } +} + +func TestNewServerWithoutCollector(t *testing.T) { + t.Setenv(endpointEnv, "") + t.Setenv(tracesEndpointEnv, "") + t.Setenv(metricsEndpointEnv, "") + relay, err := NewServer(context.Background(), filepath.Join(t.TempDir(), "r.sock")) + if err != nil { + t.Fatalf("NewServer with no collector configured: %v", err) + } + if relay != nil { + t.Error("NewServer with no collector configured returned a server; it would accept spans and drop them") + } +} + +func TestDialMissingSocketFallsBack(t *testing.T) { + conn, err := Dial(context.Background(), filepath.Join(t.TempDir(), "absent.sock")) + if err != nil { + t.Fatalf("Dial on an absent socket: %v, want the fallback", err) + } + if conn != nil { + conn.Close() + t.Error("Dial on an absent socket returned a connection, want nil so the caller exports directly") + } +} + +func TestDialEmptyPath(t *testing.T) { + conn, err := Dial(context.Background(), "") + if err != nil { + t.Fatalf("Dial(\"\"): %v", err) + } + if conn != nil { + conn.Close() + t.Error("Dial(\"\") returned a connection, want nil") + } +} + +// A rejected export is dropped by the SDK without a retry, so the node log is +// the only place the misconfiguration shows up — and it has to show up exactly +// once per name, because the rejected exporter keeps retrying for the life of +// the pod. +func TestSourceGateLogsEachRejectionOnce(t *testing.T) { + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + gate := &sourceGate{} + for range 3 { + if err := gate.check(context.Background(), serviceResource("actor")); status.Code(err) != codes.PermissionDenied { + t.Fatalf("gate.check = %v, want PermissionDenied every time", err) + } + } + if err := gate.check(context.Background(), serviceResource("atelet")); err == nil { + t.Fatal("gate.check accepted atelet") + } + if err := gate.check(context.Background(), serviceResource("ateom-gvisor")); err != nil { + t.Fatalf("gate.check rejected an ateom: %v", err) + } + + if got := strings.Count(buf.String(), "rejected telemetry"); got != 2 { + t.Errorf("logged %d rejections, want 2 (one per distinct service.name):\n%s", got, buf.String()) + } + // The operator has to be able to tell which override to go looking for. + if !strings.Contains(buf.String(), "OTEL_SERVICE_NAME") { + t.Errorf("rejection log does not name the env var that causes it:\n%s", buf.String()) + } +} + +// TestRelayRefusesNonAteomSource is the scoping contract. The empty +// service.name case is the one worth keeping: that is the shape telemetry takes +// when identity has not been injected, which is the actor situation in #761. +func TestRelayRefusesNonAteomSource(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + for _, tc := range []struct { + name string + service string + }{ + {name: "another substrate component", service: "atelet"}, + {name: "actor telemetry", service: "actor"}, + {name: "no service name at all", service: ""}, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(tc.service)}}, + }) + if got := status.Code(err); got != codes.PermissionDenied { + t.Errorf("trace Export from service.name %q = code %v (%v), want %v", tc.service, got, err, codes.PermissionDenied) + } + + _, err = colmetricspb.NewMetricsServiceClient(conn).Export(context.Background(), &colmetricspb.ExportMetricsServiceRequest{ + ResourceMetrics: []*metricspb.ResourceMetrics{{Resource: serviceResource(tc.service)}}, + }) + if got := status.Code(err); got != codes.PermissionDenied { + t.Errorf("metric Export from service.name %q = code %v (%v), want %v", tc.service, got, err, codes.PermissionDenied) + } + }) + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.traces) != 0 || len(sink.metrics) != 0 { + t.Errorf("collector received %d traces and %d metrics from refused sources, want none to be forwarded", len(sink.traces), len(sink.metrics)) + } +} + +// TestRelayRefusesMixedBatch pins the all-or-nothing choice: dropping just the +// foreign resource would return success to a sender that lost telemetry. +func TestRelayRefusesMixedBatch(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + _, err = coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{ + {Resource: serviceResource("ateom-gvisor")}, + {Resource: serviceResource("actor")}, + }, + }) + if got := status.Code(err); got != codes.PermissionDenied { + t.Errorf("Export of a mixed batch = code %v (%v), want %v", got, err, codes.PermissionDenied) + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.traces) != 0 { + t.Errorf("collector received %d exports from a mixed batch, want the batch refused whole", len(sink.traces)) + } +} + +// TestRelayAcceptsEveryAteomService guards against the allowlist drifting from +// the binaries in a way that silently drops all of one runtime's telemetry. +func TestRelayAcceptsEveryAteomService(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + for service := range ateomServices { + if _, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(service)}}, + }); err != nil { + t.Errorf("Export from allowlisted service %q: %v", service, err) + continue + } + select { + case <-sink.got: + case <-time.After(5 * time.Second): + t.Errorf("collector never received the export from %q", service) + } + } +} + +// TestAteomServicesMatchTheAteomBinaries keeps the allowlist honest. A typo in +// it would otherwise be invisible: every real ateom export would be refused +// while every test here still passed, because they would share the typo. +func TestAteomServicesMatchTheAteomBinaries(t *testing.T) { + // Matches `const serviceName = "..."` in each ateom main package. + decl := regexp.MustCompile(`(?m)^\s*const\s+serviceName\s*=\s*"([^"]+)"`) + + found := map[string]bool{} + for _, main := range []string{"../../cmd/ateom-gvisor/main.go", "../../cmd/ateom-microvm/main.go"} { + src, err := os.ReadFile(main) + if err != nil { + t.Fatalf("reading %s: %v", main, err) + } + m := decl.FindSubmatch(src) + if m == nil { + t.Fatalf("no `const serviceName = \"...\"` found in %s; if it moved, this test and ateomServices both need updating", main) + } + name := string(m[1]) + found[name] = true + if !ateomServices[name] { + t.Errorf("%s reports service.name %q, which ateomServices does not allow; the relay would refuse all of its telemetry", main, name) + } + } + + for name := range ateomServices { + if !found[name] { + t.Errorf("ateomServices allows %q, but no ateom binary declares it", name) + } + } +} + +// TestDialRejectsRelativeSocketPath: a relative path was accepted here and then +// failed lazily at the first export, with the spans already gone. +func TestDialRejectsRelativeSocketPath(t *testing.T) { + conn, err := Dial(context.Background(), "relative/r.sock") + if conn != nil { + conn.Close() + } + if err == nil { + t.Fatal("Dial with a relative socket path returned no error; it would dial a misparsed target and lose telemetry per export") + } + if !strings.Contains(err.Error(), "absolute") { + t.Errorf("Dial error = %v, want it to say the path must be absolute", err) + } +} + +func TestNewServerRejectsRelativeSocketPath(t *testing.T) { + t.Setenv(endpointEnv, "collector:4317") + relay, err := NewServer(context.Background(), "relative/r.sock") + if relay != nil { + relay.Stop() + } + if err == nil { + t.Fatal("NewServer with a relative socket path returned no error; it would listen somewhere no ateom can name") + } + if !strings.Contains(err.Error(), "absolute") { + t.Errorf("NewServer error = %v, want it to say the path must be absolute", err) + } +} + +// TestServeLeavesAPopulatedDirectoryAlone is the RemoveAll regression: a flag +// value naming the directory must fail rather than empty it. +func TestServeLeavesAPopulatedDirectoryAlone(t *testing.T) { + _, collector := startFakeCollector(t) + t.Setenv(endpointEnv, collector) + + dir := filepath.Join(t.TempDir(), "basepath") + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + occupant := filepath.Join(dir, "ateom.sock") + if err := os.WriteFile(occupant, nil, 0o600); err != nil { + t.Fatalf("planting a neighbouring socket: %v", err) + } + + relay, err := NewServer(context.Background(), dir) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + t.Cleanup(relay.Stop) + + if err := relay.Serve(context.Background()); err == nil { + t.Error("Serve on a populated directory returned no error, want it to refuse") + } + if _, err := os.Stat(occupant); err != nil { + t.Errorf("os.Stat(%q) = %v, want the neighbouring socket untouched", occupant, err) + } +} + +func TestNormalizeEndpoint(t *testing.T) { + for _, tc := range []struct { + name string + in string + want string + wantErr string + }{ + {name: "host and port", in: "otel-collector.ate-system.svc:4317", want: "otel-collector.ate-system.svc:4317"}, + {name: "bare host defaults the port", in: "otel-collector", want: "otel-collector:" + otlpDefaultPort}, + {name: "http url", in: "http://otel-collector:4317", want: "otel-collector:4317"}, + {name: "http url without port", in: "http://otel-collector", want: "otel-collector:" + otlpDefaultPort}, + {name: "ipv6 literal", in: "[::1]:4317", want: "[::1]:4317"}, + {name: "ipv6 literal without port", in: "[::1]", want: "[::1]:" + otlpDefaultPort}, + {name: "https rejected", in: "https://otel-collector:4317", wantErr: "https"}, + {name: "unknown scheme rejected", in: "grpc://otel-collector:4317", wantErr: "unsupported scheme"}, + {name: "empty host rejected", in: "http://:4317", wantErr: "names no host"}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := normalizeEndpoint(tc.in) + if tc.wantErr != "" { + if err == nil { + t.Fatalf("normalizeEndpoint(%q) = %q, want an error containing %q", tc.in, got, tc.wantErr) + } + if !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("normalizeEndpoint(%q) error = %v, want it to mention %q", tc.in, err, tc.wantErr) + } + return + } + if err != nil { + t.Fatalf("normalizeEndpoint(%q): %v", tc.in, err) + } + if got != tc.want { + t.Errorf("normalizeEndpoint(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} + +func TestUpstreamTarget(t *testing.T) { + for _, tc := range []struct { + name string + generic string + traces string + metrics string + want string + wantErr bool + }{ + {name: "unset", want: ""}, + {name: "generic only", generic: "collector:4317", want: "collector:4317"}, + {name: "signal specific overrides generic", generic: "generic:4317", traces: "specific:4317", metrics: "specific:4317", want: "specific:4317"}, + {name: "generic conflicts with different traces specific", generic: "generic:4317", traces: "traces-only:4317", wantErr: true}, + {name: "generic conflicts with different metrics specific", generic: "generic:4317", metrics: "metrics-only:4317", wantErr: true}, + {name: "matching generic and signal specific", generic: "collector:4317", traces: "collector:4317", metrics: "collector:4317", want: "collector:4317"}, + {name: "traces specific alone", traces: "specific:4317", want: "specific:4317"}, + {name: "metrics specific alone", metrics: "specific:4317", want: "specific:4317"}, + {name: "conflicting signals rejected", traces: "a:4317", metrics: "b:4317", wantErr: true}, + {name: "whitespace trimmed", generic: " collector:4317 ", want: "collector:4317"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(endpointEnv, tc.generic) + t.Setenv(tracesEndpointEnv, tc.traces) + t.Setenv(metricsEndpointEnv, tc.metrics) + got, err := upstreamTarget() + if tc.wantErr { + if err == nil { + t.Fatalf("upstreamTarget() = %q, want an error", got) + } + return + } + if err != nil { + t.Fatalf("upstreamTarget(): %v", err) + } + if got != tc.want { + t.Errorf("upstreamTarget() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestUpstreamCompression(t *testing.T) { + for _, tc := range []struct { + name string + generic string + traces string + metrics string + want string + wantErr bool + }{ + {name: "unset", want: "none"}, + {name: "generic gzip", generic: "gzip", want: "gzip"}, + {name: "generic none", generic: "none", want: "none"}, + {name: "traces specific gzip", traces: "gzip", want: "gzip"}, + {name: "metrics specific gzip", metrics: "gzip", want: "gzip"}, + {name: "both specific gzip", traces: "gzip", metrics: "gzip", want: "gzip"}, + {name: "conflicting compression rejected", traces: "gzip", metrics: "none", wantErr: true}, + {name: "generic conflicts with traces", generic: "none", traces: "gzip", wantErr: true}, + {name: "invalid compression rejected", generic: "zstd", wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(compressionEnv, tc.generic) + t.Setenv(tracesCompressionEnv, tc.traces) + t.Setenv(metricsCompressionEnv, tc.metrics) + got, err := upstreamCompression() + if tc.wantErr { + if err == nil { + t.Fatalf("upstreamCompression() = %q, want error", got) + } + return + } + if err != nil { + t.Fatalf("upstreamCompression(): %v", err) + } + if got != tc.want { + t.Errorf("upstreamCompression() = %q, want %q", got, tc.want) + } + }) + } +} + +// exportBoth sends one empty trace batch and one empty metric batch from +// service through the relay, and returns the metadata each arrived with. +func exportBoth(t *testing.T, sink *fakeCollector, sock, service string, ctx context.Context) (traceMD, metricMD metadata.MD) { + t.Helper() + conn, err := Dial(context.Background(), sock) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + if _, err := coltracepb.NewTraceServiceClient(conn).Export(ctx, &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(service)}}, + }); err != nil { + t.Fatalf("traceClient.Export: %v", err) + } + if _, err := colmetricspb.NewMetricsServiceClient(conn).Export(ctx, &colmetricspb.ExportMetricsServiceRequest{ + ResourceMetrics: []*metricspb.ResourceMetrics{{Resource: serviceResource(service)}}, + }); err != nil { + t.Fatalf("metricClient.Export: %v", err) + } + + sink.mu.Lock() + defer sink.mu.Unlock() + if len(sink.traceMD) == 0 { + t.Fatal("collector received no metadata for trace export") + } + if len(sink.metricMD) == 0 { + t.Fatal("collector received no metadata for metrics export") + } + return sink.traceMD[0], sink.metricMD[0] +} + +// The upstream leg is atelet's connection, so its credentials are atelet's. An +// ateom that sets a header of its own must not get to choose what atelet +// presents to the collector. +func TestExportDropsClientMetadata(t *testing.T) { + sink, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + ctx := metadata.AppendToOutgoingContext(context.Background(), + "authorization", "Bearer client-token", + "custom-header", "custom-value", + ) + traceMD, metricMD := exportBoth(t, sink, sock, "ateom-gvisor", ctx) + + for _, tc := range []struct { + signal string + md metadata.MD + }{{"trace", traceMD}, {"metric", metricMD}} { + for _, key := range []string{"authorization", "custom-header"} { + if got := tc.md.Get(key); len(got) != 0 { + t.Errorf("%s export reached the collector with the client's %s = %v, want it dropped", tc.signal, key, got) + } + } + } +} + +// ...and the headers atelet is configured with are attached in its place, per +// signal, exactly as the SDK exporter the relay stands in for would have. +func TestExportAttachesAteletHeaders(t *testing.T) { + sink, collector := startFakeCollector(t) + // Set before startRelay: NewServer resolves headers once, at construction. + t.Setenv(headersEnv, "authorization=Bearer atelet-token,x-tenant=substrate") + t.Setenv(metricsHeadersEnv, "authorization=Bearer metrics-token") + sock := startRelay(t, collector) + + // The client sends its own, which must lose to atelet's rather than + // appending a second value the collector might pick either way. + ctx := metadata.AppendToOutgoingContext(context.Background(), "authorization", "Bearer client-token") + traceMD, metricMD := exportBoth(t, sink, sock, "ateom-gvisor", ctx) + + if got := traceMD.Get("authorization"); len(got) != 1 || got[0] != "Bearer atelet-token" { + t.Errorf("trace export authorization = %v, want exactly [Bearer atelet-token]", got) + } + if got := traceMD.Get("x-tenant"); len(got) != 1 || got[0] != "substrate" { + t.Errorf("trace export x-tenant = %v, want [substrate]", got) + } + // The metrics-specific variable replaces the generic one whole, so x-tenant + // is deliberately absent here. + if got := metricMD.Get("authorization"); len(got) != 1 || got[0] != "Bearer metrics-token" { + t.Errorf("metric export authorization = %v, want exactly [Bearer metrics-token]", got) + } + if got := metricMD.Get("x-tenant"); len(got) != 0 { + t.Errorf("metric export x-tenant = %v, want absent: %s replaces %s rather than merging", got, metricsHeadersEnv, headersEnv) + } +} + +func TestParseHeaders(t *testing.T) { + for _, tc := range []struct { + name string + raw string + want map[string][]string + wantErr bool + }{ + {name: "empty", raw: "", want: map[string][]string{}}, + {name: "single", raw: "api-key=secret", want: map[string][]string{"api-key": {"secret"}}}, + {name: "multiple with spaces", raw: "api-key=secret, x-tenant = sub ", want: map[string][]string{"api-key": {"secret"}, "x-tenant": {"sub"}}}, + // gRPC metadata keys are case-insensitive and stored lower-cased; a + // mixed-case key here would be invisible to metadata.Get. + {name: "key lower-cased", raw: "X-Tenant=sub", want: map[string][]string{"x-tenant": {"sub"}}}, + // Percent-decoding is what lets a base64 token containing "=" through. + {name: "percent-encoded value", raw: "authorization=Bearer%20abc%3D%3D", want: map[string][]string{"authorization": {"Bearer abc=="}}}, + {name: "empty value kept", raw: "x-tenant=", want: map[string][]string{"x-tenant": {""}}}, + {name: "trailing comma tolerated", raw: "api-key=secret,", want: map[string][]string{"api-key": {"secret"}}}, + {name: "missing equals rejected", raw: "api-key", wantErr: true}, + {name: "empty name rejected", raw: "=secret", wantErr: true}, + {name: "bad percent-encoding rejected", raw: "api-key=%zz", wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := parseHeaders(tc.raw) + if tc.wantErr { + if err == nil { + t.Fatalf("parseHeaders(%q) = %v, want error", tc.raw, got) + } + // A credential must not end up in a log line or a test failure. + if strings.Contains(err.Error(), "secret") { + t.Errorf("parseHeaders error leaks the header value: %v", err) + } + return + } + if err != nil { + t.Fatalf("parseHeaders(%q): %v", tc.raw, err) + } + if len(got) != len(tc.want) { + t.Fatalf("parseHeaders(%q) = %v, want %v", tc.raw, got, tc.want) + } + for k, want := range tc.want { + if diff := got.Get(k); len(diff) != len(want) || (len(want) == 1 && diff[0] != want[0]) { + t.Errorf("parseHeaders(%q)[%q] = %v, want %v", tc.raw, k, diff, want) + } + } + }) + } +} + +// The signal-specific variable replaces the generic one, per the OTLP spec. +func TestUpstreamHeadersSignalOverride(t *testing.T) { + t.Setenv(headersEnv, "api-key=generic") + t.Setenv(tracesHeadersEnv, "x-tenant=traces") + + traces, err := upstreamHeaders(tracesHeadersEnv) + if err != nil { + t.Fatalf("upstreamHeaders(traces): %v", err) + } + if got := traces.Get("api-key"); len(got) != 0 { + t.Errorf("traces api-key = %v, want absent; the signal-specific variable replaces the generic one", got) + } + if got := traces.Get("x-tenant"); len(got) != 1 || got[0] != "traces" { + t.Errorf("traces x-tenant = %v, want [traces]", got) + } + + // metrics has no override, so it falls back to the generic variable. + metrics, err := upstreamHeaders(metricsHeadersEnv) + if err != nil { + t.Fatalf("upstreamHeaders(metrics): %v", err) + } + if got := metrics.Get("api-key"); len(got) != 1 || got[0] != "generic" { + t.Errorf("metrics api-key = %v, want [generic]", got) + } +} + +// A header set atelet cannot parse fails startup rather than becoming a +// per-export failure once the socket is already live. +func TestNewServerRejectsUnparseableHeaders(t *testing.T) { + _, collector := startFakeCollector(t) + t.Setenv(endpointEnv, collector) + t.Setenv(headersEnv, "not-a-pair") + + relay, err := NewServer(context.Background(), filepath.Join(t.TempDir(), "r.sock")) + if err == nil { + if relay != nil { + relay.Stop() + } + t.Fatal("NewServer accepted an unparseable OTEL_EXPORTER_OTLP_HEADERS") + } +} + +func TestSocketPermissions(t *testing.T) { + _, collector := startFakeCollector(t) + sock := startRelay(t, collector) + + info, err := os.Stat(sock) + if err != nil { + t.Fatalf("Stat(%q): %v", sock, err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Errorf("socket permissions = %04o, want 0600", perm) + } +} diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index c9a434d15..236dd3477 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -32,6 +32,7 @@ import ( "github.com/google/uuid" "github.com/prometheus/client_golang/prometheus/promhttp" "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" "go.opentelemetry.io/otel/exporters/prometheus" @@ -40,6 +41,7 @@ import ( "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.40.0" + "google.golang.org/grpc" ) // InitLogger sets the global slog logger to a JSON handler wrapped in @@ -83,16 +85,18 @@ var serviceInstanceID = uuid.NewString() // newResource builds the resource shared by the tracer and meter providers. // WithFromEnv is last so OTEL_* env vars override the defaults. -func newResource(ctx context.Context, serviceName string) (*resource.Resource, error) { +func newResource(ctx context.Context, serviceName string, extraAttrs ...attribute.KeyValue) (*resource.Resource, error) { + attrs := []attribute.KeyValue{ + semconv.ServiceName(serviceName), + semconv.ServiceInstanceID(serviceInstanceID), + } + attrs = append(attrs, extraAttrs...) res, err := resource.New(ctx, resource.WithTelemetrySDK(), // Must track the schema version the SDK's own detectors emit, else the // merge drops the schema URL with ErrSchemaURLConflict (tolerated below). resource.WithSchemaURL(semconv.SchemaURL), - resource.WithAttributes( - semconv.ServiceName(serviceName), - semconv.ServiceInstanceID(serviceInstanceID), - ), + resource.WithAttributes(attrs...), resource.WithFromEnv(), ) if errors.Is(err, resource.ErrPartialResource) || errors.Is(err, resource.ErrSchemaURLConflict) { @@ -103,6 +107,38 @@ func newResource(ctx context.Context, serviceName string) (*resource.Resource, e return res, nil } +// relayAttrKey records which OTLP export path a signal took. It only makes +// sense for the components that have a relay to take or miss — the ateoms — +// so relayAttrs leaves it off everything else rather than labeling, say, +// atecontroller "direct" for a relay it was never offered. +// +// The name is spelled here rather than taken from internal/ateattr on purpose: +// serverboot is the bottom of the dependency graph (it imports one other +// agent-substrate package) and every binary's main links it, while ateattr +// pulls in pkg/api/v1alpha1, ateapipb, internal/resources and ateletpb. Move it +// to ateattr if that cost ever drops, or if a second non-ate.* consumer appears. +const relayAttrKey = "ate.otlp.relay" + +// relayAttrs describes the export path taken by a component that could have +// used the relay. relayCapable false means the component never had one, and +// gets no attribute at all; true means it did, and conn says whether it got it. +// +// The distinction matters because a nil conn on a relay-capable component is +// exactly the degraded case worth alerting on: the ateom asked for the relay, +// could not dial it, and is now exporting over the worker pod's own network. +// Collapsing that into the same "no attribute" bucket as atecontroller would +// hide it. +func relayAttrs(relayCapable bool, conn *grpc.ClientConn) []attribute.KeyValue { + if !relayCapable { + return nil + } + status := "direct" + if conn != nil { + status = "relay" + } + return []attribute.KeyValue{attribute.String(relayAttrKey, status)} +} + // TracingOptions configures InitTracing. type TracingOptions struct { // ServiceName is required; populates resource.semconv ServiceName. @@ -111,6 +147,19 @@ type TracingOptions struct { // OTEL_TRACES_SAMPLER / OTEL_TRACES_SAMPLER_ARG override the component // default. Sampling TraceSampling + // ExporterConn, when non-nil, is the connection the OTLP exporter sends + // over, instead of dialing OTEL_EXPORTER_OTLP_ENDPOINT itself. ateom passes + // the unix socket to atelet's relay (internal/otlprelay) so a worker pod + // exports without a network path of its own; nil keeps the direct dial. + // + // The caller owns the connection: the exporter's Shutdown does not close a + // connection it did not create. + ExporterConn *grpc.ClientConn + // RelayCapable marks a component that is meant to export through the relay, + // whether or not it managed to (see relayAttrs). Only the ateoms set it. It + // is separate from ExporterConn because a nil conn on its own cannot tell + // "the ateom tried and fell back" from "this component never had a relay". + RelayCapable bool } // InitTracing registers a global TracerProvider with the given options @@ -122,7 +171,7 @@ func InitTracing(ctx context.Context, opts TracingOptions) (*sdktrace.TracerProv if opts.Sampling.sampler == nil { return nil, fmt.Errorf("TracingOptions.Sampling is required") } - res, err := newResource(ctx, opts.ServiceName) + res, err := newResource(ctx, opts.ServiceName, relayAttrs(opts.RelayCapable, opts.ExporterConn)...) if err != nil { return nil, fmt.Errorf("create tracer resource: %w", err) } @@ -138,10 +187,16 @@ func InitTracing(ctx context.Context, opts TracingOptions) (*sdktrace.TracerProv sdktrace.WithResource(res), sdktrace.WithSampler(opts.Sampling.Sampler()), } - exporter, err := otlptracegrpc.New(ctx, + expOpts := []otlptracegrpc.Option{ // GKE managed traces doesn't support validating the TLS certs of the collector. otlptracegrpc.WithInsecure(), - ) + } + if opts.ExporterConn != nil { + // WithGRPCConn takes precedence over endpoint/credential options, so + // WithInsecure above is inert on this path. + expOpts = append(expOpts, otlptracegrpc.WithGRPCConn(opts.ExporterConn)) + } + exporter, err := otlptracegrpc.New(ctx, expOpts...) if err != nil { return nil, fmt.Errorf("create OTLP exporter: %w", err) } @@ -166,7 +221,7 @@ func InitMetrics(ctx context.Context, serviceName string) (*sdkmetric.MeterProvi if err != nil { return nil, fmt.Errorf("create Prometheus metric exporter: %w", err) } - return newMeterProvider(ctx, serviceName, nil, promExporter) + return newMeterProvider(ctx, serviceName, false, nil, nil, promExporter) } // InitMetricsPushOnly is InitMetrics without the Prometheus reader, for binaries @@ -175,18 +230,45 @@ func InitMetrics(ctx context.Context, serviceName string) (*sdkmetric.MeterProvi // recorded outside the OTel SDK on the same push path; atecontroller bridges // controller-runtime's registry that way. func InitMetricsPushOnly(ctx context.Context, serviceName string, producers ...sdkmetric.Producer) (*sdkmetric.MeterProvider, error) { - return newMeterProvider(ctx, serviceName, producers) + return newMeterProvider(ctx, serviceName, false, nil, producers) } -func newMeterProvider(ctx context.Context, serviceName string, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) { +// InitMetricsPushOnlyVia is InitMetricsPushOnly with an explicit exporter +// connection: the metrics counterpart of TracingOptions.ExporterConn. ateom +// passes atelet's relay socket (internal/otlprelay) so the worker pod needs no +// network path of its own; a nil conn keeps the direct dial to +// OTEL_EXPORTER_OTLP_ENDPOINT. +// +// The caller owns the connection: the meter provider's Shutdown does not close +// a connection it did not create. +// +// Calling this at all marks the component relay-capable, so its metrics carry +// the relay attribute (see relayAttrs) either way — "relay" with a conn, +// "direct" without one. It is the metrics counterpart of +// TracingOptions.RelayCapable, implied rather than a parameter because only a +// caller that has a relay to pass reaches for this function in the first place. +func InitMetricsPushOnlyVia(ctx context.Context, serviceName string, conn *grpc.ClientConn, producers ...sdkmetric.Producer) (*sdkmetric.MeterProvider, error) { + return newMeterProvider(ctx, serviceName, true, conn, producers) +} + +func newMeterProvider(ctx context.Context, serviceName string, relayCapable bool, conn *grpc.ClientConn, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) { if serviceName == "" { return nil, fmt.Errorf("serviceName is required") } - otlpExporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithInsecure()) + expOpts := []otlpmetricgrpc.Option{ + // GKE managed metrics doesn't support validating the TLS certs of the collector. + otlpmetricgrpc.WithInsecure(), + } + if conn != nil { + // WithGRPCConn takes precedence over endpoint/credential options, so + // WithInsecure above is inert on this path. + expOpts = append(expOpts, otlpmetricgrpc.WithGRPCConn(conn)) + } + otlpExporter, err := otlpmetricgrpc.New(ctx, expOpts...) if err != nil { return nil, fmt.Errorf("create OTLP metric exporter: %w", err) } - res, err := newResource(ctx, serviceName) + res, err := newResource(ctx, serviceName, relayAttrs(relayCapable, conn)...) if err != nil { return nil, fmt.Errorf("create metric resource: %w", err) } diff --git a/internal/serverboot/serverboot_test.go b/internal/serverboot/serverboot_test.go index 56118a54b..5217d29dc 100644 --- a/internal/serverboot/serverboot_test.go +++ b/internal/serverboot/serverboot_test.go @@ -25,8 +25,12 @@ import ( "time" "github.com/prometheus/client_golang/prometheus/promhttp" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.opentelemetry.io/otel/sdk/resource" semconv "go.opentelemetry.io/otel/semconv/v1.21.0" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) func resourceAttrs(res *resource.Resource) map[string]string { @@ -67,6 +71,104 @@ func TestNewResourceEnvWins(t *testing.T) { } } +// lazyConn is a ClientConn that never dials: grpc.NewClient connects on first +// RPC, and relayAttrs only cares whether the pointer is nil. +func lazyConn(t *testing.T) *grpc.ClientConn { + t.Helper() + conn, err := grpc.NewClient("passthrough:///unused", grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("grpc.NewClient: %v", err) + } + t.Cleanup(func() { conn.Close() }) + return conn +} + +// The three cases relayAttrs exists to separate: a component that got the +// relay, one that wanted it and fell back, and one that was never offered one. +// The last must carry no attribute at all rather than a misleading "direct". +func TestRelayAttrs(t *testing.T) { + for _, tc := range []struct { + name string + relayCapable bool + conn bool + want string // "" means the attribute must be absent + }{ + {name: "relay capable with conn", relayCapable: true, conn: true, want: "relay"}, + {name: "relay capable fell back", relayCapable: true, conn: false, want: "direct"}, + {name: "not relay capable", relayCapable: false, conn: false}, + // atecontroller stays unlabeled even if some future caller hands it a + // connection for another reason: capability, not the conn, is the gate. + {name: "not relay capable with conn", relayCapable: false, conn: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var conn *grpc.ClientConn + if tc.conn { + conn = lazyConn(t) + } + res, err := newResource(context.Background(), "ateom-gvisor", relayAttrs(tc.relayCapable, conn)...) + if err != nil { + t.Fatalf("newResource: %v", err) + } + got, ok := resourceAttrs(res)[relayAttrKey] + if tc.want == "" { + if ok { + t.Errorf("%s = %q, want absent", relayAttrKey, got) + } + return + } + if got != tc.want { + t.Errorf("%s = %q, want %q", relayAttrKey, got, tc.want) + } + }) + } +} + +// collectedResource reads back the resource a meter provider actually stamps on +// its exports, by attaching a ManualReader alongside the OTLP one. The provider +// does not expose its resource any other way, and asserting on newResource's +// return would only re-test relayAttrs. +func collectedResource(t *testing.T, relayCapable bool, conn *grpc.ClientConn) map[string]string { + t.Helper() + reader := sdkmetric.NewManualReader() + mp, err := newMeterProvider(context.Background(), "ateom-gvisor", relayCapable, conn, nil, reader) + if err != nil { + t.Fatalf("newMeterProvider: %v", err) + } + t.Cleanup(func() { + // Shutdown flushes the OTLP reader too, and no collector is listening + // here: a live context spends the exporter's full retry budget (~10s) + // per provider. A cancelled one skips the flush, which is all this test + // wants from Shutdown anyway. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + mp.Shutdown(ctx) + }) + // A meter with no instruments still collects, carrying the resource. + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("Collect: %v", err) + } + return resourceAttrs(rm.Resource) +} + +// The metric path's half of the decision, asserted on what the provider exports +// rather than on what relayAttrs returns: a wiring mistake in newMeterProvider +// (passing the wrong flag, dropping the attrs) fails here and not in +// TestRelayAttrs. +func TestMeterProviderRelayAttribute(t *testing.T) { + if got, ok := collectedResource(t, true, lazyConn(t))[relayAttrKey]; !ok || got != "relay" { + t.Errorf("%s = %q (present %t), want relay", relayAttrKey, got, ok) + } + if got, ok := collectedResource(t, true, nil)[relayAttrKey]; !ok || got != "direct" { + t.Errorf("%s = %q (present %t), want direct", relayAttrKey, got, ok) + } + // atecontroller and ateapi: no relay was ever offered, so no claim is made + // about which path they took. + if got, ok := collectedResource(t, false, nil)[relayAttrKey]; ok { + t.Errorf("%s = %q, want absent for a component with no relay", relayAttrKey, got) + } +} + func TestReadyzDrainsWhileHealthzStaysUp(t *testing.T) { readiness := &Readiness{} mux := metricsMux(MetricsServerOptions{