From 693321f5f8389cec25195ba2447ec3bc3c1c4c47 Mon Sep 17 00:00:00 2001 From: Ulzii Otgonbaatar Date: Thu, 30 Jul 2026 20:52:11 -0600 Subject: [PATCH] Count Chromium net::ERR_* failures per VM We can't currently tell how often sessions hit network failures like ERR_HTTP2_PROTOCOL_ERROR. Chrome tracks this in the UMA histogram Net.ErrorCodesForMainFrame4, but that is recorded in the renderer process and so is invisible to Browser.getHistograms on these images, and the cdpmonitor events that do carry errorText only flow when a caller opts into network telemetry. Tap the CDP proxy instead, the one vantage point that is on for every session, and tally Network.loadingFailed by error text and resource type. Cancellations and ERR_ABORTED are dropped so real failures aren't buried; route.abort() with its default reason is reported as ERR_FAILED with canceled false, which is indistinguishable from a genuine failure and is counted in its own series rather than dropped. The counts surface on /metrics as kernel_chromium_net_errors_total, with a throttled log line per error class for per-VM triage. The tap is passive and prescreens each frame with a substring scan before decoding, so non-failure frames cost ~108ns and zero allocations. The "->"/"<-" direction strings the pump already passed to MessageTransform are now named constants, so the tap's upstream-only check reads as intent rather than as a magic string. Co-authored-by: Cursor --- server/cmd/api/main.go | 9 +- server/lib/devtoolsproxy/proxy.go | 17 ++- server/lib/devtoolsproxy/proxy_test.go | 95 ++++++++++++- server/lib/metrics/neterror.go | 52 +++++++ server/lib/metrics/neterror_test.go | 57 ++++++++ server/lib/neterror/neterror.go | 179 ++++++++++++++++++++++++ server/lib/neterror/neterror_test.go | 180 +++++++++++++++++++++++++ server/lib/wsproxy/wsproxy.go | 15 ++- 8 files changed, 592 insertions(+), 12 deletions(-) create mode 100644 server/lib/metrics/neterror.go create mode 100644 server/lib/metrics/neterror_test.go create mode 100644 server/lib/neterror/neterror.go create mode 100644 server/lib/neterror/neterror_test.go diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index 1196e890..06832727 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -34,6 +34,7 @@ import ( "github.com/kernel/kernel-images/server/lib/logger" "github.com/kernel/kernel-images/server/lib/metrics" "github.com/kernel/kernel-images/server/lib/nekoclient" + "github.com/kernel/kernel-images/server/lib/neterror" oapi "github.com/kernel/kernel-images/server/lib/oapi" "github.com/kernel/kernel-images/server/lib/recorder" "github.com/kernel/kernel-images/server/lib/scaletozero" @@ -297,6 +298,11 @@ func main() { os.Exit(1) } + // Tallies net::ERR_* failures seen on relayed CDP traffic. Shared by every + // proxy session and read by the metrics endpoint, so it lives for the + // process and its counts are cumulative per VM. + netErrorTracker := neterror.NewTracker(slogger) + rDevtools := chi.NewRouter() rDevtools.Use( chiMiddleware.Logger, @@ -322,7 +328,7 @@ func main() { rDevtools.Get("/json/list", jsonTargetHandler) rDevtools.Get("/json/list/", jsonTargetHandler) rDevtools.Get("/*", func(w http.ResponseWriter, r *http.Request) { - devtoolsproxy.WebSocketProxyHandler(upstreamMgr, slogger, config.LogCDPMessages, stz, telemetrySession.Publish, wsRegistry).ServeHTTP(w, r) + devtoolsproxy.WebSocketProxyHandler(upstreamMgr, slogger, config.LogCDPMessages, stz, telemetrySession.Publish, wsRegistry, netErrorTracker).ServeHTTP(w, r) }) srvDevtools := &http.Server{ @@ -364,6 +370,7 @@ func main() { rMetrics.Use(chiMiddleware.Recoverer) metricsCollectors := []metrics.Collector{ metrics.NewChromeCollector(upstreamMgr), + metrics.NewNetErrorCollector(netErrorTracker), metrics.NewGPUCollector(), metrics.NewSystemCollector(), } diff --git a/server/lib/devtoolsproxy/proxy.go b/server/lib/devtoolsproxy/proxy.go index df8b47eb..f57044ca 100644 --- a/server/lib/devtoolsproxy/proxy.go +++ b/server/lib/devtoolsproxy/proxy.go @@ -305,12 +305,20 @@ func maybePauseAfterCurrentRead(ctx context.Context, logger *slog.Logger, r *htt // so it can be wired directly; the proxy ignores the returns. type EventPublisher func(ev events.Event) (events.Envelope, bool) +// NetErrorObserver is offered every CDP frame Chromium sends to the client so +// it can tally net::ERR_* failures. Satisfied by neterror.Tracker; nil +// disables the tap. +type NetErrorObserver interface { + Observe(msg []byte) +} + // WebSocketProxyHandler returns an http.Handler that upgrades incoming connections and // proxies them to the current upstream websocket URL. It expects only websocket requests. // If logCDPMessages is true, all CDP messages will be logged with their direction. // publish is invoked on accept (cdp_connect) and on teardown (cdp_disconnect); pass -// nil to disable emission. -func WebSocketProxyHandler(mgr *UpstreamManager, logger *slog.Logger, logCDPMessages bool, ctrl scaletozero.Controller, publish EventPublisher, reg *wsdrain.Registry) http.Handler { +// nil to disable emission. netErrors, when non-nil, observes upstream frames; it +// must not mutate them. +func WebSocketProxyHandler(mgr *UpstreamManager, logger *slog.Logger, logCDPMessages bool, ctrl scaletozero.Controller, publish EventPublisher, reg *wsdrain.Registry, netErrors NetErrorObserver) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Counts every relayed message so cdp_disconnect can report message_count. var msgCount atomic.Int64 @@ -318,6 +326,11 @@ func WebSocketProxyHandler(mgr *UpstreamManager, logger *slog.Logger, logCDPMess if logCDPMessages { logCDPMessage(logger, direction, mt, msg) } + // Events only travel upstream-to-client, so client frames are + // skipped rather than scanned for a needle they cannot contain. + if netErrors != nil && direction == wsproxy.DirectionUpstreamToClient { + netErrors.Observe(msg) + } msgCount.Add(1) return msg } diff --git a/server/lib/devtoolsproxy/proxy_test.go b/server/lib/devtoolsproxy/proxy_test.go index 5956a555..9e6a08c4 100644 --- a/server/lib/devtoolsproxy/proxy_test.go +++ b/server/lib/devtoolsproxy/proxy_test.go @@ -20,6 +20,9 @@ import ( "time" "github.com/coder/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/kernel/kernel-images/server/lib/events" oapi "github.com/kernel/kernel-images/server/lib/oapi" "github.com/kernel/kernel-images/server/lib/scaletozero" @@ -27,6 +30,86 @@ import ( "github.com/kernel/kernel-images/server/lib/wsproxy" ) +// recordingObserver captures every frame handed to the net error tap. +type recordingObserver struct { + mu sync.Mutex + frames []string +} + +func (o *recordingObserver) Observe(msg []byte) { + o.mu.Lock() + defer o.mu.Unlock() + o.frames = append(o.frames, string(msg)) +} + +func (o *recordingObserver) snapshot() []string { + o.mu.Lock() + defer o.mu.Unlock() + return append([]string(nil), o.frames...) +} + +// The tap must see Chromium's frames and only Chromium's frames: a reversed +// direction check would silently scan client commands, which can never carry +// a CDP event, and count nothing. +func TestWebSocketProxyHandler_ObservesOnlyUpstreamFrames(t *testing.T) { + const clientFrame = "client-to-upstream" + const upstreamFrame = "upstream-to-client" + + upstreamSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, &websocket.AcceptOptions{OriginPatterns: []string{"*"}}) + if err != nil { + return + } + defer c.Close(websocket.StatusNormalClosure, "") + // Reply with a distinct payload so the two directions are never + // confusable in the recorded frames. + if _, _, err := c.Read(r.Context()); err != nil { + return + } + _ = c.Write(r.Context(), websocket.MessageText, []byte(upstreamFrame)) + <-r.Context().Done() + })) + defer upstreamSrv.Close() + + u, _ := url.Parse(upstreamSrv.URL) + logger := silentLogger() + mgr := NewUpstreamManager("/dev/null", logger) + mgr.setCurrent((&url.URL{Scheme: "ws", Host: u.Host, Path: "/"}).String()) + + obs := &recordingObserver{} + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, nil, obs)) + defer proxySrv.Close() + + pu, _ := url.Parse(proxySrv.URL) + pu.Scheme = "ws" + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + conn, _, err := websocket.Dial(ctx, pu.String(), nil) + require.NoError(t, err) + defer conn.Close(websocket.StatusNormalClosure, "") + + require.NoError(t, conn.Write(ctx, websocket.MessageText, []byte(clientFrame))) + + // The tap runs before the frame is written to the client, so a successful + // read means the observer has already been offered it. + _, got, err := conn.Read(ctx) + require.NoError(t, err) + require.Equal(t, upstreamFrame, string(got)) + + assert.Equal(t, []string{upstreamFrame}, obs.snapshot()) +} + +func TestWebSocketProxyHandler_NilObserverIsAllowed(t *testing.T) { + logger := silentLogger() + mgr := NewUpstreamManager("/dev/null", logger) + mgr.setCurrent("ws://127.0.0.1:1/") + + assert.NotPanics(t, func() { + WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, nil, nil) + }) +} + func silentLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } @@ -133,7 +216,7 @@ func TestWebSocketProxyHandler_ProxiesEcho(t *testing.T) { // seed current upstream to echo server including path/query (bypass tailing) mgr.setCurrent((&url.URL{Scheme: u.Scheme, Host: u.Host, Path: u.Path, RawQuery: u.RawQuery}).String()) - proxy := WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, nil) + proxy := WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, nil, nil) proxySrv := httptest.NewServer(proxy) defer proxySrv.Close() @@ -191,7 +274,7 @@ func TestWebSocketProxyHandler_RegistryClosesClientWithGoingAway(t *testing.T) { mgr.setCurrent((&url.URL{Scheme: "ws", Host: u.Host, Path: "/echo"}).String()) reg := wsdrain.New() - proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, reg)) + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), nil, reg, nil)) defer proxySrv.Close() pu, _ := url.Parse(proxySrv.URL) @@ -520,7 +603,7 @@ func TestWebSocketProxyHandler_EmitsConnectAndDisconnect(t *testing.T) { mgr.setCurrent(u.String()) rp := &recordingPublisher{} - proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil)) + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil, nil)) defer proxySrv.Close() pu, _ := url.Parse(proxySrv.URL) @@ -695,7 +778,7 @@ func TestWebSocketProxyHandler_EmitsUpstreamChangedOnMidStreamRestart(t *testing mgr.setCurrent(urlA.String()) rp := &recordingPublisher{} - proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil)) + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil, nil)) defer proxySrv.Close() pu, _ := url.Parse(proxySrv.URL) @@ -779,7 +862,7 @@ func TestWebSocketProxyHandler_KicksClientOffStaleUpstreamOnURLChange(t *testing mgr.setCurrent(urlA.String()) rp := &recordingPublisher{} - proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil)) + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil, nil)) defer proxySrv.Close() pu, _ := url.Parse(proxySrv.URL) @@ -831,7 +914,7 @@ func TestWebSocketProxyHandler_EmitsUpstreamErrorOnDialFailure(t *testing.T) { mgr.setCurrent(deadURL) rp := &recordingPublisher{} - proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil)) + proxySrv := httptest.NewServer(WebSocketProxyHandler(mgr, logger, false, scaletozero.NewNoopController(), rp.publish, nil, nil)) defer proxySrv.Close() pu, _ := url.Parse(proxySrv.URL) diff --git a/server/lib/metrics/neterror.go b/server/lib/metrics/neterror.go new file mode 100644 index 00000000..fa1d7537 --- /dev/null +++ b/server/lib/metrics/neterror.go @@ -0,0 +1,52 @@ +package metrics + +import ( + "context" + + "github.com/kernel/kernel-images/server/lib/neterror" +) + +// NetErrorSource reports network failure counts observed on relayed CDP +// traffic. Satisfied by neterror.Tracker. +type NetErrorSource interface { + Snapshot() (map[neterror.Key]int64, int64) +} + +// NetErrorCollector exposes Chromium net::ERR_* failures as counters. Unlike +// the Chrome collector it touches neither Chrome nor the network on scrape: +// the counts are accumulated by the CDP proxy as traffic flows and only read +// here. +type NetErrorCollector struct { + source NetErrorSource +} + +func NewNetErrorCollector(source NetErrorSource) *NetErrorCollector { + return &NetErrorCollector{source: source} +} + +func (c *NetErrorCollector) Name() string { return "neterror" } + +func (c *NetErrorCollector) Collect(_ context.Context, w *Writer) error { + counts, overflow := c.source.Snapshot() + + // A VM that has seen no failures emits no samples here, so this family is + // simply absent from queries until something fails. Use the unlabelled + // _dropped_total below to tell "no failures" apart from "not scraped". + w.Metric("kernel_chromium_net_errors_total", + "Chromium network request failures by error text and resource type, cumulative since VM start. Client-cancelled requests (net::ERR_ABORTED) are excluded; deliberate route.abort() blocking is not separable and lands in the net::ERR_FAILED series.", + "counter") + for _, k := range neterror.SortedKeys(counts) { + w.Sample("kernel_chromium_net_errors_total", []Label{ + {"error", k.Error}, + {"resource_type", k.ResourceType}, + }, float64(counts[k])) + } + + // Always sampled, so this doubles as the proof that the tap is running on + // a VM reporting no failures. + w.Metric("kernel_chromium_net_errors_dropped_total", + "Network failures not counted because the distinct error/resource-type cap was reached.", + "counter") + w.Sample("kernel_chromium_net_errors_dropped_total", nil, float64(overflow)) + return nil +} diff --git a/server/lib/metrics/neterror_test.go b/server/lib/metrics/neterror_test.go new file mode 100644 index 00000000..488e2335 --- /dev/null +++ b/server/lib/metrics/neterror_test.go @@ -0,0 +1,57 @@ +package metrics + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/kernel/kernel-images/server/lib/neterror" +) + +type stubNetErrorSource struct { + counts map[neterror.Key]int64 + overflow int64 +} + +func (s stubNetErrorSource) Snapshot() (map[neterror.Key]int64, int64) { + return s.counts, s.overflow +} + +func TestNetErrorCollector(t *testing.T) { + c := NewNetErrorCollector(stubNetErrorSource{ + counts: map[neterror.Key]int64{ + {Error: "net::ERR_TIMED_OUT", ResourceType: "XHR"}: 3, + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Document"}: 12, + }, + overflow: 2, + }) + + w := &Writer{} + require.NoError(t, c.Collect(context.Background(), w)) + + out := string(w.Bytes()) + assert.Contains(t, out, "# TYPE kernel_chromium_net_errors_total counter\n") + // Samples are ordered by error then resource type so an unchanged tracker + // scrapes byte-identically. + assert.Contains(t, out, `kernel_chromium_net_errors_total{error="net::ERR_HTTP2_PROTOCOL_ERROR",resource_type="Document"} 12 +kernel_chromium_net_errors_total{error="net::ERR_TIMED_OUT",resource_type="XHR"} 3 +`) + assert.Contains(t, out, "kernel_chromium_net_errors_dropped_total 2\n") +} + +// With no failures the labelled family has no samples and so is absent from +// queries; the unlabelled dropped counter is what proves the VM reported. +func TestNetErrorCollectorWithNoFailures(t *testing.T) { + c := NewNetErrorCollector(stubNetErrorSource{}) + + w := &Writer{} + require.NoError(t, c.Collect(context.Background(), w)) + + out := string(w.Bytes()) + assert.NotContains(t, out, "kernel_chromium_net_errors_total{") + assert.Contains(t, out, "kernel_chromium_net_errors_dropped_total 0\n") +} + +var _ NetErrorSource = (*neterror.Tracker)(nil) diff --git a/server/lib/neterror/neterror.go b/server/lib/neterror/neterror.go new file mode 100644 index 00000000..29b17d56 --- /dev/null +++ b/server/lib/neterror/neterror.go @@ -0,0 +1,179 @@ +// Package neterror tallies Chromium network failures (net::ERR_*) observed on +// the DevTools traffic relayed by the CDP proxy. +// +// Chrome records net errors in the UMA histogram Net.ErrorCodesForMainFrame4, +// but that is written from the renderer process and so is invisible to +// Browser.getHistograms on these images. The CDP proxy is the only always-on +// vantage point that sees failures for every session, so the counts are +// derived from the Network.loadingFailed events already flowing across it. +// +// The tap is passive: it reads relayed frames and never injects or rewrites +// CDP traffic, so it is invisible to automation running in the browser. It +// observes what the client's own CDP session sees, which means a session that +// never sends Network.enable contributes nothing. Playwright, Puppeteer, and +// the CUA app all enable the Network domain by default. +// +// Deliberate client aborts are only partly separable from real failures. +// Cancellations and net::ERR_ABORTED are dropped, but a resource blocker +// calling Playwright route.abort() with its default reason lands on +// net::ERR_FAILED with canceled false and is indistinguishable from a genuine +// generic failure. Rather than drop a real error class, net::ERR_FAILED is +// counted and kept in its own series, so a session that blocks images inflates +// only that series and never the specific ones (ERR_HTTP2_PROTOCOL_ERROR and +// friends) these counts exist to track. +package neterror + +import ( + "bytes" + "encoding/json" + "log/slog" + "sort" + "strings" + "sync" + "time" +) + +// loadingFailedMethod is the CDP event carrying the net error. It doubles as +// the prescreen needle: every relayed frame is scanned for it, so the JSON +// decode below only runs on the rare frames that actually report a failure. +var loadingFailedMethod = []byte(`"Network.loadingFailed"`) + +// netErrorPrefix is the namespace Chromium stamps on the errorText of a +// network failure. errorText is only loosely specified by the protocol, so +// requiring the prefix keeps anything unexpected from consuming a label slot. +const netErrorPrefix = "net::ERR_" + +// abortedError is how a client's own cancellation is reported: navigating +// away, closing a tab, or Playwright route.abort({errorCode: 'aborted'}). +const abortedError = "net::ERR_ABORTED" + +// maxKeys bounds the tracked (error, resource type) pairs. Both fields are +// Chromium enums, so the real ceiling is well under this; the cap only exists +// so a malformed or hostile upstream cannot grow the map without limit and +// blow up scrape cardinality. Failures past the cap still increment overflow. +const maxKeys = 256 + +// logInterval throttles the per-key log line. Net errors arrive in bursts — +// one failing page can fail every subresource — and the metric already +// carries the volume, so the log only needs to show that a given failure is +// happening and roughly when. +const logInterval = time.Minute + +// Key identifies one class of network failure. +type Key struct { + // Error is Chromium's error text, e.g. "net::ERR_HTTP2_PROTOCOL_ERROR". + Error string + // ResourceType is the CDP resource type, e.g. "Document" or "XHR". + ResourceType string +} + +type entry struct { + count int64 + lastLogAt time.Time +} + +// Tracker counts network failures. Counts are cumulative for the lifetime of +// the process, matching Prometheus counter semantics. Safe for concurrent use. +type Tracker struct { + logger *slog.Logger + now func() time.Time + + mu sync.Mutex + entries map[Key]*entry + overflow int64 +} + +func NewTracker(logger *slog.Logger) *Tracker { + if logger == nil { + logger = slog.Default() + } + return &Tracker{ + logger: logger, + now: time.Now, + entries: make(map[Key]*entry), + } +} + +// Observe inspects one relayed CDP frame and records a failure if it is a +// Network.loadingFailed event worth counting. Frames that are not failures +// cost a single substring scan. +func (t *Tracker) Observe(msg []byte) { + if !bytes.Contains(msg, loadingFailedMethod) { + return + } + var frame struct { + Method string `json:"method"` + Params struct { + Type string `json:"type"` + ErrorText string `json:"errorText"` + Canceled bool `json:"canceled"` + } `json:"params"` + } + // The needle can also appear inside an unrelated payload (a response body + // quoting the event name, say), so the decoded method still has to match. + if err := json.Unmarshal(msg, &frame); err != nil || frame.Method != "Network.loadingFailed" { + return + } + p := frame.Params + if p.Canceled || p.ErrorText == abortedError || !strings.HasPrefix(p.ErrorText, netErrorPrefix) { + return + } + t.record(Key{Error: p.ErrorText, ResourceType: p.Type}) +} + +func (t *Tracker) record(k Key) { + t.mu.Lock() + e, ok := t.entries[k] + if !ok { + if len(t.entries) >= maxKeys { + t.overflow++ + t.mu.Unlock() + return + } + e = &entry{} + t.entries[k] = e + } + e.count++ + now := t.now() + shouldLog := now.Sub(e.lastLogAt) >= logInterval + if shouldLog { + e.lastLogAt = now + } + count := e.count + t.mu.Unlock() + + if shouldLog { + t.logger.Warn("chromium network request failed", + slog.String("error_text", k.Error), + slog.String("resource_type", k.ResourceType), + slog.Int64("count", count)) + } +} + +// Snapshot returns the current counts sorted by key, plus the number of +// failures dropped because the key cap was reached. +func (t *Tracker) Snapshot() (counts map[Key]int64, overflow int64) { + t.mu.Lock() + defer t.mu.Unlock() + counts = make(map[Key]int64, len(t.entries)) + for k, e := range t.entries { + counts[k] = e.count + } + return counts, t.overflow +} + +// SortedKeys returns the keys of counts in a stable order so scrapes of an +// unchanged tracker are byte-identical. +func SortedKeys(counts map[Key]int64) []Key { + keys := make([]Key, 0, len(counts)) + for k := range counts { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].Error != keys[j].Error { + return keys[i].Error < keys[j].Error + } + return keys[i].ResourceType < keys[j].ResourceType + }) + return keys +} diff --git a/server/lib/neterror/neterror_test.go b/server/lib/neterror/neterror_test.go new file mode 100644 index 00000000..b68bbe82 --- /dev/null +++ b/server/lib/neterror/neterror_test.go @@ -0,0 +1,180 @@ +package neterror + +import ( + "context" + "fmt" + "log/slog" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newTestTracker() *Tracker { + return NewTracker(slog.New(slog.DiscardHandler)) +} + +func loadingFailed(errorText, resourceType string, canceled bool) []byte { + return fmt.Appendf(nil, + `{"method":"Network.loadingFailed","params":{"requestId":"1","timestamp":1.5,"type":%q,"errorText":%q,"canceled":%t}}`, + resourceType, errorText, canceled) +} + +func TestObserveCountsHTTP2ProtocolError(t *testing.T) { + tr := newTestTracker() + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false)) + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false)) + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "XHR", false)) + + counts, overflow := tr.Snapshot() + assert.Zero(t, overflow) + assert.Equal(t, map[Key]int64{ + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Document"}: 2, + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "XHR"}: 1, + }, counts) +} + +func TestObserveIgnoresNonFailures(t *testing.T) { + tr := newTestTracker() + for _, msg := range []string{ + `{"method":"Network.responseReceived","params":{"requestId":"1"}}`, + `{"id":7,"result":{}}`, + // The needle appears in a payload that is not the event itself. + `{"method":"Network.dataReceived","params":{"body":"\"Network.loadingFailed\""}}`, + // Malformed JSON that still contains the needle. + `{"method":"Network.loadingFailed","params":{`, + } { + tr.Observe([]byte(msg)) + } + + counts, overflow := tr.Snapshot() + assert.Empty(t, counts) + assert.Zero(t, overflow) +} + +func TestObserveSkipsCancellations(t *testing.T) { + tr := newTestTracker() + // Client-initiated aborts show up either as canceled or as ERR_ABORTED. + tr.Observe(loadingFailed("net::ERR_ABORTED", "Fetch", false)) + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Fetch", true)) + + counts, _ := tr.Snapshot() + assert.Empty(t, counts) +} + +func TestObserveRequiresNetErrorPrefix(t *testing.T) { + tr := newTestTracker() + for _, errorText := range []string{"", "Failed", "ERR_HTTP2_PROTOCOL_ERROR", "net::OK", "blocked"} { + tr.Observe(loadingFailed(errorText, "Fetch", false)) + } + + counts, overflow := tr.Snapshot() + assert.Empty(t, counts) + assert.Zero(t, overflow) +} + +// route.abort() with Playwright's default reason is reported as ERR_FAILED +// with canceled false, so it is indistinguishable from a genuine generic +// failure and is deliberately counted. It must stay in its own series rather +// than contaminate the specific errors these counts exist to track. +func TestObserveCountsErrFailedSeparately(t *testing.T) { + tr := newTestTracker() + for range 50 { + tr.Observe(loadingFailed("net::ERR_FAILED", "Image", false)) + } + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false)) + + counts, _ := tr.Snapshot() + assert.EqualValues(t, 50, counts[Key{Error: "net::ERR_FAILED", ResourceType: "Image"}]) + assert.EqualValues(t, 1, counts[Key{Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Document"}]) +} + +func TestRecordCapsDistinctKeys(t *testing.T) { + tr := newTestTracker() + for i := range maxKeys + 10 { + tr.Observe(loadingFailed(fmt.Sprintf("net::ERR_SYNTHETIC_%d", i), "Other", false)) + } + // A key already tracked still counts after the cap is hit. + tr.Observe(loadingFailed("net::ERR_SYNTHETIC_0", "Other", false)) + + counts, overflow := tr.Snapshot() + assert.Len(t, counts, maxKeys) + assert.EqualValues(t, 10, overflow) + assert.EqualValues(t, 2, counts[Key{Error: "net::ERR_SYNTHETIC_0", ResourceType: "Other"}]) +} + +func TestRecordThrottlesLogging(t *testing.T) { + var logged int + tr := newTestTracker() + tr.logger = slog.New(countingHandler{n: &logged}) + + now := time.Now() + tr.now = func() time.Time { return now } + + // The first failure logs; the rest of the burst is absorbed by the metric. + for range 5 { + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false)) + } + require.Equal(t, 1, logged) + + now = now.Add(logInterval) + tr.Observe(loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false)) + assert.Equal(t, 2, logged) +} + +func TestObserveIsConcurrencySafe(t *testing.T) { + tr := newTestTracker() + msg := loadingFailed("net::ERR_HTTP2_PROTOCOL_ERROR", "Document", false) + + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + for range 100 { + tr.Observe(msg) + } + }) + } + wg.Wait() + + counts, _ := tr.Snapshot() + assert.EqualValues(t, 800, counts[Key{Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Document"}]) +} + +func TestSortedKeysIsStable(t *testing.T) { + counts := map[Key]int64{ + {Error: "net::ERR_TIMED_OUT", ResourceType: "XHR"}: 1, + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "XHR"}: 1, + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Doc"}: 1, + } + assert.Equal(t, []Key{ + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "Doc"}, + {Error: "net::ERR_HTTP2_PROTOCOL_ERROR", ResourceType: "XHR"}, + {Error: "net::ERR_TIMED_OUT", ResourceType: "XHR"}, + }, SortedKeys(counts)) +} + +type countingHandler struct { + slog.Handler + n *int +} + +func (h countingHandler) Enabled(context.Context, slog.Level) bool { return true } + +func (h countingHandler) Handle(context.Context, slog.Record) error { + *h.n++ + return nil +} + +func (h countingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h countingHandler) WithGroup(string) slog.Handler { return h } + +func BenchmarkObserveNonFailure(b *testing.B) { + tr := newTestTracker() + // A typical relayed frame: no needle, so it pays only the substring scan. + msg := []byte(`{"method":"Network.responseReceived","params":{"requestId":"42","loaderId":"7","timestamp":1234.5,"type":"Script","response":{"url":"https://example.com/app.js","status":200,"mimeType":"text/javascript"}}}`) + for b.Loop() { + tr.Observe(msg) + } +} diff --git a/server/lib/wsproxy/wsproxy.go b/server/lib/wsproxy/wsproxy.go index 8a7e5940..572d8b3c 100644 --- a/server/lib/wsproxy/wsproxy.go +++ b/server/lib/wsproxy/wsproxy.go @@ -17,8 +17,17 @@ type Conn interface { Close(statusCode websocket.StatusCode, reason string) error } +const ( + // DirectionClientToUpstream marks a message travelling from the CDP + // client to Chromium. + DirectionClientToUpstream = "->" + // DirectionUpstreamToClient marks a message travelling from Chromium to + // the CDP client. CDP events only ever travel this way. + DirectionUpstreamToClient = "<-" +) + // MessageTransform is called for every message flowing through the proxy. -// direction is "->" for client-to-upstream and "<-" for upstream-to-client. +// direction is one of DirectionClientToUpstream or DirectionUpstreamToClient. // It returns the (possibly modified) message bytes to forward. type MessageTransform func(direction string, mt websocket.MessageType, msg []byte) []byte @@ -67,7 +76,7 @@ func Pump(ctx context.Context, client, upstream Conn, onClose func(cause PumpExi return } if transform != nil { - msg = transform("->", mt, msg) + msg = transform(DirectionClientToUpstream, mt, msg) } if err := upstream.Write(ctx, mt, msg); err != nil { logger.Error("upstream write error", slog.String("err", err.Error())) @@ -86,7 +95,7 @@ func Pump(ctx context.Context, client, upstream Conn, onClose func(cause PumpExi return } if transform != nil { - msg = transform("<-", mt, msg) + msg = transform(DirectionUpstreamToClient, mt, msg) } if err := client.Write(ctx, mt, msg); err != nil { logger.Error("client write error", slog.String("err", err.Error()))