From f667d495fd4fa49c4109692fe89b43bfc5c31b8d Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Sat, 8 Aug 2026 01:32:16 +0000 Subject: [PATCH 1/6] ateom: export telemetry through an atelet unix-socket OTLP relay ateom runs inside the worker pod that hosts the actor, and exported OTLP straight to the collector over the pod's network. This adds a node-local relay: ateom pushes OTLP/gRPC over a unix socket that atelet serves and forwards to the collector, so a worker pod needs no network path of its own to export spans and metrics. Four things motivate it: - Blast radius. The pod runs untrusted agent code, so allowing it egress to the collector makes the collector reachable to anything that escapes the sandbox. A unix socket cannot leave the node. - Connection count. Worker pods are heavily oversubscribed; N ateoms per node each held their own collector connection. They collapse into atelet's single per-node one. - Interference. ateom transparently redirects actor egress to its own atunnel listener, and its own outbound traffic has to stay clear of the rules it installs. A unix socket is not IP traffic. - Shutdown loss. Teardown frees the actor's network and then the pod goes away, which is when the spans describing teardown are still queued in the batch processor. atelet outlives the worker pod. The relay forwards the OTLP request verbatim rather than decoding and re-exporting, so each ateom's own resource (service.name, service.instance.id) survives instead of being absorbed into atelet's. It is best-effort: an ateom that finds no socket at startup logs it and exports directly to OTEL_EXPORTER_OTLP_ENDPOINT as before, so this is a no-op for a cluster running an older atelet. atelet likewise declines to serve a relay when no collector is configured, since it would accept spans only to drop them. Both halves stay off with --otlp-relay-socket="". The socket lives in ateompath.BasePath, the hostPath already mounted at the same path into atelet and into every ateom pod, so no new volume or controller change is needed. Also includes: - End-to-end tests covering the full serverboot-to-collector path. - Observability documentation updates for Jaeger tracing. --- cmd/atelet/main.go | 24 ++ cmd/ateom-gvisor/main.go | 22 +- cmd/ateom-microvm/main.go | 22 +- docs/observability.md | 16 +- internal/ateompath/ateompath.go | 16 ++ internal/ateompath/ateompath_test.go | 22 ++ internal/otlprelay/client.go | 65 +++++ internal/otlprelay/e2e_test.go | 110 ++++++++ internal/otlprelay/relay.go | 254 +++++++++++++++++ internal/otlprelay/relay_test.go | 390 +++++++++++++++++++++++++++ internal/serverboot/serverboot.go | 48 +++- 11 files changed, 976 insertions(+), 13 deletions(-) create mode 100644 internal/otlprelay/client.go create mode 100644 internal/otlprelay/e2e_test.go create mode 100644 internal/otlprelay/relay.go create mode 100644 internal/otlprelay/relay_test.go diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 60cca4c3c..d8f7e83eb 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 { + serverboot.Fatal(ctx, "Failed to create the OTLP relay", 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..7eebddeef 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,28 @@ 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. + relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket) + if err != nil { + serverboot.Fatal(ctx, "Failed to connect to the OTLP relay", 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, }) 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..08dd802d3 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,28 @@ 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. + relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket) + if err != nil { + serverboot.Fatal(ctx, "Failed to connect to the OTLP relay", 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, }) 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..2fabf6fe4 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,18 @@ 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 itself. 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 no longer needs egress to the collector; 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. + --- ## 5. Dashboards diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 680e329c1..a807fb428 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -53,6 +53,22 @@ 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. +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..a31db64c5 --- /dev/null +++ b/internal/otlprelay/client.go @@ -0,0 +1,65 @@ +// 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 := 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..6e894ee8c --- /dev/null +++ b/internal/otlprelay/e2e_test.go @@ -0,0 +1,110 @@ +// 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" + "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) + 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, + }) + 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 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() + } + } + for _, ss := range rs.GetScopeSpans() { + for _, s := range ss.GetSpans() { + gotSpan = s.GetName() + } + } + } + } + t.Logf("collector received span %q from service %q", gotSpan, gotService) + + // 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") + } +} diff --git a/internal/otlprelay/relay.go b/internal/otlprelay/relay.go new file mode 100644 index 000000000..a1b929395 --- /dev/null +++ b/internal/otlprelay/relay.go @@ -0,0 +1,254 @@ +// 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. 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. +package otlprelay + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/url" + "os" + "path/filepath" + "strings" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" + coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/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" + + // otlpDefaultPort is the OTLP/gRPC default, used when an endpoint names a + // host with no port. Matches atenet's normalizeOtlpCollector. + otlpDefaultPort = "4317" + + // socketMode keeps the relay socket reachable by the ateom pods on the node + // (which do not necessarily share atelet's uid) while staying off-node by + // construction. The socket is inside BasePath, a root-owned host directory. + socketMode = 0o666 + + // 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 +} + +// 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 +} + +// 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. +func (t *traceRelay) Export(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) { + return t.upstream.Export(ctx, req) +} + +type metricRelay struct { + colmetricspb.UnimplementedMetricsServiceServer + upstream colmetricspb.MetricsServiceClient +} + +// Export forwards a batch of metric datapoints to the collector unchanged. +func (m *metricRelay) Export(ctx context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) { + return m.upstream.Export(ctx, req) +} + +// 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 + } + 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 + } + + // 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, + grpc.WithTransportCredentials(insecure.NewCredentials())) + 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)), + } + coltracepb.RegisterTraceServiceServer(s.grpc, &traceRelay{upstream: coltracepb.NewTraceServiceClient(upstream)}) + colmetricspb.RegisterMetricsServiceServer(s.grpc, &metricRelay{upstream: colmetricspb.NewMetricsServiceClient(upstream)}) + slog.InfoContext(ctx, "OTLP relay forwarding to collector", slog.String("collector", target)) + 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.RemoveAll(s.sockPath); err != nil { + 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 and closes the upstream connection. +func (s *Server) Stop() { + s.grpc.GracefulStop() + _ = s.upstream.Close() + _ = os.Remove(s.sockPath) +} + +// 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)) + + resolved := generic + for _, specific := range []string{traces, metrics} { + if specific == "" { + continue + } + if resolved != "" && resolved != generic && specific != resolved { + return "", fmt.Errorf("%s and %s name different collectors (%q vs %q); the relay carries both signals over one connection", + tracesEndpointEnv, metricsEndpointEnv, resolved, specific) + } + resolved = specific + } + 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..609470e83 --- /dev/null +++ b/internal/otlprelay/relay_test.go @@ -0,0 +1,390 @@ +// 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" + "io/fs" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc" + "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 + 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) + 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) + 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) +} + +// 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: &resourcepb.Resource{ + Attributes: []*commonpb.KeyValue{{ + Key: "service.name", + Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "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{{ + 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") + } +} + +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: "one signal specific", generic: "generic:4317", metrics: "specific:4317", want: "specific:4317"}, + {name: "signal specific alone", traces: "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) + } + }) + } +} diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index c9a434d15..33d6d8726 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -40,6 +40,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 @@ -111,6 +112,14 @@ 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 } // InitTracing registers a global TracerProvider with the given options @@ -138,10 +147,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 +181,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, nil, nil, promExporter) } // InitMetricsPushOnly is InitMetrics without the Prometheus reader, for binaries @@ -175,14 +190,35 @@ 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, nil, producers) +} + +// 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. +func InitMetricsPushOnlyVia(ctx context.Context, serviceName string, conn *grpc.ClientConn, producers ...sdkmetric.Producer) (*sdkmetric.MeterProvider, error) { + return newMeterProvider(ctx, serviceName, conn, producers) } -func newMeterProvider(ctx context.Context, serviceName string, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) { +func newMeterProvider(ctx context.Context, serviceName string, 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) } From 648f5a63eb99d805ce9cbe8379ee164d018c31c3 Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Mon, 10 Aug 2026 16:43:43 -0700 Subject: [PATCH 2/6] go.mod: record go.opentelemetry.io/proto/otlp as a direct dependency internal/otlprelay imports the OTLP collector protos directly, but go.mod still carried the module as indirect: go mod tidy was never rerun after the relay landed, so hack/verify/go-modules.sh regenerates, finds a diff and fails run-tests. Nothing else moves. vendor/modules.txt already marked the module explicit, which is why every local build and test stayed green and only CI caught it. --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From fd57f4e9e6909242f2cf12f8a8e162ec9db14294 Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Mon, 10 Aug 2026 16:44:11 -0700 Subject: [PATCH 3/6] otlprelay: scope the relay to ateom, and fix two socket path bugs Addresses review on #809: Krisztian's scoping request and Benjamin's two bugs. Scope the relay to ateom sources. Forwarding a request untouched is not a property of relaying in general; it is correct only for a source whose resource is already correct. ateom's is, so the relay now says so and refuses everything else with PermissionDenied, a resource declaring no service.name included. The case this excludes is actor telemetry: actors share a hostname ("runsc") and an interior IP, so their series merge unless identity is injected from outside the actor (#761), and rewriting in flight is the negation of pass-through. The two do not share this socket today -- #761 is actor->ateom and this relay is ateom->atelet -- but naming the contract while it still holds makes that path an added branch later rather than a re-argument about whether pass-through was ever safe. A batch is refused whole rather than having the offending resource dropped, since a partial success the sender reads as success loses telemetry silently. The allowlist duplicates the serviceName constants in cmd/ateom-*, which are package main and cannot be imported, so a test reads them back out of the source: a typo there would otherwise refuse every real ateom export while the package's own tests, sharing the typo, kept passing. Remove the stale socket with os.Remove rather than os.RemoveAll. The path comes from a flag and lives in BasePath, the host directory that also holds every ateom's own socket, so a value naming the directory would have had RemoveAll empty it. os.Remove refuses a populated directory, turning that into a startup error instead of data loss. Reject a relative socket path in both halves. gRPC does not resolve one: "unix://foo/r.sock" parses as authority "foo" and path "/r.sock". Because grpc.NewClient is lazy, that wrong target was accepted at startup and failed per export afterwards with the spans already gone, defeating the existence check in Dial that makes the direct-export fallback a startup decision. atelet and ateom would also resolve a relative path against different working directories, so it cannot name one socket for both halves. An absent socket still falls back quietly; only a malformed path errors. --- docs/observability.md | 2 + internal/otlprelay/client.go | 3 + internal/otlprelay/relay.go | 80 +++++++++++- internal/otlprelay/relay_test.go | 215 ++++++++++++++++++++++++++++++- 4 files changed, 289 insertions(+), 11 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index 2fabf6fe4..0344c9560 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -235,6 +235,8 @@ The socket sits in the `BasePath` hostPath already mounted into both, so nothing 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. +It forwards each request verbatim rather than decoding and re-exporting, which is what keeps every ateom its own service in Jaeger instead of being absorbed into `atelet`'s. That is safe only because ateom's resource is already right, so the relay carries ateom telemetry only and refuses anything else with `PermissionDenied`, a resource declaring no `service.name` included. 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 is the opposite of pass-through. + --- ## 5. Dashboards diff --git a/internal/otlprelay/client.go b/internal/otlprelay/client.go index a31db64c5..fef990666 100644 --- a/internal/otlprelay/client.go +++ b/internal/otlprelay/client.go @@ -43,6 +43,9 @@ 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", diff --git a/internal/otlprelay/relay.go b/internal/otlprelay/relay.go index a1b929395..1b58bd499 100644 --- a/internal/otlprelay/relay.go +++ b/internal/otlprelay/relay.go @@ -38,12 +38,16 @@ // The relay forwards the OTLP request message verbatim rather than decoding it // into SDK records and re-exporting. 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. +// spans stay attributed to ateom instead of being absorbed into atelet's. That +// holds only for a source whose resource is already right, so the relay carries +// ateom telemetry only; see ateomServices. package otlprelay import ( "context" + "errors" "fmt" + "io/fs" "log/slog" "net" "net/url" @@ -52,10 +56,14 @@ import ( "strings" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "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 ( @@ -66,8 +74,7 @@ const ( tracesEndpointEnv = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" metricsEndpointEnv = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" - // otlpDefaultPort is the OTLP/gRPC default, used when an endpoint names a - // host with no port. Matches atenet's normalizeOtlpCollector. + // otlpDefaultPort matches atenet's normalizeOtlpCollector. otlpDefaultPort = "4317" // socketMode keeps the relay socket reachable by the ateom pods on the node @@ -89,6 +96,42 @@ type Server struct { sockPath string } +// ateomServices are the only sources this relay carries, keyed by the +// service.name their resource declares. Mirrors the serviceName constants in +// cmd/ateom-gvisor and cmd/ateom-microvm, which are package main and cannot be +// imported; TestAteomServicesMatchTheAteomBinaries guards the duplication. +// +// Actor telemetry is what the allowlist excludes: actors share a hostname +// ("runsc") and an interior IP, so their series merge unless identity is +// injected from outside the actor (#761) -- a rewrite, not a forward. Nothing +// but ateom reaches this socket today; naming the contract now makes that path +// an added branch later rather than a re-argument about pass-through. +var ateomServices = map[string]bool{ + "ateom-gvisor": true, + "ateom-microvm": true, +} + +// checkAteomSource rejects a missing service.name along with an unrecognized +// one: an unidentified source is the one the relay cannot vouch for. +func checkAteomSource(r *resourcepb.Resource) error { + name := resourceServiceName(r) + if ateomServices[name] { + return nil + } + 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", + name) +} + +func resourceServiceName(r *resourcepb.Resource) string { + for _, attr := range r.GetAttributes() { + if attr.GetKey() == string(semconv.ServiceNameKey) { + return attr.GetValue().GetStringValue() + } + } + return "" +} + // 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 @@ -103,7 +146,15 @@ type traceRelay struct { // // 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 := checkAteomSource(rs.GetResource()); err != nil { + return nil, err + } + } return t.upstream.Export(ctx, req) } @@ -114,9 +165,25 @@ type metricRelay struct { // 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 := checkAteomSource(rm.GetResource()); err != nil { + return nil, err + } + } return m.upstream.Export(ctx, 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 @@ -126,6 +193,9 @@ 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 @@ -164,7 +234,7 @@ func (s *Server) Serve(ctx context.Context) error { } // A socket left behind by a previous atelet would make Listen fail with // EADDRINUSE even though nothing holds it. - if err := os.RemoveAll(s.sockPath); err != nil { + 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) @@ -183,7 +253,7 @@ func (s *Server) Serve(ctx context.Context) error { return s.grpc.Serve(lis) } -// Stop drains the relay and closes the upstream connection. +// Stop drains the relay, closes the upstream connection and removes the socket. func (s *Server) Stop() { s.grpc.GracefulStop() _ = s.upstream.Close() diff --git a/internal/otlprelay/relay_test.go b/internal/otlprelay/relay_test.go index 609470e83..4d69b31e3 100644 --- a/internal/otlprelay/relay_test.go +++ b/internal/otlprelay/relay_test.go @@ -21,6 +21,7 @@ import ( "net" "os" "path/filepath" + "regexp" "strings" "sync" "testing" @@ -28,6 +29,8 @@ import ( "github.com/google/go-cmp/cmp" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/testing/protocmp" colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1" @@ -134,6 +137,20 @@ func waitForSocket(t *testing.T, sock string, serveErr <-chan error) { 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. @@ -152,12 +169,7 @@ func TestRelayForwardsTracesVerbatim(t *testing.T) { req := &coltracepb.ExportTraceServiceRequest{ ResourceSpans: []*tracepb.ResourceSpans{{ - Resource: &resourcepb.Resource{ - Attributes: []*commonpb.KeyValue{{ - Key: "service.name", - Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "ateom-microvm"}}, - }}, - }, + Resource: serviceResource("ateom-microvm"), ScopeSpans: []*tracepb.ScopeSpans{{ Spans: []*tracepb.Span{{ Name: "RunWorkload", @@ -199,6 +211,7 @@ func TestRelayForwardsMetrics(t *testing.T) { req := &colmetricspb.ExportMetricsServiceRequest{ ResourceMetrics: []*metricspb.ResourceMetrics{{ + Resource: serviceResource("ateom-microvm"), ScopeMetrics: []*metricspb.ScopeMetrics{{ Metrics: []*metricspb.Metric{{Name: "ateom.workload.runs"}}, }}, @@ -313,6 +326,196 @@ func TestDialEmptyPath(t *testing.T) { } } +// 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 From 8ae4b5d57afdce47bbb5705a870338b5df7c106d Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Tue, 11 Aug 2026 23:34:02 +0000 Subject: [PATCH 4/6] otlprelay: address PR review feedback on metadata, compression, fallback, and permissions - Forward incoming gRPC metadata (headers, auth tokens) in trace and metric relays to upstream - Add upstream compression support (gzip / none) - Fix signal-specific and generic OTLP endpoint conflict resolution in upstreamTarget - Restrict socket mode to 0600 - Add substrate.otlp.relay resource attribute in serverboot for fleet-wide fallback observability - Make NewServer failure non-fatal in atelet - Ensure test isolation in e2e test with unroutable generic endpoint - Update documentation on ateom pass-through, k8sattributes preservation, and Phase 2 egress lockdown --- cmd/atelet/main.go | 2 +- docs/observability.md | 12 ++- internal/otlprelay/e2e_test.go | 14 ++- internal/otlprelay/relay.go | 132 ++++++++++++++++++------ internal/otlprelay/relay_test.go | 137 +++++++++++++++++++++++-- internal/serverboot/serverboot.go | 25 +++-- internal/serverboot/serverboot_test.go | 19 ++++ 7 files changed, 292 insertions(+), 49 deletions(-) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index d8f7e83eb..0ab4d8310 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -159,7 +159,7 @@ func main() { // 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 { - serverboot.Fatal(ctx, "Failed to create the OTLP relay", err) + 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 diff --git a/docs/observability.md b/docs/observability.md index 0344c9560..5d158c44b 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -225,17 +225,21 @@ Telemetry is emitted the same way everywhere; only the backend differs between a ### The ateom OTLP relay -ateom is the one component that does not talk to the collector itself. 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 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 no longer needs egress to the collector; 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 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. +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`). -It forwards each request verbatim rather than decoding and re-exporting, which is what keeps every ateom its own service in Jaeger instead of being absorbed into `atelet`'s. That is safe only because ateom's resource is already right, so the relay carries ateom telemetry only and refuses anything else with `PermissionDenied`, a resource declaring no `service.name` included. 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 is the opposite of pass-through. +> **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. --- diff --git a/internal/otlprelay/e2e_test.go b/internal/otlprelay/e2e_test.go index 6e894ee8c..bfbc48630 100644 --- a/internal/otlprelay/e2e_test.go +++ b/internal/otlprelay/e2e_test.go @@ -39,6 +39,10 @@ import ( 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) @@ -82,13 +86,16 @@ func TestEndToEndThroughServerboot(t *testing.T) { t.Fatal("collector recorded no trace exports") } - var gotService, gotSpan string + 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() == "substrate.otlp.relay" { + gotRelay = attr.GetValue().GetStringValue() + } } for _, ss := range rs.GetScopeSpans() { for _, s := range ss.GetSpans() { @@ -97,7 +104,7 @@ func TestEndToEndThroughServerboot(t *testing.T) { } } } - t.Logf("collector received span %q from service %q", gotSpan, gotService) + 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. @@ -107,4 +114,7 @@ func TestEndToEndThroughServerboot(t *testing.T) { if gotSpan != "RunWorkload" { t.Errorf("span arrived named %q, want %q", gotSpan, "RunWorkload") } + if gotRelay != "relay" { + t.Errorf("span arrived with substrate.otlp.relay %q, want %q", gotRelay, "relay") + } } diff --git a/internal/otlprelay/relay.go b/internal/otlprelay/relay.go index 1b58bd499..445d6dd3c 100644 --- a/internal/otlprelay/relay.go +++ b/internal/otlprelay/relay.go @@ -36,11 +36,12 @@ // 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. 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. That -// holds only for a source whose resource is already right, so the relay carries -// ateom telemetry only; see ateomServices. +// 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. package otlprelay import ( @@ -58,6 +59,8 @@ import ( "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" @@ -74,13 +77,19 @@ const ( 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" + // otlpDefaultPort matches atenet's normalizeOtlpCollector. otlpDefaultPort = "4317" - // socketMode keeps the relay socket reachable by the ateom pods on the node - // (which do not necessarily share atelet's uid) while staying off-node by - // construction. The socket is inside BasePath, a root-owned host directory. - socketMode = 0o666 + // 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 @@ -101,11 +110,12 @@ type Server struct { // cmd/ateom-gvisor and cmd/ateom-microvm, which are package main and cannot be // imported; TestAteomServicesMatchTheAteomBinaries guards the duplication. // -// Actor telemetry is what the allowlist excludes: actors share a hostname -// ("runsc") and an interior IP, so their series merge unless identity is -// injected from outside the actor (#761) -- a rewrite, not a forward. Nothing -// but ateom reaches this socket today; naming the contract now makes that path -// an added branch later rather than a re-argument about pass-through. +// 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, @@ -132,6 +142,15 @@ func resourceServiceName(r *resourcepb.Resource) string { return "" } +// forwardContext propagates incoming gRPC metadata (headers, auth tokens) +// to outgoing context so upstream calls preserve client headers. +func forwardContext(ctx context.Context) context.Context { + if md, ok := metadata.FromIncomingContext(ctx); ok { + return metadata.NewOutgoingContext(ctx, md.Copy()) + } + return ctx +} + // 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 @@ -155,7 +174,7 @@ func (t *traceRelay) Export(ctx context.Context, req *coltracepb.ExportTraceServ return nil, err } } - return t.upstream.Export(ctx, req) + return t.upstream.Export(forwardContext(ctx), req) } type metricRelay struct { @@ -170,7 +189,7 @@ func (m *metricRelay) Export(ctx context.Context, req *colmetricspb.ExportMetric return nil, err } } - return m.upstream.Export(ctx, req) + return m.upstream.Export(forwardContext(ctx), req) } // validateSocketPath rejects a relative path, which gRPC does not resolve: @@ -206,10 +225,23 @@ func NewServer(ctx context.Context, sockPath string) (*Server, error) { return nil, nil } + comp, err := upstreamCompression() + 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, - grpc.WithTransportCredentials(insecure.NewCredentials())) + upstream, err := grpc.NewClient(target, dialOpts...) if err != nil { return nil, fmt.Errorf("while dialing OTLP collector %q: %w", target, err) } @@ -221,7 +253,7 @@ func NewServer(ctx context.Context, sockPath string) (*Server, error) { } coltracepb.RegisterTraceServiceServer(s.grpc, &traceRelay{upstream: coltracepb.NewTraceServiceClient(upstream)}) colmetricspb.RegisterMetricsServiceServer(s.grpc, &metricRelay{upstream: colmetricspb.NewMetricsServiceClient(upstream)}) - slog.InfoContext(ctx, "OTLP relay forwarding to collector", slog.String("collector", target)) + slog.InfoContext(ctx, "OTLP relay forwarding to collector", slog.String("collector", target), slog.String("compression", comp)) return s, nil } @@ -260,6 +292,41 @@ func (s *Server) Stop() { _ = os.Remove(s.sockPath) } +// 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. // @@ -272,16 +339,23 @@ func upstreamTarget() (string, error) { traces := strings.TrimSpace(os.Getenv(tracesEndpointEnv)) metrics := strings.TrimSpace(os.Getenv(metricsEndpointEnv)) - resolved := generic - for _, specific := range []string{traces, metrics} { - if specific == "" { - continue - } - if resolved != "" && resolved != generic && specific != resolved { - return "", fmt.Errorf("%s and %s name different collectors (%q vs %q); the relay carries both signals over one connection", - tracesEndpointEnv, metricsEndpointEnv, resolved, specific) - } - resolved = specific + 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 diff --git a/internal/otlprelay/relay_test.go b/internal/otlprelay/relay_test.go index 4d69b31e3..6a5a8eb89 100644 --- a/internal/otlprelay/relay_test.go +++ b/internal/otlprelay/relay_test.go @@ -30,6 +30,7 @@ import ( "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" @@ -46,15 +47,20 @@ import ( type fakeCollector struct { coltracepb.UnimplementedTraceServiceServer - mu sync.Mutex - traces []*coltracepb.ExportTraceServiceRequest - metrics []*colmetricspb.ExportMetricsServiceRequest - got chan struct{} + 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 @@ -70,6 +76,9 @@ type metricsSink struct { 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 @@ -566,8 +575,11 @@ func TestUpstreamTarget(t *testing.T) { {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: "one signal specific", generic: "generic:4317", metrics: "specific:4317", want: "specific:4317"}, - {name: "signal specific alone", traces: "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"}, } { @@ -591,3 +603,116 @@ func TestUpstreamTarget(t *testing.T) { }) } } + +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) + } + }) + } +} + +func TestExportForwardsMetadata(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() + + ctx := metadata.AppendToOutgoingContext(context.Background(), + "authorization", "Bearer test-token", + "custom-header", "custom-value", + ) + + // Send trace export with metadata + traceClient := coltracepb.NewTraceServiceClient(conn) + _, err = traceClient.Export(ctx, &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{ + Resource: serviceResource("ateom-gvisor"), + }}, + }) + if err != nil { + t.Fatalf("traceClient.Export: %v", err) + } + + // Send metrics export with metadata + metricClient := colmetricspb.NewMetricsServiceClient(conn) + _, err = metricClient.Export(ctx, &colmetricspb.ExportMetricsServiceRequest{ + ResourceMetrics: []*metricspb.ResourceMetrics{{ + Resource: serviceResource("ateom-gvisor"), + }}, + }) + if 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 got := sink.traceMD[0].Get("authorization"); len(got) == 0 || got[0] != "Bearer test-token" { + t.Errorf("trace export authorization metadata = %v, want Bearer test-token", got) + } + if got := sink.traceMD[0].Get("custom-header"); len(got) == 0 || got[0] != "custom-value" { + t.Errorf("trace export custom-header metadata = %v, want custom-value", got) + } + + if len(sink.metricMD) == 0 { + t.Fatal("collector received no metadata for metrics export") + } + if got := sink.metricMD[0].Get("authorization"); len(got) == 0 || got[0] != "Bearer test-token" { + t.Errorf("metric export authorization metadata = %v, want Bearer test-token", got) + } + if got := sink.metricMD[0].Get("custom-header"); len(got) == 0 || got[0] != "custom-value" { + t.Errorf("metric export custom-header metadata = %v, want custom-value", got) + } +} + +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 33d6d8726..d4387da4e 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" @@ -84,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) { @@ -131,7 +134,11 @@ 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) + relayStatus := "direct" + if opts.ExporterConn != nil { + relayStatus = "relay" + } + res, err := newResource(ctx, opts.ServiceName, attribute.String("substrate.otlp.relay", relayStatus)) if err != nil { return nil, fmt.Errorf("create tracer resource: %w", err) } @@ -222,7 +229,11 @@ func newMeterProvider(ctx context.Context, serviceName string, conn *grpc.Client if err != nil { return nil, fmt.Errorf("create OTLP metric exporter: %w", err) } - res, err := newResource(ctx, serviceName) + relayStatus := "direct" + if conn != nil { + relayStatus = "relay" + } + res, err := newResource(ctx, serviceName, attribute.String("substrate.otlp.relay", relayStatus)) 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..e0b45fea1 100644 --- a/internal/serverboot/serverboot_test.go +++ b/internal/serverboot/serverboot_test.go @@ -25,6 +25,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus/promhttp" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" semconv "go.opentelemetry.io/otel/semconv/v1.21.0" ) @@ -67,6 +68,24 @@ func TestNewResourceEnvWins(t *testing.T) { } } +func TestResourceRelayAttribute(t *testing.T) { + resDirect, err := newResource(context.Background(), "ateom-gvisor", attribute.String("substrate.otlp.relay", "direct")) + if err != nil { + t.Fatalf("newResource: %v", err) + } + if got := resourceAttrs(resDirect)["substrate.otlp.relay"]; got != "direct" { + t.Errorf("substrate.otlp.relay = %q, want direct", got) + } + + resRelay, err := newResource(context.Background(), "ateom-gvisor", attribute.String("substrate.otlp.relay", "relay")) + if err != nil { + t.Fatalf("newResource: %v", err) + } + if got := resourceAttrs(resRelay)["substrate.otlp.relay"]; got != "relay" { + t.Errorf("substrate.otlp.relay = %q, want relay", got) + } +} + func TestReadyzDrainsWhileHealthzStaysUp(t *testing.T) { readiness := &Readiness{} mux := metricsMux(MetricsServerOptions{ From ae8b1ae01ec6936c8253b630b32df4355c1df723 Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Wed, 12 Aug 2026 14:49:09 -0700 Subject: [PATCH 5/6] otlprelay: authenticate the upstream leg as atelet, and scope the relay attribute to ateom Addresses Krisztian's second round on #809. Attach atelet's own OTEL_EXPORTER_OTLP_HEADERS upstream and drop the client's. The previous commit forwarded the incoming metadata verbatim, which answered the question it was asked -- an ateom with headers set could authenticate -- by handing the answer to the wrong party. The upstream leg is atelet's connection to the collector, so its credentials are atelet's, exactly as they were for the SDK exporter the relay stands in for. Forwarding let anything that reached the socket choose what atelet presents upstream, and a worker pod runs untrusted code. This is not the same trade-off as the resource: a claimed service.name that is wrong misfiles a series, while a header is an actual credential. Verbatim now describes the payload, not the call around it. Headers are held per signal rather than per connection, since unlike the endpoint and the compression they are per-call metadata and TRACES_/METRICS_ may legitimately differ; per the OTLP spec the signal-specific variable replaces the generic one rather than merging. An unparseable value fails NewServer instead of becoming a per-export rejection against a collector that refuses unauthenticated calls, and ateom then finds no socket and exports directly. Nothing is allow-listed through: atecontroller injects only OTEL_EXPORTER_OTLP_ENDPOINT into worker pods, so no ateom has a header to lose, and an allowlist can be added when one does. Parse errors name the offending entry by position and the log names the header keys only, because the values are credentials. Report the first rejection of each service.name. resource.WithFromEnv() runs last in newResource, so OTEL_SERVICE_NAME or an OTEL_RESOURCE_ATTRIBUTES entry on a worker pod overrides the ateom's own name, the allowlist stops recognizing it, and every export is refused with PermissionDenied -- which the SDK does not retry. The ateom keeps running and its telemetry silently stops, with nothing on the collector side to distinguish that from an idle node. The refusal is still correct; what was missing was anyone saying so, since the whole diagnosis lives on the relay side and was being discarded. Both the log and the gRPC error now name the two environment variables that produce this, so it is reachable from the node and from the ateom's own error handler. Dedup is keyed by name because a refused exporter retries on its own schedule for the pod's whole life, and a node runs many of them: unbounded logging would be the worse failure. Traces and metrics share one gate so a misnamed ateom is reported once rather than per signal. Rename substrate.otlp.relay to ate.otlp.relay and set it only where it means something. Custom attributes live under ate.*, and newResource is shared, so every component was carrying "direct" forever for a relay it will never be offered. Capability and connection are now separate inputs: nil alone cannot distinguish an ateom that tried and fell back from an atecontroller that never had a relay, and the fallback is precisely the state the attribute was added to make observable -- collapsing the two would hide a node whose atelet never came up. Only the ateoms set RelayCapable; calling InitMetricsPushOnlyVia implies it, since only a caller holding a relay reaches for that function. The key is spelled locally rather than taken from internal/ateattr: serverboot sits at the bottom of the dependency graph with one agent-substrate import and every main links it, while ateattr would pull in pkg/api/v1alpha1, ateapipb, internal/resources and ateletpb. Worth revisiting if that cost drops. Degrade instead of exiting when the relay cannot be dialed, matching atelet. Both ends of this decision already agreed and the ateoms disagreed: Dial 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 a failed dial loses is the node-local export path, not the ability to run actors, so failing the worker pod over its telemetry route would turn a misconfigured flag into an outage. Test the decision rather than the plumbing. The relay attribute test asserted what it had just passed in and would have kept passing if InitTracing stopped setting the attribute; it now covers all four combinations of capability and connection, two of which assert the attribute is absent, and reads the metric resource back off the provider through a ManualReader so a wiring mistake in newMeterProvider fails there and not in the unit test beside it. The e2e gains the direct half it never had: no socket, nil conn, and the span still reaches the collector carrying "direct". The metadata test asserted the behaviour this commit removes and is replaced by its inverse plus one for the attached headers. Record why the socket is not in a read-only subdirectory. It would be an improvement, but worker pods mount BasePath itself writable, so it needs its own volume and a controller change -- against this PR's premise of reusing the existing mount -- and the pods keep CAP_SYS_ADMIN, so it guards against mistakes rather than malice. It is also not specific to this socket: CredentialBrokerSocket and the image cache share the same writable directory, so the fix belongs to the mount, not to the newest thing under it. --- cmd/ateom-gvisor/main.go | 13 +- cmd/ateom-microvm/main.go | 13 +- internal/ateompath/ateompath.go | 8 +- internal/otlprelay/e2e_test.go | 86 +++++++++- internal/otlprelay/relay.go | 207 +++++++++++++++++++++-- internal/otlprelay/relay_test.go | 223 +++++++++++++++++++++---- internal/serverboot/serverboot.go | 63 +++++-- internal/serverboot/serverboot_test.go | 103 ++++++++++-- 8 files changed, 633 insertions(+), 83 deletions(-) diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 7eebddeef..d03208377 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -109,9 +109,17 @@ func do(ctx context.Context) error { // 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 { - serverboot.Fatal(ctx, "Failed to connect to the OTLP relay", err) + 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() @@ -121,6 +129,9 @@ func do(ctx context.Context) error { 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) diff --git a/cmd/ateom-microvm/main.go b/cmd/ateom-microvm/main.go index 08dd802d3..86b1d9345 100644 --- a/cmd/ateom-microvm/main.go +++ b/cmd/ateom-microvm/main.go @@ -108,9 +108,17 @@ func do(ctx context.Context) error { // 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 { - serverboot.Fatal(ctx, "Failed to connect to the OTLP relay", err) + 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() @@ -120,6 +128,9 @@ func do(ctx context.Context) error { 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) diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index a807fb428..38d14e709 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -61,7 +61,13 @@ func GVisorReleaseDir(sha256 string) string { // // 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. +// 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, diff --git a/internal/otlprelay/e2e_test.go b/internal/otlprelay/e2e_test.go index bfbc48630..8c64258cf 100644 --- a/internal/otlprelay/e2e_test.go +++ b/internal/otlprelay/e2e_test.go @@ -16,6 +16,7 @@ package otlprelay import ( "context" + "path/filepath" "testing" "time" @@ -60,6 +61,7 @@ func TestEndToEndThroughServerboot(t *testing.T) { // 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) @@ -93,7 +95,7 @@ func TestEndToEndThroughServerboot(t *testing.T) { if attr.GetKey() == "service.name" { gotService = attr.GetValue().GetStringValue() } - if attr.GetKey() == "substrate.otlp.relay" { + if attr.GetKey() == relayAttrKey { gotRelay = attr.GetValue().GetStringValue() } } @@ -115,6 +117,86 @@ func TestEndToEndThroughServerboot(t *testing.T) { t.Errorf("span arrived named %q, want %q", gotSpan, "RunWorkload") } if gotRelay != "relay" { - t.Errorf("span arrived with substrate.otlp.relay %q, want %q", 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 index 445d6dd3c..b68b65ebc 100644 --- a/internal/otlprelay/relay.go +++ b/internal/otlprelay/relay.go @@ -42,6 +42,11 @@ // 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 ( @@ -54,7 +59,9 @@ import ( "net/url" "os" "path/filepath" + "sort" "strings" + "sync" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -83,6 +90,14 @@ const ( 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" @@ -106,7 +121,9 @@ type Server struct { } // ateomServices are the only sources this relay carries, keyed by the -// service.name their resource declares. Mirrors the serviceName constants in +// 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. // @@ -121,18 +138,54 @@ var ateomServices = map[string]bool{ "ateom-microvm": true, } -// checkAteomSource rejects a missing service.name along with an unrecognized -// one: an unidentified source is the one the relay cannot vouch for. -func checkAteomSource(r *resourcepb.Resource) error { +// 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", + "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) { @@ -142,13 +195,84 @@ func resourceServiceName(r *resourcepb.Resource) string { return "" } -// forwardContext propagates incoming gRPC metadata (headers, auth tokens) -// to outgoing context so upstream calls preserve client headers. -func forwardContext(ctx context.Context) context.Context { - if md, ok := metadata.FromIncomingContext(ctx); ok { - return metadata.NewOutgoingContext(ctx, md.Copy()) +// 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 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 @@ -159,6 +283,12 @@ func forwardContext(ctx context.Context) context.Context { 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. @@ -170,26 +300,28 @@ type traceRelay struct { // 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 := checkAteomSource(rs.GetResource()); err != nil { + if err := t.gate.check(ctx, rs.GetResource()); err != nil { return nil, err } } - return t.upstream.Export(forwardContext(ctx), req) + 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 := checkAteomSource(rm.GetResource()); err != nil { + if err := m.gate.check(ctx, rm.GetResource()); err != nil { return nil, err } } - return m.upstream.Export(forwardContext(ctx), req) + return m.upstream.Export(upstreamContext(ctx, m.headers), req) } // validateSocketPath rejects a relative path, which gRPC does not resolve: @@ -230,6 +362,19 @@ func NewServer(ctx context.Context, sockPath string) (*Server, error) { 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. @@ -251,9 +396,23 @@ func NewServer(ctx context.Context, sockPath string) (*Server, error) { sockPath: sockPath, grpc: grpc.NewServer(grpc.MaxRecvMsgSize(maxRecvMsgSize)), } - coltracepb.RegisterTraceServiceServer(s.grpc, &traceRelay{upstream: coltracepb.NewTraceServiceClient(upstream)}) - colmetricspb.RegisterMetricsServiceServer(s.grpc, &metricRelay{upstream: colmetricspb.NewMetricsServiceClient(upstream)}) - slog.InfoContext(ctx, "OTLP relay forwarding to collector", slog.String("collector", target), slog.String("compression", comp)) + 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 } @@ -292,6 +451,18 @@ func (s *Server) Stop() { _ = 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) { diff --git a/internal/otlprelay/relay_test.go b/internal/otlprelay/relay_test.go index 6a5a8eb89..5aaccf2f4 100644 --- a/internal/otlprelay/relay_test.go +++ b/internal/otlprelay/relay_test.go @@ -15,9 +15,11 @@ package otlprelay import ( + "bytes" "context" "errors" "io/fs" + "log/slog" "net" "os" "path/filepath" @@ -335,6 +337,38 @@ func TestDialEmptyPath(t *testing.T) { } } +// 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. @@ -644,40 +678,24 @@ func TestUpstreamCompression(t *testing.T) { } } -func TestExportForwardsMetadata(t *testing.T) { - sink, collector := startFakeCollector(t) - sock := startRelay(t, collector) - +// 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() - ctx := metadata.AppendToOutgoingContext(context.Background(), - "authorization", "Bearer test-token", - "custom-header", "custom-value", - ) - - // Send trace export with metadata - traceClient := coltracepb.NewTraceServiceClient(conn) - _, err = traceClient.Export(ctx, &coltracepb.ExportTraceServiceRequest{ - ResourceSpans: []*tracepb.ResourceSpans{{ - Resource: serviceResource("ateom-gvisor"), - }}, - }) - if err != nil { + if _, err := coltracepb.NewTraceServiceClient(conn).Export(ctx, &coltracepb.ExportTraceServiceRequest{ + ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(service)}}, + }); err != nil { t.Fatalf("traceClient.Export: %v", err) } - - // Send metrics export with metadata - metricClient := colmetricspb.NewMetricsServiceClient(conn) - _, err = metricClient.Export(ctx, &colmetricspb.ExportMetricsServiceRequest{ - ResourceMetrics: []*metricspb.ResourceMetrics{{ - Resource: serviceResource("ateom-gvisor"), - }}, - }) - if err != nil { + if _, err := colmetricspb.NewMetricsServiceClient(conn).Export(ctx, &colmetricspb.ExportMetricsServiceRequest{ + ResourceMetrics: []*metricspb.ResourceMetrics{{Resource: serviceResource(service)}}, + }); err != nil { t.Fatalf("metricClient.Export: %v", err) } @@ -686,21 +704,154 @@ func TestExportForwardsMetadata(t *testing.T) { if len(sink.traceMD) == 0 { t.Fatal("collector received no metadata for trace export") } - if got := sink.traceMD[0].Get("authorization"); len(got) == 0 || got[0] != "Bearer test-token" { - t.Errorf("trace export authorization metadata = %v, want Bearer test-token", got) + if len(sink.metricMD) == 0 { + t.Fatal("collector received no metadata for metrics export") } - if got := sink.traceMD[0].Get("custom-header"); len(got) == 0 || got[0] != "custom-value" { - t.Errorf("trace export custom-header metadata = %v, want custom-value", got) + 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) + } + } } +} - if len(sink.metricMD) == 0 { - t.Fatal("collector received no metadata for metrics export") +// ...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) } - if got := sink.metricMD[0].Get("authorization"); len(got) == 0 || got[0] != "Bearer test-token" { - t.Errorf("metric export authorization metadata = %v, want Bearer test-token", 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) } - if got := sink.metricMD[0].Get("custom-header"); len(got) == 0 || got[0] != "custom-value" { - t.Errorf("metric export custom-header metadata = %v, want custom-value", 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") } } diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index d4387da4e..5f608ca09 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -107,6 +107,38 @@ func newResource(ctx context.Context, serviceName string, extraAttrs ...attribut 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 labelling, 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. @@ -123,6 +155,11 @@ type TracingOptions struct { // 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 @@ -134,11 +171,7 @@ func InitTracing(ctx context.Context, opts TracingOptions) (*sdktrace.TracerProv if opts.Sampling.sampler == nil { return nil, fmt.Errorf("TracingOptions.Sampling is required") } - relayStatus := "direct" - if opts.ExporterConn != nil { - relayStatus = "relay" - } - res, err := newResource(ctx, opts.ServiceName, attribute.String("substrate.otlp.relay", relayStatus)) + res, err := newResource(ctx, opts.ServiceName, relayAttrs(opts.RelayCapable, opts.ExporterConn)...) if err != nil { return nil, fmt.Errorf("create tracer resource: %w", err) } @@ -188,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, nil, promExporter) + return newMeterProvider(ctx, serviceName, false, nil, nil, promExporter) } // InitMetricsPushOnly is InitMetrics without the Prometheus reader, for binaries @@ -197,7 +230,7 @@ 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, nil, producers) + return newMeterProvider(ctx, serviceName, false, nil, producers) } // InitMetricsPushOnlyVia is InitMetricsPushOnly with an explicit exporter @@ -208,11 +241,17 @@ func InitMetricsPushOnly(ctx context.Context, serviceName string, producers ...s // // 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, conn, producers) + return newMeterProvider(ctx, serviceName, true, conn, producers) } -func newMeterProvider(ctx context.Context, serviceName string, conn *grpc.ClientConn, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) { +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") } @@ -229,11 +268,7 @@ func newMeterProvider(ctx context.Context, serviceName string, conn *grpc.Client if err != nil { return nil, fmt.Errorf("create OTLP metric exporter: %w", err) } - relayStatus := "direct" - if conn != nil { - relayStatus = "relay" - } - res, err := newResource(ctx, serviceName, attribute.String("substrate.otlp.relay", relayStatus)) + 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 e0b45fea1..f1777d9fb 100644 --- a/internal/serverboot/serverboot_test.go +++ b/internal/serverboot/serverboot_test.go @@ -25,9 +25,12 @@ import ( "time" "github.com/prometheus/client_golang/prometheus/promhttp" - "go.opentelemetry.io/otel/attribute" + 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 { @@ -68,21 +71,101 @@ func TestNewResourceEnvWins(t *testing.T) { } } -func TestResourceRelayAttribute(t *testing.T) { - resDirect, err := newResource(context.Background(), "ateom-gvisor", attribute.String("substrate.otlp.relay", "direct")) +// 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("newResource: %v", err) + t.Fatalf("grpc.NewClient: %v", err) } - if got := resourceAttrs(resDirect)["substrate.otlp.relay"]; got != "direct" { - t.Errorf("substrate.otlp.relay = %q, want direct", got) + 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 unlabelled 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) + } + }) } +} - resRelay, err := newResource(context.Background(), "ateom-gvisor", attribute.String("substrate.otlp.relay", "relay")) +// 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("newResource: %v", err) + 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) } - if got := resourceAttrs(resRelay)["substrate.otlp.relay"]; got != "relay" { - t.Errorf("substrate.otlp.relay = %q, want relay", got) + 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) } } From 20931ccc750dd1f58cbb2f9da851e964899cc4f3 Mon Sep 17 00:00:00 2001 From: Chenyi Wang Date: Thu, 13 Aug 2026 04:49:59 +0000 Subject: [PATCH 6/6] serverboot: fix typo in comment to satisfy misspell linter --- internal/serverboot/serverboot.go | 2 +- internal/serverboot/serverboot_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index 5f608ca09..236dd3477 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -109,7 +109,7 @@ func newResource(ctx context.Context, serviceName string, extraAttrs ...attribut // 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 labelling, say, +// 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: diff --git a/internal/serverboot/serverboot_test.go b/internal/serverboot/serverboot_test.go index f1777d9fb..5217d29dc 100644 --- a/internal/serverboot/serverboot_test.go +++ b/internal/serverboot/serverboot_test.go @@ -96,7 +96,7 @@ func TestRelayAttrs(t *testing.T) { {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 unlabelled even if some future caller hands it a + // 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}, } {