diff --git a/internal/parser/caddy.go b/internal/parser/caddy.go index 219b10e..235547a 100644 --- a/internal/parser/caddy.go +++ b/internal/parser/caddy.go @@ -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 +// <2xx> <3xx> <4xx> <5xx> +// HOSTS +// +// IPS +// +// TOTALS +// +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 +} diff --git a/internal/sshclient/client.go b/internal/sshclient/client.go index 1bab54e..ce32f2b 100644 --- a/internal/sshclient/client.go +++ b/internal/sshclient/client.go @@ -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) } diff --git a/internal/tui/app.go b/internal/tui/app.go index d29deab..413a7d8 100644 --- a/internal/tui/app.go +++ b/internal/tui/app.go @@ -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.")) }