Send instance name as x-api-key to the OTLP relay

server/cmd/api/main.go

@@ -140,10 +140,11 @@ func main() {
 	var otlpExporter api.OTLPExporter
 	var otlpMetrics *events.OTLPMetrics
 	if config.OTLPEndpoint != "" {
-		headers := map[string]string{}
-		if config.InstanceJWT != "" {
-			headers["Authorization"] = "Bearer " + config.InstanceJWT
-		}
+		// The relay authenticates the VM by its instance JWT, sent as a bearer
+		// token. The credential is resolved per request: on a forked VM the boot
+		// env JWT is stale, so it is re-read from the applied fork-identity
+		// payload (see instanceJWTProvider).
+		jwtProvider := newInstanceJWTProvider(config.InstanceJWT)
 		slogger.Info("OTLP export available", "endpoint", config.OTLPEndpoint, "path", config.OTLPPath)
 		// The OTel log SDK reports batch-queue drops through its global logger at
 		// logr V(1), which slog renders just below Info, so a default Info handler
@@ -156,7 +157,7 @@ func main() {
 			Endpoint:       config.OTLPEndpoint,
 			URLPath:        config.OTLPPath,
 			Insecure:       config.OTLPInsecure,
-			Headers:        headers,
+			AuthTokenFunc:  jwtProvider.Token,
 			ServiceName:    config.OTLPServiceName,
 			InstanceName:   config.InstanceName,
 			Metro:          config.MetroName,

server/cmd/api/otlp_credential.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"sync"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/forkidentity"
+)
+
+// instanceJWTCacheTTL bounds how often the provider re-reads the applied
+// fork-identity payload from disk. Export requests are frequent; the JWT changes
+// at most once per fork, so a few seconds of staleness is harmless and avoids a
+// filesystem read on every request.
+const instanceJWTCacheTTL = 3 * time.Second
+
+// instanceJWTProvider resolves the OTLP relay bearer credential per request.
+//
+// In fork-identity-wait mode the platform delivers a fresh KERNEL_INSTANCE_JWT
+// in the applied identity payload after kernel-images-api has already started,
+// and the process is not restarted, so the boot-time env JWT is stale (empty on
+// an unbound fork). This provider reads the JWT from the applied payload instead,
+// falling back to the boot env otherwise. Outside fork mode it returns the boot
+// JWT unchanged.
+type instanceJWTProvider struct {
+	bootJWT     string
+	forkEnabled bool
+
+	mu       sync.Mutex
+	cached   string
+	cachedAt time.Time
+}
+
+func newInstanceJWTProvider(bootJWT string) *instanceJWTProvider {
+	enabled, _ := forkidentity.WaitEnabled()
+	return &instanceJWTProvider{bootJWT: bootJWT, forkEnabled: enabled}
+}
+
+// Token returns the current bearer credential, or "" when none is available yet
+// (the caller then sends no Authorization header rather than an empty bearer).
+func (p *instanceJWTProvider) Token() string {
+	if !p.forkEnabled {
+		return p.bootJWT
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if !p.cachedAt.IsZero() && time.Since(p.cachedAt) < instanceJWTCacheTTL {
+		return p.cached
+	}
+	p.cached = p.resolve()
+	p.cachedAt = time.Now()
+	return p.cached
+}
+
+// resolve reads the applied fork-identity payload for the JWT. It only reads once
+// the applied marker is present, to avoid a torn read while the wrapper is
+// mid-apply, and falls back to the boot JWT on any miss.
+func (p *instanceJWTProvider) resolve() string {
+	applied, err := forkidentity.ReadAppliedMarker()
+	if err != nil || applied == "" {
+		return p.bootJWT
+	}
+	payload, err := forkidentity.ReadPayload()
+	if err != nil {
+		return p.bootJWT
+	}
+	if jwt := payload.Get("kernel_instance_jwt"); jwt != "" {
+		return jwt
+	}
+	return p.bootJWT
+}

server/cmd/api/otlp_credential_test.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/kernel/kernel-images/server/lib/forkidentity"
+)
+
+// withForkIdentityFiles points the forkidentity package vars at temp paths for
+// the duration of a test and restores them after.
+func withForkIdentityFiles(t *testing.T) {
+	t.Helper()
+	dir := t.TempDir()
+	payload, applied := forkidentity.PayloadFile, forkidentity.AppliedFile
+	forkidentity.PayloadFile = filepath.Join(dir, "fork-identity.json")
+	forkidentity.AppliedFile = filepath.Join(dir, "fork-identity-applied")
+	t.Cleanup(func() {
+		forkidentity.PayloadFile = payload
+		forkidentity.AppliedFile = applied
+	})
+}
+
+func writeAppliedPayload(t *testing.T, jwt string) {
+	t.Helper()
+	p := forkidentity.Payload{
+		"instance_name":       "browser-fork-1",
+		"session_intel_url":   "https://intel.example",
+		"kernel_instance_jwt": jwt,
+	}
+	if err := forkidentity.WritePayload(p); err != nil {
+		t.Fatalf("write payload: %v", err)
+	}
+	if err := forkidentity.WriteAppliedMarker(p.InstanceName()); err != nil {
+		t.Fatalf("write applied marker: %v", err)
+	}
+}
+
+func TestInstanceJWTProvider(t *testing.T) {
+	t.Run("non-fork returns boot jwt", func(t *testing.T) {
+		withForkIdentityFiles(t)
+		t.Setenv(forkidentity.WaitEnv, "")
+		// Even with an applied payload present, non-fork mode ignores it.
+		writeAppliedPayload(t, "payload-jwt")
+		p := newInstanceJWTProvider("boot-jwt")
+		if got := p.Token(); got != "boot-jwt" {
+			t.Errorf("Token() = %q, want boot-jwt", got)
+		}
+	})
+
+	t.Run("fork before apply falls back to boot jwt", func(t *testing.T) {
+		withForkIdentityFiles(t)
+		t.Setenv(forkidentity.WaitEnv, "true")
+		// No payload/applied marker written yet: mid-apply, must not read a torn value.
+		p := newInstanceJWTProvider("boot-jwt")
+		if got := p.Token(); got != "boot-jwt" {
+			t.Errorf("Token() = %q, want boot-jwt", got)
+		}
+	})
+
+	t.Run("fork after apply returns fresh payload jwt", func(t *testing.T) {
+		withForkIdentityFiles(t)
+		t.Setenv(forkidentity.WaitEnv, "true")
+		writeAppliedPayload(t, "fresh-jwt")
+		// Boot jwt is empty, mirroring an unbound fork; the fresh credential must
+		// come from the applied payload.
+		p := newInstanceJWTProvider("")
+		if got := p.Token(); got != "fresh-jwt" {
+			t.Errorf("Token() = %q, want fresh-jwt", got)
+		}
+	})
+}

server/lib/events/otlpstorage.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"net/http"
 	"sync"
 	"time"
 
@@ -43,8 +44,14 @@ type OTLPConfig struct {
 	URLPath string
 	// Insecure sends over plaintext HTTP. Development only.
 	Insecure bool
-	// Headers are attached to every export request (e.g. the instance JWT).
+	// Headers are attached to every export request.
 	Headers map[string]string
+	// AuthTokenFunc, when set, is called on every export request to resolve the
+	// bearer credential, set as "Authorization: Bearer <token>". It is read per
+	// request (not captured once) so a credential that changes after start, e.g.
+	// the instance JWT refreshed from an applied fork-identity payload, takes
+	// effect without restarting the process. An empty return sends no header.
+	AuthTokenFunc func() string
 
 	// ServiceName, InstanceName, and Metro populate the OTLP Resource.
 	ServiceName  string
@@ -152,6 +159,25 @@ type otlpStorage struct {
 	logger   log.Logger
 }
 
+// bearerRoundTripper sets the Authorization header from token() on each request,
+// so a credential that changes after the exporter is built (the fork-refreshed
+// instance JWT) is picked up per request rather than frozen at construction.
+type bearerRoundTripper struct {
+	base  http.RoundTripper
+	token func() string
+}
+
+func (t *bearerRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	tok := t.token()
+	if tok == "" {
+		return t.base.RoundTrip(req)
+	}
+	// RoundTrip must not mutate the caller's request; clone before setting.
+	req = req.Clone(req.Context())
+	req.Header.Set("Authorization", "Bearer "+tok)
+	return t.base.RoundTrip(req)
+}
+
 func newOTLPStorage(ctx context.Context, cfg OTLPConfig, log *slog.Logger) (*otlpStorage, error) {
 	if cfg.Endpoint == "" {
 		return nil, fmt.Errorf("otlp storage: endpoint is required")
@@ -167,6 +193,11 @@ func newOTLPStorage(ctx context.Context, cfg OTLPConfig, log *slog.Logger) (*otl
 	if len(cfg.Headers) > 0 {
 		opts = append(opts, otlploghttp.WithHeaders(cfg.Headers))
 	}
+	if cfg.AuthTokenFunc != nil {
+		opts = append(opts, otlploghttp.WithHTTPClient(&http.Client{
+			Transport: &bearerRoundTripper{base: http.DefaultTransport, token: cfg.AuthTokenFunc},
+		}))
+	}
 
 	exporter, err := otlploghttp.New(ctx, opts...)
 	if err != nil {

server/lib/events/otlpstorage_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -89,13 +90,13 @@ func TestOTLPStorageWriter_ExportsEvents(t *testing.T) {
 	require.NoError(t, err)
 
 	cfg := OTLPConfig{
-		Endpoint:     strings.TrimPrefix(srv.URL, "http://"),
-		URLPath:      "/otlp-relay/v1/logs",
-		Insecure:     true,
-		Headers:      map[string]string{"Authorization": "Bearer test-jwt"},
-		ServiceName:  "kernel-browser",
-		InstanceName: "browser-1",
-		Metro:        "dev-iad",
+		Endpoint:      strings.TrimPrefix(srv.URL, "http://"),
+		URLPath:       "/otlp-relay/v1/logs",
+		Insecure:      true,
+		AuthTokenFunc: func() string { return "test-jwt" },
+		ServiceName:   "kernel-browser",
+		InstanceName:  "browser-1",
+		Metro:         "dev-iad",
 	}
 	wtr := NewOTLPStorageWriter(es, cfg, slog.Default())
 
@@ -115,6 +116,7 @@ func TestOTLPStorageWriter_ExportsEvents(t *testing.T) {
 	defer mu.Unlock()
 	require.NotEmpty(t, paths, "expected at least one export request")
 	assert.Equal(t, "/otlp-relay/v1/logs", paths[0])
+	// The relay authenticates the VM by its instance JWT, sent as a bearer token.
 	assert.Equal(t, "Bearer test-jwt", auths[0])
 	// The excluded screenshot must not reach the receiver; only the network event does.
 	assert.Equal(t, []string{"network_response"}, eventNames)
@@ -239,3 +241,64 @@ func TestChunkBySize(t *testing.T) {
 		assert.Len(t, chunks[0], 1)
 	})
 }
+
+// TestOTLPStorageWriter_RefreshesAuthToken confirms AuthTokenFunc is read per
+// request, not captured once: a token that changes after the exporter starts is
+// reflected on later requests. This is what lets a forked VM export with the
+// fresh instance JWT from its applied identity payload without a restart.
+func TestOTLPStorageWriter_RefreshesAuthToken(t *testing.T) {
+	var mu sync.Mutex
+	var auths []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		auths = append(auths, r.Header.Get("Authorization"))
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	var token atomic.Value
+	token.Store("jwt-1")
+
+	es, err := NewEventStream(EventStreamConfig{RingCapacity: 64})
+	require.NoError(t, err)
+
+	cfg := OTLPConfig{
+		Endpoint:       strings.TrimPrefix(srv.URL, "http://"),
+		URLPath:        "/otlp-relay/v1/logs",
+		Insecure:       true,
+		AuthTokenFunc:  func() string { return token.Load().(string) },
+		ServiceName:    "kernel-browser",
+		ExportInterval: 20 * time.Millisecond,
+	}
+	wtr := NewOTLPStorageWriter(es, cfg, slog.Default())
+	ctx, cancel := context.WithCancel(context.Background())
+	require.NoError(t, wtr.Start(ctx))
+
+	seen := func(want string) func() bool {
+		return func() bool {
+			mu.Lock()
+			defer mu.Unlock()
+			for _, a := range auths {
+				if a == want {
+					return true
+				}
+			}
+			return false
+		}
+	}
+
+	es.Publish(Envelope{Event: Event{Ts: 1, Type: "network_response", Category: Network,
+		Data: []byte(`{"method":"GET","url":"https://x","status":200}`)}})
+	require.Eventually(t, seen("Bearer jwt-1"), 3*time.Second, 10*time.Millisecond, "first export should use jwt-1")
+
+	token.Store("jwt-2")
+	es.Publish(Envelope{Event: Event{Ts: 2, Type: "network_response", Category: Network,
+		Data: []byte(`{"method":"GET","url":"https://y","status":200}`)}})
+	require.Eventually(t, seen("Bearer jwt-2"), 3*time.Second, 10*time.Millisecond, "later export should pick up the refreshed jwt-2")
+
+	cancel()
+	stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer stopCancel()
+	require.NoError(t, wtr.Stop(stopCtx))
+}