fix: rework history aggregation, live streaming and visual refresh

History was effectively frozen: the find pattern 'access.log*' never
matched Caddy's native rotation naming (access-<ts>.log.gz), so rotated
files were invisible and the server-side cache - invalidated only by new
.gz files - never refreshed after the first load.

- match both Caddy native (access-*.log.gz) and logrotate (access.log.N)
  rotation naming
- cache aggregation per rotated file (immutable, scanned exactly once);
  always scan the current log fresh, so history never goes stale
- always bucket hourly on the server, re-bucket client-side so all tabs
  share the same caches
- cap IPs at top 50 per file so output stays small on busy servers
- aggregate request duration so history shows real avg response time
- tail -F instead of -f: Caddy rotates by rename, which silently killed
  the live stream at every rotation on high-traffic servers
- process live lines in batches instead of one Update cycle per line
- fetch CPU/MEM stats every 5s instead of every second
- stable sort in topN: equal counts no longer flicker and swap rows
- drop the 1 Month tab (Live / 1 Day / 1 Week remain)
- retro green-phosphor CRT look: double borders, ▓░ bars, amber accents

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
SW 2026-06-12 21:30:38 +02:00
parent fb09f6e028
commit 7c6adbc17e
3 changed files with 262 additions and 208 deletions

View file

@ -3,6 +3,7 @@ package parser
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"sort"
"strconv" "strconv"
"strings" "strings"
"time" "time"
@ -47,17 +48,6 @@ func ParseLine(line string) *LogEntry {
return &entry return &entry
} }
// ParseLines parses multiple lines and returns valid entries.
func ParseLines(lines []string) []*LogEntry {
entries := make([]*LogEntry, 0, len(lines))
for _, l := range lines {
if e := ParseLine(l); e != nil {
entries = append(entries, e)
}
}
return entries
}
// -- Stats (used for live view and derived from AggregatedHistory for history) -- // -- Stats (used for live view and derived from AggregatedHistory for history) --
type Stats struct { type Stats struct {
@ -100,14 +90,6 @@ func (s *Stats) BytesPerSec(windowSecs float64) float64 {
return float64(s.TotalBytes) / windowSecs return float64(s.TotalBytes) / windowSecs
} }
func ComputeStats(entries []*LogEntry) *Stats {
s := NewStats()
for _, e := range entries {
s.Add(e)
}
return s
}
// -- Bucket (time-series bar chart data) -------------------------------------- // -- Bucket (time-series bar chart data) --------------------------------------
type Bucket struct { type Bucket struct {
@ -116,43 +98,47 @@ type Bucket struct {
Bytes int64 Bytes int64
} }
func Bucketize(entries []*LogEntry, bucketSize time.Duration) []Bucket {
if len(entries) == 0 {
return nil
}
buckets := map[int64]*Bucket{}
for _, e := range entries {
t := e.Time().Truncate(bucketSize)
key := t.Unix()
if _, ok := buckets[key]; !ok {
buckets[key] = &Bucket{Time: t}
}
buckets[key].Requests++
buckets[key].Bytes += e.Size
}
result := make([]Bucket, 0, len(buckets))
for _, b := range buckets {
result = append(result, *b)
}
return result
}
// -- AggregatedHistory (server-side aggregation result) ----------------------- // -- AggregatedHistory (server-side aggregation result) -----------------------
// AggregatedBucket is one time bucket returned by server-side awk aggregation. // AggregatedBucket is one time bucket returned by server-side awk aggregation.
type AggregatedBucket struct { type AggregatedBucket struct {
Time time.Time Time time.Time
Requests, Bytes int64 Requests, Bytes int64
Status2xx, Status3xx, Status4xx, Status5xx int64 Status2xx, Status3xx, Status4xx, Status5xx int64
} }
// AggregatedHistory holds the full result of a server-side history aggregation. // AggregatedHistory holds the merged result of per-file server-side aggregations.
type AggregatedHistory struct { type AggregatedHistory struct {
Buckets []AggregatedBucket Buckets []AggregatedBucket
TopHosts map[string]int64 TopHosts map[string]int64
TopIPs map[string]int64 TopIPs map[string]int64
TotalReqs int64 TotalReqs int64
TotalBytes int64 TotalBytes int64
DurSumSecs float64
}
// FilterSince drops buckets older than the window and recomputes totals from
// the remaining buckets, so the chart and the totals always agree. TopHosts and
// TopIPs come from whole rotated files and may slightly overcount at the oldest
// edge of the window — acceptable for a monitoring view.
func (h *AggregatedHistory) FilterSince(since int64) {
avgDur := 0.0
if h.TotalReqs > 0 {
avgDur = h.DurSumSecs / float64(h.TotalReqs)
}
kept := h.Buckets[:0]
var reqs, bytes int64
for _, b := range h.Buckets {
if b.Time.Unix() >= since {
kept = append(kept, b)
reqs += b.Requests
bytes += b.Bytes
}
}
h.Buckets = kept
h.TotalReqs = reqs
h.TotalBytes = bytes
h.DurSumSecs = avgDur * float64(reqs)
} }
// ToStats converts aggregated history to a Stats object for use in the TUI panels. // ToStats converts aggregated history to a Stats object for use in the TUI panels.
@ -162,6 +148,9 @@ func (h *AggregatedHistory) ToStats() *Stats {
s.TotalBytes = h.TotalBytes s.TotalBytes = h.TotalBytes
s.TopHosts = h.TopHosts s.TopHosts = h.TopHosts
s.TopIPs = h.TopIPs s.TopIPs = h.TopIPs
if h.TotalReqs > 0 {
s.AvgDurationMs = h.DurSumSecs / float64(h.TotalReqs) * 1000
}
for _, b := range h.Buckets { for _, b := range h.Buckets {
s.StatusCodes[200] += b.Status2xx s.StatusCodes[200] += b.Status2xx
s.StatusCodes[301] += b.Status3xx s.StatusCodes[301] += b.Status3xx
@ -171,18 +160,37 @@ func (h *AggregatedHistory) ToStats() *Stats {
return s return s
} }
// ToBuckets converts aggregated buckets to the Bucket slice used by the bar chart. // ToBuckets re-buckets the hourly server-side buckets into the granularity the
func (h *AggregatedHistory) ToBuckets() []Bucket { // chart wants (e.g. 6h for the week view) and returns them sorted by time.
out := make([]Bucket, len(h.Buckets)) func (h *AggregatedHistory) ToBuckets(size time.Duration) []Bucket {
for i, ab := range h.Buckets { secs := int64(size.Seconds())
out[i] = Bucket{Time: ab.Time, Requests: ab.Requests, Bytes: ab.Bytes} if secs < 1 {
secs = 1
} }
merged := map[int64]*Bucket{}
for _, ab := range h.Buckets {
ts := ab.Time.Unix() / secs * secs
b, ok := merged[ts]
if !ok {
b = &Bucket{Time: time.Unix(ts, 0)}
merged[ts] = b
}
b.Requests += ab.Requests
b.Bytes += ab.Bytes
}
out := make([]Bucket, 0, len(merged))
for _, b := range merged {
out = append(out, *b)
}
sort.Slice(out, func(i, j int) bool { return out[i].Time.Before(out[j].Time) })
return out return out
} }
// ParseAggregatedOutput parses the output of the server-side awk aggregation script. // ParseAggregatedOutput parses the concatenated output of per-file server-side
// awk aggregations. Sections may repeat (one set per log file) and are merged:
// bucket counts sum by timestamp, host/IP counts sum by key, totals sum.
// //
// Format: // Per-file format:
// //
// BUCKETS // BUCKETS
// <unix_ts> <count> <bytes> <2xx> <3xx> <4xx> <5xx> // <unix_ts> <count> <bytes> <2xx> <3xx> <4xx> <5xx>
@ -191,14 +199,16 @@ func (h *AggregatedHistory) ToBuckets() []Bucket {
// IPS // IPS
// <count> <ip> // <count> <ip>
// TOTALS // TOTALS
// <total_reqs> <total_bytes> // <total_reqs> <total_bytes> <duration_sum_secs>
func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) { func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) {
h := &AggregatedHistory{ h := &AggregatedHistory{
TopHosts: make(map[string]int64), TopHosts: make(map[string]int64),
TopIPs: make(map[string]int64), TopIPs: make(map[string]int64),
} }
bucketsByTS := map[int64]*AggregatedBucket{}
section := "" section := ""
sawSection := false
for _, line := range strings.Split(strings.TrimSpace(raw), "\n") { for _, line := range strings.Split(strings.TrimSpace(raw), "\n") {
line = strings.TrimSpace(line) line = strings.TrimSpace(line)
if line == "" { if line == "" {
@ -207,6 +217,7 @@ func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) {
switch line { switch line {
case "BUCKETS", "HOSTS", "IPS", "TOTALS": case "BUCKETS", "HOSTS", "IPS", "TOTALS":
section = line section = line
sawSection = true
continue continue
} }
@ -220,14 +231,18 @@ func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) {
if err != nil { if err != nil {
continue continue
} }
ab := AggregatedBucket{Time: time.Unix(ts, 0)} ab, ok := bucketsByTS[ts]
ab.Requests, _ = strconv.ParseInt(fields[1], 10, 64) if !ok {
ab.Bytes, _ = strconv.ParseInt(fields[2], 10, 64) ab = &AggregatedBucket{Time: time.Unix(ts, 0)}
ab.Status2xx, _ = strconv.ParseInt(fields[3], 10, 64) bucketsByTS[ts] = ab
ab.Status3xx, _ = strconv.ParseInt(fields[4], 10, 64) }
ab.Status4xx, _ = strconv.ParseInt(fields[5], 10, 64) n := func(i int) int64 { v, _ := strconv.ParseInt(fields[i], 10, 64); return v }
ab.Status5xx, _ = strconv.ParseInt(fields[6], 10, 64) ab.Requests += n(1)
h.Buckets = append(h.Buckets, ab) ab.Bytes += n(2)
ab.Status2xx += n(3)
ab.Status3xx += n(4)
ab.Status4xx += n(5)
ab.Status5xx += n(6)
case "HOSTS": case "HOSTS":
if len(fields) < 2 { if len(fields) < 2 {
@ -235,26 +250,38 @@ func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) {
} }
count, _ := strconv.ParseInt(fields[0], 10, 64) count, _ := strconv.ParseInt(fields[0], 10, 64)
host := strings.Join(fields[1:], " ") host := strings.Join(fields[1:], " ")
h.TopHosts[host] = count h.TopHosts[host] += count
case "IPS": case "IPS":
if len(fields) < 2 { if len(fields) < 2 {
continue continue
} }
count, _ := strconv.ParseInt(fields[0], 10, 64) count, _ := strconv.ParseInt(fields[0], 10, 64)
h.TopIPs[fields[1]] = count h.TopIPs[fields[1]] += count
case "TOTALS": case "TOTALS":
if len(fields) < 2 { if len(fields) < 2 {
continue continue
} }
h.TotalReqs, _ = strconv.ParseInt(fields[0], 10, 64) reqs, _ := strconv.ParseInt(fields[0], 10, 64)
h.TotalBytes, _ = strconv.ParseInt(fields[1], 10, 64) bytes, _ := strconv.ParseInt(fields[1], 10, 64)
h.TotalReqs += reqs
h.TotalBytes += bytes
if len(fields) >= 3 {
dur, _ := strconv.ParseFloat(fields[2], 64)
h.DurSumSecs += dur
}
} }
} }
if section == "" { if !sawSection {
return nil, fmt.Errorf("no aggregation output received from server") return nil, fmt.Errorf("no aggregation output received from server")
} }
h.Buckets = make([]AggregatedBucket, 0, len(bucketsByTS))
for _, ab := range bucketsByTS {
h.Buckets = append(h.Buckets, *ab)
}
sort.Slice(h.Buckets, func(i, j int) bool { return h.Buckets[i].Time.Before(h.Buckets[j].Time) })
return h, nil return h, nil
} }

View file

@ -64,20 +64,7 @@ func (c *Client) RunCommand(cmd string) (string, error) {
return string(out), err return string(out), err
} }
// TailLines reads the last n lines of a remote file. // StreamLines streams new lines from a remote file (like tail -F) into the channel.
func (c *Client) TailLines(path string, n int) ([]string, error) {
out, err := c.RunCommand(fmt.Sprintf("tail -n %d %s 2>/dev/null", n, path))
if err != nil {
return nil, fmt.Errorf("cannot read %s: %w", path, err)
}
lines := strings.Split(strings.TrimSpace(out), "\n")
if len(lines) == 1 && lines[0] == "" {
return nil, nil
}
return lines, nil
}
// StreamLines streams new lines from a remote file (like tail -f) into the channel.
func (c *Client) StreamLines(path string, lines chan<- string, done <-chan struct{}) error { func (c *Client) StreamLines(path string, lines chan<- string, done <-chan struct{}) error {
session, err := c.conn.NewSession() session, err := c.conn.NewSession()
if err != nil { if err != nil {
@ -90,7 +77,9 @@ func (c *Client) StreamLines(path string, lines chan<- string, done <-chan struc
return err return err
} }
if err := session.Start(fmt.Sprintf("tail -n 0 -f %s 2>/dev/null", path)); err != nil { // -F follows by name: Caddy rotates by renaming the file, which would
// silently kill a plain -f stream at every rotation.
if err := session.Start(fmt.Sprintf("tail -n 0 -F %s 2>/dev/null", path)); err != nil {
session.Close() session.Close()
return err return err
} }
@ -98,6 +87,7 @@ func (c *Client) StreamLines(path string, lines chan<- string, done <-chan struc
go func() { go func() {
defer session.Close() defer session.Close()
scanner := bufio.NewScanner(stdout) scanner := bufio.NewScanner(stdout)
scanner.Buffer(make([]byte, 64*1024), 1024*1024)
for scanner.Scan() { for scanner.Scan() {
select { select {
case <-done: case <-done:
@ -241,27 +231,11 @@ func parseCPUStat(raw string) float64 {
return (totalDiff - idleDiff) / totalDiff * 100 return (totalDiff - idleDiff) / totalDiff * 100
} }
// LoadAggregatedHistory aggregates log data on the remote server using awk, // historyAwkProg aggregates Caddy JSON log lines into hourly summary buckets.
// returning only summary buckets instead of raw lines. This is critical for // It runs entirely on the server and uses only POSIX awk constructs (works
// high-traffic servers where raw line transfer would be impractical. // with gawk, mawk and busybox awk). IPs are capped to the top 50 per file so
// // the output stays small even on servers with millions of unique clients.
// bucketSecs: size of each time bucket in seconds (e.g. 3600 for hourly) const historyAwkProg = `
// maxAgeDays: find only files modified within this many days (performance filter)
func (c *Client) LoadAggregatedHistory(logPath string, since, bucketSecs int64, maxAgeDays int) (*parser.AggregatedHistory, error) {
lastSlash := strings.LastIndex(logPath, "/")
dir, base := "/", logPath
if lastSlash >= 0 {
dir = logPath[:lastSlash]
if dir == "" {
dir = "/"
}
base = logPath[lastSlash+1:]
}
// awk aggregation script — runs entirely on the server.
// Parses JSON log lines, groups by time bucket, outputs summary.
// Uses only POSIX-compatible awk constructs.
awkProg := `
BEGIN { OFS=" " } BEGIN { OFS=" " }
{ {
if (index($0,"\"msg\":\"handled request\"") == 0) next if (index($0,"\"msg\":\"handled request\"") == 0) next
@ -274,6 +248,8 @@ BEGIN { OFS=" " }
st = (p>0) ? substr($0,p+9,5)+0 : 0 st = (p>0) ? substr($0,p+9,5)+0 : 0
p = index($0,"\"size\":") p = index($0,"\"size\":")
sz = (p>0) ? substr($0,p+7,15)+0 : 0 sz = (p>0) ? substr($0,p+7,15)+0 : 0
p = index($0,"\"duration\":")
du = (p>0) ? substr($0,p+11,14)+0 : 0
p = index($0,"\"host\":\"") p = index($0,"\"host\":\"")
if (p>0) { rest=substr($0,p+8); q=index(rest,"\""); ho=(q>0)?substr(rest,1,q-1):"?" } else ho="?" if (p>0) { rest=substr($0,p+8); q=index(rest,"\""); ho=(q>0)?substr(rest,1,q-1):"?" } else ho="?"
p = index($0,"\"remote_ip\":\"") p = index($0,"\"remote_ip\":\"")
@ -284,7 +260,7 @@ BEGIN { OFS=" " }
else if (st>=300) bs3[b]++ else if (st>=300) bs3[b]++
else if (st>=200) bs2[b]++ else if (st>=200) bs2[b]++
hc[ho]++; ic[ip]++ hc[ho]++; ic[ip]++
tot++; totb+=sz tot++; totb+=sz; totd+=du
} }
END { END {
print "BUCKETS" print "BUCKETS"
@ -292,35 +268,69 @@ END {
print "HOSTS" print "HOSTS"
for (h in hc) print hc[h], h for (h in hc) print hc[h], h
print "IPS" print "IPS"
for (i in ic) print ic[i], i k = 50
for (i in ic) n++
while (k > 0 && n > 0) {
max = -1; mi = ""
for (i in ic) if (ic[i] > max) { max = ic[i]; mi = i }
print ic[mi], mi
delete ic[mi]
k--; n--
}
print "TOTALS" print "TOTALS"
print tot+0, totb+0 print tot+0, totb+0, totd+0
}` }`
// Cache file is keyed by log path + bucket size (not since, which changes every second). // LoadAggregatedHistory aggregates log data on the remote server using awk,
// Cache is valid as long as no new .gz file appeared since it was written. // returning only summary buckets instead of raw lines. This is critical for
// A log rotation (every ~9 min on high-traffic servers) is the only invalidation event. // high-traffic servers where raw line transfer would be impractical.
cacheKey := strings.NewReplacer("/", "_", ".", "_", "-", "_").Replace(base) //
cachePath := fmt.Sprintf("/tmp/caddyview_%s_%d.cache", cacheKey, bucketSecs) // Rotated log files are immutable, so each one is aggregated exactly once and
// the result is cached on the server (/tmp/caddyview_*.cache); after a
// rotation only the single new file is scanned. The current log file is always
// scanned fresh, so the view never goes stale. Buckets are always hourly; the
// caller re-buckets client-side, which lets all tabs share the same caches.
//
// since: unix cutoff for the current (still growing) log file
// maxAgeDays: only look at files modified within this many days
func (c *Client) LoadAggregatedHistory(logPath string, since int64, maxAgeDays int) (*parser.AggregatedHistory, error) {
lastSlash := strings.LastIndex(logPath, "/")
dir, base := "/", logPath
if lastSlash >= 0 {
dir = logPath[:lastSlash]
if dir == "" {
dir = "/"
}
base = logPath[lastSlash+1:]
}
findPipe := fmt.Sprintf( // Match both Caddy's native rotation naming (access-2026-06-12T10-30-00.000.log
`find %s -name '%s*' -type f -mtime -%d 2>/dev/null | sort -r | `+ // and .log.gz next to access.log) and classic logrotate naming (access.log.1.gz).
`while read f; do case "$f" in *.gz) zcat "$f" ;; *) cat "$f" ;; esac; done 2>/dev/null | `+ stem, ext := base, ""
`awk -v cutoff=%d -v bucket=%d '%s'`, if dot := strings.LastIndex(base, "."); dot > 0 {
dir, base, maxAgeDays+1, since, bucketSecs, awkProg, stem, ext = base[:dot], base[dot:]
}
namePattern := fmt.Sprintf(
`\( -name '%s' -o -name '%s.*' -o -name '%s-*%s' -o -name '%s-*%s.gz' \)`,
base, base, stem, ext, stem, ext,
) )
// Wrap with server-side cache: hit → instant cat; miss → run awk and tee to cache atomically.
cmd := fmt.Sprintf( cmd := fmt.Sprintf(
`CACHE=%s; `+ `find /tmp -maxdepth 1 -name 'caddyview_*' -mtime +8 -delete 2>/dev/null; `+
`LATEST=$(find %s -name '%s*.gz' -mtime -%d 2>/dev/null | sort | tail -1); `+ `find %s %s -type f -mtime -%d 2>/dev/null | sort | while read -r f; do `+
`if [ -f "$CACHE" ] && { [ -z "$LATEST" ] || [ "$CACHE" -nt "$LATEST" ]; }; then `+ `if [ "$f" = "%s" ]; then `+
`cat "$CACHE"; `+ `awk -v cutoff=%d -v bucket=3600 '%s' < "$f"; `+
`else `+ `else `+
`TMPFILE=$(mktemp /tmp/caddyview.XXXXXX 2>/dev/null || echo /tmp/caddyview_fallback.tmp); `+ `key=$(printf '%%s' "$f" | tr -c 'A-Za-z0-9' '_'); c="/tmp/caddyview_${key}.cache"; `+
`%s | tee "$TMPFILE" && mv "$TMPFILE" "$CACHE" 2>/dev/null || cat "$TMPFILE" 2>/dev/null; `+ `if [ -s "$c" ] && [ "$c" -nt "$f" ]; then cat "$c"; else `+
`fi`, `t=$(mktemp /tmp/caddyview.XXXXXX 2>/dev/null) || t="/tmp/caddyview_tmp.$$"; `+
cachePath, dir, base, maxAgeDays+1, findPipe, `case "$f" in *.gz) zcat "$f" ;; *) cat "$f" ;; esac 2>/dev/null `+
`| awk -v cutoff=0 -v bucket=3600 '%s' > "$t" && mv "$t" "$c" && cat "$c" || cat "$t" 2>/dev/null; `+
`fi; fi; done`,
dir, namePattern, maxAgeDays+1,
logPath,
since, historyAwkProg,
historyAwkProg,
) )
out, err := c.RunCommand(cmd) out, err := c.RunCommand(cmd)

View file

@ -18,67 +18,65 @@ import (
type viewTab int type viewTab int
const ( const (
tabLive viewTab = 0 tabLive viewTab = 0
tabDay viewTab = 1 tabDay viewTab = 1
tabWeek viewTab = 2 tabWeek viewTab = 2
tabMonth viewTab = 3
) )
var tabLabels = []string{"[1] Live", "[2] 1 Day", "[3] 1 Week", "[4] 1 Month"} var tabLabels = []string{"[1] LIVE", "[2] 1 DAY", "[3] 1 WEEK"}
// tabWindow returns how far back each history tab reaches. // tabWindow returns how far back each history tab reaches.
func tabWindow(t viewTab) time.Duration { func tabWindow(t viewTab) time.Duration {
switch t { if t == tabWeek {
case tabDay:
return 24 * time.Hour
case tabWeek:
return 7 * 24 * time.Hour return 7 * 24 * time.Hour
case tabMonth:
return 30 * 24 * time.Hour
} }
return 0 return 24 * time.Hour
} }
// tabBucketSize returns the bucket granularity for the bar chart. // tabBucketSize returns the bucket granularity for the bar chart.
func tabBucketSize(t viewTab) time.Duration { func tabBucketSize(t viewTab) time.Duration {
switch t { if t == tabWeek {
case tabDay:
return time.Hour
case tabWeek:
return 6 * time.Hour return 6 * time.Hour
case tabMonth:
return 24 * time.Hour
} }
return time.Hour return time.Hour
} }
// -- styles ------------------------------------------------------------------- // -- styles (green phosphor / amber CRT palette) --------------------------------
var ( var (
colGreen = lipgloss.Color("40") // primary phosphor green
colBright = lipgloss.Color("46") // highlights
colDark = lipgloss.Color("28") // borders
colFaint = lipgloss.Color("22") // empty bar fill
colDim = lipgloss.Color("65") // secondary text
colAmber = lipgloss.Color("214") // warnings
colRed = lipgloss.Color("160") // errors
styleBorder = lipgloss.NewStyle(). styleBorder = lipgloss.NewStyle().
Border(lipgloss.RoundedBorder()). Border(lipgloss.DoubleBorder()).
BorderForeground(lipgloss.Color("62")). BorderForeground(colDark).
Padding(0, 1) Padding(0, 1)
styleTitle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("212")) styleTitle = lipgloss.NewStyle().Bold(true).Foreground(colBright)
styleHeader = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("39")) styleHeader = lipgloss.NewStyle().Bold(true).Foreground(colGreen)
styleTabActive = lipgloss.NewStyle(). styleTabActive = lipgloss.NewStyle().
Bold(true). Bold(true).
Foreground(lipgloss.Color("16")). Foreground(lipgloss.Color("16")).
Background(lipgloss.Color("62")). Background(colGreen).
Padding(0, 1) Padding(0, 1)
styleTabInactive = lipgloss.NewStyle(). styleTabInactive = lipgloss.NewStyle().
Foreground(lipgloss.Color("240")). Foreground(colDim).
Padding(0, 1) Padding(0, 1)
styleGood = lipgloss.NewStyle().Foreground(lipgloss.Color("42")) styleGood = lipgloss.NewStyle().Foreground(colGreen)
styleWarn = lipgloss.NewStyle().Foreground(lipgloss.Color("214")) styleWarn = lipgloss.NewStyle().Foreground(colAmber)
styleBad = lipgloss.NewStyle().Foreground(lipgloss.Color("196")) styleBad = lipgloss.NewStyle().Foreground(colRed)
styleDim = lipgloss.NewStyle().Foreground(lipgloss.Color("240")) styleDim = lipgloss.NewStyle().Foreground(colDim)
styleBar = lipgloss.NewStyle().Foreground(lipgloss.Color("62")) styleBar = lipgloss.NewStyle().Foreground(colGreen)
styleBarEmpty = lipgloss.NewStyle().Foreground(colFaint)
) )
// -- history state ------------------------------------------------------------ // -- history state ------------------------------------------------------------
@ -97,16 +95,16 @@ type historyState struct {
// -- messages ----------------------------------------------------------------- // -- messages -----------------------------------------------------------------
type tickMsg time.Time type tickMsg time.Time
type newLineMsg string type newLinesMsg []string
type sysStatsMsg struct { type sysStatsMsg struct {
cpu float64 cpu float64
memUsed int64 memUsed int64
memTotal int64 memTotal int64
} }
type historyLoadedMsg struct { type historyLoadedMsg struct {
tab viewTab tab viewTab
agg *parser.AggregatedHistory agg *parser.AggregatedHistory
err error err error
} }
type errMsg struct{ err error } type errMsg struct{ err error }
@ -118,17 +116,17 @@ type model struct {
lineCh chan string lineCh chan string
doneCh chan struct{} doneCh chan struct{}
liveEntries []*parser.LogEntry liveStats *parser.Stats
liveStats *parser.Stats liveStart time.Time
liveStart time.Time
activeTab viewTab activeTab viewTab
history [3]*historyState history [2]*historyState
cpu float64 cpu float64
memUsed int64 memUsed int64
memTotal int64 memTotal int64
ticks int
width int width int
height int height int
err error err error
@ -138,7 +136,7 @@ func initialModel(client *sshclient.Client, logPath string) model {
m := model{ m := model{
client: client, client: client,
logPath: logPath, logPath: logPath,
lineCh: make(chan string, 256), lineCh: make(chan string, 8192),
doneCh: make(chan struct{}), doneCh: make(chan struct{}),
liveStats: parser.NewStats(), liveStats: parser.NewStats(),
liveStart: time.Now(), liveStart: time.Now(),
@ -176,6 +174,8 @@ func tick() tea.Cmd {
return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) }) return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) })
} }
const sysStatsEvery = 5 // seconds; each fetch opens SSH sessions and sleeps 0.5s server-side
func fetchSysStats(client *sshclient.Client) tea.Cmd { func fetchSysStats(client *sshclient.Client) tea.Cmd {
return func() tea.Msg { return func() tea.Msg {
cpu, used, total, err := client.SysStats() cpu, used, total, err := client.SysStats()
@ -186,26 +186,38 @@ func fetchSysStats(client *sshclient.Client) tea.Cmd {
} }
} }
// drainLines grabs everything currently buffered in the channel as one batch,
// so a traffic burst costs one Update/render cycle instead of one per line.
func drainLines(ch chan string) tea.Cmd { func drainLines(ch chan string) tea.Cmd {
return func() tea.Msg { return func() tea.Msg {
select { var batch []string
case line := <-ch: for {
return newLineMsg(line) select {
default: case line := <-ch:
return nil batch = append(batch, line)
if len(batch) >= 10000 {
return newLinesMsg(batch)
}
default:
if len(batch) == 0 {
return nil
}
return newLinesMsg(batch)
}
} }
} }
} }
func loadHistory(client *sshclient.Client, logPath string, tab viewTab) tea.Cmd { func loadHistory(client *sshclient.Client, logPath string, tab viewTab) tea.Cmd {
since := time.Now().Add(-tabWindow(tab)).Unix() window := tabWindow(tab)
bucketSecs := int64(tabBucketSize(tab).Seconds()) since := time.Now().Add(-window).Unix()
maxAgeDays := int(tabWindow(tab).Hours()/24) + 1 maxAgeDays := int(window.Hours()/24) + 1
return func() tea.Msg { return func() tea.Msg {
agg, err := client.LoadAggregatedHistory(logPath, since, bucketSecs, maxAgeDays) agg, err := client.LoadAggregatedHistory(logPath, since, maxAgeDays)
if err != nil { if err != nil {
return historyLoadedMsg{tab: tab, err: err} return historyLoadedMsg{tab: tab, err: err}
} }
agg.FilterSince(since)
return historyLoadedMsg{tab: tab, agg: agg} return historyLoadedMsg{tab: tab, agg: agg}
} }
} }
@ -232,13 +244,14 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
case "3": case "3":
m.activeTab = tabWeek m.activeTab = tabWeek
return m, m.ensureHistory(tabWeek) return m, m.ensureHistory(tabWeek)
case "4":
m.activeTab = tabMonth
return m, m.ensureHistory(tabMonth)
} }
case tickMsg: case tickMsg:
cmds := []tea.Cmd{tick(), drainLines(m.lineCh), fetchSysStats(m.client)} m.ticks++
cmds := []tea.Cmd{tick(), drainLines(m.lineCh)}
if m.ticks%sysStatsEvery == 0 {
cmds = append(cmds, fetchSysStats(m.client))
}
if m.activeTab != tabLive { if m.activeTab != tabLive {
if cmd := m.maybeRefreshHistory(m.activeTab); cmd != nil { if cmd := m.maybeRefreshHistory(m.activeTab); cmd != nil {
cmds = append(cmds, cmd) cmds = append(cmds, cmd)
@ -246,12 +259,10 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
} }
return m, tea.Batch(cmds...) return m, tea.Batch(cmds...)
case newLineMsg: case newLinesMsg:
if e := parser.ParseLine(string(msg)); e != nil { for _, line := range msg {
m.liveStats.Add(e) if e := parser.ParseLine(line); e != nil {
m.liveEntries = append(m.liveEntries, e) m.liveStats.Add(e)
if len(m.liveEntries) > 10000 {
m.liveEntries = m.liveEntries[len(m.liveEntries)-10000:]
} }
} }
return m, drainLines(m.lineCh) return m, drainLines(m.lineCh)
@ -268,7 +279,7 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
h.err = msg.err h.err = msg.err
if msg.err == nil && msg.agg != nil { if msg.err == nil && msg.agg != nil {
h.stats = msg.agg.ToStats() h.stats = msg.agg.ToStats()
h.buckets = msg.agg.ToBuckets() h.buckets = msg.agg.ToBuckets(tabBucketSize(msg.tab))
} }
case errMsg: case errMsg:
@ -311,7 +322,7 @@ func (m *model) maybeRefreshHistory(tab viewTab) tea.Cmd {
func (m model) View() string { func (m model) View() string {
if m.err != nil { if m.err != nil {
return styleBad.Render(fmt.Sprintf("Error: %v\n\nPress q to quit.", m.err)) return styleBad.Render(fmt.Sprintf("ERROR: %v\n\nPress q to quit.", m.err))
} }
parts := []string{ parts := []string{
m.renderHeader(), m.renderHeader(),
@ -321,13 +332,11 @@ func (m model) View() string {
case tabLive: case tabLive:
parts = append(parts, m.renderLive()) parts = append(parts, m.renderLive())
case tabDay: case tabDay:
parts = append(parts, m.renderHistory(0, "last 24 hours")) parts = append(parts, m.renderHistory(0, "LAST 24 HOURS"))
case tabWeek: case tabWeek:
parts = append(parts, m.renderHistory(1, "last 7 days")) parts = append(parts, m.renderHistory(1, "LAST 7 DAYS"))
case tabMonth:
parts = append(parts, m.renderHistory(2, "last 30 days"))
} }
parts = append(parts, styleDim.Render(" 1 live 2 day 3 week 4 month q quit")) parts = append(parts, styleDim.Render(" [1] LIVE [2] 1 DAY [3] 1 WEEK [Q] QUIT"))
return lipgloss.JoinVertical(lipgloss.Left, parts...) return lipgloss.JoinVertical(lipgloss.Left, parts...)
} }
@ -349,21 +358,21 @@ func (m model) renderHeader() string {
memColor = styleWarn memColor = styleWarn
} }
row := strings.Join([]string{ row := strings.Join([]string{
styleTitle.Render("caddyview"), styleTitle.Render("░▒▓ CADDYVIEW ▓▒░"),
" ", " ",
styleHeader.Render("host: ") + m.client.Host(), styleHeader.Render("HOST: ") + m.client.Host(),
" ", " ",
styleHeader.Render("log: ") + styleDim.Render(m.logPath), styleHeader.Render("LOG: ") + styleDim.Render(m.logPath),
" ", " ",
styleHeader.Render("up: ") + time.Since(m.liveStart).Round(time.Second).String(), styleHeader.Render("UP: ") + time.Since(m.liveStart).Round(time.Second).String(),
" ", " ",
styleHeader.Render("CPU: ") + cpuColor.Render(fmt.Sprintf("%.1f%%", m.cpu)), styleHeader.Render("CPU: ") + cpuColor.Render(fmt.Sprintf("%.1f%%", m.cpu)),
" ", " ",
styleHeader.Render("MEM: ") + memColor.Render(fmt.Sprintf("%d/%d MB (%.0f%%)", m.memUsed, m.memTotal, memPct)), styleHeader.Render("MEM: ") + memColor.Render(fmt.Sprintf("%d/%d MB (%.0f%%)", m.memUsed, m.memTotal, memPct)),
}, "") }, "")
return lipgloss.NewStyle(). return lipgloss.NewStyle().
BorderBottom(true).BorderStyle(lipgloss.NormalBorder()). BorderBottom(true).BorderStyle(lipgloss.DoubleBorder()).
BorderForeground(lipgloss.Color("62")).Width(m.width). BorderForeground(colDark).Width(m.width).
Render(row) Render(row)
} }
@ -378,7 +387,7 @@ func (m model) renderTabBar() string {
} }
return lipgloss.NewStyle(). return lipgloss.NewStyle().
BorderBottom(true).BorderStyle(lipgloss.NormalBorder()). BorderBottom(true).BorderStyle(lipgloss.NormalBorder()).
BorderForeground(lipgloss.Color("62")).Width(m.width). BorderForeground(colDark).Width(m.width).
Render(strings.Join(tabs, " ")) Render(strings.Join(tabs, " "))
} }
@ -391,7 +400,7 @@ func (m model) renderLive() string {
rps = float64(m.liveStats.TotalRequests) / elapsed rps = float64(m.liveStats.TotalRequests) / elapsed
} }
statsBox := styleBorder.Render(strings.Join([]string{ statsBox := styleBorder.Render(strings.Join([]string{
styleHeader.Render("Live Stats"), styleHeader.Render("LIVE STATS"),
"", "",
fmt.Sprintf("Requests : %s", styleGood.Render(formatComma(m.liveStats.TotalRequests))), fmt.Sprintf("Requests : %s", styleGood.Render(formatComma(m.liveStats.TotalRequests))),
fmt.Sprintf("Req/sec : %s", styleGood.Render(fmt.Sprintf("%.2f", rps))), fmt.Sprintf("Req/sec : %s", styleGood.Render(fmt.Sprintf("%.2f", rps))),
@ -413,14 +422,14 @@ func (m model) renderHistory(idx int, label string) string {
h := m.history[idx] h := m.history[idx]
if h.loading { if h.loading {
return lipgloss.NewStyle().Padding(2, 4). return lipgloss.NewStyle().Padding(2, 4).
Render(styleWarn.Render("Loading historical data…")) Render(styleWarn.Render("▒▒ LOADING HISTORY ▒▒"))
} }
if !h.loaded { if !h.loaded {
return lipgloss.NewStyle().Padding(2, 4).Render(styleDim.Render("No data loaded.")) return lipgloss.NewStyle().Padding(2, 4).Render(styleDim.Render("No data loaded."))
} }
if h.err != nil { if h.err != nil {
return lipgloss.NewStyle().Padding(2, 4). return lipgloss.NewStyle().Padding(2, 4).
Render(styleBad.Render(fmt.Sprintf("Error: %v", h.err))) Render(styleBad.Render(fmt.Sprintf("ERROR: %v", h.err)))
} }
if h.stats == nil || h.stats.TotalRequests == 0 { if h.stats == nil || h.stats.TotalRequests == 0 {
return lipgloss.NewStyle().Padding(2, 4). return lipgloss.NewStyle().Padding(2, 4).
@ -476,7 +485,7 @@ func (m model) renderBarChart(buckets []parser.Bucket, bucketSize time.Duration,
barMax = 10 barMax = 10
} }
title := fmt.Sprintf("Requests / %s — %s — total: %s req, %s", title := fmt.Sprintf("REQUESTS / %s — %s — TOTAL: %s req, %s",
bucketSizeLabel(bucketSize), label, bucketSizeLabel(bucketSize), label,
formatComma(stats.TotalRequests), formatBytes(stats.TotalBytes), formatComma(stats.TotalRequests), formatBytes(stats.TotalBytes),
) )
@ -487,10 +496,11 @@ func (m model) renderBarChart(buckets []parser.Bucket, bucketSize time.Duration,
if maxReqs > 0 { if maxReqs > 0 {
barLen = int(math.Round(float64(b.Requests) / float64(maxReqs) * float64(barMax))) barLen = int(math.Round(float64(b.Requests) / float64(maxReqs) * float64(barMax)))
} }
bar := strings.Repeat("█", barLen) bar := styleBar.Render(strings.Repeat("▓", barLen)) +
styleBarEmpty.Render(strings.Repeat("░", barMax-barLen))
rows = append(rows, fmt.Sprintf("%-*s %s %s", rows = append(rows, fmt.Sprintf("%-*s %s %s",
labelW, b.Time.Format(timeFmt), labelW, b.Time.Format(timeFmt),
styleBar.Render(fmt.Sprintf("%-*s", barMax, bar)), bar,
styleDim.Render(formatComma(b.Requests)), styleDim.Render(formatComma(b.Requests)),
)) ))
} }
@ -502,7 +512,7 @@ func (m model) renderBarChart(buckets []parser.Bucket, bucketSize time.Duration,
func (m model) renderPeriodStats(stats *parser.Stats) string { func (m model) renderPeriodStats(stats *parser.Stats) string {
return styleBorder.Render(strings.Join([]string{ return styleBorder.Render(strings.Join([]string{
styleHeader.Render("Period Stats"), styleHeader.Render("PERIOD STATS"),
"", "",
fmt.Sprintf("Requests : %s", styleGood.Render(formatComma(stats.TotalRequests))), fmt.Sprintf("Requests : %s", styleGood.Render(formatComma(stats.TotalRequests))),
fmt.Sprintf("Transfer : %s", formatBytes(stats.TotalBytes)), fmt.Sprintf("Transfer : %s", formatBytes(stats.TotalBytes)),
@ -511,7 +521,7 @@ func (m model) renderPeriodStats(stats *parser.Stats) string {
} }
func (m model) renderTopHosts(stats *parser.Stats) string { func (m model) renderTopHosts(stats *parser.Stats) string {
rows := []string{styleHeader.Render("Top Hosts"), ""} rows := []string{styleHeader.Render("TOP HOSTS"), ""}
for _, p := range topN(stats.TopHosts, 8) { for _, p := range topN(stats.TopHosts, 8) {
host := p.key host := p.key
if len(host) > 30 { if len(host) > 30 {
@ -526,7 +536,7 @@ func (m model) renderTopHosts(stats *parser.Stats) string {
} }
func (m model) renderTopIPs(stats *parser.Stats) string { func (m model) renderTopIPs(stats *parser.Stats) string {
rows := []string{styleHeader.Render("Top IPs"), ""} rows := []string{styleHeader.Render("TOP IPS"), ""}
for _, p := range topN(stats.TopIPs, 8) { for _, p := range topN(stats.TopIPs, 8) {
rows = append(rows, fmt.Sprintf("%-20s %s", p.key, styleGood.Render(formatComma(p.val)))) rows = append(rows, fmt.Sprintf("%-20s %s", p.key, styleGood.Render(formatComma(p.val))))
} }
@ -550,7 +560,7 @@ func (m model) renderStatusCodes(stats *parser.Stats) string {
groups["5xx"] += n groups["5xx"] += n
} }
} }
rows := []string{styleHeader.Render("Status Codes"), ""} rows := []string{styleHeader.Render("STATUS CODES"), ""}
for _, g := range []string{"2xx", "3xx", "4xx", "5xx"} { for _, g := range []string{"2xx", "3xx", "4xx", "5xx"} {
s := formatComma(groups[g]) s := formatComma(groups[g])
var colored string var colored string
@ -601,12 +611,19 @@ type kv struct {
val int64 val int64
} }
// topN returns the n highest entries. Ties are broken alphabetically so rows
// with equal counts keep a stable order between renders instead of flickering.
func topN(m map[string]int64, n int) []kv { func topN(m map[string]int64, n int) []kv {
pairs := make([]kv, 0, len(m)) pairs := make([]kv, 0, len(m))
for k, v := range m { for k, v := range m {
pairs = append(pairs, kv{k, v}) pairs = append(pairs, kv{k, v})
} }
sort.Slice(pairs, func(i, j int) bool { return pairs[i].val > pairs[j].val }) sort.Slice(pairs, func(i, j int) bool {
if pairs[i].val != pairs[j].val {
return pairs[i].val > pairs[j].val
}
return pairs[i].key < pairs[j].key
})
if len(pairs) > n { if len(pairs) > n {
return pairs[:n] return pairs[:n]
} }
@ -616,11 +633,11 @@ func topN(m map[string]int64, n int) []kv {
func bucketSizeLabel(d time.Duration) string { func bucketSizeLabel(d time.Duration) string {
switch { switch {
case d >= 24*time.Hour: case d >= 24*time.Hour:
return "day" return "DAY"
case d >= time.Hour: case d >= time.Hour:
return fmt.Sprintf("%dh", int(d.Hours())) return fmt.Sprintf("%dH", int(d.Hours()))
default: default:
return fmt.Sprintf("%dm", int(d.Minutes())) return fmt.Sprintf("%dM", int(d.Minutes()))
} }
} }