ateom: export telemetry through an atelet unix-socket OTLP relay (#809)

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.

> It's a good idea to open an issue first for discussion.

- [x] Tests pass
- [x] Appropriate changes to documentation are included in the PR

cmd/atelet/main.go

@@ -44,6 +44,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"
@@ -98,6 +99,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.")
 )
@@ -153,6 +156,27 @@ func main() {
 		EnableHealthz: true,
 	})
 
+	// The OTLP relay lets the ateom pods on this node export telemetry over a
+	// unix socket instead of their own network (see internal/otlprelay). Started
+	// early: an ateom that finds no socket at startup falls back to exporting
+	// directly for its whole life, so the socket should exist before any worker
+	// pod on this node boots.
+	if relay, err := otlprelay.NewServer(ctx, *otlpRelaySocket); err != nil {
+		slog.ErrorContext(ctx, "Failed to create the OTLP relay; ateoms will export directly", slog.Any("err", err))
+	} else if relay != nil {
+		// Deferred rather than tied to the drain: the relay carries other
+		// processes' telemetry, so it should outlive atelet's own RPC serving
+		// and stay up while the ateoms it serves are themselves shutting down.
+		defer relay.Stop()
+		go func() {
+			if err := relay.Serve(ctx); err != nil {
+				// Not fatal: atelet's actual job does not depend on the relay,
+				// and the ateoms fall back to exporting directly.
+				slog.ErrorContext(ctx, "OTLP relay stopped", slog.Any("err", err))
+			}
+		}()
+	}
+
 	ateomDialer := &AteomDialer{
 		conns: lru.New(256),
 	}

cmd/ateom-gvisor/main.go

@@ -43,6 +43,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"
@@ -74,6 +75,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
 )
 
@@ -114,16 +118,39 @@ func do(ctx context.Context) error {
 	slog.InfoContext(ctx, "ateom booting")
 
 	const serviceName = "ateom-gvisor"
+	// Export through atelet's node-local relay when it is there, so telemetry
+	// never touches the worker pod's network. A nil conn means it is not, and
+	// both providers fall back to dialing the collector directly.
+	//
+	// A relay that cannot be dialed is logged rather than fatal, matching both
+	// ends of the same decision: Dial already treats an absent socket as a
+	// fallback rather than an error, and atelet logs and keeps going when it
+	// cannot serve the relay at all. What is lost here is the node-local export
+	// path, not the ateom's ability to run actors, and failing the worker pod
+	// over its telemetry route would turn a misconfigured flag into an outage.
+	relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to connect to the OTLP relay; exporting telemetry directly over the pod network",
+			slog.String("socket", *otlpRelaySocket), slog.Any("err", err))
+	}
+	if relayConn != nil {
+		defer relayConn.Close()
+	}
+
 	tp, err := serverboot.InitTracing(ctx, serverboot.TracingOptions{
-		ServiceName: serviceName,
-		Sampling:    serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)),
+		ServiceName:  serviceName,
+		Sampling:     serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)),
+		ExporterConn: relayConn,
+		// So the spans say which path they took, including when relayConn is nil
+		// because the dial above failed and this ateom is exporting directly.
+		RelayCapable: true,
 	})
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to initialize tracing", err)
 	}
 	defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown)
 
-	mp, err := serverboot.InitMetricsPushOnly(ctx, serviceName)
+	mp, err := serverboot.InitMetricsPushOnlyVia(ctx, serviceName, relayConn)
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to initialize metrics", err)
 	}

cmd/ateom-microvm/main.go

@@ -44,6 +44,7 @@ import (
 	"github.com/agent-substrate/substrate/internal/ateompath"
 	"github.com/agent-substrate/substrate/internal/ateomstats"
 	"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"
@@ -63,6 +64,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")
@@ -104,16 +108,39 @@ func do(ctx context.Context) error {
 	slog.InfoContext(ctx, "ateom-microvm booting", slog.String("version", version.String()))
 
 	const serviceName = "ateom-microvm"
+	// Export through atelet's node-local relay when it is there, so telemetry
+	// never touches the worker pod's network. A nil conn means it is not, and
+	// both providers fall back to dialing the collector directly.
+	//
+	// A relay that cannot be dialed is logged rather than fatal, matching both
+	// ends of the same decision: Dial already treats an absent socket as a
+	// fallback rather than an error, and atelet logs and keeps going when it
+	// cannot serve the relay at all. What is lost here is the node-local export
+	// path, not the ateom's ability to run actors, and failing the worker pod
+	// over its telemetry route would turn a misconfigured flag into an outage.
+	relayConn, err := otlprelay.Dial(ctx, *otlpRelaySocket)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to connect to the OTLP relay; exporting telemetry directly over the pod network",
+			slog.String("socket", *otlpRelaySocket), slog.Any("err", err))
+	}
+	if relayConn != nil {
+		defer relayConn.Close()
+	}
+
 	tp, err := serverboot.InitTracing(ctx, serverboot.TracingOptions{
-		ServiceName: serviceName,
-		Sampling:    serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)),
+		ServiceName:  serviceName,
+		Sampling:     serverboot.ResolveTraceSampling(ctx, serverboot.ParentRatioSampling(serverboot.ControlPlaneTraceRatio)),
+		ExporterConn: relayConn,
+		// So the spans say which path they took, including when relayConn is nil
+		// because the dial above failed and this ateom is exporting directly.
+		RelayCapable: true,
 	})
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to initialize tracing", err)
 	}
 	defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown)
 
-	mp, err := serverboot.InitMetricsPushOnly(ctx, serviceName)
+	mp, err := serverboot.InitMetricsPushOnlyVia(ctx, serviceName, relayConn)
 	if err != nil {
 		serverboot.Fatal(ctx, "Failed to initialize metrics", err)
 	}

docs/observability.md

@@ -211,7 +211,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. To check whether ateom exported its spans through the [OTLP relay](#the-ateom-otlp-relay) rather than falling back to direct network egress, inspect the span's resource attributes: `ate.otlp.relay` will be set to `"relay"` (instead of `"direct"`).
 
 > **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.
 
@@ -234,6 +236,24 @@ Telemetry is emitted the same way everywhere; only the backend differs between a
 >
 > ateom workers don't read the ConfigMap at all — `ate-controller` copies the value into each worker pod at creation. A new endpoint reaches them only once the controller itself restarts, and that restart then rolls every WorkerPool Deployment, replacing the running workers along with the actors on them.
 
+### The ateom OTLP relay
+
+ateom is the one component that does not talk to the collector directly. It exports over a unix socket at `/var/lib/ateom-gvisor/atelet-otlp.sock`, which `atelet` serves and forwards to the collector on the node's network ([`internal/otlprelay`](../internal/otlprelay)):
+
+```
+ateom ──OTLP/gRPC over unix socket──► atelet relay ──OTLP/gRPC──► collector
+```
+
+The socket sits in the `BasePath` hostPath already mounted into both, so nothing new is mounted. `atelet` is a DaemonSet, so every ateom on a node shares one relay, and the many per-pod collector connections collapse into one per node. Four things motivate it: the worker pod runs untrusted agent code and will not need egress to the collector once direct fallback is phased out; the connection count drops; ateom's own telemetry stays clear of the transparent egress redirect it installs for the actor; and `atelet` outlives the worker pod, so spans still queued at teardown are not lost with it.
+
+The relay is best-effort. If the socket is absent when ateom starts — `atelet` not up yet, `--otlp-relay-socket=""`, or no collector configured for the relay to forward to — ateom logs it and exports directly to `OTEL_EXPORTER_OTLP_ENDPOINT` as before. That fallback is decided once at startup, not per export, and is stamped on telemetry as the `ate.otlp.relay` resource attribute (`relay` vs `direct`).
+
+> **Note on Network Egress Lockdown:** Complete network policy lockdown of worker pod egress to the collector is planned as a Phase 2 milestone once the relay path is fully proven and direct fallback is deprecated. While the fallback path remains active, worker pods retain network egress to the collector and `ate-controller` continues to inject `OTEL_EXPORTER_OTLP_ENDPOINT`.
+
+For verified ateom sources, the relay forwards each request verbatim rather than decoding and re-exporting, which is what keeps every ateom its own service in Jaeger/GCP Trace instead of being absorbed into `atelet`'s. `ate-controller` injects `k8s.pod.name`, `k8s.namespace.name`, `k8s.pod.uid`, and `service.instance.id` directly into `OTEL_RESOURCE_ATTRIBUTES` via the Kubernetes Downward API; because the relay preserves resources verbatim, Kubernetes attributes remain intact even though the TCP connection to the collector originates from `atelet` rather than the worker pod IP (bypassing reliance on collector-side IP-based `k8sattributes` enrichment).
+
+Verbatim forwarding is restricted to known ateom sources and refuses anything else with `PermissionDenied`. Actor telemetry is what that excludes: actors share a hostname (`runsc`) and an interior IP, so their series merge unless identity is injected from outside the actor ([#761](https://github.com/agent-substrate/substrate/issues/761)) — a rewrite, which will be implemented as an explicit rewriting path alongside this forwarder.
+
 ---
 
 ## 5. Dashboards

go.mod

@@ -49,6 +49,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
@@ -200,7 +201,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

internal/ateompath/ateompath.go

@@ -58,6 +58,28 @@ func GVisorReleaseDir(sha256 string) string {
 	return filepath.Join(StaticFilesDir, "gvisor-"+sha256)
 }
 
+// AteletOTLPSocketPath is the node-scoped unix socket atelet serves the OTLP
+// relay on (see internal/otlprelay). It is node-scoped rather than per-pod
+// because every ateom on the node pushes into the same relay: atelet is a
+// DaemonSet, so one socket collapses N per-pod collector connections into one
+// per-node connection.
+//
+// It sits directly under BasePath, which is the host directory already mounted
+// at the same path into atelet and into every ateom pod, so no new volume is
+// needed for ateom to reach it. Note that BasePath is mounted writable
+// (workerpool_apply.go) and shared with CredentialBrokerSocket and the image
+// cache, so a worker pod can unlink or replace this socket. Confining
+// atelet-owned sockets to a subdirectory mounted read-only would be an
+// improvement, but it is a property of the whole BasePath mount rather than of
+// this socket — a read-only subdir needs its own volume and mount, and the pod
+// keeps CAP_SYS_ADMIN. Tracked separately rather than solved here.
+func AteletOTLPSocketPath() string {
+	return filepath.Join(
+		BasePath,
+		"atelet-otlp.sock",
+	)
+}
+
 func AteomPath(podUID string) string {
 	return filepath.Join(
 		BasePath,

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"

internal/otlprelay/client.go

@@ -0,0 +1,68 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlprelay
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"os"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// Dial opens the ateom half of the relay: a gRPC connection over atelet's unix
+// socket, to be handed to the OTLP exporters via serverboot's ExporterConn.
+//
+// It returns (nil, nil) when sockPath is empty or absent, which the caller reads
+// as "export directly instead". The existence check is what makes the fallback
+// deterministic at startup: grpc.NewClient is lazy, so a connection to a missing
+// socket would be created happily and only fail later, per export, with the
+// telemetry already lost. Losing spans is not worth failing ateom over either,
+// hence a fallback rather than an error.
+//
+// The connection is plaintext by design. A unix socket cannot leave the node, so
+// there is no transport to protect; access is controlled by the socket's file
+// permissions instead (see socketMode).
+func Dial(ctx context.Context, sockPath string) (*grpc.ClientConn, error) {
+	if sockPath == "" {
+		return nil, nil
+	}
+	if err := validateSocketPath(sockPath); err != nil {
+		return nil, err
+	}
+	if _, err := os.Stat(sockPath); err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			slog.WarnContext(ctx, "OTLP relay socket absent, exporting telemetry directly over the pod network",
+				slog.String("socket", sockPath))
+			return nil, nil
+		}
+		return nil, fmt.Errorf("while checking the OTLP relay socket %q: %w", sockPath, err)
+	}
+
+	// gRPC resolves a "unix://" target to a unix socket dialer natively, so the
+	// OTLP exporters above this connection are unchanged: OTLP is gRPC, and gRPC
+	// needs only a reliable byte stream.
+	conn, err := grpc.NewClient("unix://"+sockPath,
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, fmt.Errorf("while dialing the OTLP relay socket %q: %w", sockPath, err)
+	}
+	slog.InfoContext(ctx, "Exporting telemetry through the atelet OTLP relay", slog.String("socket", sockPath))
+	return conn, nil
+}

internal/otlprelay/e2e_test.go

@@ -0,0 +1,202 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlprelay
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/internal/serverboot"
+)
+
+// TestEndToEndThroughServerboot exercises the whole path an ateom span actually
+// takes, rather than the relay in isolation:
+//
+//	serverboot.InitTracing → OTLP exporter → unix socket → relay → collector
+//
+// The relay tests above speak the collector protocol directly, so they would
+// still pass if TracingOptions.ExporterConn were wired up wrong and the exporter
+// quietly kept dialing OTEL_EXPORTER_OTLP_ENDPOINT. This one would not: the
+// endpoint variable points at the fake collector *through* the relay only, and
+// the assertion is that the span arrived carrying ateom's own service.name.
+//
+// Run it on its own to watch the hop happen:
+//
+//	go test ./internal/otlprelay/ -run TestEndToEndThroughServerboot -v
+func TestEndToEndThroughServerboot(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+	// Re-point the generic endpoint at an unroutable address so an exporter
+	// that ignored ExporterConn would fail deterministically instead of dialing
+	// the test collector directly.
+	t.Setenv(endpointEnv, "http://127.0.0.1:1")
+	t.Logf("fake collector on %s, relay socket %s", collector, sock)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	if conn == nil {
+		t.Fatal("Dial returned no connection; the exporter would have bypassed the relay")
+	}
+	defer conn.Close()
+
+	const serviceName = "ateom-microvm"
+	tp, err := serverboot.InitTracing(context.Background(), serverboot.TracingOptions{
+		ServiceName: serviceName,
+		// Ratio 1.0: this test asserts on delivery, not on sampling.
+		Sampling:     serverboot.ParentRatioSampling(1.0),
+		ExporterConn: conn,
+		RelayCapable: true,
+	})
+	if err != nil {
+		t.Fatalf("InitTracing: %v", err)
+	}
+
+	_, span := tp.Tracer("relay-e2e").Start(context.Background(), "RunWorkload")
+	span.End()
+
+	// Shutdown flushes the batch processor, which is what actually puts the
+	// span on the wire.
+	if err := tp.Shutdown(context.Background()); err != nil {
+		t.Fatalf("TracerProvider.Shutdown: %v", err)
+	}
+
+	select {
+	case <-sink.got:
+	case <-time.After(10 * time.Second):
+		t.Fatal("collector never received a span through the relay")
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.traces) == 0 {
+		t.Fatal("collector recorded no trace exports")
+	}
+
+	var gotService, gotSpan, gotRelay string
+	for _, req := range sink.traces {
+		for _, rs := range req.GetResourceSpans() {
+			for _, attr := range rs.GetResource().GetAttributes() {
+				if attr.GetKey() == "service.name" {
+					gotService = attr.GetValue().GetStringValue()
+				}
+				if attr.GetKey() == relayAttrKey {
+					gotRelay = attr.GetValue().GetStringValue()
+				}
+			}
+			for _, ss := range rs.GetScopeSpans() {
+				for _, s := range ss.GetSpans() {
+					gotSpan = s.GetName()
+				}
+			}
+		}
+	}
+	t.Logf("collector received span %q from service %q (relay=%q)", gotSpan, gotService, gotRelay)
+
+	// The point of forwarding the request verbatim: the span is still ateom's,
+	// not atelet's.
+	if gotService != serviceName {
+		t.Errorf("span arrived with service.name %q, want %q; the relay must not re-attribute it", gotService, serviceName)
+	}
+	if gotSpan != "RunWorkload" {
+		t.Errorf("span arrived named %q, want %q", gotSpan, "RunWorkload")
+	}
+	if gotRelay != "relay" {
+		t.Errorf("span arrived with %s %q, want %q", relayAttrKey, gotRelay, "relay")
+	}
+}
+
+// relayAttrKey duplicates serverboot's unexported constant. Keeping a literal
+// here is the point: if serverboot renames the attribute, the dashboards and
+// alerts keyed on it break too, and this test is where that shows up.
+const relayAttrKey = "ate.otlp.relay"
+
+// TestEndToEndFallsBackToDirect is the other half of TestEndToEndThroughServerboot:
+// the ateom asked for the relay, atelet was not serving one, and the exporter
+// must fall back to the network path rather than dropping telemetry.
+//
+// This is the case the ateoms degrade into instead of exiting (see the Dial call
+// in cmd/ateom-*/main.go), so it needs to be more than a nil check: the span has
+// to reach the collector, and it has to be distinguishable from a relayed one at
+// query time — hence the "direct" attribute.
+func TestEndToEndFallsBackToDirect(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	// No relay: the socket path is inside a fresh temp dir nothing created.
+	sock := filepath.Join(t.TempDir(), "absent-atelet-otlp.sock")
+	// The direct path is the exporter dialing this itself, which is exactly what
+	// the relay test points at an unroutable address to rule out.
+	t.Setenv(endpointEnv, "http://"+collector)
+	t.Logf("fake collector on %s, absent relay socket %s", collector, sock)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial with no relay present must not fail: %v", err)
+	}
+	if conn != nil {
+		conn.Close()
+		t.Fatal("Dial returned a connection for a socket that does not exist")
+	}
+
+	const serviceName = "ateom-microvm"
+	tp, err := serverboot.InitTracing(context.Background(), serverboot.TracingOptions{
+		ServiceName:  serviceName,
+		Sampling:     serverboot.ParentRatioSampling(1.0),
+		ExporterConn: conn, // nil: the fallback
+		RelayCapable: true,
+	})
+	if err != nil {
+		t.Fatalf("InitTracing: %v", err)
+	}
+
+	_, span := tp.Tracer("relay-e2e").Start(context.Background(), "RunWorkload")
+	span.End()
+
+	if err := tp.Shutdown(context.Background()); err != nil {
+		t.Fatalf("TracerProvider.Shutdown: %v", err)
+	}
+
+	select {
+	case <-sink.got:
+	case <-time.After(10 * time.Second):
+		t.Fatal("collector never received a span over the direct path")
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	var gotService, gotRelay string
+	for _, req := range sink.traces {
+		for _, rs := range req.GetResourceSpans() {
+			for _, attr := range rs.GetResource().GetAttributes() {
+				switch attr.GetKey() {
+				case "service.name":
+					gotService = attr.GetValue().GetStringValue()
+				case relayAttrKey:
+					gotRelay = attr.GetValue().GetStringValue()
+				}
+			}
+		}
+	}
+	if gotService != serviceName {
+		t.Errorf("span arrived with service.name %q, want %q", gotService, serviceName)
+	}
+	// Without this, a node whose atelet never came up looks identical to a
+	// healthy one in the trace store.
+	if gotRelay != "direct" {
+		t.Errorf("span arrived with %s %q, want %q", relayAttrKey, gotRelay, "direct")
+	}
+}

internal/otlprelay/relay.go

@@ -0,0 +1,580 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package otlprelay carries ateom's OTLP telemetry to the collector over a unix
+// socket served by atelet, so a worker pod needs no network path of its own to
+// export spans and metrics.
+//
+// Motivation. ateom runs inside the worker pod that hosts the actor, and until
+// now exported OTLP straight to the collector over the pod's network (the
+// endpoint is injected by atecontroller, see workerpool_apply.go). That has four
+// costs the relay removes:
+//
+//   - Blast radius. The pod runs untrusted agent code. Exporting over the pod
+//     network means the pod must be allowed egress to the collector, which is
+//     reachable to anything that escapes the sandbox. A unix socket cannot leave
+//     the node, so the pod can be denied network egress entirely.
+//   - Connection count. Worker pods are heavily oversubscribed, so a node runs
+//     many ateoms, each holding its own gRPC connection to the collector. They
+//     collapse into atelet's single per-node connection.
+//   - Interference. ateom installs a transparent redirect of actor egress to its
+//     own atunnel listener; its own outbound traffic has to stay clear of the
+//     rules it installs. A unix socket is not IP traffic and cannot be caught.
+//   - Shutdown loss. Teardown frees the actor's network and then the pod goes
+//     away, which is exactly when the spans describing teardown are still queued
+//     in the batch processor. atelet outlives the worker pod.
+//
+// The relay forwards the OTLP request message verbatim rather than decoding it
+// into SDK records and re-exporting. Verbatim pass-through keeps each ateom's
+// own resource (service.name, service.instance.id, pod attributes) intact, so
+// its spans stay attributed to ateom instead of being absorbed into atelet's.
+// Restricting this pass-through to verified ateom sources ensures that future
+// actor telemetry requiring identity rewrites (#761) will be added as an
+// explicit rewriting path alongside this forwarder; see ateomServices.
+//
+// Verbatim applies to the payload, not to the call around it. The request's
+// metadata is dropped and replaced with the headers atelet resolves from its own
+// OTEL_EXPORTER_OTLP_HEADERS, since the upstream leg is atelet's connection and
+// authenticating it is atelet's business; see upstreamContext.
+package otlprelay
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"net"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"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"
+	"k8s.io/utils/lru"
+
+	semconv "go.opentelemetry.io/otel/semconv/v1.40.0"
+	colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+	coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
+)
+
+const (
+	// endpointEnv and its signal-specific overrides are the standard OTLP
+	// exporter variables. The relay resolves them itself because it dials the
+	// collector directly rather than through an OTel SDK exporter.
+	endpointEnv        = "OTEL_EXPORTER_OTLP_ENDPOINT"
+	tracesEndpointEnv  = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+	metricsEndpointEnv = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
+
+	// compressionEnv and its signal-specific overrides configure upstream
+	// gRPC compression (gzip or none).
+	compressionEnv        = "OTEL_EXPORTER_OTLP_COMPRESSION"
+	tracesCompressionEnv  = "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION"
+	metricsCompressionEnv = "OTEL_EXPORTER_OTLP_METRICS_COMPRESSION"
+
+	// headersEnv and its signal-specific overrides carry the headers the
+	// collector expects (an API key, a tenant id). Unlike the endpoint and the
+	// compression, these are per-call metadata rather than per-connection, so
+	// traces and metrics may legitimately differ and are resolved separately.
+	headersEnv        = "OTEL_EXPORTER_OTLP_HEADERS"
+	tracesHeadersEnv  = "OTEL_EXPORTER_OTLP_TRACES_HEADERS"
+	metricsHeadersEnv = "OTEL_EXPORTER_OTLP_METRICS_HEADERS"
+
+	// otlpDefaultPort matches atenet's normalizeOtlpCollector.
+	otlpDefaultPort = "4317"
+
+	// socketMode keeps the relay socket private to root: both atelet and the
+	// ateom worker pods run as root (runAsUser: 0). The socket lives inside
+	// BasePath, a root-owned host directory.
+	socketMode = 0o600
+
+	// maxRecvMsgSize bounds a single Export payload. One misbehaving ateom
+	// should not be able to make atelet allocate without limit; the OTel SDK's
+	// batch processor emits far smaller messages than this.
+	maxRecvMsgSize = 16 << 20 // 16 MiB
+)
+
+// Server is the atelet half of the relay: an OTLP receiver on a unix socket
+// that forwards to the real collector over the node's network.
+type Server struct {
+	upstream *grpc.ClientConn
+	grpc     *grpc.Server
+	sockPath string
+}
+
+// ateomServices are the only sources this relay carries, keyed by the
+// service.name their resource declares — which the OTEL_* environment can
+// override out from under an ateom; see sourceGate for what that looks like.
+// Mirrors the serviceName constants in
+// cmd/ateom-gvisor and cmd/ateom-microvm, which are package main and cannot be
+// imported; TestAteomServicesMatchTheAteomBinaries guards the duplication.
+//
+// This allowlist is a protocol contract rather than a security boundary:
+// service.name is client-provided, so a compromised process could claim an
+// ateom name. Its purpose is to prevent accidental misuse (e.g. an actor SDK
+// pointed at the socket) and keep the pass-through contract explicit for #761.
+// Peer authentication, if needed, would require per-pod sockets or UDS peer
+// credentials (SO_PEERCRED) tied to #741.
+var ateomServices = map[string]bool{
+	"ateom-gvisor":  true,
+	"ateom-microvm": true,
+}
+
+// sourceGate applies the ateomServices allowlist and reports the first
+// rejection of each service.name.
+//
+// The log matters because of how this failure presents. service.name is
+// whatever the resource declares, and resource.WithFromEnv() runs last in
+// serverboot.newResource, so OTEL_SERVICE_NAME or an OTEL_RESOURCE_ATTRIBUTES
+// entry set on a worker pod overrides the ateom's own name. The relay then
+// refuses every export with PermissionDenied, which the OTel SDK does not retry
+// — it drops the batch and reports through the SDK error handler. Telemetry from
+// that ateom simply stops, with nothing on the collector side to say why. One
+// line per distinct name on the node makes it greppable; the ateom itself is
+// unaffected, so this is a diagnosability problem rather than an outage.
+//
+// Dedup is keyed by name with an LRU cache bounded at 256 entries to prevent
+// memory growth from arbitrary client-provided service names, while still
+// suppressing spam from a rejected exporter that retries repeatedly for the
+// pod's whole life.
+type sourceGate struct {
+	logged *lru.Cache // service.name -> struct{}
+}
+
+func newSourceGate() *sourceGate {
+	return &sourceGate{
+		logged: lru.New(256),
+	}
+}
+
+// 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 g.logged == nil {
+		g.logged = lru.New(256)
+	}
+	if _, seen := g.logged.Get(name); !seen {
+		g.logged.Add(name, struct{}{})
+		slog.WarnContext(ctx, "OTLP relay rejected telemetry from an unrecognized source; it is being dropped, not retried. If this is an ateom, check whether OTEL_SERVICE_NAME or OTEL_RESOURCE_ATTRIBUTES on the worker pod is overriding its service.name",
+			slog.String("service.name", name),
+			slog.Any("allowed", allowedServices()),
+			slog.String("note", "logged once per distinct service.name"))
+	}
+	return status.Errorf(codes.PermissionDenied,
+		"the OTLP relay carries ateom telemetry only, got service.name %q; a source whose identity has to be rewritten (#761) must not be forwarded verbatim. If this is an ateom, an OTEL_SERVICE_NAME or OTEL_RESOURCE_ATTRIBUTES override on the worker pod would produce exactly this",
+		name)
+}
+
+func allowedServices() []string {
+	names := make([]string, 0, len(ateomServices))
+	for name := range ateomServices {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}
+
+func resourceServiceName(r *resourcepb.Resource) string {
+	for _, attr := range r.GetAttributes() {
+		if attr.GetKey() == string(semconv.ServiceNameKey) {
+			return attr.GetValue().GetStringValue()
+		}
+	}
+	return ""
+}
+
+// upstreamContext builds the metadata for the upstream call from atelet's own
+// configuration, dropping whatever the ateom sent.
+//
+// The upstream leg is atelet's connection to the collector, so its credentials
+// belong to atelet: the relay resolves OTEL_EXPORTER_OTLP_HEADERS from its own
+// environment, exactly as the SDK exporter it replaces would have. Forwarding
+// the client's headers instead would let anything that reached the socket choose
+// what atelet presents to the collector — a header set is not telemetry to be
+// passed through verbatim the way the resource is, and unlike service.name (see
+// ateomServices) it is not merely claimed identity but an actual credential.
+//
+// Nothing is allow-listed through. atecontroller injects only
+// OTEL_EXPORTER_OTLP_ENDPOINT into worker pods (workerpool_apply.go), so no
+// ateom has a header to lose today; add an allowlist here, not a blanket
+// forward, if one ever needs to reach the collector.
+//
+// The incoming metadata is dropped by simply not copying it: gRPC never
+// propagates incoming metadata to an outgoing call on its own.
+func upstreamContext(ctx context.Context, md metadata.MD) context.Context {
+	if len(md) == 0 {
+		return ctx
+	}
+	return metadata.NewOutgoingContext(ctx, md)
+}
+
+// parseHeaders reads the W3C-Baggage-shaped list the OTLP headers variables
+// carry ("key1=value1,key2=value2", values percent-encoded), as the OTel SDK
+// exporters do.
+//
+// Keys are lower-cased because gRPC metadata keys are case-insensitive and
+// metadata.MD is documented to hold them lower-cased; a mixed-case key set here
+// would otherwise be invisible to metadata.Get.
+func parseHeaders(raw string) (metadata.MD, error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return nil, nil
+	}
+	md := metadata.MD{}
+	// Errors name the position rather than the offending text: any of these
+	// entries may be a credential, and this error reaches a log line.
+	for i, pair := range strings.Split(raw, ",") {
+		pair = strings.TrimSpace(pair)
+		if pair == "" {
+			continue
+		}
+		key, value, found := strings.Cut(pair, "=")
+		if !found {
+			return nil, fmt.Errorf("OTLP header %d is not in key=value form", i+1)
+		}
+		key = strings.ToLower(strings.TrimSpace(key))
+		if key == "" {
+			return nil, fmt.Errorf("OTLP header %d has an empty name", i+1)
+		}
+		// Percent-decoding is what makes a value containing "," or "=" (a base64
+		// token, say) expressible in this format at all. PathUnescape preserves '+'
+		// as a literal character rather than converting it to a space.
+		decoded, err := url.PathUnescape(strings.TrimSpace(value))
+		if err != nil {
+			// The value is deliberately not in the message: these are credentials.
+			return nil, fmt.Errorf("OTLP header %q has a value that is not valid percent-encoding: %w", key, err)
+		}
+		md.Append(key, decoded)
+	}
+	return md, nil
+}
+
+// upstreamHeaders resolves the headers for one signal. Per the OTLP spec the
+// signal-specific variable replaces the generic one rather than merging with
+// it, so a component that sets both gets exactly what the SDK would have sent.
+func upstreamHeaders(signalEnv string) (metadata.MD, error) {
+	env, raw := signalEnv, strings.TrimSpace(os.Getenv(signalEnv))
+	if raw == "" {
+		env, raw = headersEnv, os.Getenv(headersEnv)
+	}
+	md, err := parseHeaders(raw)
+	if err != nil {
+		return nil, fmt.Errorf("while reading %s: %w", env, err)
+	}
+	return md, nil
+}
+
+// The two OTLP services both declare a method named Export, with different
+// request types, so one type cannot implement both: the embedded Unimplemented
+// structs would give Server an ambiguous promoted Export and satisfy neither
+// interface. Each service gets its own tiny forwarder instead.
+
+type traceRelay struct {
+	coltracepb.UnimplementedTraceServiceServer
+	upstream coltracepb.TraceServiceClient
+	// headers atelet presents to the collector; see upstreamContext. Resolved
+	// once at construction: they come from atelet's environment, not the call.
+	headers metadata.MD
+	// gate is shared with metricRelay so a misnamed ateom is reported once, not
+	// once per signal.
+	gate *sourceGate
+}
+
+// Export forwards a batch of spans to the collector unchanged.
+//
+// Deliberately not wrapped in a span of atelet's own: the relay must not inject
+// itself into the trace it is carrying.
+//
+// A batch is refused whole rather than having the offending resource dropped: a
+// partial success the sender reads as success loses telemetry silently.
+func (t *traceRelay) Export(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) {
+	for _, rs := range req.GetResourceSpans() {
+		if err := t.gate.check(ctx, rs.GetResource()); err != nil {
+			return nil, err
+		}
+	}
+	return t.upstream.Export(upstreamContext(ctx, t.headers), req)
+}
+
+type metricRelay struct {
+	colmetricspb.UnimplementedMetricsServiceServer
+	upstream colmetricspb.MetricsServiceClient
+	headers  metadata.MD
+	gate     *sourceGate
+}
+
+// Export forwards a batch of metric datapoints to the collector unchanged.
+func (m *metricRelay) Export(ctx context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) {
+	for _, rm := range req.GetResourceMetrics() {
+		if err := m.gate.check(ctx, rm.GetResource()); err != nil {
+			return nil, err
+		}
+	}
+	return m.upstream.Export(upstreamContext(ctx, m.headers), req)
+}
+
+// validateSocketPath rejects a relative path, which gRPC does not resolve:
+// "unix://foo/r.sock" parses as authority "foo", path "/r.sock". grpc.NewClient
+// being lazy, that wrong target is accepted at startup and fails per export
+// afterwards, which is why this errors rather than falling back.
+func validateSocketPath(sockPath string) error {
+	if !filepath.IsAbs(sockPath) {
+		return fmt.Errorf("the OTLP relay socket path %q is relative; it must be absolute, since atelet and ateom would otherwise resolve it against different working directories", sockPath)
+	}
+	return nil
+}
+
+// NewServer builds a relay that forwards to the collector named by the standard
+// OTLP endpoint environment variables. It returns (nil, nil) when sockPath is
+// empty (the relay is switched off) or when no endpoint is configured: a relay
+// with nowhere to forward to would accept an ateom's spans and drop them, which
+// is worse than ateom finding no socket and falling back to a direct export.
+func NewServer(ctx context.Context, sockPath string) (*Server, error) {
+	if sockPath == "" {
+		return nil, nil
+	}
+	if err := validateSocketPath(sockPath); err != nil {
+		return nil, err
+	}
+	target, err := upstreamTarget()
+	if err != nil {
+		return nil, err
+	}
+	if target == "" {
+		slog.InfoContext(ctx, "OTLP relay disabled: no collector endpoint configured",
+			slog.String("env", endpointEnv))
+		return nil, nil
+	}
+
+	comp, err := upstreamCompression()
+	if err != nil {
+		return nil, err
+	}
+
+	// Resolved before the socket exists: a header set atelet cannot parse would
+	// otherwise become a per-export failure against a collector that rejects the
+	// unauthenticated calls, which is harder to read than refusing to start the
+	// relay. ateom then finds no socket and exports directly.
+	traceHeaders, err := upstreamHeaders(tracesHeadersEnv)
+	if err != nil {
+		return nil, err
+	}
+	metricHeaders, err := upstreamHeaders(metricsHeadersEnv)
+	if err != nil {
+		return nil, err
+	}
+
+	dialOpts := []grpc.DialOption{
+		// Plaintext by design today; TLS support for the upstream leg will be added
+		// in tandem with #741.
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}
+	if comp == "gzip" {
+		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name)))
+	}
+
+	// Lazy by design: grpc.NewClient does not block on the collector being up,
+	// so atelet startup does not depend on the collector's readiness.
+	upstream, err := grpc.NewClient(target, dialOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("while dialing OTLP collector %q: %w", target, err)
+	}
+
+	s := &Server{
+		upstream: upstream,
+		sockPath: sockPath,
+		grpc:     grpc.NewServer(grpc.MaxRecvMsgSize(maxRecvMsgSize)),
+	}
+	gate := newSourceGate()
+	coltracepb.RegisterTraceServiceServer(s.grpc, &traceRelay{
+		upstream: coltracepb.NewTraceServiceClient(upstream),
+		headers:  traceHeaders,
+		gate:     gate,
+	})
+	colmetricspb.RegisterMetricsServiceServer(s.grpc, &metricRelay{
+		upstream: colmetricspb.NewMetricsServiceClient(upstream),
+		headers:  metricHeaders,
+		gate:     gate,
+	})
+	// Header names only: the values are credentials.
+	slog.InfoContext(ctx, "OTLP relay forwarding to collector",
+		slog.String("collector", target),
+		slog.String("compression", comp),
+		slog.Any("traceHeaders", headerNames(traceHeaders)),
+		slog.Any("metricHeaders", headerNames(metricHeaders)))
+	return s, nil
+}
+
+// Serve listens on the relay socket and blocks until the server stops. Designed
+// to be `go`-launched; it returns an error only if the socket cannot be opened
+// or serving fails.
+func (s *Server) Serve(ctx context.Context) error {
+	if err := os.MkdirAll(filepath.Dir(s.sockPath), 0o755); err != nil {
+		return fmt.Errorf("while creating the OTLP relay socket directory: %w", err)
+	}
+	// A socket left behind by a previous atelet would make Listen fail with
+	// EADDRINUSE even though nothing holds it.
+	if err := os.Remove(s.sockPath); err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return fmt.Errorf("while removing a stale OTLP relay socket %q: %w", s.sockPath, err)
+	}
+	lis, err := net.Listen("unix", s.sockPath)
+	if err != nil {
+		return fmt.Errorf("while opening the OTLP relay socket %q: %w", s.sockPath, err)
+	}
+	// net.Listen applies the umask, which on atelet would typically leave the
+	// socket group/other-unwritable and unreachable from an ateom running as a
+	// different uid. Widen it explicitly.
+	if err := os.Chmod(s.sockPath, socketMode); err != nil {
+		_ = lis.Close()
+		return fmt.Errorf("while setting the OTLP relay socket mode: %w", err)
+	}
+
+	slog.InfoContext(ctx, "OTLP relay serving", slog.String("socket", s.sockPath))
+	return s.grpc.Serve(lis)
+}
+
+// Stop drains the relay, closes the upstream connection and removes the socket.
+func (s *Server) Stop() {
+	s.grpc.GracefulStop()
+	_ = s.upstream.Close()
+	_ = os.Remove(s.sockPath)
+}
+
+// headerNames lists the configured header names, sorted, for logging. It exists
+// so an operator can confirm the relay picked up the headers without the values
+// reaching the node's logs.
+func headerNames(md metadata.MD) []string {
+	names := make([]string, 0, len(md))
+	for k := range md {
+		names = append(names, k)
+	}
+	sort.Strings(names)
+	return names
+}
+
+// upstreamCompression resolves the compression algorithm (gzip or none) to use
+// for upstream export.
+func upstreamCompression() (string, error) {
+	generic := strings.TrimSpace(os.Getenv(compressionEnv))
+	traces := strings.TrimSpace(os.Getenv(tracesCompressionEnv))
+	metrics := strings.TrimSpace(os.Getenv(metricsCompressionEnv))
+
+	traceComp := generic
+	if traces != "" {
+		traceComp = traces
+	}
+	metricComp := generic
+	if metrics != "" {
+		metricComp = metrics
+	}
+
+	if traceComp != "" && metricComp != "" && traceComp != metricComp {
+		return "", fmt.Errorf("signal-specific compression settings conflict (%q for traces vs %q for metrics); the relay carries both signals over one connection",
+			traceComp, metricComp)
+	}
+
+	resolved := traceComp
+	if resolved == "" {
+		resolved = metricComp
+	}
+	switch resolved {
+	case "", "none":
+		return "none", nil
+	case "gzip":
+		return "gzip", nil
+	default:
+		return "", fmt.Errorf("unsupported OTLP compression %q, want gzip or none", resolved)
+	}
+}
+
+// upstreamTarget resolves the collector address the relay forwards to, from the
+// standard OTLP endpoint variables, into the bare host:port grpc.NewClient wants.
+//
+// The signal-specific variables must agree: the relay carries traces and metrics
+// over one connection, so it cannot honor two different collectors. Configuring
+// both differently is a misconfiguration rather than something to silently pick
+// a winner for.
+func upstreamTarget() (string, error) {
+	generic := strings.TrimSpace(os.Getenv(endpointEnv))
+	traces := strings.TrimSpace(os.Getenv(tracesEndpointEnv))
+	metrics := strings.TrimSpace(os.Getenv(metricsEndpointEnv))
+
+	traceTarget := generic
+	if traces != "" {
+		traceTarget = traces
+	}
+	metricTarget := generic
+	if metrics != "" {
+		metricTarget = metrics
+	}
+
+	if traceTarget != "" && metricTarget != "" && traceTarget != metricTarget {
+		return "", fmt.Errorf("signal-specific endpoints conflict (%q for traces vs %q for metrics); the relay carries both signals over one connection",
+			traceTarget, metricTarget)
+	}
+
+	resolved := traceTarget
+	if resolved == "" {
+		resolved = metricTarget
+	}
+	if resolved == "" {
+		return "", nil
+	}
+	return normalizeEndpoint(resolved)
+}
+
+// normalizeEndpoint accepts both a bare "host:port" and the URL form the OTLP
+// environment variables carry, and returns the host:port grpc.NewClient dials.
+//
+// https is rejected rather than downgraded: the relay dials with insecure
+// credentials, so honoring it would ship telemetry in plaintext to an endpoint
+// that asked for TLS.
+func normalizeEndpoint(addr string) (string, error) {
+	hostport := addr
+	if strings.Contains(addr, "://") {
+		u, err := url.Parse(addr)
+		if err != nil {
+			return "", fmt.Errorf("parse OTLP collector endpoint %q: %w", addr, err)
+		}
+		switch u.Scheme {
+		case "http":
+		case "https":
+			return "", fmt.Errorf("OTLP collector endpoint %q uses https, which the relay does not support: it forwards over an insecure gRPC connection. Point it at an http:// endpoint", addr)
+		default:
+			return "", fmt.Errorf("OTLP collector endpoint %q has unsupported scheme %q, want http", addr, u.Scheme)
+		}
+		hostport = u.Host
+	}
+
+	host, port, err := net.SplitHostPort(hostport)
+	if err != nil {
+		host = strings.Trim(hostport, "[]")
+		port = otlpDefaultPort
+	}
+	if host == "" {
+		return "", fmt.Errorf("OTLP collector endpoint %q names no host", addr)
+	}
+	return net.JoinHostPort(host, port), nil
+}

internal/otlprelay/relay_test.go

@@ -0,0 +1,871 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otlprelay
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io/fs"
+	"log/slog"
+	"net"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/testing/protocmp"
+
+	colmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+	coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	commonpb "go.opentelemetry.io/proto/otlp/common/v1"
+	metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
+	resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
+	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+)
+
+// fakeCollector is a stand-in for the real OTLP collector: it records what the
+// relay forwards so the tests can assert the payload arrived unchanged.
+type fakeCollector struct {
+	coltracepb.UnimplementedTraceServiceServer
+
+	mu       sync.Mutex
+	traces   []*coltracepb.ExportTraceServiceRequest
+	metrics  []*colmetricspb.ExportMetricsServiceRequest
+	traceMD  []metadata.MD
+	metricMD []metadata.MD
+	got      chan struct{}
+}
+
+func (f *fakeCollector) Export(ctx context.Context, req *coltracepb.ExportTraceServiceRequest) (*coltracepb.ExportTraceServiceResponse, error) {
+	f.mu.Lock()
+	f.traces = append(f.traces, req)
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		f.traceMD = append(f.traceMD, md.Copy())
+	}
+	f.mu.Unlock()
+	f.got <- struct{}{}
+	return &coltracepb.ExportTraceServiceResponse{}, nil
+}
+
+// metricsSink exists because the two OTLP services both declare Export with
+// different request types, the same collision the relay itself works around.
+type metricsSink struct {
+	colmetricspb.UnimplementedMetricsServiceServer
+	parent *fakeCollector
+}
+
+func (m *metricsSink) Export(ctx context.Context, req *colmetricspb.ExportMetricsServiceRequest) (*colmetricspb.ExportMetricsServiceResponse, error) {
+	m.parent.mu.Lock()
+	m.parent.metrics = append(m.parent.metrics, req)
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		m.parent.metricMD = append(m.parent.metricMD, md.Copy())
+	}
+	m.parent.mu.Unlock()
+	m.parent.got <- struct{}{}
+	return &colmetricspb.ExportMetricsServiceResponse{}, nil
+}
+
+// startFakeCollector serves the OTLP collector services on a loopback TCP port
+// (the shape the relay forwards to) and returns the sink and its host:port.
+func startFakeCollector(t *testing.T) (*fakeCollector, string) {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	sink := &fakeCollector{got: make(chan struct{}, 8)}
+	srv := grpc.NewServer()
+	coltracepb.RegisterTraceServiceServer(srv, sink)
+	colmetricspb.RegisterMetricsServiceServer(srv, &metricsSink{parent: sink})
+	go func() { _ = srv.Serve(lis) }()
+	t.Cleanup(srv.Stop)
+	return sink, lis.Addr().String()
+}
+
+// startRelay brings up a relay on a socket in a temp dir, wired to collector.
+func startRelay(t *testing.T, collector string) string {
+	t.Helper()
+	t.Setenv(endpointEnv, collector)
+
+	// Short filename: a unix socket path is capped at ~104 bytes and the test
+	// temp dir already eats most of that on darwin.
+	sock := filepath.Join(t.TempDir(), "r.sock")
+	relay, err := NewServer(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	if relay == nil {
+		t.Fatal("NewServer returned nil with a collector endpoint set")
+	}
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- relay.Serve(context.Background()) }()
+	t.Cleanup(relay.Stop)
+
+	// Serve creates the socket asynchronously; Dial's existence check needs it.
+	waitForSocket(t, sock, serveErr)
+	return sock
+}
+
+func waitForSocket(t *testing.T, sock string, serveErr <-chan error) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		select {
+		case err := <-serveErr:
+			t.Fatalf("relay.Serve returned early: %v", err)
+		default:
+		}
+		// Closed immediately: a probe connection that never speaks HTTP/2 sits
+		// in the server's handshake path until its 120s timeout, and
+		// GracefulStop would wait the whole of it.
+		if c, err := net.Dial("unix", sock); err == nil {
+			_ = c.Close()
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("relay socket %q never became connectable", sock)
+}
+
+// serviceResource builds the one resource attribute the relay's scoping looks
+// at. Passing "" yields a resource that declares no service.name at all.
+func serviceResource(name string) *resourcepb.Resource {
+	if name == "" {
+		return &resourcepb.Resource{}
+	}
+	return &resourcepb.Resource{
+		Attributes: []*commonpb.KeyValue{{
+			Key:   "service.name",
+			Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: name}},
+		}},
+	}
+}
+
+// TestRelayForwardsTracesVerbatim is the property the whole design rests on:
+// what an ateom exports is what the collector sees, including the resource
+// attributes that attribute the spans to that ateom rather than to atelet.
+func TestRelayForwardsTracesVerbatim(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	if conn == nil {
+		t.Fatal("Dial returned no connection for an existing socket")
+	}
+	defer conn.Close()
+
+	req := &coltracepb.ExportTraceServiceRequest{
+		ResourceSpans: []*tracepb.ResourceSpans{{
+			Resource: serviceResource("ateom-microvm"),
+			ScopeSpans: []*tracepb.ScopeSpans{{
+				Spans: []*tracepb.Span{{
+					Name:    "RunWorkload",
+					TraceId: []byte("0123456789abcdef"),
+					SpanId:  []byte("01234567"),
+				}},
+			}},
+		}},
+	}
+	if _, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), req); err != nil {
+		t.Fatalf("Export through the relay: %v", err)
+	}
+
+	select {
+	case <-sink.got:
+	case <-time.After(5 * time.Second):
+		t.Fatal("collector never received the forwarded trace export")
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.traces) != 1 {
+		t.Fatalf("collector got %d trace exports, want 1", len(sink.traces))
+	}
+	if diff := cmp.Diff(req, sink.traces[0], protocmp.Transform()); diff != "" {
+		t.Errorf("forwarded request differs from what was sent (-sent +received):\n%s", diff)
+	}
+}
+
+func TestRelayForwardsMetrics(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+
+	req := &colmetricspb.ExportMetricsServiceRequest{
+		ResourceMetrics: []*metricspb.ResourceMetrics{{
+			Resource: serviceResource("ateom-microvm"),
+			ScopeMetrics: []*metricspb.ScopeMetrics{{
+				Metrics: []*metricspb.Metric{{Name: "ateom.workload.runs"}},
+			}},
+		}},
+	}
+	if _, err := colmetricspb.NewMetricsServiceClient(conn).Export(context.Background(), req); err != nil {
+		t.Fatalf("Export through the relay: %v", err)
+	}
+
+	select {
+	case <-sink.got:
+	case <-time.After(5 * time.Second):
+		t.Fatal("collector never received the forwarded metric export")
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.metrics) != 1 {
+		t.Fatalf("collector got %d metric exports, want 1", len(sink.metrics))
+	}
+	if diff := cmp.Diff(req, sink.metrics[0], protocmp.Transform()); diff != "" {
+		t.Errorf("forwarded request differs from what was sent (-sent +received):\n%s", diff)
+	}
+}
+
+// TestStopRemovesSocket matters for the restart path: a leftover socket makes
+// the next atelet's Listen fail with EADDRINUSE, and in the meantime makes
+// every ateom on the node believe a relay is there.
+func TestStopRemovesSocket(t *testing.T) {
+	_, collector := startFakeCollector(t)
+	t.Setenv(endpointEnv, collector)
+
+	sock := filepath.Join(t.TempDir(), "r.sock")
+	relay, err := NewServer(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- relay.Serve(context.Background()) }()
+	waitForSocket(t, sock, serveErr)
+
+	relay.Stop()
+	if _, err := os.Stat(sock); !errors.Is(err, fs.ErrNotExist) {
+		t.Errorf("os.Stat(%q) after Stop = %v, want the socket to be gone", sock, err)
+	}
+}
+
+// TestServeReplacesStaleSocket covers the atelet-crashed-and-restarted case:
+// the socket file survives the process, and Listen would refuse to reuse it.
+func TestServeReplacesStaleSocket(t *testing.T) {
+	_, collector := startFakeCollector(t)
+	t.Setenv(endpointEnv, collector)
+
+	sock := filepath.Join(t.TempDir(), "r.sock")
+	if err := os.WriteFile(sock, nil, 0o600); err != nil {
+		t.Fatalf("planting a stale socket file: %v", err)
+	}
+
+	relay, err := NewServer(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- relay.Serve(context.Background()) }()
+	t.Cleanup(relay.Stop)
+	waitForSocket(t, sock, serveErr)
+}
+
+func TestNewServerDisabled(t *testing.T) {
+	t.Setenv(endpointEnv, "otel-collector:4317")
+	relay, err := NewServer(context.Background(), "")
+	if err != nil {
+		t.Fatalf("NewServer with an empty socket path: %v", err)
+	}
+	if relay != nil {
+		t.Error("NewServer with an empty socket path returned a server, want nil (relay disabled)")
+	}
+}
+
+func TestNewServerWithoutCollector(t *testing.T) {
+	t.Setenv(endpointEnv, "")
+	t.Setenv(tracesEndpointEnv, "")
+	t.Setenv(metricsEndpointEnv, "")
+	relay, err := NewServer(context.Background(), filepath.Join(t.TempDir(), "r.sock"))
+	if err != nil {
+		t.Fatalf("NewServer with no collector configured: %v", err)
+	}
+	if relay != nil {
+		t.Error("NewServer with no collector configured returned a server; it would accept spans and drop them")
+	}
+}
+
+func TestDialMissingSocketFallsBack(t *testing.T) {
+	conn, err := Dial(context.Background(), filepath.Join(t.TempDir(), "absent.sock"))
+	if err != nil {
+		t.Fatalf("Dial on an absent socket: %v, want the fallback", err)
+	}
+	if conn != nil {
+		conn.Close()
+		t.Error("Dial on an absent socket returned a connection, want nil so the caller exports directly")
+	}
+}
+
+func TestDialEmptyPath(t *testing.T) {
+	conn, err := Dial(context.Background(), "")
+	if err != nil {
+		t.Fatalf("Dial(\"\"): %v", err)
+	}
+	if conn != nil {
+		conn.Close()
+		t.Error("Dial(\"\") returned a connection, want nil")
+	}
+}
+
+// A rejected export is dropped by the SDK without a retry, so the node log is
+// the only place the misconfiguration shows up — and it has to show up exactly
+// once per name, because the rejected exporter keeps retrying for the life of
+// the pod.
+func TestSourceGateLogsEachRejectionOnce(t *testing.T) {
+	var buf bytes.Buffer
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&buf, nil)))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	gate := newSourceGate()
+	for range 3 {
+		if err := gate.check(context.Background(), serviceResource("actor")); status.Code(err) != codes.PermissionDenied {
+			t.Fatalf("gate.check = %v, want PermissionDenied every time", err)
+		}
+	}
+	if err := gate.check(context.Background(), serviceResource("atelet")); err == nil {
+		t.Fatal("gate.check accepted atelet")
+	}
+	if err := gate.check(context.Background(), serviceResource("ateom-gvisor")); err != nil {
+		t.Fatalf("gate.check rejected an ateom: %v", err)
+	}
+
+	if got := strings.Count(buf.String(), "rejected telemetry"); got != 2 {
+		t.Errorf("logged %d rejections, want 2 (one per distinct service.name):\n%s", got, buf.String())
+	}
+	// The operator has to be able to tell which override to go looking for.
+	if !strings.Contains(buf.String(), "OTEL_SERVICE_NAME") {
+		t.Errorf("rejection log does not name the env var that causes it:\n%s", buf.String())
+	}
+}
+
+// TestRelayRefusesNonAteomSource is the scoping contract. The empty
+// service.name case is the one worth keeping: that is the shape telemetry takes
+// when identity has not been injected, which is the actor situation in #761.
+func TestRelayRefusesNonAteomSource(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+
+	for _, tc := range []struct {
+		name    string
+		service string
+	}{
+		{name: "another substrate component", service: "atelet"},
+		{name: "actor telemetry", service: "actor"},
+		{name: "no service name at all", service: ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{
+				ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(tc.service)}},
+			})
+			if got := status.Code(err); got != codes.PermissionDenied {
+				t.Errorf("trace Export from service.name %q = code %v (%v), want %v", tc.service, got, err, codes.PermissionDenied)
+			}
+
+			_, err = colmetricspb.NewMetricsServiceClient(conn).Export(context.Background(), &colmetricspb.ExportMetricsServiceRequest{
+				ResourceMetrics: []*metricspb.ResourceMetrics{{Resource: serviceResource(tc.service)}},
+			})
+			if got := status.Code(err); got != codes.PermissionDenied {
+				t.Errorf("metric Export from service.name %q = code %v (%v), want %v", tc.service, got, err, codes.PermissionDenied)
+			}
+		})
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.traces) != 0 || len(sink.metrics) != 0 {
+		t.Errorf("collector received %d traces and %d metrics from refused sources, want none to be forwarded", len(sink.traces), len(sink.metrics))
+	}
+}
+
+// TestRelayRefusesMixedBatch pins the all-or-nothing choice: dropping just the
+// foreign resource would return success to a sender that lost telemetry.
+func TestRelayRefusesMixedBatch(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+
+	_, err = coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{
+		ResourceSpans: []*tracepb.ResourceSpans{
+			{Resource: serviceResource("ateom-gvisor")},
+			{Resource: serviceResource("actor")},
+		},
+	})
+	if got := status.Code(err); got != codes.PermissionDenied {
+		t.Errorf("Export of a mixed batch = code %v (%v), want %v", got, err, codes.PermissionDenied)
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.traces) != 0 {
+		t.Errorf("collector received %d exports from a mixed batch, want the batch refused whole", len(sink.traces))
+	}
+}
+
+// TestRelayAcceptsEveryAteomService guards against the allowlist drifting from
+// the binaries in a way that silently drops all of one runtime's telemetry.
+func TestRelayAcceptsEveryAteomService(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+
+	for service := range ateomServices {
+		if _, err := coltracepb.NewTraceServiceClient(conn).Export(context.Background(), &coltracepb.ExportTraceServiceRequest{
+			ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(service)}},
+		}); err != nil {
+			t.Errorf("Export from allowlisted service %q: %v", service, err)
+			continue
+		}
+		select {
+		case <-sink.got:
+		case <-time.After(5 * time.Second):
+			t.Errorf("collector never received the export from %q", service)
+		}
+	}
+}
+
+// TestAteomServicesMatchTheAteomBinaries keeps the allowlist honest. A typo in
+// it would otherwise be invisible: every real ateom export would be refused
+// while every test here still passed, because they would share the typo.
+func TestAteomServicesMatchTheAteomBinaries(t *testing.T) {
+	// Matches `const serviceName = "..."` in each ateom main package.
+	decl := regexp.MustCompile(`(?m)^\s*const\s+serviceName\s*=\s*"([^"]+)"`)
+
+	found := map[string]bool{}
+	for _, main := range []string{"../../cmd/ateom-gvisor/main.go", "../../cmd/ateom-microvm/main.go"} {
+		src, err := os.ReadFile(main)
+		if err != nil {
+			t.Fatalf("reading %s: %v", main, err)
+		}
+		m := decl.FindSubmatch(src)
+		if m == nil {
+			t.Fatalf("no `const serviceName = \"...\"` found in %s; if it moved, this test and ateomServices both need updating", main)
+		}
+		name := string(m[1])
+		found[name] = true
+		if !ateomServices[name] {
+			t.Errorf("%s reports service.name %q, which ateomServices does not allow; the relay would refuse all of its telemetry", main, name)
+		}
+	}
+
+	for name := range ateomServices {
+		if !found[name] {
+			t.Errorf("ateomServices allows %q, but no ateom binary declares it", name)
+		}
+	}
+}
+
+// TestDialRejectsRelativeSocketPath: a relative path was accepted here and then
+// failed lazily at the first export, with the spans already gone.
+func TestDialRejectsRelativeSocketPath(t *testing.T) {
+	conn, err := Dial(context.Background(), "relative/r.sock")
+	if conn != nil {
+		conn.Close()
+	}
+	if err == nil {
+		t.Fatal("Dial with a relative socket path returned no error; it would dial a misparsed target and lose telemetry per export")
+	}
+	if !strings.Contains(err.Error(), "absolute") {
+		t.Errorf("Dial error = %v, want it to say the path must be absolute", err)
+	}
+}
+
+func TestNewServerRejectsRelativeSocketPath(t *testing.T) {
+	t.Setenv(endpointEnv, "collector:4317")
+	relay, err := NewServer(context.Background(), "relative/r.sock")
+	if relay != nil {
+		relay.Stop()
+	}
+	if err == nil {
+		t.Fatal("NewServer with a relative socket path returned no error; it would listen somewhere no ateom can name")
+	}
+	if !strings.Contains(err.Error(), "absolute") {
+		t.Errorf("NewServer error = %v, want it to say the path must be absolute", err)
+	}
+}
+
+// TestServeLeavesAPopulatedDirectoryAlone is the RemoveAll regression: a flag
+// value naming the directory must fail rather than empty it.
+func TestServeLeavesAPopulatedDirectoryAlone(t *testing.T) {
+	_, collector := startFakeCollector(t)
+	t.Setenv(endpointEnv, collector)
+
+	dir := filepath.Join(t.TempDir(), "basepath")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatalf("MkdirAll: %v", err)
+	}
+	occupant := filepath.Join(dir, "ateom.sock")
+	if err := os.WriteFile(occupant, nil, 0o600); err != nil {
+		t.Fatalf("planting a neighbouring socket: %v", err)
+	}
+
+	relay, err := NewServer(context.Background(), dir)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	t.Cleanup(relay.Stop)
+
+	if err := relay.Serve(context.Background()); err == nil {
+		t.Error("Serve on a populated directory returned no error, want it to refuse")
+	}
+	if _, err := os.Stat(occupant); err != nil {
+		t.Errorf("os.Stat(%q) = %v, want the neighbouring socket untouched", occupant, err)
+	}
+}
+
+func TestNormalizeEndpoint(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		in      string
+		want    string
+		wantErr string
+	}{
+		{name: "host and port", in: "otel-collector.ate-system.svc:4317", want: "otel-collector.ate-system.svc:4317"},
+		{name: "bare host defaults the port", in: "otel-collector", want: "otel-collector:" + otlpDefaultPort},
+		{name: "http url", in: "http://otel-collector:4317", want: "otel-collector:4317"},
+		{name: "http url without port", in: "http://otel-collector", want: "otel-collector:" + otlpDefaultPort},
+		{name: "ipv6 literal", in: "[::1]:4317", want: "[::1]:4317"},
+		{name: "ipv6 literal without port", in: "[::1]", want: "[::1]:" + otlpDefaultPort},
+		{name: "https rejected", in: "https://otel-collector:4317", wantErr: "https"},
+		{name: "unknown scheme rejected", in: "grpc://otel-collector:4317", wantErr: "unsupported scheme"},
+		{name: "empty host rejected", in: "http://:4317", wantErr: "names no host"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := normalizeEndpoint(tc.in)
+			if tc.wantErr != "" {
+				if err == nil {
+					t.Fatalf("normalizeEndpoint(%q) = %q, want an error containing %q", tc.in, got, tc.wantErr)
+				}
+				if !strings.Contains(err.Error(), tc.wantErr) {
+					t.Errorf("normalizeEndpoint(%q) error = %v, want it to mention %q", tc.in, err, tc.wantErr)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("normalizeEndpoint(%q): %v", tc.in, err)
+			}
+			if got != tc.want {
+				t.Errorf("normalizeEndpoint(%q) = %q, want %q", tc.in, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestUpstreamTarget(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		generic string
+		traces  string
+		metrics string
+		want    string
+		wantErr bool
+	}{
+		{name: "unset", want: ""},
+		{name: "generic only", generic: "collector:4317", want: "collector:4317"},
+		{name: "signal specific overrides generic", generic: "generic:4317", traces: "specific:4317", metrics: "specific:4317", want: "specific:4317"},
+		{name: "generic conflicts with different traces specific", generic: "generic:4317", traces: "traces-only:4317", wantErr: true},
+		{name: "generic conflicts with different metrics specific", generic: "generic:4317", metrics: "metrics-only:4317", wantErr: true},
+		{name: "matching generic and signal specific", generic: "collector:4317", traces: "collector:4317", metrics: "collector:4317", want: "collector:4317"},
+		{name: "traces specific alone", traces: "specific:4317", want: "specific:4317"},
+		{name: "metrics specific alone", metrics: "specific:4317", want: "specific:4317"},
+		{name: "conflicting signals rejected", traces: "a:4317", metrics: "b:4317", wantErr: true},
+		{name: "whitespace trimmed", generic: "  collector:4317  ", want: "collector:4317"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv(endpointEnv, tc.generic)
+			t.Setenv(tracesEndpointEnv, tc.traces)
+			t.Setenv(metricsEndpointEnv, tc.metrics)
+			got, err := upstreamTarget()
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("upstreamTarget() = %q, want an error", got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("upstreamTarget(): %v", err)
+			}
+			if got != tc.want {
+				t.Errorf("upstreamTarget() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestUpstreamCompression(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		generic string
+		traces  string
+		metrics string
+		want    string
+		wantErr bool
+	}{
+		{name: "unset", want: "none"},
+		{name: "generic gzip", generic: "gzip", want: "gzip"},
+		{name: "generic none", generic: "none", want: "none"},
+		{name: "traces specific gzip", traces: "gzip", want: "gzip"},
+		{name: "metrics specific gzip", metrics: "gzip", want: "gzip"},
+		{name: "both specific gzip", traces: "gzip", metrics: "gzip", want: "gzip"},
+		{name: "conflicting compression rejected", traces: "gzip", metrics: "none", wantErr: true},
+		{name: "generic conflicts with traces", generic: "none", traces: "gzip", wantErr: true},
+		{name: "invalid compression rejected", generic: "zstd", wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv(compressionEnv, tc.generic)
+			t.Setenv(tracesCompressionEnv, tc.traces)
+			t.Setenv(metricsCompressionEnv, tc.metrics)
+			got, err := upstreamCompression()
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("upstreamCompression() = %q, want error", got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("upstreamCompression(): %v", err)
+			}
+			if got != tc.want {
+				t.Errorf("upstreamCompression() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// exportBoth sends one empty trace batch and one empty metric batch from
+// service through the relay, and returns the metadata each arrived with.
+func exportBoth(t *testing.T, sink *fakeCollector, sock, service string, ctx context.Context) (traceMD, metricMD metadata.MD) {
+	t.Helper()
+	conn, err := Dial(context.Background(), sock)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	defer conn.Close()
+
+	if _, err := coltracepb.NewTraceServiceClient(conn).Export(ctx, &coltracepb.ExportTraceServiceRequest{
+		ResourceSpans: []*tracepb.ResourceSpans{{Resource: serviceResource(service)}},
+	}); err != nil {
+		t.Fatalf("traceClient.Export: %v", err)
+	}
+	if _, err := colmetricspb.NewMetricsServiceClient(conn).Export(ctx, &colmetricspb.ExportMetricsServiceRequest{
+		ResourceMetrics: []*metricspb.ResourceMetrics{{Resource: serviceResource(service)}},
+	}); err != nil {
+		t.Fatalf("metricClient.Export: %v", err)
+	}
+
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	if len(sink.traceMD) == 0 {
+		t.Fatal("collector received no metadata for trace export")
+	}
+	if len(sink.metricMD) == 0 {
+		t.Fatal("collector received no metadata for metrics export")
+	}
+	return sink.traceMD[0], sink.metricMD[0]
+}
+
+// The upstream leg is atelet's connection, so its credentials are atelet's. An
+// ateom that sets a header of its own must not get to choose what atelet
+// presents to the collector.
+func TestExportDropsClientMetadata(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	ctx := metadata.AppendToOutgoingContext(context.Background(),
+		"authorization", "Bearer client-token",
+		"custom-header", "custom-value",
+	)
+	traceMD, metricMD := exportBoth(t, sink, sock, "ateom-gvisor", ctx)
+
+	for _, tc := range []struct {
+		signal string
+		md     metadata.MD
+	}{{"trace", traceMD}, {"metric", metricMD}} {
+		for _, key := range []string{"authorization", "custom-header"} {
+			if got := tc.md.Get(key); len(got) != 0 {
+				t.Errorf("%s export reached the collector with the client's %s = %v, want it dropped", tc.signal, key, got)
+			}
+		}
+	}
+}
+
+// ...and the headers atelet is configured with are attached in its place, per
+// signal, exactly as the SDK exporter the relay stands in for would have.
+func TestExportAttachesAteletHeaders(t *testing.T) {
+	sink, collector := startFakeCollector(t)
+	// Set before startRelay: NewServer resolves headers once, at construction.
+	t.Setenv(headersEnv, "authorization=Bearer atelet-token,x-tenant=substrate")
+	t.Setenv(metricsHeadersEnv, "authorization=Bearer metrics-token")
+	sock := startRelay(t, collector)
+
+	// The client sends its own, which must lose to atelet's rather than
+	// appending a second value the collector might pick either way.
+	ctx := metadata.AppendToOutgoingContext(context.Background(), "authorization", "Bearer client-token")
+	traceMD, metricMD := exportBoth(t, sink, sock, "ateom-gvisor", ctx)
+
+	if got := traceMD.Get("authorization"); len(got) != 1 || got[0] != "Bearer atelet-token" {
+		t.Errorf("trace export authorization = %v, want exactly [Bearer atelet-token]", got)
+	}
+	if got := traceMD.Get("x-tenant"); len(got) != 1 || got[0] != "substrate" {
+		t.Errorf("trace export x-tenant = %v, want [substrate]", got)
+	}
+	// The metrics-specific variable replaces the generic one whole, so x-tenant
+	// is deliberately absent here.
+	if got := metricMD.Get("authorization"); len(got) != 1 || got[0] != "Bearer metrics-token" {
+		t.Errorf("metric export authorization = %v, want exactly [Bearer metrics-token]", got)
+	}
+	if got := metricMD.Get("x-tenant"); len(got) != 0 {
+		t.Errorf("metric export x-tenant = %v, want absent: %s replaces %s rather than merging", got, metricsHeadersEnv, headersEnv)
+	}
+}
+
+func TestParseHeaders(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		raw     string
+		want    map[string][]string
+		wantErr bool
+	}{
+		{name: "empty", raw: "", want: map[string][]string{}},
+		{name: "single", raw: "api-key=secret", want: map[string][]string{"api-key": {"secret"}}},
+		{name: "multiple with spaces", raw: "api-key=secret, x-tenant = sub ", want: map[string][]string{"api-key": {"secret"}, "x-tenant": {"sub"}}},
+		// gRPC metadata keys are case-insensitive and stored lower-cased; a
+		// mixed-case key here would be invisible to metadata.Get.
+		{name: "key lower-cased", raw: "X-Tenant=sub", want: map[string][]string{"x-tenant": {"sub"}}},
+		// Percent-decoding is what lets a base64 token containing "=" through.
+		// PathUnescape keeps '+' as a literal rather than converting it to a space.
+		{name: "percent-encoded value", raw: "authorization=Bearer%20abc%3D%3D", want: map[string][]string{"authorization": {"Bearer abc=="}}},
+		{name: "plus preserved", raw: "authorization=Bearer+token%20val", want: map[string][]string{"authorization": {"Bearer+token val"}}},
+		{name: "empty value kept", raw: "x-tenant=", want: map[string][]string{"x-tenant": {""}}},
+		{name: "trailing comma tolerated", raw: "api-key=secret,", want: map[string][]string{"api-key": {"secret"}}},
+		{name: "missing equals rejected", raw: "api-key", wantErr: true},
+		{name: "empty name rejected", raw: "=secret", wantErr: true},
+		{name: "bad percent-encoding rejected", raw: "api-key=%zz", wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := parseHeaders(tc.raw)
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("parseHeaders(%q) = %v, want error", tc.raw, got)
+				}
+				// A credential must not end up in a log line or a test failure.
+				if strings.Contains(err.Error(), "secret") {
+					t.Errorf("parseHeaders error leaks the header value: %v", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("parseHeaders(%q): %v", tc.raw, err)
+			}
+			if len(got) != len(tc.want) {
+				t.Fatalf("parseHeaders(%q) = %v, want %v", tc.raw, got, tc.want)
+			}
+			for k, want := range tc.want {
+				if diff := got.Get(k); len(diff) != len(want) || (len(want) == 1 && diff[0] != want[0]) {
+					t.Errorf("parseHeaders(%q)[%q] = %v, want %v", tc.raw, k, diff, want)
+				}
+			}
+		})
+	}
+}
+
+// The signal-specific variable replaces the generic one, per the OTLP spec.
+func TestUpstreamHeadersSignalOverride(t *testing.T) {
+	t.Setenv(headersEnv, "api-key=generic")
+	t.Setenv(tracesHeadersEnv, "x-tenant=traces")
+
+	traces, err := upstreamHeaders(tracesHeadersEnv)
+	if err != nil {
+		t.Fatalf("upstreamHeaders(traces): %v", err)
+	}
+	if got := traces.Get("api-key"); len(got) != 0 {
+		t.Errorf("traces api-key = %v, want absent; the signal-specific variable replaces the generic one", got)
+	}
+	if got := traces.Get("x-tenant"); len(got) != 1 || got[0] != "traces" {
+		t.Errorf("traces x-tenant = %v, want [traces]", got)
+	}
+
+	// metrics has no override, so it falls back to the generic variable.
+	metrics, err := upstreamHeaders(metricsHeadersEnv)
+	if err != nil {
+		t.Fatalf("upstreamHeaders(metrics): %v", err)
+	}
+	if got := metrics.Get("api-key"); len(got) != 1 || got[0] != "generic" {
+		t.Errorf("metrics api-key = %v, want [generic]", got)
+	}
+}
+
+// A header set atelet cannot parse fails startup rather than becoming a
+// per-export failure once the socket is already live.
+func TestNewServerRejectsUnparseableHeaders(t *testing.T) {
+	_, collector := startFakeCollector(t)
+	t.Setenv(endpointEnv, collector)
+	t.Setenv(headersEnv, "not-a-pair")
+
+	relay, err := NewServer(context.Background(), filepath.Join(t.TempDir(), "r.sock"))
+	if err == nil {
+		if relay != nil {
+			relay.Stop()
+		}
+		t.Fatal("NewServer accepted an unparseable OTEL_EXPORTER_OTLP_HEADERS")
+	}
+}
+
+func TestSocketPermissions(t *testing.T) {
+	_, collector := startFakeCollector(t)
+	sock := startRelay(t, collector)
+
+	info, err := os.Stat(sock)
+	if err != nil {
+		t.Fatalf("Stat(%q): %v", sock, err)
+	}
+	if perm := info.Mode().Perm(); perm != 0o600 {
+		t.Errorf("socket permissions = %04o, want 0600", perm)
+	}
+}

internal/serverboot/serverboot.go

@@ -32,6 +32,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/exporters/prometheus"
@@ -40,6 +41,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.40.0"
+	"google.golang.org/grpc"
 )
 
 // InitLogger sets the global slog logger to a JSON handler wrapped in
@@ -83,16 +85,18 @@ var serviceInstanceID = uuid.NewString()
 
 // newResource builds the resource shared by the tracer and meter providers.
 // WithFromEnv is last so OTEL_* env vars override the defaults.
-func newResource(ctx context.Context, serviceName string) (*resource.Resource, error) {
+func newResource(ctx context.Context, serviceName string, extraAttrs ...attribute.KeyValue) (*resource.Resource, error) {
+	attrs := []attribute.KeyValue{
+		semconv.ServiceName(serviceName),
+		semconv.ServiceInstanceID(serviceInstanceID),
+	}
+	attrs = append(attrs, extraAttrs...)
 	res, err := resource.New(ctx,
 		resource.WithTelemetrySDK(),
 		// Must track the schema version the SDK's own detectors emit, else the
 		// merge drops the schema URL with ErrSchemaURLConflict (tolerated below).
 		resource.WithSchemaURL(semconv.SchemaURL),
-		resource.WithAttributes(
-			semconv.ServiceName(serviceName),
-			semconv.ServiceInstanceID(serviceInstanceID),
-		),
+		resource.WithAttributes(attrs...),
 		resource.WithFromEnv(),
 	)
 	if errors.Is(err, resource.ErrPartialResource) || errors.Is(err, resource.ErrSchemaURLConflict) {
@@ -103,6 +107,38 @@ func newResource(ctx context.Context, serviceName string) (*resource.Resource, e
 	return res, nil
 }
 
+// relayAttrKey records which OTLP export path a signal took. It only makes
+// sense for the components that have a relay to take or miss — the ateoms —
+// so relayAttrs leaves it off everything else rather than labeling, say,
+// atecontroller "direct" for a relay it was never offered.
+//
+// The name is spelled here rather than taken from internal/ateattr on purpose:
+// serverboot is the bottom of the dependency graph (it imports one other
+// agent-substrate package) and every binary's main links it, while ateattr
+// pulls in pkg/api/v1alpha1, ateapipb, internal/resources and ateletpb. Move it
+// to ateattr if that cost ever drops, or if a second non-ate.* consumer appears.
+const relayAttrKey = "ate.otlp.relay"
+
+// relayAttrs describes the export path taken by a component that could have
+// used the relay. relayCapable false means the component never had one, and
+// gets no attribute at all; true means it did, and conn says whether it got it.
+//
+// The distinction matters because a nil conn on a relay-capable component is
+// exactly the degraded case worth alerting on: the ateom asked for the relay,
+// could not dial it, and is now exporting over the worker pod's own network.
+// Collapsing that into the same "no attribute" bucket as atecontroller would
+// hide it.
+func relayAttrs(relayCapable bool, conn *grpc.ClientConn) []attribute.KeyValue {
+	if !relayCapable {
+		return nil
+	}
+	status := "direct"
+	if conn != nil {
+		status = "relay"
+	}
+	return []attribute.KeyValue{attribute.String(relayAttrKey, status)}
+}
+
 // TracingOptions configures InitTracing.
 type TracingOptions struct {
 	// ServiceName is required; populates resource.semconv ServiceName.
@@ -111,6 +147,19 @@ type TracingOptions struct {
 	// OTEL_TRACES_SAMPLER / OTEL_TRACES_SAMPLER_ARG override the component
 	// default.
 	Sampling TraceSampling
+	// ExporterConn, when non-nil, is the connection the OTLP exporter sends
+	// over, instead of dialing OTEL_EXPORTER_OTLP_ENDPOINT itself. ateom passes
+	// the unix socket to atelet's relay (internal/otlprelay) so a worker pod
+	// exports without a network path of its own; nil keeps the direct dial.
+	//
+	// The caller owns the connection: the exporter's Shutdown does not close a
+	// connection it did not create.
+	ExporterConn *grpc.ClientConn
+	// RelayCapable marks a component that is meant to export through the relay,
+	// whether or not it managed to (see relayAttrs). Only the ateoms set it. It
+	// is separate from ExporterConn because a nil conn on its own cannot tell
+	// "the ateom tried and fell back" from "this component never had a relay".
+	RelayCapable bool
 }
 
 // InitTracing registers a global TracerProvider with the given options
@@ -122,7 +171,7 @@ func InitTracing(ctx context.Context, opts TracingOptions) (*sdktrace.TracerProv
 	if opts.Sampling.sampler == nil {
 		return nil, fmt.Errorf("TracingOptions.Sampling is required")
 	}
-	res, err := newResource(ctx, opts.ServiceName)
+	res, err := newResource(ctx, opts.ServiceName, relayAttrs(opts.RelayCapable, opts.ExporterConn)...)
 	if err != nil {
 		return nil, fmt.Errorf("create tracer resource: %w", err)
 	}
@@ -138,10 +187,16 @@ func InitTracing(ctx context.Context, opts TracingOptions) (*sdktrace.TracerProv
 		sdktrace.WithResource(res),
 		sdktrace.WithSampler(opts.Sampling.Sampler()),
 	}
-	exporter, err := otlptracegrpc.New(ctx,
+	expOpts := []otlptracegrpc.Option{
 		// GKE managed traces doesn't support validating the TLS certs of the collector.
 		otlptracegrpc.WithInsecure(),
-	)
+	}
+	if opts.ExporterConn != nil {
+		// WithGRPCConn takes precedence over endpoint/credential options, so
+		// WithInsecure above is inert on this path.
+		expOpts = append(expOpts, otlptracegrpc.WithGRPCConn(opts.ExporterConn))
+	}
+	exporter, err := otlptracegrpc.New(ctx, expOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("create OTLP exporter: %w", err)
 	}
@@ -166,7 +221,7 @@ func InitMetrics(ctx context.Context, serviceName string) (*sdkmetric.MeterProvi
 	if err != nil {
 		return nil, fmt.Errorf("create Prometheus metric exporter: %w", err)
 	}
-	return newMeterProvider(ctx, serviceName, nil, promExporter)
+	return newMeterProvider(ctx, serviceName, false, nil, nil, promExporter)
 }
 
 // InitMetricsPushOnly is InitMetrics without the Prometheus reader, for binaries
@@ -175,18 +230,45 @@ func InitMetrics(ctx context.Context, serviceName string) (*sdkmetric.MeterProvi
 // recorded outside the OTel SDK on the same push path; atecontroller bridges
 // controller-runtime's registry that way.
 func InitMetricsPushOnly(ctx context.Context, serviceName string, producers ...sdkmetric.Producer) (*sdkmetric.MeterProvider, error) {
-	return newMeterProvider(ctx, serviceName, producers)
+	return newMeterProvider(ctx, serviceName, false, nil, producers)
 }
 
-func newMeterProvider(ctx context.Context, serviceName string, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) {
+// InitMetricsPushOnlyVia is InitMetricsPushOnly with an explicit exporter
+// connection: the metrics counterpart of TracingOptions.ExporterConn. ateom
+// passes atelet's relay socket (internal/otlprelay) so the worker pod needs no
+// network path of its own; a nil conn keeps the direct dial to
+// OTEL_EXPORTER_OTLP_ENDPOINT.
+//
+// The caller owns the connection: the meter provider's Shutdown does not close
+// a connection it did not create.
+//
+// Calling this at all marks the component relay-capable, so its metrics carry
+// the relay attribute (see relayAttrs) either way — "relay" with a conn,
+// "direct" without one. It is the metrics counterpart of
+// TracingOptions.RelayCapable, implied rather than a parameter because only a
+// caller that has a relay to pass reaches for this function in the first place.
+func InitMetricsPushOnlyVia(ctx context.Context, serviceName string, conn *grpc.ClientConn, producers ...sdkmetric.Producer) (*sdkmetric.MeterProvider, error) {
+	return newMeterProvider(ctx, serviceName, true, conn, producers)
+}
+
+func newMeterProvider(ctx context.Context, serviceName string, relayCapable bool, conn *grpc.ClientConn, producers []sdkmetric.Producer, extraReaders ...sdkmetric.Reader) (*sdkmetric.MeterProvider, error) {
 	if serviceName == "" {
 		return nil, fmt.Errorf("serviceName is required")
 	}
-	otlpExporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithInsecure())
+	expOpts := []otlpmetricgrpc.Option{
+		// GKE managed metrics doesn't support validating the TLS certs of the collector.
+		otlpmetricgrpc.WithInsecure(),
+	}
+	if conn != nil {
+		// WithGRPCConn takes precedence over endpoint/credential options, so
+		// WithInsecure above is inert on this path.
+		expOpts = append(expOpts, otlpmetricgrpc.WithGRPCConn(conn))
+	}
+	otlpExporter, err := otlpmetricgrpc.New(ctx, expOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("create OTLP metric exporter: %w", err)
 	}
-	res, err := newResource(ctx, serviceName)
+	res, err := newResource(ctx, serviceName, relayAttrs(relayCapable, conn)...)
 	if err != nil {
 		return nil, fmt.Errorf("create metric resource: %w", err)
 	}

internal/serverboot/serverboot_test.go

@@ -25,8 +25,12 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.opentelemetry.io/otel/sdk/resource"
 	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 func resourceAttrs(res *resource.Resource) map[string]string {
@@ -67,6 +71,104 @@ func TestNewResourceEnvWins(t *testing.T) {
 	}
 }
 
+// lazyConn is a ClientConn that never dials: grpc.NewClient connects on first
+// RPC, and relayAttrs only cares whether the pointer is nil.
+func lazyConn(t *testing.T) *grpc.ClientConn {
+	t.Helper()
+	conn, err := grpc.NewClient("passthrough:///unused", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	return conn
+}
+
+// The three cases relayAttrs exists to separate: a component that got the
+// relay, one that wanted it and fell back, and one that was never offered one.
+// The last must carry no attribute at all rather than a misleading "direct".
+func TestRelayAttrs(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		relayCapable bool
+		conn         bool
+		want         string // "" means the attribute must be absent
+	}{
+		{name: "relay capable with conn", relayCapable: true, conn: true, want: "relay"},
+		{name: "relay capable fell back", relayCapable: true, conn: false, want: "direct"},
+		{name: "not relay capable", relayCapable: false, conn: false},
+		// atecontroller stays unlabeled even if some future caller hands it a
+		// connection for another reason: capability, not the conn, is the gate.
+		{name: "not relay capable with conn", relayCapable: false, conn: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var conn *grpc.ClientConn
+			if tc.conn {
+				conn = lazyConn(t)
+			}
+			res, err := newResource(context.Background(), "ateom-gvisor", relayAttrs(tc.relayCapable, conn)...)
+			if err != nil {
+				t.Fatalf("newResource: %v", err)
+			}
+			got, ok := resourceAttrs(res)[relayAttrKey]
+			if tc.want == "" {
+				if ok {
+					t.Errorf("%s = %q, want absent", relayAttrKey, got)
+				}
+				return
+			}
+			if got != tc.want {
+				t.Errorf("%s = %q, want %q", relayAttrKey, got, tc.want)
+			}
+		})
+	}
+}
+
+// collectedResource reads back the resource a meter provider actually stamps on
+// its exports, by attaching a ManualReader alongside the OTLP one. The provider
+// does not expose its resource any other way, and asserting on newResource's
+// return would only re-test relayAttrs.
+func collectedResource(t *testing.T, relayCapable bool, conn *grpc.ClientConn) map[string]string {
+	t.Helper()
+	reader := sdkmetric.NewManualReader()
+	mp, err := newMeterProvider(context.Background(), "ateom-gvisor", relayCapable, conn, nil, reader)
+	if err != nil {
+		t.Fatalf("newMeterProvider: %v", err)
+	}
+	t.Cleanup(func() {
+		// Shutdown flushes the OTLP reader too, and no collector is listening
+		// here: a live context spends the exporter's full retry budget (~10s)
+		// per provider. A cancelled one skips the flush, which is all this test
+		// wants from Shutdown anyway.
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		mp.Shutdown(ctx)
+	})
+	// A meter with no instruments still collects, carrying the resource.
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("Collect: %v", err)
+	}
+	return resourceAttrs(rm.Resource)
+}
+
+// The metric path's half of the decision, asserted on what the provider exports
+// rather than on what relayAttrs returns: a wiring mistake in newMeterProvider
+// (passing the wrong flag, dropping the attrs) fails here and not in
+// TestRelayAttrs.
+func TestMeterProviderRelayAttribute(t *testing.T) {
+	if got, ok := collectedResource(t, true, lazyConn(t))[relayAttrKey]; !ok || got != "relay" {
+		t.Errorf("%s = %q (present %t), want relay", relayAttrKey, got, ok)
+	}
+	if got, ok := collectedResource(t, true, nil)[relayAttrKey]; !ok || got != "direct" {
+		t.Errorf("%s = %q (present %t), want direct", relayAttrKey, got, ok)
+	}
+	// atecontroller and ateapi: no relay was ever offered, so no claim is made
+	// about which path they took.
+	if got, ok := collectedResource(t, false, nil)[relayAttrKey]; ok {
+		t.Errorf("%s = %q, want absent for a component with no relay", relayAttrKey, got)
+	}
+}
+
 func TestReadyzDrainsWhileHealthzStaysUp(t *testing.T) {
 	readiness := &Readiness{}
 	mux := metricsMux(MetricsServerOptions{