Skip to content

Commit a213e33

Browse files
archandattaarchandatta
committed
review: address should-fix and nit feedback on telemetry stream
1 parent 1c71174 commit a213e33

3 files changed

Lines changed: 108 additions & 27 deletions

File tree

‎README.md‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,8 @@ Commands with JSON output support:
128128
- **Apps**: `list`, `history`
129129
- **Deploy**: `deploy` (JSONL streaming), `history`
130130
- **Invoke**: `invoke` (JSONL streaming), `history`
131-
- **Browser Sub-commands**: `replays list/start`, `process exec/spawn`, `fs file-info/list-files`, `telemetry stream`
131+
- **Browser Sub-commands**: `replays list/start`, `process exec/spawn`, `fs file-info/list-files`
132+
- **Browser NDJSON streaming**: `telemetry stream`
132133

133134
### Authentication
134135

@@ -211,8 +212,7 @@ Commands with JSON output support:
211212
- `--start-url <url>` - Initial page to open on launch
212213
- `--pool-id <id>` - Acquire a browser from the specified pool (mutually exclusive with --pool-name; ignores other session flags)
213214
- `--pool-name <name>` - Acquire a browser from the pool name (mutually exclusive with --pool-id; ignores other session flags)
214-
- `--telemetry=all` - Enable telemetry for all categories
215-
- `--telemetry=<list>` - Per-category config, e.g. `--telemetry=network=on,page=off`
215+
- `--telemetry=all` - Enable telemetry for all categories; `--telemetry=off` to disable; `--telemetry=network=on,page=off` for per-category
216216
- `--output json`, `-o json` - Output raw JSON object
217217
- _Note: When a pool is specified, omit other session configuration flags—pool settings determine profile, proxy, viewport, etc._
218218
- `kernel browsers delete <id>` - Delete a browser
@@ -221,7 +221,9 @@ Commands with JSON output support:
221221
- `kernel browsers get <id>` - Get detailed browser session info
222222
- `--output json`, `-o json` - Output raw JSON object
223223
- `kernel browsers update <id>` - Update a running browser session
224-
- `--telemetry=all` to enable all categories; `--telemetry=off` to disable; `--telemetry=network=on,page=off` for per-category
224+
- `--telemetry=all` - Enable telemetry for all categories
225+
- `--telemetry=off` - Disable telemetry
226+
- `--telemetry=<list>` - Per-category config, e.g. `--telemetry=network=on,page=off`
225227
- `--output json`, `-o json` - Output raw JSON object
226228
- `kernel browsers curl <id> <url>` - Make HTTP requests through a browser session's Chrome network stack
227229
- `-X, --request <method>` - HTTP method (default: GET; defaults to POST when `--data` is set)
@@ -293,11 +295,12 @@ Telemetry config is a sub-field of the browser session. Use `browsers update` to
293295
- Disable: `kernel browsers update <id> --telemetry=off`
294296
- Per-category: `kernel browsers update <id> --telemetry=network=on,page=off` (valid: `console`, `interaction`, `network`, `page`; `system` always emits and cannot be toggled)
295297

296-
- `kernel browsers telemetry stream <id>` - Stream live telemetry events
298+
- `kernel browsers telemetry stream <id>` - Stream live telemetry events (NDJSON with `-o json`)
297299
- `--categories <list>` - Filter by API event category (console,network,page,interaction,system); `system` covers all `monitor_*` events
298300
- `--types <list>` - Filter by event type (e.g. network_response,console_error)
299-
- `--seq <n>` - Resume stream from sequence number (Last-Event-ID)
301+
- `--seq <n>` - Resume stream from sequence number (Last-Event-ID); `--seq=0` resumes from the beginning
300302
- `-o, --output json` - Output newline-delimited JSON envelopes
303+
- Default output: `15:04:05 [network] network_response`
301304

302305
### Browser Process Control
303306

‎cmd/browsers_telemetry.go‎

Lines changed: 14 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -84,9 +84,6 @@ func applyTelemetryParam(s string) (kernel.BrowserTelemetryRequestConfigParam, e
8484
// "system" is always-on and cannot be toggled, but is valid as a --categories stream filter.
8585
var settableCategories = []string{"console", "interaction", "network", "page"}
8686

87-
// streamableCategories extends settableCategories with "system" for --categories stream filtering.
88-
var streamableCategories = append(settableCategories[:len(settableCategories):len(settableCategories)], "system")
89-
9087
// eventCategory derives the category from the event type prefix.
9188
// "monitor_*" maps to "system"; all others use the prefix before the first "_".
9289
// TODO(sdk): kernel-go-sdk should surface Category directly on BrowserTelemetryEventUnion.
@@ -113,39 +110,39 @@ func shouldEmit(ev kernel.BrowserTelemetryEventUnion, categories, types []string
113110
}
114111

115112
func (b BrowsersCmd) TelemetryStream(ctx context.Context, in BrowsersTelemetryStreamInput) error {
113+
if b.telemetry == nil {
114+
return fmt.Errorf("telemetry service not available")
115+
}
116116
if in.Output != "" && in.Output != "json" {
117117
return fmt.Errorf("unsupported --output value: use 'json'")
118118
}
119-
for _, c := range in.Categories {
120-
if !slices.Contains(streamableCategories, c) {
121-
return fmt.Errorf("unknown category %q: must be one of %s", c, strings.Join(streamableCategories, ", "))
122-
}
123-
}
124-
if b.telemetry == nil {
125-
pterm.Error.Println("telemetry service not available")
126-
return nil
127-
}
128119
br, err := b.browsers.Get(ctx, in.Identifier, kernel.BrowserGetParams{})
129120
if err != nil {
130121
return util.CleanedUpSdkError{Err: err}
131122
}
132123
params := kernel.BrowserTelemetryStreamParams{}
133-
if in.Seq > 0 {
124+
if in.Seq >= 0 {
134125
params.LastEventID = kernel.Opt(strconv.FormatInt(in.Seq, 10))
135126
}
136127
stream := b.telemetry.StreamStreaming(ctx, br.SessionID, params)
137128
defer stream.Close()
138129
for stream.Next() {
139130
ev := stream.Current()
140-
if !shouldEmit(ev.Event, in.Categories, in.Types) {
131+
cat := eventCategory(ev.Event)
132+
if len(in.Categories) > 0 && !slices.Contains(in.Categories, cat) {
133+
continue
134+
}
135+
if len(in.Types) > 0 && !slices.Contains(in.Types, ev.Event.Type) {
141136
continue
142137
}
143138
if in.Output == "json" {
144-
_ = util.PrintCompactJSONLine(ev)
139+
if err := util.PrintCompactJSONLine(ev); err != nil {
140+
return err
141+
}
145142
continue
146143
}
147144
ts := time.UnixMicro(ev.Event.Ts).Local().Format("15:04:05")
148-
pterm.Printf("%s [%s] %s\n", ts, eventCategory(ev.Event), ev.Event.Type)
145+
pterm.Printf("%s %-11s %s\n", ts, "["+cat+"]", ev.Event.Type)
149146
}
150147
if err := stream.Err(); err != nil {
151148
return util.CleanedUpSdkError{Err: err}
@@ -159,7 +156,7 @@ func init() {
159156
telemetryStream := &cobra.Command{Use: "stream <id>", Short: "Stream live telemetry events", Args: cobra.ExactArgs(1), RunE: runBrowsersTelemetryStream}
160157
telemetryStream.Flags().StringSlice("categories", []string{}, "Filter by API event category (console,network,page,interaction,system); system covers all monitor_* events")
161158
telemetryStream.Flags().StringSlice("types", []string{}, "Filter by event type (e.g. network_response,console_error)")
162-
telemetryStream.Flags().Int64("seq", 0, "Resume stream from sequence number (Last-Event-ID)")
159+
telemetryStream.Flags().Int64("seq", -1, "Resume stream from sequence number (Last-Event-ID); 0 means from the beginning")
163160
telemetryStream.Flags().StringP("output", "o", "", "Output format: json for newline-delimited JSON envelopes")
164161
telemetryRoot.AddCommand(telemetryStream)
165162
browsersCmd.AddCommand(telemetryRoot)

‎cmd/browsers_telemetry_test.go‎

Lines changed: 85 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,22 +11,103 @@ import (
1111
"github.com/stretchr/testify/assert"
1212
)
1313

14-
type FakeBrowserTelemetryService struct{}
14+
type FakeBrowserTelemetryService struct {
15+
StreamFunc func() *ssestream.Stream[kernel.BrowserTelemetryStreamResponse]
16+
}
1517

1618
func (f *FakeBrowserTelemetryService) StreamStreaming(ctx context.Context, id string, query kernel.BrowserTelemetryStreamParams, opts ...option.RequestOption) *ssestream.Stream[kernel.BrowserTelemetryStreamResponse] {
19+
if f.StreamFunc != nil {
20+
return f.StreamFunc()
21+
}
1722
return makeStream([]kernel.BrowserTelemetryStreamResponse{})
1823
}
1924

20-
func TestTelemetryStream_UnknownCategoryErrors(t *testing.T) {
25+
func TestTelemetryStream_NilTelemetryErrors(t *testing.T) {
2126
b := BrowsersCmd{browsers: &FakeBrowsersService{}}
2227

2328
err := b.TelemetryStream(context.Background(), BrowsersTelemetryStreamInput{
2429
Identifier: "session123",
25-
Categories: []string{"invalid"},
2630
})
2731

2832
assert.Error(t, err)
29-
assert.Contains(t, err.Error(), "unknown category")
33+
assert.Contains(t, err.Error(), "telemetry service not available")
34+
}
35+
36+
func TestTelemetryStream_UnknownCategoryPassesThrough(t *testing.T) {
37+
setupStdoutCapture(t)
38+
fake := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
39+
return &kernel.BrowserGetResponse{SessionID: id}, nil
40+
}}
41+
b := BrowsersCmd{browsers: fake, telemetry: &FakeBrowserTelemetryService{}}
42+
43+
err := b.TelemetryStream(context.Background(), BrowsersTelemetryStreamInput{
44+
Identifier: "session123",
45+
Categories: []string{"future_category"},
46+
Seq: -1,
47+
})
48+
49+
assert.NoError(t, err)
50+
}
51+
52+
func TestTelemetryStream_SystemCategoryAccepted(t *testing.T) {
53+
setupStdoutCapture(t)
54+
fake := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
55+
return &kernel.BrowserGetResponse{SessionID: id}, nil
56+
}}
57+
b := BrowsersCmd{browsers: fake, telemetry: &FakeBrowserTelemetryService{}}
58+
59+
err := b.TelemetryStream(context.Background(), BrowsersTelemetryStreamInput{
60+
Identifier: "session123",
61+
Categories: []string{"system"},
62+
Seq: -1,
63+
})
64+
65+
assert.NoError(t, err)
66+
}
67+
68+
func TestTelemetryStream_EventsFlow(t *testing.T) {
69+
setupStdoutCapture(t)
70+
event := kernel.BrowserTelemetryStreamResponse{}
71+
if err := json.Unmarshal([]byte(`{"event":{"type":"network_response","ts":1000000}}`), &event); err != nil {
72+
t.Fatalf("unmarshal: %v", err)
73+
}
74+
fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
75+
return &kernel.BrowserGetResponse{SessionID: id}, nil
76+
}}
77+
fakeTelemetry := &FakeBrowserTelemetryService{StreamFunc: func() *ssestream.Stream[kernel.BrowserTelemetryStreamResponse] {
78+
return makeStream([]kernel.BrowserTelemetryStreamResponse{event})
79+
}}
80+
b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
81+
82+
err := b.TelemetryStream(context.Background(), BrowsersTelemetryStreamInput{
83+
Identifier: "session123",
84+
Seq: -1,
85+
})
86+
87+
assert.NoError(t, err)
88+
assert.Contains(t, outBuf.String(), "network_response")
89+
}
90+
91+
func TestTelemetryStream_EventsFlow_JSON(t *testing.T) {
92+
event := kernel.BrowserTelemetryStreamResponse{}
93+
if err := json.Unmarshal([]byte(`{"event":{"type":"network_response","ts":1000000}}`), &event); err != nil {
94+
t.Fatalf("unmarshal: %v", err)
95+
}
96+
fakeBrowsers := &FakeBrowsersService{GetFunc: func(ctx context.Context, id string, query kernel.BrowserGetParams, opts ...option.RequestOption) (*kernel.BrowserGetResponse, error) {
97+
return &kernel.BrowserGetResponse{SessionID: id}, nil
98+
}}
99+
fakeTelemetry := &FakeBrowserTelemetryService{StreamFunc: func() *ssestream.Stream[kernel.BrowserTelemetryStreamResponse] {
100+
return makeStream([]kernel.BrowserTelemetryStreamResponse{event})
101+
}}
102+
b := BrowsersCmd{browsers: fakeBrowsers, telemetry: fakeTelemetry}
103+
104+
err := b.TelemetryStream(context.Background(), BrowsersTelemetryStreamInput{
105+
Identifier: "session123",
106+
Output: "json",
107+
Seq: -1,
108+
})
109+
110+
assert.NoError(t, err)
30111
}
31112

32113
func makeEvent(t *testing.T, raw string) kernel.BrowserTelemetryEventUnion {

0 commit comments

Comments
 (0)