seth/Caddyview
1
0
Fork 0
Code Issues Pull requests Projects Releases 17 Packages Wiki Activity Actions

perf: server-side awk aggregation for history views

Browse source 
High-traffic servers (100+ req/s) generated millions of log lines
per day. Transferring raw lines over SSH was impractical (GB of data).

Instead, run an awk script on the server that aggregates all matching
log files (including .gz rotated ones) into time buckets and returns
only a ~50-line summary. For a server doing 180 req/s this reduces
SSH transfer from ~7GB to <1KB per history request.

The awk script filters by timestamp, groups by bucket, and outputs:
- per-bucket request counts, bytes, and status code groups
- top hosts and IPs with counts
- overall totals

Files are pre-filtered with find -mtime to skip archives older than
the requested window.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
SW 2026-06-11 14:50:51 +02:00
parent da6a128882
commit 2be12c24cd
3 changed files with 200 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

157
internal/parser/caddy.go
View file

@ -2,6 +2,8 @@ package parser
import (
"encoding/json"
"fmt"
"strconv"
"strings"
"time"
)
@ -12,10 +14,10 @@ type LogEntry struct {
Level string `json:"level"`
Logger string `json:"logger"`
Request struct {
Method string `json:"method"`
URI string `json:"uri"`
Host string `json:"host"`
Proto string `json:"proto"`
Method string `json:"method"`
URI string `json:"uri"`
Host string `json:"host"`
Proto string `json:"proto"`
RemoteIP string `json:"remote_ip"`
} `json:"request"`
Duration float64 `json:"duration"`
@ -39,7 +41,7 @@ func ParseLine(line string) *LogEntry {
if err := json.Unmarshal([]byte(line), &entry); err != nil {
return nil
}
if entry.Request.Host == "" {
if entry.Request.Host == "" || entry.Status == 0 {
return nil
}
return &entry
@ -56,7 +58,8 @@ func ParseLines(lines []string) []*LogEntry {
return entries
}
// Stats holds aggregated metrics for a time window.
// -- Stats (used for live view and derived from AggregatedHistory for history) --
type Stats struct {
TotalRequests int64
TotalBytes int64
@ -90,7 +93,6 @@ func (s *Stats) Add(e *LogEntry) {
}
}
// BytesPerSec calculates transfer rate given a window duration in seconds.
func (s *Stats) BytesPerSec(windowSecs float64) float64 {
if windowSecs <= 0 {
return 0
@ -98,14 +100,6 @@ func (s *Stats) BytesPerSec(windowSecs float64) float64 {
return float64(s.TotalBytes) / windowSecs
}
// Bucket aggregates entries into fixed time buckets for the history view.
type Bucket struct {
Time time.Time
Requests int64
Bytes int64
}
// ComputeStats builds a Stats object from a slice of log entries.
func ComputeStats(entries []*LogEntry) *Stats {
s := NewStats()
for _, e := range entries {
@ -114,12 +108,18 @@ func ComputeStats(entries []*LogEntry) *Stats {
return s
}
// Bucketize groups entries into buckets of the given duration.
// -- Bucket (time-series bar chart data) --------------------------------------
type Bucket struct {
Time time.Time
Requests 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)
@ -130,10 +130,131 @@ func Bucketize(entries []*LogEntry, bucketSize time.Duration) []Bucket {
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) -----------------------
// AggregatedBucket is one time bucket returned by server-side awk aggregation.
type AggregatedBucket struct {
Time time.Time
Requests, Bytes int64
Status2xx, Status3xx, Status4xx, Status5xx int64
}
// AggregatedHistory holds the full result of a server-side history aggregation.
type AggregatedHistory struct {
Buckets []AggregatedBucket
TopHosts map[string]int64
TopIPs map[string]int64
TotalReqs int64
TotalBytes int64
}
// ToStats converts aggregated history to a Stats object for use in the TUI panels.
func (h *AggregatedHistory) ToStats() *Stats {
s := NewStats()
s.TotalRequests = h.TotalReqs
s.TotalBytes = h.TotalBytes
s.TopHosts = h.TopHosts
s.TopIPs = h.TopIPs
for _, b := range h.Buckets {
s.StatusCodes[200] += b.Status2xx
s.StatusCodes[301] += b.Status3xx
s.StatusCodes[404] += b.Status4xx
s.StatusCodes[500] += b.Status5xx
}
return s
}
// ToBuckets converts aggregated buckets to the Bucket slice used by the bar chart.
func (h *AggregatedHistory) ToBuckets() []Bucket {
out := make([]Bucket, len(h.Buckets))
for i, ab := range h.Buckets {
out[i] = Bucket{Time: ab.Time, Requests: ab.Requests, Bytes: ab.Bytes}
}
return out
}
// ParseAggregatedOutput parses the output of the server-side awk aggregation script.
//
// Format:
//
// BUCKETS
// <unix_ts> <count> <bytes> <2xx> <3xx> <4xx> <5xx>
// HOSTS
// <count> <host>
// IPS
// <count> <ip>
// TOTALS
// <total_reqs> <total_bytes>
func ParseAggregatedOutput(raw string) (*AggregatedHistory, error) {
h := &AggregatedHistory{
TopHosts: make(map[string]int64),
TopIPs: make(map[string]int64),
}
section := ""
for _, line := range strings.Split(strings.TrimSpace(raw), "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
switch line {
case "BUCKETS", "HOSTS", "IPS", "TOTALS":
section = line
continue
}
fields := strings.Fields(line)
switch section {
case "BUCKETS":
if len(fields) < 7 {
continue
}
ts, err := strconv.ParseInt(fields[0], 10, 64)
if err != nil {
continue
}
ab := AggregatedBucket{Time: time.Unix(ts, 0)}
ab.Requests, _ = strconv.ParseInt(fields[1], 10, 64)
ab.Bytes, _ = strconv.ParseInt(fields[2], 10, 64)
ab.Status2xx, _ = strconv.ParseInt(fields[3], 10, 64)
ab.Status3xx, _ = strconv.ParseInt(fields[4], 10, 64)
ab.Status4xx, _ = strconv.ParseInt(fields[5], 10, 64)
ab.Status5xx, _ = strconv.ParseInt(fields[6], 10, 64)
h.Buckets = append(h.Buckets, ab)
case "HOSTS":
if len(fields) < 2 {
continue
}
count, _ := strconv.ParseInt(fields[0], 10, 64)
host := strings.Join(fields[1:], " ")
h.TopHosts[host] = count
case "IPS":
if len(fields) < 2 {
continue
}
count, _ := strconv.ParseInt(fields[0], 10, 64)
h.TopIPs[fields[1]] = count
case "TOTALS":
if len(fields) < 2 {
continue
}
h.TotalReqs, _ = strconv.ParseInt(fields[0], 10, 64)
h.TotalBytes, _ = strconv.ParseInt(fields[1], 10, 64)
}
}
if section == "" {
return nil, fmt.Errorf("no aggregation output received from server")
}
return h, nil
}

66
internal/sshclient/client.go
View file

@ -8,6 +8,7 @@ import (
"path/filepath"
"strings"
"github.com/janvete/caddyview/internal/parser"
"golang.org/x/crypto/ssh"
"golang.org/x/crypto/ssh/agent"
)
@ -240,9 +241,13 @@ func parseCPUStat(raw string) float64 {
return (totalDiff - idleDiff) / totalDiff * 100
}
// LoadHistoricalLines reads log lines newer than `since` from the main log file
// and any rotated versions (access.log.1, access.log.2.gz, etc.).
func (c *Client) LoadHistoricalLines(logPath string, since int64) ([]string, error) {
// LoadAggregatedHistory aggregates log data on the remote server using awk,
// returning only summary buckets instead of raw lines. This is critical for
// high-traffic servers where raw line transfer would be impractical.
//
// bucketSecs: size of each time bucket in seconds (e.g. 3600 for hourly)
// 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 {
@ -253,22 +258,55 @@ func (c *Client) LoadHistoricalLines(logPath string, since int64) ([]string, err
base = logPath[lastSlash+1:]
}
// Find all rotated log files, cat/zcat them, filter JSON lines by ts field.
// awk: find offset of "ts": then parse the numeric value after it.
// 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=" " }
{
if (index($0,"\"msg\":\"handled request\"") == 0) next
p = index($0,"\"ts\":")
if (p == 0) next
ts = substr($0,p+5,20)+0
if (ts < cutoff) next
b = int(ts/bucket)*bucket
p = index($0,"\"status\":")
st = (p>0) ? substr($0,p+9,5)+0 : 0
p = index($0,"\"size\":")
sz = (p>0) ? substr($0,p+7,15)+0 : 0
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="?"
p = index($0,"\"remote_ip\":\"")
if (p>0) { rest=substr($0,p+13); q=index(rest,"\""); ip=(q>0)?substr(rest,1,q-1):"?" } else ip="?"
bc[b]++; bb[b]+=sz
if (st>=500) bs5[b]++
else if (st>=400) bs4[b]++
else if (st>=300) bs3[b]++
else if (st>=200) bs2[b]++
hc[ho]++; ic[ip]++
tot++; totb+=sz
}
END {
print "BUCKETS"
for (b in bc) print b, bc[b], bb[b], bs2[b]+0, bs3[b]+0, bs4[b]+0, bs5[b]+0
print "HOSTS"
for (h in hc) print hc[h], h
print "IPS"
for (i in ic) print ic[i], i
print "TOTALS"
print tot+0, totb+0
}`
cmd := fmt.Sprintf(
`find %s -name '%s*' -type f 2>/dev/null | sort -r | `+
`find %s -name '%s*' -type f -mtime -%d 2>/dev/null | sort -r | `+
`while read f; do case "$f" in *.gz) zcat "$f" ;; *) cat "$f" ;; esac; done 2>/dev/null | `+
`awk -v c=%d '{p=index($0,"\"ts\":"); if(p>0){v=substr($0,p+5,20)+0; if(v>=c) print}}'`,
dir, base, since,
`awk -v cutoff=%d -v bucket=%d '%s'`,
dir, base, maxAgeDays+1, since, bucketSecs, awkProg,
)
out, err := c.RunCommand(cmd)
if err != nil && strings.TrimSpace(out) == "" {
return nil, fmt.Errorf("failed to read historical logs: %w", err)
return nil, fmt.Errorf("aggregation failed: %w", err)
}
out = strings.TrimSpace(out)
if out == "" {
return nil, nil
}
return strings.Split(out, "\n"), nil
return parser.ParseAggregatedOutput(out)
}

18
internal/tui/app.go
View file

@ -86,7 +86,6 @@ var (
const historyRefreshInterval = 5 * time.Minute
type historyState struct {
entries []*parser.LogEntry
stats *parser.Stats
buckets []parser.Bucket
loaded bool
@ -106,7 +105,7 @@ type sysStatsMsg struct {
}
type historyLoadedMsg struct {
tab viewTab
entries []*parser.LogEntry
agg *parser.AggregatedHistory
err error
}
type errMsg struct{ err error }
@ -200,12 +199,14 @@ func drainLines(ch chan string) tea.Cmd {
func loadHistory(client *sshclient.Client, logPath string, tab viewTab) tea.Cmd {
since := time.Now().Add(-tabWindow(tab)).Unix()
bucketSecs := int64(tabBucketSize(tab).Seconds())
maxAgeDays := int(tabWindow(tab).Hours()/24) + 1
return func() tea.Msg {
lines, err := client.LoadHistoricalLines(logPath, since)
agg, err := client.LoadAggregatedHistory(logPath, since, bucketSecs, maxAgeDays)
if err != nil {
return historyLoadedMsg{tab: tab, err: err}
}
return historyLoadedMsg{tab: tab, entries: parser.ParseLines(lines)}
return historyLoadedMsg{tab: tab, agg: agg}
}
}
@ -265,10 +266,9 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
h.loaded = true
h.loadedAt = time.Now()
h.err = msg.err
if msg.err == nil {
h.entries = msg.entries
h.stats = parser.ComputeStats(msg.entries)
h.buckets = parser.Bucketize(msg.entries, tabBucketSize(msg.tab))
if msg.err == nil && msg.agg != nil {
h.stats = msg.agg.ToStats()
h.buckets = msg.agg.ToBuckets()
}
case errMsg:
@ -422,7 +422,7 @@ func (m model) renderHistory(idx int, label string) string {
return lipgloss.NewStyle().Padding(2, 4).
Render(styleBad.Render(fmt.Sprintf("Error: %v", h.err)))
}
if len(h.entries) == 0 {
if h.stats == nil || h.stats.TotalRequests == 0 {
return lipgloss.NewStyle().Padding(2, 4).
Render(styleDim.Render("No log entries found for this period."))
}