feat: add GET /telemetry/events to read archived telemetry from S2

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: archandatta

server/cmd/api/api/api.go

@@ -92,6 +92,17 @@ type ApiService struct {
 	monitorMu        sync.Mutex
 	lifecycleCtx     context.Context
 	lifecycleCancel  context.CancelFunc
+
+	// Durable S2 telemetry storage. All three must be set for reads to hit S2;
+	// they mirror the values that gate the S2 storage writer.
+	s2Basin       string
+	s2AccessToken string
+	s2Stream      string
+}
+
+// s2Enabled reports whether durable S2 telemetry storage is configured.
+func (s *ApiService) s2Enabled() bool {
+	return s.s2Basin != "" && s.s2AccessToken != "" && s.s2Stream != ""
 }
 
 var _ oapi.StrictServerInterface = (*ApiService)(nil)
@@ -105,6 +116,9 @@ func New(
 	telemetrySession *telemetry.TelemetrySession,
 	eventStream *events.EventStream,
 	displayNum int,
+	s2Basin string,
+	s2AccessToken string,
+	s2Stream string,
 ) (*ApiService, error) {
 	switch {
 	case recordManager == nil:
@@ -140,6 +154,9 @@ func New(
 		cdpMonitor:        mon,
 		lifecycleCtx:      ctx,
 		lifecycleCancel:   cancel,
+		s2Basin:           s2Basin,
+		s2AccessToken:     s2AccessToken,
+		s2Stream:          s2Stream,
 	}, nil
 }
 

server/cmd/api/api/api_test.go

@@ -318,7 +318,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, "", "", "")
 }
 
 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, "", "", "")
 	require.NoError(t, err)
 	return svc
 }

server/cmd/api/api/events.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/kernel/kernel-images/server/lib/events"
+	"github.com/kernel/kernel-images/server/lib/logger"
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
 )
 
@@ -123,6 +124,115 @@ func (s *ApiService) StreamTelemetryEvents(ctx context.Context, req oapi.StreamT
 	return oapi.StreamTelemetryEvents200TexteventStreamResponse{Body: pr, Headers: headers}, nil
 }
 
+// defaultReadWindow bounds a read that supplies no since/until.
+const defaultReadWindow = 5 * time.Minute
+
+// maxReadLimit caps the number of envelopes a single read returns.
+const maxReadLimit = 1000
+
+// ReadTelemetryEvents handles GET /telemetry/events.
+// Reads archived telemetry envelopes for the current session from durable S2
+// storage, applies category and limit filters, and returns them in ascending
+// sequence order. Returns an empty list when S2 storage is not configured.
+func (s *ApiService) ReadTelemetryEvents(ctx context.Context, req oapi.ReadTelemetryEventsRequestObject) (oapi.ReadTelemetryEventsResponseObject, error) {
+	log := logger.FromContext(ctx)
+
+	if !s.s2Enabled() {
+		return readTelemetryEventsOKResponse{}, nil
+	}
+
+	startSeq := s.telemetrySession.SessionStartSeq()
+	envs, err := events.Read(ctx, s.s2Basin, s.s2AccessToken, s.s2Stream, buildReadOptions(req.Params), log)
+	if err != nil {
+		log.Error("failed to read telemetry events from S2", "err", err)
+		return oapi.ReadTelemetryEvents500JSONResponse{InternalErrorJSONResponse: oapi.InternalErrorJSONResponse{Message: "failed to read telemetry events"}}, nil
+	}
+
+	envs = dropPriorSessions(envs, startSeq)
+	envs = filterByCategory(envs, req.Params.Category)
+	envs = capLimit(envs, req.Params.Limit)
+
+	return readTelemetryEventsOKResponse{envs: envs}, nil
+}
+
+// buildReadOptions maps query params to a bounded read window. since/until are
+// the start/end of the window; the window defaults to the last defaultReadWindow.
+// limit is applied after category filtering, not pushed into the S2 read.
+func buildReadOptions(p oapi.ReadTelemetryEventsParams) events.ReadOptions {
+	var opts events.ReadOptions
+	if p.Since != nil {
+		start := uint64(*p.Since)
+		opts.Timestamp = &start
+	} else {
+		start := uint64(time.Now().Add(-defaultReadWindow).UnixMilli())
+		opts.Timestamp = &start
+	}
+	if p.Until != nil {
+		until := uint64(*p.Until)
+		opts.Until = &until
+	}
+	return opts
+}
+
+// dropPriorSessions removes envelopes from before the current session's start.
+// startSeq is 0 when no session has run, in which case nothing is dropped.
+func dropPriorSessions(envs []events.Envelope, startSeq uint64) []events.Envelope {
+	if startSeq == 0 {
+		return envs
+	}
+	out := make([]events.Envelope, 0, len(envs))
+	for _, e := range envs {
+		if e.Seq >= startSeq {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+func filterByCategory(envs []events.Envelope, cats *[]oapi.TelemetryEventCategory) []events.Envelope {
+	if cats == nil || len(*cats) == 0 {
+		return envs
+	}
+	want := make(map[oapi.TelemetryEventCategory]struct{}, len(*cats))
+	for _, c := range *cats {
+		want[c] = struct{}{}
+	}
+	out := make([]events.Envelope, 0, len(envs))
+	for _, e := range envs {
+		if _, ok := want[e.Event.Category]; ok {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+func capLimit(envs []events.Envelope, limit *int) []events.Envelope {
+	n := maxReadLimit
+	if limit != nil && *limit > 0 && *limit < n {
+		n = *limit
+	}
+	if len(envs) > n {
+		return envs[:n]
+	}
+	return envs
+}
+
+// readTelemetryEventsOKResponse serializes events.Envelope directly so the
+// response shape matches the SSE stream frames and the publish endpoint.
+type readTelemetryEventsOKResponse struct{ envs []events.Envelope }
+
+func (r readTelemetryEventsOKResponse) VisitReadTelemetryEventsResponse(w http.ResponseWriter) error {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	envs := r.envs
+	if envs == nil {
+		envs = []events.Envelope{}
+	}
+	return json.NewEncoder(w).Encode(struct {
+		Events []events.Envelope `json:"events"`
+	}{Events: envs})
+}
+
 // publishTelemetryEventOKResponse serializes events.Envelope directly so the response
 // is identical in shape to the SSE stream frames.
 type publishTelemetryEventOKResponse struct{ env events.Envelope }

server/cmd/api/api/events_test.go

@@ -4,6 +4,8 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
@@ -155,3 +157,69 @@ func TestPublishDroppedWhenCategoryDisabled(t *testing.T) {
 	require.NoError(t, err)
 	assert.IsType(t, oapi.PublishTelemetryEvent204Response{}, resp, "events in disabled categories should return 204")
 }
+
+func TestReadTelemetryEventsS2Disabled(t *testing.T) {
+	t.Parallel()
+	svc := newTestService(t, newMockRecordManager()) // s2 creds empty -> disabled
+
+	resp, err := svc.ReadTelemetryEvents(context.Background(), oapi.ReadTelemetryEventsRequestObject{})
+	require.NoError(t, err)
+	ok, isOK := resp.(readTelemetryEventsOKResponse)
+	require.True(t, isOK, "expected 200 response when S2 is disabled")
+
+	rec := httptest.NewRecorder()
+	require.NoError(t, ok.VisitReadTelemetryEventsResponse(rec))
+	assert.Equal(t, http.StatusOK, rec.Code)
+	// Empty result must serialize as [] not null, or the Python SDK chokes.
+	assert.JSONEq(t, `{"events":[]}`, rec.Body.String())
+}
+
+func TestDropPriorSessions(t *testing.T) {
+	t.Parallel()
+	envs := []events.Envelope{{Seq: 1}, {Seq: 2}, {Seq: 3}}
+
+	got := dropPriorSessions(envs, 2)
+	require.Len(t, got, 2)
+	assert.Equal(t, uint64(2), got[0].Seq)
+
+	// startSeq 0 means no session ran; keep everything.
+	assert.Len(t, dropPriorSessions(envs, 0), 3)
+}
+
+func TestFilterByCategory(t *testing.T) {
+	t.Parallel()
+	mk := func(c oapi.TelemetryEventCategory) events.Envelope {
+		return events.Envelope{Event: events.Event{Category: c}}
+	}
+	envs := []events.Envelope{mk(events.Console), mk(events.Network), mk(events.Console)}
+
+	assert.Len(t, filterByCategory(envs, nil), 3, "nil filter keeps everything")
+
+	cats := []oapi.TelemetryEventCategory{events.Console}
+	assert.Len(t, filterByCategory(envs, &cats), 2)
+}
+
+func TestCapLimit(t *testing.T) {
+	t.Parallel()
+	envs := make([]events.Envelope, 5)
+
+	assert.Len(t, capLimit(envs, nil), 5, "no limit returns all under the ceiling")
+
+	three := 3
+	assert.Len(t, capLimit(envs, &three), 3)
+}
+
+func TestBuildReadOptions(t *testing.T) {
+	t.Parallel()
+	// No params: defaults the start to roughly defaultReadWindow ago, no end bound.
+	opts := buildReadOptions(oapi.ReadTelemetryEventsParams{})
+	require.NotNil(t, opts.Timestamp)
+	assert.Nil(t, opts.Until)
+
+	since, until := int64(1000), int64(2000)
+	opts = buildReadOptions(oapi.ReadTelemetryEventsParams{Since: &since, Until: &until})
+	require.NotNil(t, opts.Timestamp)
+	require.NotNil(t, opts.Until)
+	assert.Equal(t, uint64(1000), *opts.Timestamp)
+	assert.Equal(t, uint64(2000), *opts.Until)
+}

server/cmd/api/api/telemetry_test.go

@@ -358,7 +358,7 @@ func (m *mockRecordManager) StopAll(_ context.Context) error
 func newTestService(t *testing.T, mgr recorder.RecordManager) *ApiService {
 	t.Helper()
 	ts, es := newTelemetrySession(t)
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0)
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), ts, es, 0, "", "", "")
 	require.NoError(t, err)
 	svc.cdpMonitor = &stubCdpMonitor{}
 	return svc

server/cmd/api/main.go

@@ -137,6 +137,9 @@ func main() {
 		telemetrySession,
 		eventStream,
 		config.DisplayNum,
+		config.S2Basin,
+		config.S2AccessToken,
+		config.S2Stream,
 	)
 	if err != nil {
 		slogger.Error("failed to create api service", "err", err)