diff --git a/cmd/hamlab/main.go b/cmd/hamlab/main.go index a3e4cd1..e9b17fd 100644 --- a/cmd/hamlab/main.go +++ b/cmd/hamlab/main.go @@ -13,6 +13,7 @@ import ( "os/user" "path/filepath" "runtime" + "runtime/debug" "strings" "syscall" "time" @@ -171,6 +172,10 @@ func serve(cfg *config.Config) error { } } + // Keep the heap tidy: the daemon's live data is tens of MB; without a limit the GC + // happily lets RSS coast at twice that. (Soft limit: exceeding it only means more GC.) + debug.SetMemoryLimit(160 << 20) + col := metrics.NewCollector() go col.Run(context.Background()) diff --git a/internal/cron/cron.go b/internal/cron/cron.go index 5bf9ec9..1d912ff 100644 --- a/internal/cron/cron.go +++ b/internal/cron/cron.go @@ -116,6 +116,35 @@ func run(ctx context.Context, stdin string, args ...string) (string, error) { return out.String(), nil } +// runLines streams a command's stdout to fn line by line, so multi-megabyte journal +// dumps never sit in memory whole. +func runLines(ctx context.Context, fn func(line []byte), args ...string) error { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, args[0], args[1:]...) + var errb bytes.Buffer + cmd.Stderr = &errb + stdout, err := cmd.StdoutPipe() + if err != nil { + return err + } + if err := cmd.Start(); err != nil { + return err + } + sc := bufio.NewScanner(stdout) + sc.Buffer(make([]byte, 0, 64<<10), 4<<20) + for sc.Scan() { + fn(sc.Bytes()) + } + if err := cmd.Wait(); err != nil { + if msg := strings.TrimSpace(errb.String()); msg != "" { + return errors.New(msg) + } + return err + } + return nil +} + func (m *Manager) sudoArgs(args ...string) []string { if m.sudo { return append([]string{"sudo", "-n"}, args...) @@ -414,18 +443,15 @@ func (m *Manager) lastRunsFromJournal(ctx context.Context) map[string]int64 { if m.journalSudo { args = append([]string{"sudo", "-n"}, args...) } - out, _ := run(ctx, "", args...) seen := m.logging - sc := bufio.NewScanner(strings.NewReader(out)) - sc.Buffer(make([]byte, 0, 64<<10), 4<<20) - for sc.Scan() { + _ = runLines(ctx, func(line []byte) { var raw struct { Msg any `json:"MESSAGE"` TS string `json:"__REALTIME_TIMESTAMP"` Cursor string `json:"__CURSOR"` } - if json.Unmarshal(sc.Bytes(), &raw) != nil { - continue + if json.Unmarshal(line, &raw) != nil { + return } seen = true if raw.Cursor != "" { @@ -434,14 +460,14 @@ func (m *Manager) lastRunsFromJournal(ctx context.Context) map[string]int64 { msg, _ := raw.Msg.(string) mm := cronLogRe.FindStringSubmatch(msg) if mm == nil { - continue + return } us, _ := strconv.ParseInt(raw.TS, 10, 64) k := mm[1] + "\x00" + mm[2] if us/1000 > res[k] { res[k] = us / 1000 } - } + }, args...) m.mu.Lock() m.lastRuns, m.lastAt, m.lastCursor, m.logging = res, time.Now(), cursor, seen m.mu.Unlock() @@ -454,28 +480,29 @@ func (m *Manager) JobLogs(ctx context.Context, j Job, n int) ([]map[string]any, if m.journalSudo { args = append([]string{"sudo", "-n"}, args...) } - out, err := run(ctx, "", args...) - if err != nil { - return nil, err - } var res []map[string]any - sc := bufio.NewScanner(strings.NewReader(out)) - sc.Buffer(make([]byte, 0, 64<<10), 4<<20) - for sc.Scan() { + needle := "(" + j.User + ") CMD (" + j.Command + ")" + err := runLines(ctx, func(line []byte) { + if !bytes.Contains(line, []byte(j.Command[:min(len(j.Command), 40)])) { // cheap pre-filter before JSON + return + } var raw struct { Msg any `json:"MESSAGE"` TS string `json:"__REALTIME_TIMESTAMP"` PID string `json:"_PID"` } - if json.Unmarshal(sc.Bytes(), &raw) != nil { - continue + if json.Unmarshal(line, &raw) != nil { + return } msg, _ := raw.Msg.(string) - if !strings.Contains(msg, "("+j.User+") CMD ("+j.Command+")") { - continue + if !strings.Contains(msg, needle) { + return } us, _ := strconv.ParseInt(raw.TS, 10, 64) res = append(res, map[string]any{"t": us / 1000, "p": 6, "ident": "CRON", "pid": raw.PID, "msg": msg}) + }, args...) + if err != nil { + return nil, err } if len(res) > n { res = res[len(res)-n:]