From 66cd008aa4b8169570dbc9a5201fa50b6509c7b6 Mon Sep 17 00:00:00 2001 From: amrik-lc <140389585+amrik-lc@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:30:21 -0400 Subject: [PATCH] mac_unified_logging: fix infinite json.Decoder error loop pegging CPU The gatherer dropped the first line of `log stream` output via a throwaway bufio.Reader, which could buffer and discard up to 64KB beyond that line. The json.Decoder then started mid-object and hit a SyntaxError, which json.Decoder never advances past: the loop spun forever printing the same error (sustained ~113% CPU) while the stdout pipe filled and `log stream` blocked permanently, so no logs were collected at all until restart. Parse the ndjson output line by line with a bufio.Scanner instead and skip lines that fail to parse (including the non-JSON header line, so the drop-first-line hack is no longer needed). A bad line can now never wedge the stream. Also: - Close logs.Channel when the stream ends so the adapter shuts down instead of hanging silently if `log stream` dies. - Make StopGathering close-based (idempotent, non-blocking) and kill the subprocess so a reader blocked in Scan() unblocks. - Report the actual stderr line instead of panicking with a stale nil err from the enclosing scope. - Buffer the signal.Notify channel (pre-existing go vet finding). Co-Authored-By: Claude Fable 5 --- mac_unified_logging/client.go | 2 +- mac_unified_logging/log.go | 88 ++++++++++++++++++++++++++--------- 2 files changed, 66 insertions(+), 24 deletions(-) diff --git a/mac_unified_logging/client.go b/mac_unified_logging/client.go index ba1e78a9..ab21e4e0 100644 --- a/mac_unified_logging/client.go +++ b/mac_unified_logging/client.go @@ -101,7 +101,7 @@ func (a *MacUnifiedLoggingAdapter) handleEvent(predicate string) uintptr { logs := NewLogs() - signalChannel := make(chan os.Signal) + signalChannel := make(chan os.Signal, 1) signal.Notify(signalChannel, os.Interrupt, syscall.SIGTERM) go func() { <-signalChannel diff --git a/mac_unified_logging/log.go b/mac_unified_logging/log.go index 6e4c342a..b4e2e2cc 100644 --- a/mac_unified_logging/log.go +++ b/mac_unified_logging/log.go @@ -10,6 +10,10 @@ import ( "sync" ) +// log stream events can carry large payloads (eventMessage, backtraces), +// well past bufio.Scanner's 64KB default line limit. +const maxLogLineSize = 4 * 1024 * 1024 + type Log struct { TraceID int64 `json:"traceID"` EventMessage string `json:"eventMessage"` @@ -41,24 +45,26 @@ type Log struct { } type Logs struct { - m sync.Mutex - Channel chan Log - exit chan bool + m sync.Mutex + Channel chan Log + exit chan struct{} + exitOnce sync.Once } func NewLogs() *Logs { return &Logs{ Channel: make(chan Log), - exit: make(chan bool), + exit: make(chan struct{}), } } func (logs *Logs) StartGathering(predicate string) error { - cmd := exec.Command("log", "stream", "--color=none", "--style=ndjson") + args := []string{"stream", "--color=none", "--style=ndjson"} if predicate != "" { - cmd = exec.Command("log", "stream", "--color=none", "--style=ndjson", "--predicate", predicate) + args = append(args, "--predicate", predicate) } + cmd := exec.Command("log", args...) stdout, err := cmd.StdoutPipe() if err != nil { @@ -70,42 +76,76 @@ func (logs *Logs) StartGathering(predicate string) error { return err } + if err := cmd.Start(); err != nil { + return err + } + + // Kill the subprocess on StopGathering so a reader blocked in + // Scan() unblocks and the gathering goroutine can exit. + go func() { + <-logs.exit + cmd.Process.Kill() + }() + go func() { logs.m.Lock() defer logs.m.Unlock() - cmd.Start() defer cmd.Process.Kill() + // Closing the channel lets consumers terminate instead of + // waiting forever on a stream that has ended. + defer close(logs.Channel) + + // Parse the ndjson output line by line. A json.Decoder is + // unsuitable here: it never advances past a SyntaxError, so a + // single bad line (like the non-JSON "Filtering the log data" + // header) would wedge it in a permanent error loop while the + // pipe fills up and `log stream` blocks forever. + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), maxLogLineSize) + + for scanner.Scan() { + select { + case <-logs.exit: + return + default: + } - // drop first message - bufio.NewReader(stdout).ReadLine() + line := scanner.Bytes() + if len(line) == 0 { + continue + } - dec := json.NewDecoder(stdout) + log := Log{} + if err := json.Unmarshal(line, &log); err != nil { + // Skip non-JSON lines (e.g. the header line) without + // stalling the stream. + fmt.Printf("skipping non-json line from log stream: %v\n", err) + continue + } - for { select { + case logs.Channel <- log: case <-logs.exit: return - default: - log := Log{} - - if err := dec.Decode(&log); err != nil { - fmt.Println("error decoding json") - fmt.Println(err.Error()) - } else { - logs.Channel <- log - } } } + + if err := scanner.Err(); err != nil { + fmt.Printf("error reading log stream output: %v\n", err) + } }() go func() { stderrBuf := bufio.NewReader(stderr) for { - line, _, _ := stderrBuf.ReadLine() + line, _, err := stderrBuf.ReadLine() if len(line) > 0 { logs.StopGathering() - panic(err) + panic(fmt.Errorf("log stream error: %s", string(line))) + } + if err != nil { + return } } }() @@ -114,5 +154,7 @@ func (logs *Logs) StartGathering(predicate string) error { } func (logs *Logs) StopGathering() { - logs.exit <- true + logs.exitOnce.Do(func() { + close(logs.exit) + }) }