diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index 19a62089..9d0bc2f4 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -140,10 +140,15 @@ 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. Identity (JWT + resource attrs) is resolved dynamically: on a + // forked VM the boot env is stale, so it is re-read from the applied + // fork-identity payload (see otlpIdentityProvider). + identity := newOTLPIdentityProvider(otlpIdentity{ + jwt: config.InstanceJWT, + instanceName: config.InstanceName, + metro: config.MetroName, + }, slogger) 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 @@ -153,17 +158,17 @@ func main() { otelDiag := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo - 1}) otel.SetLogger(logr.FromSlogHandler(events.NewDropCountingHandler(otelDiag, otlpMetrics))) otlpExporter = events.NewOTLPExportController(eventStream, events.OTLPConfig{ - Endpoint: config.OTLPEndpoint, - URLPath: config.OTLPPath, - Insecure: config.OTLPInsecure, - Headers: headers, - ServiceName: config.OTLPServiceName, - InstanceName: config.InstanceName, - Metro: config.MetroName, - MaxQueueSize: config.OTLPMaxQueueSize, - ExportInterval: config.OTLPExportInterval, - ExportTimeout: config.OTLPExportTimeout, - Metrics: otlpMetrics, + Endpoint: config.OTLPEndpoint, + URLPath: config.OTLPPath, + Insecure: config.OTLPInsecure, + AuthTokenFunc: identity.Token, + ServiceName: config.OTLPServiceName, + InstanceNameFunc: identity.InstanceName, + MetroFunc: identity.Metro, + MaxQueueSize: config.OTLPMaxQueueSize, + ExportInterval: config.OTLPExportInterval, + ExportTimeout: config.OTLPExportTimeout, + Metrics: otlpMetrics, }, slogger) } diff --git a/server/cmd/api/otlp_credential.go b/server/cmd/api/otlp_credential.go new file mode 100644 index 00000000..f58b1708 --- /dev/null +++ b/server/cmd/api/otlp_credential.go @@ -0,0 +1,97 @@ +package main + +import ( + "log/slog" + "sync" + "time" + + "github.com/kernel/kernel-images/server/lib/forkidentity" +) + +// otlpIdentityCacheTTL bounds how often the provider re-reads the applied +// fork-identity payload from disk. Export requests are frequent; the identity +// changes at most once per fork, so a few seconds of staleness is harmless and +// avoids a filesystem read on every request. Only a payload-sourced identity is +// cached; a boot fallback during the pre-apply window is returned uncached so +// the fresh identity is picked up on the very next request. +const otlpIdentityCacheTTL = 3 * time.Second + +// otlpIdentity is the platform identity stamped onto OTLP export: the relay +// bearer credential plus the resource attributes. +type otlpIdentity struct { + jwt string + instanceName string + metro string +} + +// otlpIdentityProvider resolves the OTLP identity per use. +// +// In fork-identity-wait mode the platform delivers a fresh KERNEL_INSTANCE_JWT +// (and INST_NAME/METRO_NAME) in the applied identity payload after +// kernel-images-api has already started, and the process is not restarted, so +// the boot-time env values are stale (empty on an unbound fork). This provider +// reads the identity from the applied payload instead, falling back to the boot +// values. Outside fork mode it returns the boot identity unchanged. +type otlpIdentityProvider struct { + boot otlpIdentity + forkEnabled bool + + mu sync.Mutex + cached otlpIdentity + cachedOK bool + cachedAt time.Time +} + +func newOTLPIdentityProvider(boot otlpIdentity, log *slog.Logger) *otlpIdentityProvider { + enabled, err := forkidentity.WaitEnabled() + if err != nil { + // Malformed flag: treat as disabled but surface it, otherwise a fork VM + // would silently keep using the stale boot identity. + log.Warn("fork-identity wait flag invalid; treating as disabled", "err", err) + } + return &otlpIdentityProvider{boot: boot, forkEnabled: enabled} +} + +func (p *otlpIdentityProvider) Token() string { return p.current().jwt } +func (p *otlpIdentityProvider) InstanceName() string { return p.current().instanceName } +func (p *otlpIdentityProvider) Metro() string { return p.current().metro } + +// current returns the identity, caching only a payload-sourced (or non-fork, +// stable) result so a boot fallback during the pre-apply window is retried on +// the next call rather than pinned for the TTL. +func (p *otlpIdentityProvider) current() otlpIdentity { + p.mu.Lock() + defer p.mu.Unlock() + if p.cachedOK && time.Since(p.cachedAt) < otlpIdentityCacheTTL { + return p.cached + } + id, ok := p.resolve() + if ok { + p.cached, p.cachedOK, p.cachedAt = id, true, time.Now() + } + return id +} + +// resolve reads the applied fork-identity payload. The bool reports whether the +// result is stable/cacheable: true for the boot identity outside fork mode or a +// successfully read payload, false for a boot fallback while a fork is mid-apply. +func (p *otlpIdentityProvider) resolve() (otlpIdentity, bool) { + if !p.forkEnabled { + return p.boot, true + } + // Only read once the applied marker is present, to avoid a torn read while + // the wrapper is mid-apply. + applied, err := forkidentity.ReadAppliedMarker() + if err != nil || applied == "" { + return p.boot, false + } + payload, err := forkidentity.ReadPayload() + if err != nil { + return p.boot, false + } + return otlpIdentity{ + jwt: forkidentity.FirstNonEmpty(payload.Get("kernel_instance_jwt"), p.boot.jwt), + instanceName: forkidentity.FirstNonEmpty(payload.InstanceName(), p.boot.instanceName), + metro: forkidentity.FirstNonEmpty(payload.Get("metro_name"), p.boot.metro), + }, true +} diff --git a/server/cmd/api/otlp_credential_test.go b/server/cmd/api/otlp_credential_test.go new file mode 100644 index 00000000..273de9dc --- /dev/null +++ b/server/cmd/api/otlp_credential_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "log/slog" + "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, instanceName, metro string) { + t.Helper() + p := forkidentity.Payload{ + "instance_name": instanceName, + "metro_name": metro, + "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 bootIdentity() otlpIdentity { + return otlpIdentity{jwt: "boot-jwt", instanceName: "boot-inst", metro: "boot-metro"} +} + +func TestOTLPIdentityProvider(t *testing.T) { + t.Run("non-fork returns boot identity", func(t *testing.T) { + withForkIdentityFiles(t) + t.Setenv(forkidentity.WaitEnv, "") + writeAppliedPayload(t, "payload-jwt", "payload-inst", "payload-metro") // ignored in non-fork + p := newOTLPIdentityProvider(bootIdentity(), slog.Default()) + if p.Token() != "boot-jwt" || p.InstanceName() != "boot-inst" || p.Metro() != "boot-metro" { + t.Errorf("got (%q,%q,%q), want boot values", p.Token(), p.InstanceName(), p.Metro()) + } + }) + + t.Run("fork before apply falls back to boot", func(t *testing.T) { + withForkIdentityFiles(t) + t.Setenv(forkidentity.WaitEnv, "true") + p := newOTLPIdentityProvider(bootIdentity(), slog.Default()) + if p.Token() != "boot-jwt" || p.InstanceName() != "boot-inst" { + t.Errorf("got (%q,%q), want boot fallback", p.Token(), p.InstanceName()) + } + }) + + t.Run("fork after apply returns fresh payload identity", func(t *testing.T) { + withForkIdentityFiles(t) + t.Setenv(forkidentity.WaitEnv, "true") + // Empty boot values mirror an unbound fork; the fresh identity comes from + // the applied payload. + p := newOTLPIdentityProvider(otlpIdentity{}, slog.Default()) + writeAppliedPayload(t, "fresh-jwt", "fresh-inst", "fresh-metro") + if p.Token() != "fresh-jwt" || p.InstanceName() != "fresh-inst" || p.Metro() != "fresh-metro" { + t.Errorf("got (%q,%q,%q), want fresh payload values", p.Token(), p.InstanceName(), p.Metro()) + } + }) + + t.Run("boot fallback is not cached; apply is picked up next call", func(t *testing.T) { + withForkIdentityFiles(t) + t.Setenv(forkidentity.WaitEnv, "true") + p := newOTLPIdentityProvider(bootIdentity(), slog.Default()) + // Pre-apply: boot fallback, and must not be cached. + if got := p.Token(); got != "boot-jwt" { + t.Fatalf("pre-apply Token() = %q, want boot-jwt", got) + } + // Apply lands; the very next call must reflect it despite the cache TTL. + writeAppliedPayload(t, "applied-jwt", "applied-inst", "applied-metro") + if got := p.Token(); got != "applied-jwt" { + t.Errorf("post-apply Token() = %q, want applied-jwt (boot fallback must not be cached)", got) + } + }) +} diff --git a/server/e2e/e2e_otlp_storage_test.go b/server/e2e/e2e_otlp_storage_test.go new file mode 100644 index 00000000..afd703cb --- /dev/null +++ b/server/e2e/e2e_otlp_storage_test.go @@ -0,0 +1,253 @@ +package e2e + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os/exec" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + "google.golang.org/protobuf/proto" + + instanceoapi "github.com/kernel/kernel-images/server/lib/oapi" +) + +// otlpMockCollector is a minimal OTLP/HTTP logs receiver served on the test host +// (0.0.0.0 so a container can reach it via host.docker.internal). It records the +// Authorization header, resource attributes, and event names of every export. +type otlpMockCollector struct { + srv *http.Server + ln net.Listener + + mu sync.Mutex + auths []string + instanceName []string + metro []string + serviceName []string + eventNames []string +} + +func startOTLPMockCollector(t *testing.T) *otlpMockCollector { + t.Helper() + ln, err := net.Listen("tcp", "0.0.0.0:0") + require.NoError(t, err, "listen for mock collector") + c := &otlpMockCollector{ln: ln} + mux := http.NewServeMux() + mux.HandleFunc("/otlp-relay/v1/logs", c.handle) + c.srv = &http.Server{Handler: mux} + go func() { _ = c.srv.Serve(ln) }() + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = c.srv.Shutdown(ctx) + }) + return c +} + +func (c *otlpMockCollector) handle(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + var req collogspb.ExportLogsServiceRequest + if err := proto.Unmarshal(body, &req); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + c.mu.Lock() + c.auths = append(c.auths, r.Header.Get("Authorization")) + for _, rl := range req.ResourceLogs { + for _, attr := range rl.Resource.GetAttributes() { + switch attr.Key { + case "kernel.instance_name": + c.instanceName = append(c.instanceName, attr.Value.GetStringValue()) + case "kernel.metro": + c.metro = append(c.metro, attr.Value.GetStringValue()) + case "service.name": + c.serviceName = append(c.serviceName, attr.Value.GetStringValue()) + } + } + for _, sl := range rl.ScopeLogs { + for _, lr := range sl.LogRecords { + c.eventNames = append(c.eventNames, lr.EventName) + } + } + } + c.mu.Unlock() + w.WriteHeader(http.StatusOK) +} + +// hostEndpoint returns host:port the container uses to reach the collector. +func (c *otlpMockCollector) hostEndpoint() string { + return fmt.Sprintf("host.docker.internal:%d", c.ln.Addr().(*net.TCPAddr).Port) +} + +func (c *otlpMockCollector) snapshot() (auths, instanceName, eventNames []string) { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.auths...), append([]string(nil), c.instanceName...), append([]string(nil), c.eventNames...) +} + +// enableControlExport turns on the control category (so api_call events flow) +// and OTLP export, then makes a few API calls to generate exportable events. +func enableControlExport(t *testing.T, ctx context.Context, client *instanceoapi.ClientWithResponses) { + t.Helper() + tr := true + resp, err := client.PutTelemetryWithResponse(ctx, instanceoapi.PutTelemetryJSONRequestBody{ + Browser: &instanceoapi.BrowserTelemetryCategoriesConfig{ + Control: &instanceoapi.BrowserTelemetryCategoryConfig{Enabled: &tr}, + }, + Export: &instanceoapi.BrowserTelemetryExportConfig{ + Otlp: &instanceoapi.BrowserTelemetryOTLPExportConfig{Enabled: &tr}, + }, + }) + require.NoError(t, err) + require.Equal(t, http.StatusCreated, resp.StatusCode(), "put telemetry: %s", string(resp.Body)) + // Each API call emits an api_call (control) event, which OTLP exports. + for i := 0; i < 3; i++ { + _, _ = client.GetTelemetryWithResponse(ctx) + time.Sleep(50 * time.Millisecond) + } +} + +// TestOTLPExport (Tier 2) runs a headless container pointed at a mock OTLP +// collector, enables export via the telemetry API, and verifies exported records +// land carrying the instance JWT as a bearer token plus the expected resource +// attributes. Skips when docker is unavailable. +func TestOTLPExport(t *testing.T) { + if _, err := exec.LookPath("docker"); err != nil { + t.Skipf("docker not available: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + collector := startOTLPMockCollector(t) + + c := NewTestContainer(t, headlessImage) + require.NoError(t, c.Start(ctx, ContainerConfig{ + HostAccess: true, + Env: map[string]string{ + "BTEL_OTLP_ENDPOINT": collector.hostEndpoint(), + "BTEL_OTLP_PATH": "/otlp-relay/v1/logs", + "BTEL_OTLP_INSECURE": "true", + "INST_NAME": "browser-e2e-1", + "METRO_NAME": "dev-iad", + "KERNEL_INSTANCE_JWT": "e2e-jwt-token", + }, + }), "failed to start container") + defer c.Stop(ctx) + require.NoError(t, c.WaitReady(ctx), "api not ready") + + client, err := c.APIClient() + require.NoError(t, err) + enableControlExport(t, ctx, client) + + require.Eventually(t, func() bool { + _, _, events := collector.snapshot() + return len(events) > 0 + }, 20*time.Second, 250*time.Millisecond, "collector received no exported records") + + auths, instanceName, eventNames := collector.snapshot() + t.Logf("collector saw %d record(s); auths=%v instance=%v events=%v", len(eventNames), auths, instanceName, eventNames) + require.NotEmpty(t, auths) + assert.Equal(t, "Bearer e2e-jwt-token", auths[0], "exporter must send the instance JWT as a bearer token") + assert.Contains(t, instanceName, "browser-e2e-1", "resource should carry the instance name") + assert.Contains(t, eventNames, "api_call", "control-category api_call events should be exported") +} + +// TestOTLPExportForkIdentityRefresh (Tier 3) starts a container in +// fork-identity-wait mode with an empty boot JWT, applies a fork identity +// carrying a fresh KERNEL_INSTANCE_JWT via the real /internal/fork-identity +// endpoint, and verifies exports pick up that fresh JWT without a restart. +func TestOTLPExportForkIdentityRefresh(t *testing.T) { + if _, err := exec.LookPath("docker"); err != nil { + t.Skipf("docker not available: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + + collector := startOTLPMockCollector(t) + + c := NewTestContainer(t, headlessImage) + require.NoError(t, c.Start(ctx, ContainerConfig{ + HostAccess: true, + Env: map[string]string{ + "KERNEL_FORK_IDENTITY_WAIT": "true", + "BTEL_OTLP_ENDPOINT": collector.hostEndpoint(), + "BTEL_OTLP_PATH": "/otlp-relay/v1/logs", + "BTEL_OTLP_INSECURE": "true", + "KERNEL_INSTANCE_JWT": "", // unbound fork: no boot credential + "METRO_NAME": "dev-iad", + }, + }), "failed to start container") + defer c.Stop(ctx) + require.NoError(t, c.WaitReady(ctx), "api not ready") + + // Apply the fork identity with a fresh JWT via the real internal endpoint. + // The wrapper clears any pre-existing payload once when it first enters its + // wait loop, so a POST that lands before that point gets wiped. Retry with a + // bounded per-attempt timeout until the apply sticks; 409 means an earlier + // attempt already applied the (identical) payload. + payload := map[string]string{ + "instance_name": "browser-fork-e2e", + "session_intel_url": "https://intel.example", + "kernel_instance_jwt": "fresh-fork-jwt", + } + buf, err := json.Marshal(payload) + require.NoError(t, err) + applied := false + for deadline := time.Now().Add(60 * time.Second); time.Now().Before(deadline); { + attemptCtx, acancel := context.WithTimeout(ctx, 5*time.Second) + req, reqErr := http.NewRequestWithContext(attemptCtx, http.MethodPost, c.APIBaseURL()+"/internal/fork-identity", bytes.NewReader(buf)) + require.NoError(t, reqErr) + req.Header.Set("Content-Type", "application/json") + resp, doErr := http.DefaultClient.Do(req) + if doErr != nil { + acancel() + continue + } + respBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + acancel() + if resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusConflict { + applied = true + break + } + t.Logf("fork-identity apply not ready yet: %d %s", resp.StatusCode, string(respBody)) + time.Sleep(500 * time.Millisecond) + } + require.True(t, applied, "fork identity never applied") + + client, err := c.APIClient() + require.NoError(t, err) + enableControlExport(t, ctx, client) + + require.Eventually(t, func() bool { + auths, _, _ := collector.snapshot() + for _, a := range auths { + if a == "Bearer fresh-fork-jwt" { + return true + } + } + return false + }, 30*time.Second, 250*time.Millisecond, "collector never saw the refreshed fork JWT") + + auths, instanceName, _ := collector.snapshot() + t.Logf("collector auths=%v instance=%v", auths, instanceName) + for _, a := range auths { + assert.Contains(t, []string{"", "Bearer fresh-fork-jwt"}, a, "only empty or the fresh fork JWT expected") + } + // The resource identity is also refreshed from the applied payload, not left + // at the empty boot INST_NAME. + assert.Contains(t, instanceName, "browser-fork-e2e", "resource instance_name should come from the applied fork payload") +} diff --git a/server/lib/events/otlpstorage.go b/server/lib/events/otlpstorage.go index 3e7f18a3..7df5875d 100644 --- a/server/lib/events/otlpstorage.go +++ b/server/lib/events/otlpstorage.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "net/http" "sync" "time" @@ -43,13 +44,27 @@ 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 ". 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 InstanceName string Metro string + // InstanceNameFunc and MetroFunc, when set, resolve the resource identity at + // exporter-build time and take precedence over the static InstanceName/Metro. + // Like AuthTokenFunc, this lets a forked VM stamp the fresh identity from its + // applied fork-identity payload instead of stale boot env. Since export is + // started per session after identity applies, build time sees the applied + // values. + InstanceNameFunc func() string + MetroFunc func() string // Batch tuning. Zero values fall back to the SDK defaults. MaxQueueSize int @@ -152,6 +167,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,18 +201,31 @@ 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 { return nil, fmt.Errorf("otlp storage: new exporter: %w", err) } + instanceName := cfg.InstanceName + if cfg.InstanceNameFunc != nil { + instanceName = cfg.InstanceNameFunc() + } + metro := cfg.Metro + if cfg.MetroFunc != nil { + metro = cfg.MetroFunc() + } attrs := []attribute.KeyValue{semconv.ServiceName(cfg.ServiceName)} - if cfg.InstanceName != "" { - attrs = append(attrs, attribute.String("kernel.instance_name", cfg.InstanceName)) + if instanceName != "" { + attrs = append(attrs, attribute.String("kernel.instance_name", instanceName)) } - if cfg.Metro != "" { - attrs = append(attrs, attribute.String("kernel.metro", cfg.Metro)) + if metro != "" { + attrs = append(attrs, attribute.String("kernel.metro", metro)) } res := resource.NewSchemaless(attrs...) diff --git a/server/lib/events/otlpstorage_test.go b/server/lib/events/otlpstorage_test.go index 50eecc8e..be1644de 100644 --- a/server/lib/events/otlpstorage_test.go +++ b/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)) +}