OTLP browser telemetry export: follow-up improvements (KERNEL-1596)

server/cmd/api/api/api.go

@@ -30,6 +30,16 @@ type cdpMonitorController interface {
 
 var _ cdpMonitorController = (*cdpmonitor.Monitor)(nil)
 
+// OTLPExporter controls the optional OTLP export sink. Nil when no export
+// endpoint is provisioned on the VM. Implemented by *events.OTLPExportController.
+type OTLPExporter interface {
+	Start(ctx context.Context) error
+	Stop(ctx context.Context) error
+	Running() bool
+}
+
+var _ OTLPExporter = (*events.OTLPExportController)(nil)
+
 type ApiService struct {
 	// defaultRecorderID is used whenever the caller doesn't specify an explicit ID.
 	defaultRecorderID string
@@ -89,6 +99,7 @@ type ApiService struct {
 	eventStream      *events.EventStream
 	telemetrySession *telemetry.TelemetrySession
 	cdpMonitor       cdpMonitorController
+	otlpExport       OTLPExporter
 	monitorMu        sync.Mutex
 	lifecycleCtx     context.Context
 	lifecycleCancel  context.CancelFunc
@@ -105,6 +116,7 @@ func New(
 	telemetrySession *telemetry.TelemetrySession,
 	eventStream *events.EventStream,
 	displayNum int,
+	otlpExport OTLPExporter,
 ) (*ApiService, error) {
 	switch {
 	case recordManager == nil:
@@ -138,6 +150,7 @@ func New(
 		eventStream:       eventStream,
 		telemetrySession:  telemetrySession,
 		cdpMonitor:        mon,
+		otlpExport:        otlpExport,
 		lifecycleCtx:      ctx,
 		lifecycleCancel:   cancel,
 	}, nil
@@ -369,5 +382,7 @@ func (s *ApiService) Shutdown(ctx context.Context) error {
 	s.cdpMonitor.Stop()
 	s.telemetrySession.Stop()
 	s.monitorMu.Unlock()
+	// The OTLP export sink is stopped by main after the servers drain, so any
+	// events they emit on the way down are still exported (mirrors s2Writer).
 	return s.recordManager.StopAll(ctx)
 }

server/cmd/api/api/api_test.go

@@ -321,7 +321,7 @@ func newTelemetrySession(t *testing.T) (*telemetry.TelemetrySession, *events.Eve
 func newSvc(t *testing.T, mgr recorder.RecordManager) (*ApiService, error) {
 	t.Helper()
 	ts, es := newTelemetrySession(t)
-	return New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0)
+	return New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0, nil)
 }
 
 func TestApiService_PatchChromiumFlags(t *testing.T) {

server/cmd/api/api/display_test.go

@@ -36,7 +36,7 @@ func testFFmpegFactory(t *testing.T, tempDir string) recorder.FFmpegRecorderFact
 func newTestServiceWithFactory(t *testing.T, mgr recorder.RecordManager, factory recorder.FFmpegRecorderFactory) *ApiService {
 	t.Helper()
 	ts, es := newTelemetrySession(t)
-	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0)
+	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0, nil)
 	require.NoError(t, err)
 	return svc
 }

server/cmd/api/api/telemetry.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"time"
 
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
 	"github.com/nrednav/cuid2"
@@ -12,6 +13,9 @@ import (
 	"github.com/kernel/kernel-images/server/lib/telemetry"
 )
 
+// otlpStopTimeout bounds how long a runtime export toggle-off waits to drain.
+const otlpStopTimeout = 5 * time.Second
+
 // GetTelemetry handles GET /telemetry.
 // Returns the current telemetry configuration. Returns 404 if telemetry is not configured.
 func (s *ApiService) GetTelemetry(_ context.Context, _ oapi.GetTelemetryRequestObject) (oapi.GetTelemetryResponseObject, error) {
@@ -41,7 +45,7 @@ func (s *ApiService) PutTelemetry(ctx context.Context, req oapi.PutTelemetryRequ
 	if allDisabled {
 		if wasActive {
 			s.telemetrySession.Stop()
-			s.stopTelemetryState()
+			s.stopTelemetryState(ctx)
 		}
 		return oapi.PutTelemetry200JSONResponse(oapi.TelemetryState{Config: disabledConfig(), Seq: int64(s.telemetrySession.Seq())}), nil
 	}
@@ -58,10 +62,11 @@ func (s *ApiService) PutTelemetry(ctx context.Context, req oapi.PutTelemetryRequ
 	}
 
 	if err := s.reconcileTelemetryState(cfg.Categories); err != nil {
-		s.rollbackTelemetry(wasActive, prev)
+		s.rollbackTelemetry(ctx, wasActive, prev)
 		logger.FromContext(ctx).Error("failed to apply telemetry state", "err", err)
 		return oapi.PutTelemetry500JSONResponse{InternalErrorJSONResponse: oapi.InternalErrorJSONResponse{Message: "failed to start telemetry"}}, nil
 	}
+	s.reconcileExport(ctx, cfg.ExportOTLP)
 
 	if wasActive {
 		return oapi.PutTelemetry200JSONResponse(s.buildTelemetryResponse()), nil
@@ -80,26 +85,29 @@ func (s *ApiService) PatchTelemetry(ctx context.Context, req oapi.PatchTelemetry
 		return oapi.PatchTelemetry404JSONResponse{NotFoundErrorJSONResponse: oapi.NotFoundErrorJSONResponse{Message: "telemetry is not configured"}}, nil
 	}
 
-	if req.Body == nil || req.Body.Browser == nil {
+	// Nothing to merge when neither the category block nor the export toggle is
+	// present; skip the reconcile and echo current state, as before.
+	if req.Body == nil || (req.Body.Browser == nil && req.Body.Export == nil) {
 		return oapi.PatchTelemetry200JSONResponse(s.buildTelemetryResponse()), nil
 	}
 
 	prev := s.telemetrySession.Config()
-	cfg, allDisabled := mergeTelemetryConfig(prev, req.Body.Browser)
+	cfg, allDisabled := mergeTelemetryConfig(prev, req.Body)
 	if allDisabled {
 		s.telemetrySession.Stop()
-		s.stopTelemetryState()
+		s.stopTelemetryState(ctx)
 		return oapi.PatchTelemetry200JSONResponse(oapi.TelemetryState{Config: disabledConfig(), Seq: int64(s.telemetrySession.Seq())}), nil
 	}
 
 	// Commit first so the filter is live before the collector emits, then
 	// reconcile and roll back on collector-start failure.
 	s.telemetrySession.UpdateConfig(cfg)
 	if err := s.reconcileTelemetryState(cfg.Categories); err != nil {
-		s.rollbackTelemetry(true, prev)
+		s.rollbackTelemetry(ctx, true, prev)
 		logger.FromContext(ctx).Error("failed to apply telemetry state", "err", err)
 		return oapi.PatchTelemetry500JSONResponse{InternalErrorJSONResponse: oapi.InternalErrorJSONResponse{Message: "failed to apply telemetry"}}, nil
 	}
+	s.reconcileExport(ctx, cfg.ExportOTLP)
 	return oapi.PatchTelemetry200JSONResponse(s.buildTelemetryResponse()), nil
 }
 
@@ -129,22 +137,50 @@ func (s *ApiService) reconcileTelemetryState(cats []oapi.TelemetryEventCategory)
 // A fresh session is torn down; an updated session is reverted to prev. Reverting
 // never requires a fallible collector start (the failed start left it stopped),
 // so the reconcile here cannot fail.
-func (s *ApiService) rollbackTelemetry(wasActive bool, prev telemetry.TelemetryConfig) {
+func (s *ApiService) rollbackTelemetry(ctx context.Context, wasActive bool, prev telemetry.TelemetryConfig) {
 	if !wasActive {
 		s.telemetrySession.Stop()
-		s.stopTelemetryState()
+		s.stopTelemetryState(ctx)
 		return
 	}
 	s.telemetrySession.UpdateConfig(prev)
 	_ = s.reconcileTelemetryState(prev.Categories)
+	s.reconcileExport(ctx, prev.ExportOTLP)
+}
+
+// reconcileExport starts or stops the OTLP export sink to match the desired
+// state. Export is best-effort: a failed start is logged, never surfaced as an
+// error, so it cannot fail a telemetry apply. No-op when no export destination
+// is configured.
+func (s *ApiService) reconcileExport(ctx context.Context, enabled bool) {
+	if s.otlpExport == nil {
+		if enabled {
+			logger.FromContext(ctx).Warn("otlp export enabled but no destination configured; export inactive")
+		}
+		return
+	}
+	switch {
+	case enabled && !s.otlpExport.Running():
+		// Root the export on the app lifecycle, not this request; only its logs
+		// carry the request context.
+		if err := s.otlpExport.Start(s.lifecycleCtx); err != nil {
+			logger.FromContext(ctx).Error("otlp export failed to start", "err", err)
+		}
+	case !enabled && s.otlpExport.Running():
+		// Draining must outlive the request, so bound it independently.
+		stopCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), otlpStopTimeout)
+		defer cancel()
+		_ = s.otlpExport.Stop(stopCtx)
+	}
 }
 
-// stopTelemetryState tears down the collector and middleware after a session is
-// cleared.
-func (s *ApiService) stopTelemetryState() {
+// stopTelemetryState tears down the collector, export sink, and middleware
+// after a session is cleared.
+func (s *ApiService) stopTelemetryState(ctx context.Context) {
 	if s.cdpMonitor.IsRunning() {
 		s.cdpMonitor.Stop()
 	}
+	s.reconcileExport(ctx, false)
 	DisableTelemetryMiddleware()
 }
 
@@ -203,11 +239,12 @@ func containsCategory(cats []oapi.TelemetryEventCategory, target oapi.TelemetryE
 // the categories explicitly enabled there are captured (anything omitted is off). Returns the
 // config, whether the result is empty (stop signal), and any error.
 func telemetryConfigFromOAPI(cfg *oapi.BrowserTelemetryConfig) (telemetry.TelemetryConfig, bool, error) {
+	exportOTLP := exportOTLPFromOAPI(cfg)
 	if cfg == nil || cfg.Browser == nil {
 		// No per-category settings: resolve to the explicit default set so the
 		// effective categories are known before the collector is reconciled.
 		cats := append([]oapi.TelemetryEventCategory(nil), events.DefaultCategories...)
-		return telemetry.TelemetryConfig{Categories: cats}, false, nil
+		return telemetry.TelemetryConfig{Categories: cats, ExportOTLP: exportOTLP}, false, nil
 	}
 
 	cats := make([]oapi.TelemetryEventCategory, 0, len(events.UserCategories))
@@ -219,13 +256,22 @@ func telemetryConfigFromOAPI(cfg *oapi.BrowserTelemetryConfig) (telemetry.Teleme
 	if len(cats) == 0 {
 		return telemetry.TelemetryConfig{}, true, nil
 	}
-	return telemetry.TelemetryConfig{Categories: cats}, false, nil
+	return telemetry.TelemetryConfig{Categories: cats, ExportOTLP: exportOTLP}, false, nil
+}
+
+// exportOTLPFromOAPI reads the OTLP export toggle from a config, defaulting to
+// false (off) when the export block is omitted.
+func exportOTLPFromOAPI(cfg *oapi.BrowserTelemetryConfig) bool {
+	if cfg != nil && cfg.Export != nil && cfg.Export.Otlp != nil && cfg.Export.Otlp.Enabled != nil {
+		return *cfg.Export.Otlp.Enabled
+	}
+	return false
 }
 
 // mergeTelemetryConfig applies patch overrides onto current, returning the merged config and
 // whether every configurable category ended up disabled (stop signal). Only categories with an
 // explicit Enabled field in patch are changed; omitted categories keep their current state.
-func mergeTelemetryConfig(current telemetry.TelemetryConfig, patch *oapi.BrowserTelemetryCategoriesConfig) (telemetry.TelemetryConfig, bool) {
+func mergeTelemetryConfig(current telemetry.TelemetryConfig, patch *oapi.BrowserTelemetryConfig) (telemetry.TelemetryConfig, bool) {
 	userCat := categorySetOf(events.UserCategories)
 	active := make(map[oapi.TelemetryEventCategory]struct{}, len(current.Categories))
 	for _, c := range current.Categories {
@@ -234,25 +280,33 @@ func mergeTelemetryConfig(current telemetry.TelemetryConfig, patch *oapi.Browser
 		}
 	}
 
-	for _, f := range categoryFields(patch) {
-		if f.config == nil || f.config.Enabled == nil {
-			continue // not mentioned in patch — keep current state
-		}
-		if *f.config.Enabled {
-			active[f.category] = struct{}{}
-		} else {
-			delete(active, f.category)
+	if patch.Browser != nil {
+		for _, f := range categoryFields(patch.Browser) {
+			if f.config == nil || f.config.Enabled == nil {
+				continue // not mentioned in patch; keep current state
+			}
+			if *f.config.Enabled {
+				active[f.category] = struct{}{}
+			} else {
+				delete(active, f.category)
+			}
 		}
 	}
 
+	// Export follows the same patch semantics: an omitted toggle is unchanged.
+	exportOTLP := current.ExportOTLP
+	if patch.Export != nil && patch.Export.Otlp != nil && patch.Export.Otlp.Enabled != nil {
+		exportOTLP = *patch.Export.Otlp.Enabled
+	}
+
 	if len(active) == 0 {
 		return telemetry.TelemetryConfig{}, true
 	}
 	cats := make([]oapi.TelemetryEventCategory, 0, len(active))
 	for c := range active {
 		cats = append(cats, c)
 	}
-	return telemetry.TelemetryConfig{Categories: cats}, false
+	return telemetry.TelemetryConfig{Categories: cats, ExportOTLP: exportOTLP}, false
 }
 
 // disabledConfig returns a BrowserTelemetryConfig with every configurable category explicitly disabled.
@@ -272,6 +326,14 @@ func disabledConfig() oapi.BrowserTelemetryConfig {
 			Screenshot:  off(),
 			Captcha:     off(),
 		},
+		Export: exportConfigToOAPI(false),
+	}
+}
+
+// exportConfigToOAPI renders the OTLP export toggle for API responses.
+func exportConfigToOAPI(enabled bool) *oapi.BrowserTelemetryExportConfig {
+	return &oapi.BrowserTelemetryExportConfig{
+		Otlp: &oapi.BrowserTelemetryOTLPExportConfig{Enabled: lo.ToPtr(enabled)},
 	}
 }
 
@@ -295,5 +357,6 @@ func telemetryConfigToOAPI(cfg telemetry.TelemetryConfig) oapi.BrowserTelemetryC
 			Screenshot:  enabled(events.Screenshot),
 			Captcha:     enabled(events.Captcha),
 		},
+		Export: exportConfigToOAPI(cfg.ExportOTLP),
 	}
 }