diff --git a/server/README.md b/server/README.md index 1f06699f..0399f0c8 100644 --- a/server/README.md +++ b/server/README.md @@ -58,7 +58,13 @@ Configure the server using environment variables: | `OUTPUT_DIR` | `.` | Directory to save recordings | | `FFMPEG_PATH` | `ffmpeg` | Path to the ffmpeg binary | | `KERNEL_METRO_API_BASE_URL` | | Metro API base URL; answers the WebAuthn page hook (`PUT /webauthn/hook`) | -| `KERNEL_INSTANCE_JWT` | | Instance credential sent to the metro API and the OTLP relay | +| `KERNEL_INSTANCE_JWT` | | Instance credential sent to the metro API, the OTLP relay, and the storage relay | +| `BTEL_STORAGE_PATH` | | Telemetry storage relay path on the metro API (e.g. `/btel-relay/v1/records`). When set, stored telemetry is posted to the relay, which appends it to the instance's S2 stream, and `S2_BASIN`/`S2_ACCESS_TOKEN` are ignored. The host is `KERNEL_METRO_API_BASE_URL` (or the fork identity's), else `BTEL_OTLP_ENDPOINT` with `BTEL_OTLP_INSECURE` picking the scheme | +| `BTEL_STORAGE_MAX_RECORDS` | `200` | Records per relay request (at most 1000) | +| `BTEL_STORAGE_MAX_BYTES` | `786432` | S2 metered bytes per relay request (at most 1 MiB) | +| `BTEL_STORAGE_LINGER` | `100ms` | How long a partial batch waits before it is sent | +| `BTEL_STORAGE_MAX_QUEUE` | `8192` | Records buffered for the relay; the oldest are dropped when full | +| `S2_BASIN`, `S2_ACCESS_TOKEN`, `S2_STREAM` | | Direct S2 storage, used when `BTEL_STORAGE_PATH` is unset. `S2_STREAM` also gates relay storage: nothing is stored until it (or the fork identity's stream) is known | #### Example Configuration diff --git a/server/cmd/api/api/telemetry_test.go b/server/cmd/api/api/telemetry_test.go index bea95116..f7318984 100644 --- a/server/cmd/api/api/telemetry_test.go +++ b/server/cmd/api/api/telemetry_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "log/slog" "sync" "testing" "time" @@ -943,6 +944,60 @@ func TestTelemetryStorageToggle(t *testing.T) { }) } +// TestTelemetryStorageWithRealControllers runs the storage toggle against both +// real storage controllers, direct S2 and the metro-api relay, rather than a +// stub: storage opens once and refuses to turn off, and stays closed while the +// fork gate resolves no stream. +func TestTelemetryStorageWithRealControllers(t *testing.T) { + ctx := context.Background() + tr := true + networkOn := func(storage bool) *oapi.BrowserTelemetryConfig { + return &oapi.BrowserTelemetryConfig{ + Browser: &oapi.BrowserTelemetryCategoriesConfig{Network: &oapi.BrowserTelemetryCategoryConfig{Enabled: &tr}}, + Storage: &oapi.BrowserTelemetryStorageConfig{Enabled: &storage}, + } + } + controllers := map[string]func(es *events.EventStream, stream func() string) *events.S2StorageController{ + "s2": func(es *events.EventStream, stream func() string) *events.S2StorageController { + return events.NewS2StorageController(es, "test-basin", "test-token", stream, events.S2Config{}, slog.Default()) + }, + "relay": func(es *events.EventStream, stream func() string) *events.S2StorageController { + return events.NewRelayStorageController(es, events.RelayConfig{ + BaseURLFunc: func() string { return "http://127.0.0.1:1" }, + Path: "/btel-relay/v1/records", + TokenFunc: func() string { return "jwt" }, + }, stream, slog.Default()) + }, + } + for name, build := range controllers { + t.Run(name, func(t *testing.T) { + svc := newTestService(t, newMockRecordManager()) + c := build(svc.eventStream, func() string { return "instance-stream" }) + svc.s2Storage = c + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + + resp, err := svc.PutTelemetry(ctx, oapi.PutTelemetryRequestObject{Body: networkOn(true)}) + require.NoError(t, err) + require.IsType(t, oapi.PutTelemetry201JSONResponse{}, resp) + assert.True(t, c.Running()) + + resp, err = svc.PutTelemetry(ctx, oapi.PutTelemetryRequestObject{Body: networkOn(false)}) + require.NoError(t, err) + require.IsType(t, oapi.PutTelemetry409JSONResponse{}, resp) + assert.True(t, c.Running()) + }) + t.Run(name+" fork gate", func(t *testing.T) { + svc := newTestService(t, newMockRecordManager()) + c := build(svc.eventStream, func() string { return "" }) + svc.s2Storage = c + + _, err := svc.PutTelemetry(ctx, oapi.PutTelemetryRequestObject{Body: networkOn(true)}) + require.NoError(t, err) + assert.False(t, c.EverStarted(), "no stream resolved, so nothing opens") + }) + } +} + // stubS2Storage records starts for storage-toggle tests. Locked because // reconcileStorage runs outside monitorMu. type stubS2Storage struct { diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index c5515778..48aa53c7 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -139,7 +139,6 @@ func main() { // for a fork identity is carrying the stream of the instance it was forked // from, so the resolver hands out no stream until that identity arrives. s2Streams := newS2StreamResolver(config, slogger) - s2Storage := events.NewS2StorageController(eventStream, config.S2Basin, config.S2AccessToken, s2Streams.Resolve, events.S2Config{}, slogger) // Optional OTLP export sink. Independent of S2; both can run together. // Constructed when an endpoint is provisioned, but left stopped: export is @@ -156,6 +155,25 @@ func main() { metroAPIURL: config.MetroAPIBaseURL, }, slogger) + // With a relay path the VM holds no S2 credential: storage posts to the + // metro-api relay with the instance JWT, and the relay appends to this + // instance's stream. The relay wins when direct S2 credentials are also set. + s2Storage := events.NewS2StorageController(eventStream, config.S2Basin, config.S2AccessToken, s2Streams.Resolve, events.S2Config{}, slogger) + var relayMetrics *events.RelayMetrics + if config.StoragePath != "" { + relayMetrics = &events.RelayMetrics{} + s2Storage = events.NewRelayStorageController(eventStream, events.RelayConfig{ + BaseURLFunc: func() string { return relayBaseURL(identity.MetroAPIURL(), config) }, + Path: config.StoragePath, + TokenFunc: identity.Token, + MaxRecords: config.StorageMaxRecords, + MaxBytes: config.StorageMaxBytes, + Linger: config.StorageLinger, + MaxQueue: config.StorageMaxQueue, + Metrics: relayMetrics, + }, s2Streams.Resolve, slogger) + } + var otlpExporter api.OTLPExporter var otlpMetrics *events.OTLPMetrics if config.OTLPEndpoint != "" { @@ -392,6 +410,9 @@ func main() { if otlpMetrics != nil { metricsCollectors = append(metricsCollectors, metrics.NewOTLPCollector(otlpMetrics)) } + if relayMetrics != nil { + metricsCollectors = append(metricsCollectors, metrics.NewRelayCollector(relayMetrics)) + } rMetrics.Method(http.MethodGet, "/metrics", metrics.Handler(slogger, metricsCollectors...)) srvMetrics := &http.Server{ Addr: fmt.Sprintf("0.0.0.0:%d", config.MetricsPort), @@ -487,6 +508,20 @@ func mustFFmpeg() { } } +// relayBaseURL is the metro-api the storage relay is reached through: the +// identity's metro-api URL when it has one, else the OTLP relay endpoint, which +// provisioning points at the same metro-api. +func relayBaseURL(metroAPIURL string, cfg *config.Config) string { + if metroAPIURL != "" { + return metroAPIURL + } + scheme := "https" + if cfg.OTLPInsecure { + scheme = "http" + } + return scheme + "://" + cfg.OTLPEndpoint +} + // s2StreamResolver uses the hook payload once applied and the persisted payload // after an API process restart. Resolve is called by the S2 controller when the // telemetry handler opens the sink, which on a fork happens after the handoff. diff --git a/server/cmd/api/main_test.go b/server/cmd/api/main_test.go index 0826c57d..d22a476a 100644 --- a/server/cmd/api/main_test.go +++ b/server/cmd/api/main_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "github.com/kernel/kernel-images/server/cmd/config" "github.com/kernel/kernel-images/server/lib/devtoolsproxy" "github.com/stretchr/testify/require" ) @@ -126,3 +127,11 @@ func TestChromeJSONProxyHandler_CancelsUpstreamRequestWithCallerContext(t *testi require.Equal(t, http.StatusBadGateway, rec.Code) } + +func TestRelayBaseURL(t *testing.T) { + cfg := &config.Config{OTLPEndpoint: "169.254.0.1:8443"} + require.Equal(t, "https://metro.example", relayBaseURL("https://metro.example", cfg), "the identity's metro-api wins") + require.Equal(t, "https://169.254.0.1:8443", relayBaseURL("", cfg)) + cfg.OTLPInsecure = true + require.Equal(t, "http://169.254.0.1:8443", relayBaseURL("", cfg)) +} diff --git a/server/cmd/config/config.go b/server/cmd/config/config.go index 7e414b70..0148d75f 100644 --- a/server/cmd/config/config.go +++ b/server/cmd/config/config.go @@ -74,6 +74,17 @@ type Config struct { OTLPMaxQueueSize int `envconfig:"BTEL_OTLP_MAX_QUEUE_SIZE" default:"2048"` OTLPExportInterval time.Duration `envconfig:"BTEL_OTLP_EXPORT_INTERVAL" default:"1s"` OTLPExportTimeout time.Duration `envconfig:"BTEL_OTLP_EXPORT_TIMEOUT" default:"30s"` + // Browser-telemetry storage through the metro-api relay. When StoragePath is + // set, storage posts event envelopes to the relay, which appends them to the + // instance's S2 stream, and the S2_* credentials are not used. The relay host + // is KERNEL_METRO_API_BASE_URL (or the fork identity's), falling back to + // BTEL_OTLP_ENDPOINT. MaxBytes is in S2 metered bytes. + StoragePath string `envconfig:"BTEL_STORAGE_PATH" default:""` + StorageMaxRecords int `envconfig:"BTEL_STORAGE_MAX_RECORDS" default:"200"` + StorageMaxBytes int `envconfig:"BTEL_STORAGE_MAX_BYTES" default:"786432"` + StorageLinger time.Duration `envconfig:"BTEL_STORAGE_LINGER" default:"100ms"` + StorageMaxQueue int `envconfig:"BTEL_STORAGE_MAX_QUEUE" default:"8192"` + // Platform-injected identity, reused to stamp the OTLP Resource. These are // the same envs the VM already receives. InstanceJWT string `envconfig:"KERNEL_INSTANCE_JWT" default:""` @@ -124,6 +135,11 @@ func (c *Config) LogValue() slog.Value { slog.Int("otlp_max_queue_size", c.OTLPMaxQueueSize), slog.Duration("otlp_export_interval", c.OTLPExportInterval), slog.Duration("otlp_export_timeout", c.OTLPExportTimeout), + slog.String("storage_path", c.StoragePath), + slog.Int("storage_max_records", c.StorageMaxRecords), + slog.Int("storage_max_bytes", c.StorageMaxBytes), + slog.Duration("storage_linger", c.StorageLinger), + slog.Int("storage_max_queue", c.StorageMaxQueue), slog.String("otlp_instance_name", c.InstanceName), slog.String("otlp_metro", c.MetroName), slog.String("metro_api_base_url", c.MetroAPIBaseURL), @@ -180,6 +196,25 @@ func validate(config *Config) error { if config.OTLPExportTimeout <= 0 { return fmt.Errorf("BTEL_OTLP_EXPORT_TIMEOUT must be greater than 0") } + if config.StoragePath != "" { + if config.MetroAPIBaseURL == "" && config.OTLPEndpoint == "" { + return fmt.Errorf("BTEL_STORAGE_PATH requires KERNEL_METRO_API_BASE_URL or BTEL_OTLP_ENDPOINT to reach the relay") + } + // The relay appends each batch as one S2 append: at most 1000 records + // and 1 MiB metered. + if config.StorageMaxRecords < 1 || config.StorageMaxRecords > 1000 { + return fmt.Errorf("BTEL_STORAGE_MAX_RECORDS must be between 1 and 1000") + } + if config.StorageMaxBytes < 1 || config.StorageMaxBytes > 1<<20 { + return fmt.Errorf("BTEL_STORAGE_MAX_BYTES must be between 1 and %d", 1<<20) + } + if config.StorageLinger <= 0 { + return fmt.Errorf("BTEL_STORAGE_LINGER must be greater than 0") + } + if config.StorageMaxQueue < config.StorageMaxRecords { + return fmt.Errorf("BTEL_STORAGE_MAX_QUEUE must be at least BTEL_STORAGE_MAX_RECORDS") + } + } return nil } diff --git a/server/cmd/config/config_test.go b/server/cmd/config/config_test.go index a1c5598b..6198305d 100644 --- a/server/cmd/config/config_test.go +++ b/server/cmd/config/config_test.go @@ -39,6 +39,10 @@ func TestLoad(t *testing.T) { OTLPMaxQueueSize: 2048, OTLPExportInterval: time.Second, OTLPExportTimeout: 30 * time.Second, + StorageMaxRecords: 200, + StorageMaxBytes: 786432, + StorageLinger: 100 * time.Millisecond, + StorageMaxQueue: 8192, }, }, { @@ -84,6 +88,10 @@ func TestLoad(t *testing.T) { OTLPMaxQueueSize: 2048, OTLPExportInterval: time.Second, OTLPExportTimeout: 30 * time.Second, + StorageMaxRecords: 200, + StorageMaxBytes: 786432, + StorageLinger: 100 * time.Millisecond, + StorageMaxQueue: 8192, }, }, { @@ -114,6 +122,10 @@ func TestLoad(t *testing.T) { OTLPMaxQueueSize: 2048, OTLPExportInterval: time.Second, OTLPExportTimeout: 30 * time.Second, + StorageMaxRecords: 200, + StorageMaxBytes: 786432, + StorageLinger: 100 * time.Millisecond, + StorageMaxQueue: 8192, }, }, { @@ -151,6 +163,40 @@ func TestLoad(t *testing.T) { }, wantErr: true, }, + { + name: "relay storage without a relay host", + env: map[string]string{ + "BTEL_STORAGE_PATH": "/btel-relay/v1/records", + }, + wantErr: true, + }, + { + name: "relay storage batch over the S2 record limit", + env: map[string]string{ + "BTEL_OTLP_ENDPOINT": "169.254.0.1:8443", + "BTEL_STORAGE_PATH": "/btel-relay/v1/records", + "BTEL_STORAGE_MAX_RECORDS": "1001", + }, + wantErr: true, + }, + { + name: "relay storage batch over the S2 byte limit", + env: map[string]string{ + "BTEL_OTLP_ENDPOINT": "169.254.0.1:8443", + "BTEL_STORAGE_PATH": "/btel-relay/v1/records", + "BTEL_STORAGE_MAX_BYTES": "1048577", + }, + wantErr: true, + }, + { + name: "relay storage queue smaller than a batch", + env: map[string]string{ + "BTEL_OTLP_ENDPOINT": "169.254.0.1:8443", + "BTEL_STORAGE_PATH": "/btel-relay/v1/records", + "BTEL_STORAGE_MAX_QUEUE": "10", + }, + wantErr: true, + }, { name: "missing ffmpeg path (set to empty)", env: map[string]string{ diff --git a/server/e2e/e2e_relay_storage_test.go b/server/e2e/e2e_relay_storage_test.go new file mode 100644 index 00000000..cd0b37c1 --- /dev/null +++ b/server/e2e/e2e_relay_storage_test.go @@ -0,0 +1,242 @@ +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" + + instanceoapi "github.com/kernel/kernel-images/server/lib/oapi" +) + +const relayStoragePath = "/btel-relay/v1/records" + +// relayMock is a minimal metro-api storage relay served on the test host. It +// records the Authorization header and records of every request. +type relayMock struct { + srv *http.Server + ln net.Listener + + mu sync.Mutex + auths []string + records []relayMockRecord +} + +type relayMockRecord struct { + TS int64 `json:"ts"` + Body json.RawMessage `json:"body"` +} + +func startRelayMock(t *testing.T) *relayMock { + t.Helper() + ln, err := net.Listen("tcp", "0.0.0.0:0") + require.NoError(t, err, "listen for mock relay") + m := &relayMock{ln: ln} + mux := http.NewServeMux() + mux.HandleFunc(relayStoragePath, m.handle) + m.srv = &http.Server{Handler: mux} + go func() { _ = m.srv.Serve(ln) }() + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = m.srv.Shutdown(ctx) + }) + return m +} + +func (m *relayMock) handle(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + var req struct { + Records []relayMockRecord `json:"records"` + } + if err := json.Unmarshal(body, &req); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + m.mu.Lock() + m.auths = append(m.auths, r.Header.Get("Authorization")) + m.records = append(m.records, req.Records...) + m.mu.Unlock() + _, _ = fmt.Fprintf(w, `{"start_seq_num":0,"end_seq_num":%d}`, len(req.Records)) +} + +func (m *relayMock) hostEndpoint() string { + return fmt.Sprintf("host.docker.internal:%d", m.ln.Addr().(*net.TCPAddr).Port) +} + +func (m *relayMock) snapshot() ([]string, []relayMockRecord) { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.auths...), append([]relayMockRecord(nil), m.records...) +} + +// enableControlStorage turns on the control category with storage, then makes +// a few browser-control calls to generate stored events. +func enableControlStorage(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.BrowserTelemetryControlConfig{Enabled: &tr}, + }, + Storage: &instanceoapi.BrowserTelemetryStorageConfig{Enabled: &tr}, + }) + require.NoError(t, err) + require.Equal(t, http.StatusCreated, resp.StatusCode(), "put telemetry: %s", string(resp.Body)) + for range 3 { + _, _ = client.TakeScreenshotWithResponse(ctx, instanceoapi.TakeScreenshotJSONRequestBody{}) + time.Sleep(50 * time.Millisecond) + } +} + +// requireEnvelopes checks each stored record is a {seq,event} envelope whose +// ts is the event time in milliseconds. +func requireEnvelopes(t *testing.T, records []relayMockRecord) { + t.Helper() + for _, rec := range records { + var env struct { + Seq uint64 `json:"seq"` + Event struct { + Ts int64 `json:"ts"` + Type string `json:"type"` + } `json:"event"` + } + require.NoError(t, json.Unmarshal(rec.Body, &env)) + require.NotEmpty(t, env.Event.Type) + assert.Equal(t, env.Event.Ts/1000, rec.TS) + } +} + +// TestRelayStorage runs a headless container with a relay path and verifies +// stored events reach the relay with the instance JWT as a bearer token. Direct +// S2 credentials are also set, to show the relay wins. Skips when docker is +// unavailable. +func TestRelayStorage(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() + + relay := startRelayMock(t) + + c := NewTestContainer(t, headlessImage) + require.NoError(t, c.Start(ctx, ContainerConfig{ + HostAccess: true, + Env: map[string]string{ + "BTEL_OTLP_ENDPOINT": relay.hostEndpoint(), + "BTEL_OTLP_INSECURE": "true", + "BTEL_STORAGE_PATH": relayStoragePath, + "S2_BASIN": "unused-basin", + "S2_ACCESS_TOKEN": "unused-token", + "S2_STREAM": "browser-e2e-1", + "INST_NAME": "browser-e2e-1", + "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) + enableControlStorage(t, ctx, client) + + require.Eventually(t, func() bool { + _, records := relay.snapshot() + return len(records) > 0 + }, 20*time.Second, 250*time.Millisecond, "relay received no records") + + auths, records := relay.snapshot() + assert.Equal(t, "Bearer e2e-jwt-token", auths[0]) + requireEnvelopes(t, records) +} + +// TestRelayStorageForkIdentity boots a fork waiting for its identity, with a +// dead relay endpoint and no credential, then applies an identity carrying a +// fresh JWT, stream, and metro-api URL. Storage must reach the relay at the +// handed metro-api with the fresh JWT. +func TestRelayStorageForkIdentity(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() + + relay := startRelayMock(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": "127.0.0.1:1", + "BTEL_OTLP_INSECURE": "true", + "BTEL_STORAGE_PATH": relayStoragePath, + "S2_STREAM": "seed-stream", + "KERNEL_INSTANCE_JWT": "", + }, + }), "failed to start container") + defer c.Stop(ctx) + require.NoError(t, c.WaitReady(ctx), "api not ready") + + payload := map[string]string{ + "instance_name": "browser-fork-e2e", + "kernel_instance_jwt": "fresh-fork-jwt", + "s2_stream": "browser-fork-e2e", + "kernel_metro_api_base_url": "http://" + relay.hostEndpoint(), + } + buf, err := json.Marshal(payload) + require.NoError(t, err) + // See TestOTLPExportForkIdentityRefresh: retry until the apply sticks. + 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) + enableControlStorage(t, ctx, client) + + require.Eventually(t, func() bool { + _, records := relay.snapshot() + return len(records) > 0 + }, 30*time.Second, 250*time.Millisecond, "relay at the handed metro-api received no records") + + auths, records := relay.snapshot() + for _, a := range auths { + assert.Equal(t, "Bearer fresh-fork-jwt", a) + } + requireEnvelopes(t, records) +} diff --git a/server/lib/events/relaymetrics.go b/server/lib/events/relaymetrics.go new file mode 100644 index 00000000..ef15c502 --- /dev/null +++ b/server/lib/events/relaymetrics.go @@ -0,0 +1,68 @@ +package events + +import ( + "maps" + "sync" + "sync/atomic" +) + +// Reasons a relay record is dropped. +const ( + relayDropQueueFull = "queue_full" + relayDropRejected = "rejected" + relayDropRetriesExhausted = "retries_exhausted" + relayDropShutdown = "shutdown" +) + +// RelayMetrics accumulates the storage relay sink's process-lifetime counters. +// Safe for concurrent use. +type RelayMetrics struct { + exported atomic.Uint64 + retries atomic.Uint64 + queueDepth atomic.Int64 + + mu sync.Mutex + dropped map[string]uint64 + failed map[string]uint64 +} + +// Exported returns records the relay acknowledged. +func (m *RelayMetrics) Exported() uint64 { return m.exported.Load() } + +// Retries returns batch retries after a retryable response. +func (m *RelayMetrics) Retries() uint64 { return m.retries.Load() } + +// QueueDepth returns records waiting to be sent. +func (m *RelayMetrics) QueueDepth() int64 { return m.queueDepth.Load() } + +// Dropped returns dropped records by reason. +func (m *RelayMetrics) Dropped() map[string]uint64 { + m.mu.Lock() + defer m.mu.Unlock() + return maps.Clone(m.dropped) +} + +// Failed returns batches given up on, by HTTP status or "error". +func (m *RelayMetrics) Failed() map[string]uint64 { + m.mu.Lock() + defer m.mu.Unlock() + return maps.Clone(m.failed) +} + +func (m *RelayMetrics) addDropped(reason string, n int) { + m.mu.Lock() + defer m.mu.Unlock() + if m.dropped == nil { + m.dropped = map[string]uint64{} + } + m.dropped[reason] += uint64(n) +} + +func (m *RelayMetrics) addFailed(status string) { + m.mu.Lock() + defer m.mu.Unlock() + if m.failed == nil { + m.failed = map[string]uint64{} + } + m.failed[status]++ +} diff --git a/server/lib/events/relaystorage.go b/server/lib/events/relaystorage.go new file mode 100644 index 00000000..1bebdad7 --- /dev/null +++ b/server/lib/events/relaystorage.go @@ -0,0 +1,424 @@ +package events + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "math/rand/v2" + "net/http" + "net/url" + "slices" + "strconv" + "sync" + "time" +) + +// Relay batch defaults. maxBytes is in S2 metered bytes (8 + body length per +// record) and stays under the relay's 1 MiB append limit. +const ( + DefaultRelayMaxRecords = 200 + DefaultRelayMaxBytes = 768 * 1024 + DefaultRelayLinger = 100 * time.Millisecond + DefaultRelayMaxQueue = 8192 + + relayMeteredOverhead = 8 + // relayExportTimeout bounds one batch, retries included, like the OTLP + // sink's BTEL_OTLP_EXPORT_TIMEOUT default. It is shorter than the retry + // policy's maxElapsed, so it is what ends a batch's retries in practice. + relayExportTimeout = 30 * time.Second + // relayDrainGrace bounds how long Close waits for the cancelled sender to + // record its drops once its ctx has expired. + relayDrainGrace = 250 * time.Millisecond +) + +// relayRetry mirrors the OTLP exporter's retry policy: exponential backoff +// from 5s up to 30s with jitter, waiting the greater of the backoff and +// Retry-After, and giving up after a minute or relayExportTimeout, whichever +// comes first. +type relayRetry struct { + initial, max, maxElapsed time.Duration +} + +var defaultRelayRetry = relayRetry{initial: 5 * time.Second, max: 30 * time.Second, maxElapsed: time.Minute} + +// RelayConfig configures the metro-api storage relay sink. +type RelayConfig struct { + // BaseURLFunc resolves the metro-api base URL per batch, so a fork posts to + // the host it was handed rather than its seed's. Only its scheme and host + // are used. + BaseURLFunc func() string + // Path is the full relay request path, any metro-api base path included. + Path string + // TokenFunc resolves the instance JWT per request. + TokenFunc func() string + + MaxRecords int + MaxBytes int + Linger time.Duration + MaxQueue int + + Client *http.Client + Metrics *RelayMetrics + retry relayRetry +} + +func (c *RelayConfig) withDefaults() { + if c.MaxRecords <= 0 { + c.MaxRecords = DefaultRelayMaxRecords + } + if c.MaxBytes <= 0 { + c.MaxBytes = DefaultRelayMaxBytes + } + if c.Linger <= 0 { + c.Linger = DefaultRelayLinger + } + if c.MaxQueue <= 0 { + c.MaxQueue = DefaultRelayMaxQueue + } + if c.Client == nil { + c.Client = &http.Client{ + // The relay is configured, not discovered, so a redirect is never + // something to follow with the instance credential attached. + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }, + } + } + if c.Metrics == nil { + c.Metrics = &RelayMetrics{} + } + if c.retry == (relayRetry{}) { + c.retry = defaultRelayRetry + } +} + +type relayRecord struct { + TS int64 `json:"ts"` + Body json.RawMessage `json:"body"` +} + +func (r relayRecord) metered() int { return relayMeteredOverhead + len(r.Body) } + +// relayStorage implements Storage by posting envelopes in batches to the +// metro-api relay, which appends them to the instance's S2 stream. Append never +// blocks: the queue is bounded and drops its oldest record when full. One +// sender goroutine posts one batch at a time, so records arrive in order. +type relayStorage struct { + cfg RelayConfig + log *slog.Logger + + mu sync.Mutex + queue []relayRecord + closing bool + wake chan struct{} + + sendCtx context.Context + cancelSend context.CancelFunc + done chan struct{} + closeOnce sync.Once +} + +func newRelayStorage(ctx context.Context, cfg RelayConfig, log *slog.Logger) (*relayStorage, error) { + if cfg.Path == "" || cfg.BaseURLFunc == nil || cfg.TokenFunc == nil { + return nil, fmt.Errorf("relay storage: path, base URL, and token are required") + } + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("relay storage: context already cancelled: %w", err) + } + cfg.withDefaults() + // The sender runs independently of ctx so SIGTERM does not cut off the + // final flush; Close bounds it instead. + sendCtx, cancel := context.WithCancel(context.Background()) + s := &relayStorage{ + cfg: cfg, + log: log, + wake: make(chan struct{}, 1), + sendCtx: sendCtx, + cancelSend: cancel, + done: make(chan struct{}), + } + go s.run() + return s, nil +} + +// Append marshals env exactly as the direct S2 sink does and queues it with its +// event time in unix milliseconds. +func (s *relayStorage) Append(ctx context.Context, env Envelope) error { + if err := ctx.Err(); err != nil { + return err + } + data, err := json.Marshal(env) + if err != nil { + return fmt.Errorf("relay storage: marshal envelope seq=%d: %w", env.Seq, err) + } + rec := relayRecord{TS: max(env.Event.Ts/1000, 0), Body: data} + + s.mu.Lock() + if s.closing { + s.mu.Unlock() + s.cfg.Metrics.addDropped(relayDropShutdown, 1) + return nil + } + if len(s.queue) >= s.cfg.MaxQueue { + s.queue = s.queue[1:] + s.cfg.Metrics.addDropped(relayDropQueueFull, 1) + } + s.queue = append(s.queue, rec) + s.cfg.Metrics.queueDepth.Store(int64(len(s.queue))) + s.mu.Unlock() + s.signal() + return nil +} + +// Close flushes what it can within ctx, then drops the rest. +func (s *relayStorage) Close(ctx context.Context) error { + s.closeOnce.Do(func() { + s.mu.Lock() + s.closing = true + s.mu.Unlock() + s.signal() + }) + select { + case <-s.done: + return nil + case <-ctx.Done(): + } + s.cancelSend() + select { + case <-s.done: + case <-time.After(relayDrainGrace): + } + return ctx.Err() +} + +func (s *relayStorage) signal() { + select { + case s.wake <- struct{}{}: + default: + } +} + +func (s *relayStorage) run() { + defer close(s.done) + defer s.dropRemaining() + for { + batch, ok := s.nextBatch() + if !ok { + return + } + s.send(batch) + if s.sendCtx.Err() != nil { + return + } + } +} + +// nextBatch waits until a batch is due: full by count or bytes, the linger +// since the sender first saw a queued record has passed, or Close was called. +// It returns false once closing with nothing left to send, or when the sender +// is cut off. +func (s *relayStorage) nextBatch() ([]relayRecord, bool) { + var timer *time.Timer + var linger <-chan time.Time + lingered := false + defer func() { + if timer != nil { + timer.Stop() + } + }() + for { + s.mu.Lock() + n, full := s.batchLenLocked() + if n == 0 && s.closing { + s.mu.Unlock() + return nil, false + } + if n > 0 && (full || lingered || s.closing) { + batch := slices.Clone(s.queue[:n]) + s.queue = s.queue[n:] + s.cfg.Metrics.queueDepth.Store(int64(len(s.queue))) + s.mu.Unlock() + return batch, true + } + s.mu.Unlock() + + if n > 0 && timer == nil { + timer = time.NewTimer(s.cfg.Linger) + linger = timer.C + } + select { + case <-s.wake: + case <-linger: + lingered = true + case <-s.sendCtx.Done(): + return nil, false + } + } +} + +// batchLenLocked returns how many queued records fit in one batch and whether +// that batch is full. A record over the byte budget on its own still ships +// alone. +func (s *relayStorage) batchLenLocked() (int, bool) { + size := 0 + for i, rec := range s.queue { + if i == s.cfg.MaxRecords { + return i, true + } + if i > 0 && size+rec.metered() > s.cfg.MaxBytes { + return i, true + } + size += rec.metered() + } + n := len(s.queue) + return n, n >= s.cfg.MaxRecords || size >= s.cfg.MaxBytes +} + +func (s *relayStorage) dropRemaining() { + s.mu.Lock() + n := len(s.queue) + s.queue = nil + s.closing = true + s.cfg.Metrics.queueDepth.Store(0) + s.mu.Unlock() + if n > 0 { + s.cfg.Metrics.addDropped(relayDropShutdown, n) + } +} + +// send posts one batch, retrying throttled and unavailable responses. Any +// other outcome drops the batch and counts it. +func (s *relayStorage) send(batch []relayRecord) { + body, err := json.Marshal(struct { + Records []relayRecord `json:"records"` + }{batch}) + if err != nil { + s.fail(batch, "marshal", relayDropRejected, err) + return + } + ctx, cancel := context.WithTimeout(s.sendCtx, relayExportTimeout) + defer cancel() + + start := time.Now() + backoff := s.cfg.retry.initial + for { + status, retryAfter, err := s.post(ctx, body) + if err == nil && status >= 200 && status < 300 { + s.cfg.Metrics.exported.Add(uint64(len(batch))) + return + } + label := strconv.Itoa(status) + if err != nil { + label = "error" + } + if s.sendCtx.Err() != nil { + s.fail(batch, label, relayDropShutdown, err) + return + } + if !relayRetryable(status, err) { + s.fail(batch, label, relayDropRejected, err) + return + } + delay := max(retryAfter, jitter(backoff)) + if time.Since(start)+delay > s.cfg.retry.maxElapsed { + s.fail(batch, label, relayDropRetriesExhausted, err) + return + } + s.cfg.Metrics.retries.Add(1) + s.log.Debug("relay storage: retrying batch", "records", len(batch), "status", label, "delay", delay) + select { + case <-time.After(delay): + case <-ctx.Done(): + reason := relayDropRetriesExhausted + if s.sendCtx.Err() != nil { + reason = relayDropShutdown + } + s.fail(batch, label, reason, ctx.Err()) + return + } + backoff = min(time.Duration(float64(backoff)*1.5), s.cfg.retry.max) + } +} + +func (s *relayStorage) fail(batch []relayRecord, status, reason string, err error) { + s.cfg.Metrics.addFailed(status) + s.cfg.Metrics.addDropped(reason, len(batch)) + s.log.Warn("relay storage: dropped batch", "records", len(batch), "status", status, "err", err) +} + +// post sends one request. The bearer is attached only to the request built for +// the resolved relay URL, and redirects are not followed, so the instance +// credential never reaches another host. +func (s *relayStorage) post(ctx context.Context, body []byte) (status int, retryAfter time.Duration, err error) { + target, err := relayURL(s.cfg.BaseURLFunc(), s.cfg.Path) + if err != nil { + return 0, 0, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, bytes.NewReader(body)) + if err != nil { + return 0, 0, err + } + req.Header.Set("Content-Type", "application/json") + if tok := s.cfg.TokenFunc(); tok != "" { + req.Header.Set("Authorization", "Bearer "+tok) + } + resp, err := s.cfg.Client.Do(req) + if err != nil { + return 0, 0, err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) + if secs, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && secs > 0 { + retryAfter = time.Duration(secs) * time.Second + } + return resp.StatusCode, retryAfter, nil +} + +// relayRetryable matches the OTLP exporter: retry 429, 502, 503, and 504, and +// temporary transport errors; everything else is final. +func relayRetryable(status int, err error) bool { + if err != nil { + var urlErr *url.Error + return errors.As(err, &urlErr) && urlErr.Temporary() + } + switch status { + case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout: + return true + } + return false +} + +func jitter(d time.Duration) time.Duration { + return time.Duration(float64(d) * (0.5 + rand.Float64())) +} + +// relayURL is base's scheme and host with path. Provisioning already folds any +// metro-api base path into path, as it does for the OTLP relay path. +func relayURL(base, path string) (string, error) { + u, err := url.Parse(base) + if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { + return "", fmt.Errorf("relay storage: invalid metro-api base URL") + } + return (&url.URL{Scheme: u.Scheme, Host: u.Host, Path: path}).String(), nil +} + +// NewRelayStorageController builds a storage controller that writes through +// the metro-api relay. It shares the S2 controller's lifecycle and fork gate: +// streamFn must resolve before storage opens, so a fork never writes while it +// still holds its seed's identity. The relay names the stream from the +// instance JWT, so the resolved value is not sent. +func NewRelayStorageController(es *EventStream, cfg RelayConfig, streamFn func() string, log *slog.Logger) *S2StorageController { + return &S2StorageController{ + es: es, + streamFn: streamFn, + open: func(ctx context.Context, _ string) (Storage, error) { + return newRelayStorage(ctx, cfg, log) + }, + logAttr: []any{"via", "metro-api relay"}, + log: log, + } +} diff --git a/server/lib/events/relaystorage_test.go b/server/lib/events/relaystorage_test.go new file mode 100644 index 00000000..73ef5a17 --- /dev/null +++ b/server/lib/events/relaystorage_test.go @@ -0,0 +1,353 @@ +package events + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const relayTestPath = "/btel-relay/v1/records" + +type relayRequest struct { + auth string + records []relayRecord +} + +// fakeRelay records each request and answers with respond(n), n counting +// from 1. A nil respond acknowledges everything. +type fakeRelay struct { + srv *httptest.Server + mu sync.Mutex + reqs []relayRequest + respond func(n int, w http.ResponseWriter) bool +} + +func newFakeRelay(t *testing.T) *fakeRelay { + f := &fakeRelay{} + f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != relayTestPath || r.Method != http.MethodPost { + http.NotFound(w, r) + return + } + raw, _ := io.ReadAll(r.Body) + var body struct { + Records []relayRecord `json:"records"` + } + require.NoError(t, json.Unmarshal(raw, &body)) + f.mu.Lock() + f.reqs = append(f.reqs, relayRequest{auth: r.Header.Get("Authorization"), records: body.Records}) + n, respond := len(f.reqs), f.respond + f.mu.Unlock() + if respond != nil && respond(n, w) { + return + } + _, _ = io.WriteString(w, `{"start_seq_num":0,"end_seq_num":1}`) + })) + t.Cleanup(f.srv.Close) + return f +} + +func (f *fakeRelay) requests() []relayRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]relayRequest(nil), f.reqs...) +} + +func (f *fakeRelay) records() []relayRecord { + var out []relayRecord + for _, r := range f.requests() { + out = append(out, r.records...) + } + return out +} + +func testRelayConfig(base string) RelayConfig { + return RelayConfig{ + BaseURLFunc: func() string { return base }, + Path: relayTestPath, + TokenFunc: func() string { return "instance-jwt" }, + Linger: 10 * time.Millisecond, + Metrics: &RelayMetrics{}, + retry: relayRetry{initial: 5 * time.Millisecond, max: 20 * time.Millisecond, maxElapsed: 5 * time.Second}, + } +} + +func newTestRelayStorage(t *testing.T, cfg RelayConfig) *relayStorage { + t.Helper() + s, err := newRelayStorage(context.Background(), cfg, slog.Default()) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close(context.Background()) }) + return s +} + +func closeRelay(t *testing.T, s *relayStorage) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, s.Close(ctx)) +} + +func relayEnvelope(seq uint64, data string) Envelope { + return Envelope{Seq: seq, Event: Event{ + Ts: 1_700_000_000_123_456 + int64(seq), + Type: "console.log", + Category: Console, + Data: json.RawMessage(data), + }} +} + +func TestRelayStorage_SendsEnvelopesInOrderUnchanged(t *testing.T) { + relay := newFakeRelay(t) + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + + envs := []Envelope{ + relayEnvelope(1, `{"text":"&"}`), + relayEnvelope(2, `{"text":"
 line"}`), + relayEnvelope(3, `{}`), + } + for _, env := range envs { + require.NoError(t, s.Append(context.Background(), env)) + } + closeRelay(t, s) + + got := relay.records() + require.Len(t, got, len(envs)) + for i, env := range envs { + want, err := json.Marshal(env) + require.NoError(t, err) + assert.Equal(t, string(want), string(got[i].Body), "body must match the direct S2 sink's record byte for byte") + assert.Equal(t, env.Event.Ts/1000, got[i].TS, "ts is the event time in milliseconds") + } + assert.Equal(t, "Bearer instance-jwt", relay.requests()[0].auth) + assert.Equal(t, uint64(3), s.cfg.Metrics.Exported()) +} + +func TestRelayStorage_FlushesOnRecordAndByteLimits(t *testing.T) { + relay := newFakeRelay(t) + cfg := testRelayConfig(relay.srv.URL) + cfg.Linger = time.Hour + cfg.MaxRecords = 3 + s := newTestRelayStorage(t, cfg) + + for i := range 3 { + require.NoError(t, s.Append(context.Background(), relayEnvelope(uint64(i), `{}`))) + } + require.Eventually(t, func() bool { return len(relay.requests()) == 1 }, 5*time.Second, 5*time.Millisecond, + "a full batch flushes without waiting for the linger") + + relay2 := newFakeRelay(t) + payload := `{"pad":"` + string(make([]byte, 0)) + `xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}` + env := relayEnvelope(1, payload) + body, _ := json.Marshal(env) + cfg = testRelayConfig(relay2.srv.URL) + cfg.Linger = time.Hour + cfg.MaxBytes = 2*(relayMeteredOverhead+len(body)) + 1 + s2 := newTestRelayStorage(t, cfg) + for range 5 { + require.NoError(t, s2.Append(context.Background(), env)) + } + require.Eventually(t, func() bool { return len(relay2.requests()) == 2 }, 5*time.Second, 5*time.Millisecond) + for _, req := range relay2.requests() { + size := 0 + for _, rec := range req.records { + size += rec.metered() + } + assert.LessOrEqual(t, size, cfg.MaxBytes) + } + closeRelay(t, s2) + assert.Len(t, relay2.records(), 5) +} + +func TestRelayStorage_RetriesRetryableStatuses(t *testing.T) { + for _, status := range []int{http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout} { + t.Run(strconv.Itoa(status), func(t *testing.T) { + relay := newFakeRelay(t) + relay.respond = func(n int, w http.ResponseWriter) bool { + if n > 2 { + return false + } + w.WriteHeader(status) + return true + } + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + closeRelay(t, s) + + assert.Len(t, relay.requests(), 3) + assert.Equal(t, uint64(2), s.cfg.Metrics.Retries()) + assert.Equal(t, uint64(1), s.cfg.Metrics.Exported()) + assert.Empty(t, s.cfg.Metrics.Dropped()) + }) + } +} + +func TestRelayStorage_HonorsRetryAfter(t *testing.T) { + relay := newFakeRelay(t) + relay.respond = func(n int, w http.ResponseWriter) bool { + if n > 1 { + return false + } + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusServiceUnavailable) + return true + } + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + start := time.Now() + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + closeRelay(t, s) + assert.GreaterOrEqual(t, time.Since(start), time.Second) + assert.Len(t, relay.requests(), 2) +} + +func TestRelayStorage_DropsNonRetryableStatuses(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, http.StatusRequestEntityTooLarge, http.StatusInternalServerError} { + t.Run(strconv.Itoa(status), func(t *testing.T) { + relay := newFakeRelay(t) + relay.respond = func(_ int, w http.ResponseWriter) bool { + w.WriteHeader(status) + return true + } + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + require.NoError(t, s.Append(context.Background(), relayEnvelope(2, `{}`))) + closeRelay(t, s) + + assert.Len(t, relay.requests(), 1, "no retry") + assert.Equal(t, map[string]uint64{relayDropRejected: 2}, s.cfg.Metrics.Dropped()) + assert.Equal(t, map[string]uint64{strconv.Itoa(status): 1}, s.cfg.Metrics.Failed()) + assert.Zero(t, s.cfg.Metrics.Retries()) + }) + } +} + +func TestRelayStorage_DropsOnConnectionFailure(t *testing.T) { + relay := newFakeRelay(t) + relay.srv.Close() + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + closeRelay(t, s) + assert.Equal(t, map[string]uint64{"error": 1}, s.cfg.Metrics.Failed()) + assert.Equal(t, map[string]uint64{relayDropRejected: 1}, s.cfg.Metrics.Dropped()) +} + +func TestRelayStorage_DoesNotFollowRedirects(t *testing.T) { + var elsewhere atomic.Int32 + other := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { elsewhere.Add(1) })) + defer other.Close() + relay := newFakeRelay(t) + relay.respond = func(_ int, w http.ResponseWriter) bool { + w.Header().Set("Location", other.URL+relayTestPath) + w.WriteHeader(http.StatusTemporaryRedirect) + return true + } + s := newTestRelayStorage(t, testRelayConfig(relay.srv.URL)) + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + closeRelay(t, s) + assert.Zero(t, elsewhere.Load(), "the credential must not follow a redirect") + assert.Equal(t, map[string]uint64{"307": 1}, s.cfg.Metrics.Failed()) +} + +// A fork learns its token and metro-api after the sink is built, so both are +// read per batch. +func TestRelayStorage_ResolvesTokenAndBaseURLPerBatch(t *testing.T) { + seed, fork := newFakeRelay(t), newFakeRelay(t) + var base, token atomic.Pointer[string] + setIdentity := func(b, tok string) { base.Store(&b); token.Store(&tok) } + setIdentity(seed.srv.URL+"/ignored-base-path", "seed-jwt") + + cfg := testRelayConfig("") + cfg.BaseURLFunc = func() string { return *base.Load() } + cfg.TokenFunc = func() string { return *token.Load() } + s := newTestRelayStorage(t, cfg) + + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + require.Eventually(t, func() bool { return len(seed.requests()) == 1 }, 5*time.Second, 5*time.Millisecond) + setIdentity(fork.srv.URL, "fork-jwt") + require.NoError(t, s.Append(context.Background(), relayEnvelope(2, `{}`))) + closeRelay(t, s) + + assert.Equal(t, "Bearer seed-jwt", seed.requests()[0].auth) + require.Len(t, fork.requests(), 1) + assert.Equal(t, "Bearer fork-jwt", fork.requests()[0].auth) +} + +func TestRelayStorage_DropsOldestWhenQueueIsFull(t *testing.T) { + release := make(chan struct{}) + relay := newFakeRelay(t) + relay.respond = func(n int, _ http.ResponseWriter) bool { + if n == 1 { + <-release + } + return false + } + cfg := testRelayConfig(relay.srv.URL) + cfg.MaxRecords = 1 + cfg.MaxQueue = 3 + s := newTestRelayStorage(t, cfg) + + require.NoError(t, s.Append(context.Background(), relayEnvelope(0, `{}`))) + require.Eventually(t, func() bool { return len(relay.requests()) == 1 }, 5*time.Second, 5*time.Millisecond) + for i := 1; i <= 10; i++ { + require.NoError(t, s.Append(context.Background(), relayEnvelope(uint64(i), `{}`))) + } + assert.Equal(t, int64(3), s.cfg.Metrics.QueueDepth()) + assert.Equal(t, map[string]uint64{relayDropQueueFull: 7}, s.cfg.Metrics.Dropped()) + close(release) + closeRelay(t, s) + + var seqs []uint64 + for _, rec := range relay.records() { + var env Envelope + require.NoError(t, json.Unmarshal(rec.Body, &env)) + seqs = append(seqs, env.Seq) + } + assert.Equal(t, []uint64{0, 8, 9, 10}, seqs, "the newest records survive") +} + +func TestRelayStorage_CloseDropsWhatItCannotFlush(t *testing.T) { + relay := newFakeRelay(t) + relay.respond = func(_ int, w http.ResponseWriter) bool { + w.WriteHeader(http.StatusServiceUnavailable) + return true + } + cfg := testRelayConfig(relay.srv.URL) + cfg.retry = relayRetry{initial: time.Hour, max: time.Hour, maxElapsed: 2 * time.Hour} + s, err := newRelayStorage(context.Background(), cfg, slog.Default()) + require.NoError(t, err) + require.NoError(t, s.Append(context.Background(), relayEnvelope(1, `{}`))) + require.Eventually(t, func() bool { return len(relay.requests()) == 1 }, 5*time.Second, 5*time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + require.ErrorIs(t, s.Close(ctx), context.DeadlineExceeded) + assert.Equal(t, uint64(1), cfg.Metrics.Dropped()[relayDropShutdown], "a batch cut off by Close is a shutdown drop") + require.NoError(t, s.Append(context.Background(), relayEnvelope(2, `{}`))) + assert.Equal(t, uint64(2), cfg.Metrics.Dropped()[relayDropShutdown], "appends after Close are dropped") +} + +func TestRelayURL(t *testing.T) { + for base, want := range map[string]string{ + "https://169.254.0.1:8443": "https://169.254.0.1:8443/metro/btel-relay/v1/records", + "http://127.0.0.1:4000/metro/": "http://127.0.0.1:4000/metro/btel-relay/v1/records", + "https://h.example/x?y=1#z": "https://h.example/metro/btel-relay/v1/records", + } { + got, err := relayURL(base, "/metro/btel-relay/v1/records") + require.NoError(t, err) + assert.Equal(t, want, got, base) + } + for _, base := range []string{"", "://", "ftp://h/", "https://"} { + _, err := relayURL(base, relayTestPath) + assert.Error(t, err, base) + } +} diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index 92222d74..68577dd1 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "slices" "sync" "sync/atomic" "time" @@ -153,16 +154,13 @@ func (s *s2Storage) Close(ctx context.Context) error { } // S2StorageWriter reads from an EventStream and forwards each event after -// afterSeq to S2. Construct with NewS2StorageWriter, call Start to begin, Stop -// to drain and shut down. +// afterSeq to the storage open returns. Construct with NewS2StorageWriter, call +// Start to begin, Stop to drain and shut down. type S2StorageWriter struct { - es *EventStream - basin string - accessToken string - streamName string - afterSeq uint64 - cfg S2Config - log *slog.Logger + es *EventStream + open func(ctx context.Context) (Storage, error) + afterSeq uint64 + log *slog.Logger mu sync.Mutex started bool @@ -173,13 +171,12 @@ type S2StorageWriter struct { func NewS2StorageWriter(es *EventStream, basin, accessToken, streamName string, afterSeq uint64, cfg S2Config, log *slog.Logger) *S2StorageWriter { return &S2StorageWriter{ - es: es, - basin: basin, - accessToken: accessToken, - streamName: streamName, - afterSeq: afterSeq, - cfg: cfg, - log: log, + es: es, + open: func(ctx context.Context) (Storage, error) { + return newS2Storage(ctx, basin, accessToken, streamName, cfg, log) + }, + afterSeq: afterSeq, + log: log, } } @@ -192,7 +189,7 @@ func (w *S2StorageWriter) Start(ctx context.Context) error { if w.started { return fmt.Errorf("s2storagewriter: Start called more than once") } - storage, err := newS2Storage(ctx, w.basin, w.accessToken, w.streamName, w.cfg, w.log) + storage, err := w.open(ctx) if err != nil { return err } @@ -236,14 +233,17 @@ func (w *S2StorageWriter) Stop(ctx context.Context) error { } // S2StorageController opens at most one writer because StorageWriter is -// single-use and an append session is bound to one stream. +// single-use and an append session is bound to one stream. The writer appends +// either straight to S2 or through the metro-api relay, which appends to the +// same stream on the VM's behalf. type S2StorageController struct { es *EventStream - basin string - token string streamFn func() string - cfg S2Config - log *slog.Logger + // open builds the storage for a resolved stream. Nil when the sink is not + // configured, which makes Start a no-op. + open func(ctx context.Context, stream string) (Storage, error) + logAttr []any + log *slog.Logger mu sync.Mutex writer *S2StorageWriter @@ -261,7 +261,13 @@ type s2ControllerStart struct { // NewS2StorageController resolves streamFn at Start because a fork learns its // stream after construction. func NewS2StorageController(es *EventStream, basin, token string, streamFn func() string, cfg S2Config, log *slog.Logger) *S2StorageController { - return &S2StorageController{es: es, basin: basin, token: token, streamFn: streamFn, cfg: cfg, log: log} + c := &S2StorageController{es: es, streamFn: streamFn, logAttr: []any{"basin", basin}, log: log} + if basin != "" && token != "" { + c.open = func(ctx context.Context, stream string) (Storage, error) { + return newS2Storage(ctx, basin, token, stream, cfg, log) + } + } + return c } // Start opens the writer, which stores only events published after afterSeq: @@ -282,7 +288,7 @@ func (c *S2StorageController) Start(parent context.Context, afterSeq uint64) (er } return start.err } - if c.basin == "" || c.token == "" { + if c.open == nil { c.mu.Unlock() return nil } @@ -302,12 +308,17 @@ func (c *S2StorageController) Start(parent context.Context, afterSeq uint64) (er return nil } runCtx, cancel := context.WithCancel(parent) - w := NewS2StorageWriter(c.es, c.basin, c.token, stream, afterSeq, c.cfg, c.log) + w := &S2StorageWriter{ + es: c.es, + open: func(ctx context.Context) (Storage, error) { return c.open(ctx, stream) }, + afterSeq: afterSeq, + log: c.log, + } if err = w.Start(runCtx); err != nil { cancel() return err } - c.log.Info("S2 storage enabled", "basin", c.basin, "stream", stream) + c.log.Info("S2 storage enabled", append(slices.Clone(c.logAttr), "stream", stream)...) c.mu.Lock() c.writer, c.cancel, c.everStarted = w, cancel, true c.mu.Unlock() diff --git a/server/lib/events/storagecontroller_test.go b/server/lib/events/storagecontroller_test.go new file mode 100644 index 00000000..cae92e75 --- /dev/null +++ b/server/lib/events/storagecontroller_test.go @@ -0,0 +1,97 @@ +package events + +import ( + "context" + "log/slog" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// storageControllers builds each storage controller over the same stream +// resolver, so both sinks are held to one lifecycle contract. +func storageControllers(t *testing.T) map[string]func(es *EventStream, streamFn func() string) *S2StorageController { + relay := newFakeRelay(t) + return map[string]func(*EventStream, func() string) *S2StorageController{ + "s2": func(es *EventStream, streamFn func() string) *S2StorageController { + return NewS2StorageController(es, "test-basin", "test-token", streamFn, S2Config{}, slog.Default()) + }, + "relay": func(es *EventStream, streamFn func() string) *S2StorageController { + return NewRelayStorageController(es, testRelayConfig(relay.srv.URL), streamFn, slog.Default()) + }, + } +} + +func TestStorageControllers_Lifecycle(t *testing.T) { + for name, build := range storageControllers(t) { + t.Run(name, func(t *testing.T) { + var resolved atomic.Int32 + c := build(newTestStream(t, 64), func() string { + resolved.Add(1) + return "test-stream" + }) + assert.Zero(t, resolved.Load(), "nothing resolves before Start") + assert.False(t, c.Running()) + + require.NoError(t, c.Start(context.Background(), 0)) + first := c.writer + require.NoError(t, c.Start(context.Background(), 0)) + assert.Same(t, first, c.writer, "a second Start keeps the writer") + assert.Equal(t, int32(1), resolved.Load()) + assert.True(t, c.Running()) + + require.NoError(t, stopController(t, c)) + assert.False(t, c.Running()) + assert.True(t, c.EverStarted(), "EverStarted survives Stop") + }) + } +} + +// A fork resolves no stream until its identity arrives, and storage must stay +// closed until then. +func TestStorageControllers_ForkGate(t *testing.T) { + for name, build := range storageControllers(t) { + t.Run(name, func(t *testing.T) { + var stream atomic.Pointer[string] + empty := "" + stream.Store(&empty) + c := build(newTestStream(t, 64), func() string { return *stream.Load() }) + + require.NoError(t, c.Start(context.Background(), 0)) + assert.False(t, c.EverStarted()) + + applied := "fork-stream" + stream.Store(&applied) + require.NoError(t, c.Start(context.Background(), 0)) + assert.True(t, c.EverStarted()) + require.NoError(t, stopController(t, c)) + }) + } +} + +func TestStorageControllers_CancelledStartRollsBack(t *testing.T) { + for name, build := range storageControllers(t) { + t.Run(name, func(t *testing.T) { + c := build(newTestStream(t, 64), func() string { return "test-stream" }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.ErrorIs(t, c.Start(ctx, 0), context.Canceled) + assert.False(t, c.EverStarted()) + }) + } +} + +func TestRelayStorageController_ForwardsEventsAfterStart(t *testing.T) { + relay := newFakeRelay(t) + es := newTestStream(t, 64) + es.Publish(Envelope{Event: makeEvent("captured before storage")}) + c := NewRelayStorageController(es, testRelayConfig(relay.srv.URL), func() string { return "s" }, slog.Default()) + require.NoError(t, c.Start(context.Background(), es.Seq())) + es.Publish(Envelope{Event: makeEvent("stored")}) + require.Eventually(t, func() bool { return len(relay.records()) == 1 }, 5*time.Second, 5*time.Millisecond) + require.NoError(t, stopController(t, c)) + assert.Len(t, relay.records(), 1) +} diff --git a/server/lib/metrics/btelrelay.go b/server/lib/metrics/btelrelay.go new file mode 100644 index 00000000..2295115b --- /dev/null +++ b/server/lib/metrics/btelrelay.go @@ -0,0 +1,57 @@ +package metrics + +import ( + "context" + "slices" +) + +// RelaySource exposes the btel storage relay sink's lifetime counters. +// Implemented by *events.RelayMetrics. +type RelaySource interface { + Exported() uint64 + Retries() uint64 + QueueDepth() int64 + Dropped() map[string]uint64 + Failed() map[string]uint64 +} + +// RelayCollector reports btel storage relay health: records exported, records +// dropped by reason, retried and failed batches, and the send queue depth. +type RelayCollector struct { + src RelaySource +} + +func NewRelayCollector(src RelaySource) *RelayCollector { + return &RelayCollector{src: src} +} + +func (c *RelayCollector) Name() string { return "btel_relay" } + +func (c *RelayCollector) Collect(_ context.Context, w *Writer) error { + w.Metric("kernel_btel_relay_exported_total", "Telemetry storage records the metro-api relay acknowledged.", "counter") + w.Sample("kernel_btel_relay_exported_total", nil, float64(c.src.Exported())) + + w.Metric("kernel_btel_relay_dropped_total", "Telemetry storage records dropped, by reason.", "counter") + writeLabeled(w, "kernel_btel_relay_dropped_total", "reason", c.src.Dropped()) + + w.Metric("kernel_btel_relay_retries_total", "Telemetry storage batches retried after a retryable response.", "counter") + w.Sample("kernel_btel_relay_retries_total", nil, float64(c.src.Retries())) + + w.Metric("kernel_btel_relay_failed_total", "Telemetry storage batches given up on, by HTTP status.", "counter") + writeLabeled(w, "kernel_btel_relay_failed_total", "status", c.src.Failed()) + + w.Metric("kernel_btel_relay_queue_depth", "Telemetry storage records waiting to be sent to the relay.", "gauge") + w.Sample("kernel_btel_relay_queue_depth", nil, float64(c.src.QueueDepth())) + return nil +} + +func writeLabeled(w *Writer, name, label string, values map[string]uint64) { + keys := make([]string, 0, len(values)) + for k := range values { + keys = append(keys, k) + } + slices.Sort(keys) + for _, k := range keys { + w.Sample(name, []Label{{Name: label, Value: k}}, float64(values[k])) + } +} diff --git a/server/lib/metrics/btelrelay_test.go b/server/lib/metrics/btelrelay_test.go new file mode 100644 index 00000000..60e4b2c4 --- /dev/null +++ b/server/lib/metrics/btelrelay_test.go @@ -0,0 +1,33 @@ +package metrics + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type stubRelaySource struct{} + +func (stubRelaySource) Exported() uint64 { return 42 } +func (stubRelaySource) Retries() uint64 { return 3 } +func (stubRelaySource) QueueDepth() int64 { return 5 } +func (stubRelaySource) Dropped() map[string]uint64 { + return map[string]uint64{"queue_full": 7, "rejected": 2} +} +func (stubRelaySource) Failed() map[string]uint64 { return map[string]uint64{"403": 1} } + +func TestRelayCollector(t *testing.T) { + w := &Writer{} + require.NoError(t, NewRelayCollector(stubRelaySource{}).Collect(context.Background(), w)) + + out := string(w.Bytes()) + assert.Contains(t, out, "kernel_btel_relay_exported_total 42") + assert.Contains(t, out, `kernel_btel_relay_dropped_total{reason="queue_full"} 7`) + assert.Contains(t, out, `kernel_btel_relay_dropped_total{reason="rejected"} 2`) + assert.Contains(t, out, "kernel_btel_relay_retries_total 3") + assert.Contains(t, out, `kernel_btel_relay_failed_total{status="403"} 1`) + assert.Contains(t, out, "kernel_btel_relay_queue_depth 5") + assert.Contains(t, out, "# TYPE kernel_btel_relay_queue_depth gauge") +}