Serve browser telemetry storage through the metro-api relay

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
 

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 {

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.

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))
+}

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
 }

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{