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()))