mirror of
https://github.com/wweir/sower.git
synced 2026-10-19 16:14:33 +00:00
feat(admin): cumulative traffic totals for domains and clients
The domain map previously doubled as the live window and dropped entries after staleDomainAge, so from-start totals were lost. It now accumulates (bounded at maxTotalDomains with least-recent eviction) while Snapshot only filters staleness, and a new independent client table (bounded at maxTotalClients) keeps per-client totals exact even when domains are evicted. record() takes explicit counter deltas so the cumulative layer sees the same increments as the window. - GET /api/totals (auth) returns top domains and clients by bytes since start; the SSE stream pushes a totals event initially and every 30s - Traffic page gains a live/totals toggle; totals view keeps byte columns regardless of the source breadcrumb, and clicking a totals client chip jumps to the live view filtered by it - per-domain client breakdown capped at maxClientsPerDomain - tests: totals survive staleness and domain eviction, both caps enforced
This commit is contained in:
@@ -107,6 +107,7 @@ func NewServer(opts Options) *Server {
|
||||
mux.HandleFunc("POST /api/rules", s.mutateGuard(s.auth(s.handleRulesAdd)))
|
||||
mux.HandleFunc("DELETE /api/rules", s.mutateGuard(s.auth(s.handleRulesRemove)))
|
||||
mux.HandleFunc("GET /api/traffic", s.mutateGuard(s.auth(s.handleTraffic)))
|
||||
mux.HandleFunc("GET /api/totals", s.mutateGuard(s.auth(s.handleTotals)))
|
||||
mux.HandleFunc("GET /api/history", s.mutateGuard(s.auth(s.handleHistory)))
|
||||
mux.HandleFunc("GET /api/stream", s.mutateGuard(s.auth(s.handleStream)))
|
||||
if s.opts.Stats != nil {
|
||||
@@ -447,6 +448,16 @@ func (s *Server) handleTraffic(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, s.opts.Stats.Snapshot(sort, source, client))
|
||||
}
|
||||
|
||||
// handleTotals serves the from-start cumulative traffic view. It ignores
|
||||
// staleness and query filters: totals are the whole picture, not a window.
|
||||
func (s *Server) handleTotals(w http.ResponseWriter, r *http.Request) {
|
||||
if s.opts.Stats == nil {
|
||||
writeError(w, http.StatusInternalServerError, "stats unavailable")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, s.opts.Stats.Totals())
|
||||
}
|
||||
|
||||
// parseTrafficQuery extracts the domain sort, source, and client filters
|
||||
// shared by /api/traffic and the SSE stream.
|
||||
func parseTrafficQuery(r *http.Request) (sort DomainSort, source Source, client string) {
|
||||
@@ -508,11 +519,14 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
|
||||
send("status", s.statusPayload())
|
||||
send("traffic", s.opts.Stats.Snapshot(sort, source, client))
|
||||
send("history", s.opts.Stats.History())
|
||||
send("totals", s.opts.Stats.Totals())
|
||||
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
historyTicker := time.NewTicker(10 * time.Second)
|
||||
defer historyTicker.Stop()
|
||||
totalsTicker := time.NewTicker(30 * time.Second)
|
||||
defer totalsTicker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
@@ -527,6 +541,8 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
|
||||
send("traffic", s.opts.Stats.Snapshot(sort, source, client))
|
||||
case <-historyTicker.C:
|
||||
send("history", s.opts.Stats.History())
|
||||
case <-totalsTicker.C:
|
||||
send("totals", s.opts.Stats.Totals())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -403,6 +403,28 @@ func TestTrafficEndpoint(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestTotalsEndpoint(t *testing.T) {
|
||||
stats := newTestStats(t)
|
||||
s := NewServer(Options{Password: "secret", Version: "v1.2.3", Date: "2026-01-01", Rules: newFakeRules(), Stats: stats})
|
||||
ts := httptest.NewServer(s.http.Handler)
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
cookie := login(t, ts, "secret")
|
||||
stats.RecordDNS("example.com.", "192.168.1.10")
|
||||
|
||||
resp := authedRequest(t, ts, http.MethodGet, "/api/totals", cookie, "")
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d", resp.StatusCode)
|
||||
}
|
||||
body := decodeBody(t, resp)
|
||||
if domains, ok := body["domains"].([]any); !ok || len(domains) != 1 {
|
||||
t.Fatalf("expected 1 domain in totals, got %v", body["domains"])
|
||||
}
|
||||
if clients, ok := body["clients"].([]any); !ok || len(clients) != 1 {
|
||||
t.Fatalf("expected 1 client in totals, got %v", body["clients"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrafficEndpointRejectsMissingStats(t *testing.T) {
|
||||
s := NewServer(Options{Password: "secret", Rules: newFakeRules()})
|
||||
ts := httptest.NewServer(s.http.Handler)
|
||||
|
||||
+137
-21
@@ -14,10 +14,19 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
maxDomainEntries = 10000
|
||||
staleDomainAge = time.Hour
|
||||
snapshotTopN = 100
|
||||
clientsTopN = 50
|
||||
// maxTotalDomains bounds the cumulative domain map. Domains are never
|
||||
// dropped for staleness (unlike the old window-only table); the map only
|
||||
// evicts its least-recently-seen entry once full, so from-start totals
|
||||
// survive while memory stays bounded.
|
||||
maxTotalDomains = 100000
|
||||
// maxTotalClients bounds the cumulative per-client table.
|
||||
maxTotalClients = 4096
|
||||
// maxClientsPerDomain guards one domain's per-client breakdown against a
|
||||
// pathological number of source IPs; extra clients are not attributed.
|
||||
maxClientsPerDomain = 64
|
||||
staleDomainAge = time.Hour
|
||||
snapshotTopN = 100
|
||||
clientsTopN = 50
|
||||
|
||||
// historyCapacity keeps one hour of samples at the 5s poll cadence.
|
||||
historyCapacity = 720
|
||||
@@ -181,6 +190,9 @@ type Stats struct {
|
||||
|
||||
mu sync.Mutex
|
||||
domains map[string]*domainStat
|
||||
// clients holds the from-start cumulative per-client totals. It is
|
||||
// independent of the domain map so client totals survive domain eviction.
|
||||
clients map[string]*clientStat
|
||||
|
||||
histMu sync.Mutex
|
||||
history []HistorySample
|
||||
@@ -236,6 +248,7 @@ func NewStats() (*Stats, error) {
|
||||
start: time.Now(),
|
||||
metrics: m,
|
||||
domains: make(map[string]*domainStat),
|
||||
clients: make(map[string]*clientStat),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -250,7 +263,7 @@ func (s *Stats) RecordDNS(qname string, clientIP string) {
|
||||
if domain == "" {
|
||||
return
|
||||
}
|
||||
s.record(domain, SourceDNS, clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
s.record(domain, SourceDNS, clientIP, 1, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.conns++
|
||||
src.conns++
|
||||
if dc != nil {
|
||||
@@ -312,7 +325,7 @@ func (s *Stats) BindConn(conn net.Conn, domain string) {
|
||||
}
|
||||
cc.bind(domain)
|
||||
source := Source(cc.kind)
|
||||
s.record(domain, source, cc.clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
s.record(domain, source, cc.clientIP, 1, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.conns++
|
||||
src.conns++
|
||||
if dc != nil {
|
||||
@@ -321,11 +334,12 @@ func (s *Stats) BindConn(conn net.Conn, domain string) {
|
||||
})
|
||||
}
|
||||
|
||||
// Snapshot returns an immutable view of the current stats, advancing the
|
||||
// history ring and evicting stale domains along the way. sort selects the
|
||||
// domain ordering, source restricts domains to one entry point, and client
|
||||
// restricts them to one client IP; invalid values fall back to
|
||||
// DomainSortBytes and SourceAll, and an empty client disables the filter.
|
||||
// Snapshot returns the window view of the current stats, advancing the
|
||||
// history ring along the way. Domains stale for over staleDomainAge are
|
||||
// hidden from the window but kept in the map for the cumulative Totals view.
|
||||
// sort selects the domain ordering, source restricts domains to one entry
|
||||
// point, and client restricts them to one client IP; invalid values fall back
|
||||
// to DomainSortBytes and SourceAll, and an empty client disables the filter.
|
||||
// Clients aggregates per-client counters over the source-filtered view and
|
||||
// is intentionally unaffected by the client filter, so the console's client
|
||||
// chips stay stable while one is selected.
|
||||
@@ -357,7 +371,8 @@ func (s *Stats) Snapshot(sort DomainSort, source Source, client string) TrafficS
|
||||
clientAgg := make(map[string]*ClientStat)
|
||||
for domain, d := range s.domains {
|
||||
if time.Since(d.lastSeen) > staleDomainAge {
|
||||
delete(s.domains, domain)
|
||||
// Window view only: the entry stays in the map so the cumulative
|
||||
// Totals view keeps its from-start counters.
|
||||
continue
|
||||
}
|
||||
ds := DomainStat{
|
||||
@@ -445,6 +460,67 @@ func topClients(agg map[string]*ClientStat) []ClientStat {
|
||||
return out
|
||||
}
|
||||
|
||||
// Totals is the from-start cumulative traffic view, independent of the
|
||||
// window's staleness filter.
|
||||
type Totals struct {
|
||||
Domains []DomainStat `json:"domains"`
|
||||
Clients []ClientStat `json:"clients"`
|
||||
}
|
||||
|
||||
// Totals returns the cumulative top domains and clients since start, each
|
||||
// ordered by total bytes. Unlike Snapshot it ignores staleness: counters
|
||||
// accumulate until the bounded maps evict their least-recently-seen entries.
|
||||
func (s *Stats) Totals() Totals {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
domains := make([]DomainStat, 0, len(s.domains))
|
||||
for domain, d := range s.domains {
|
||||
ds := DomainStat{
|
||||
Domain: domain,
|
||||
Conns: d.conns,
|
||||
BytesUp: d.bytesUp,
|
||||
BytesDown: d.bytesDown,
|
||||
LastSeen: d.lastSeen,
|
||||
}
|
||||
var latest time.Time
|
||||
for ip, c := range d.byClient {
|
||||
if c.lastSeen.After(latest) {
|
||||
latest, ds.LastClientIP = c.lastSeen, ip
|
||||
}
|
||||
}
|
||||
domains = append(domains, ds)
|
||||
}
|
||||
sortDomains(domains, DomainSortBytes)
|
||||
if len(domains) > snapshotTopN {
|
||||
domains = domains[:snapshotTopN]
|
||||
}
|
||||
|
||||
clients := make([]ClientStat, 0, len(s.clients))
|
||||
for ip, c := range s.clients {
|
||||
clients = append(clients, ClientStat{
|
||||
IP: ip,
|
||||
Conns: c.conns,
|
||||
BytesUp: c.bytesUp,
|
||||
BytesDown: c.bytesDown,
|
||||
LastSeen: c.lastSeen,
|
||||
})
|
||||
}
|
||||
sort.Slice(clients, func(i, j int) bool {
|
||||
ti := clients[i].BytesUp + clients[i].BytesDown
|
||||
tj := clients[j].BytesUp + clients[j].BytesDown
|
||||
if ti == tj {
|
||||
return clients[i].IP < clients[j].IP
|
||||
}
|
||||
return ti > tj
|
||||
})
|
||||
if len(clients) > clientsTopN {
|
||||
clients = clients[:clientsTopN]
|
||||
}
|
||||
|
||||
return Totals{Domains: domains, Clients: clients}
|
||||
}
|
||||
|
||||
// sortDomains orders domains by the selected mode: total bytes (default),
|
||||
// most recently seen, or connection count. Ties break alphabetically.
|
||||
func sortDomains(ds []DomainStat, mode DomainSort) {
|
||||
@@ -562,14 +638,16 @@ func (s *Stats) rates() (up, down, dns, conns float64) {
|
||||
}
|
||||
|
||||
// record attributes one event to a domain, its source, and optionally its
|
||||
// client IP. The client IP may be empty when it cannot be determined; the
|
||||
// per-client stat is then nil and callers must guard.
|
||||
func (s *Stats) record(domain string, source Source, clientIP string, update func(*domainStat, *sourceStat, *clientStat)) {
|
||||
// client IP, with the given counter deltas. The client IP may be empty when it
|
||||
// cannot be determined; the per-client stat is then nil and callers must
|
||||
// guard. The deltas also feed the independent cumulative client table, so
|
||||
// client totals are exact regardless of domain eviction.
|
||||
func (s *Stats) record(domain string, source Source, clientIP string, conns, up, down uint64, update func(*domainStat, *sourceStat, *clientStat)) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
d := s.domains[domain]
|
||||
if d == nil {
|
||||
if len(s.domains) >= maxDomainEntries {
|
||||
if len(s.domains) >= maxTotalDomains {
|
||||
s.evictOldestLocked()
|
||||
}
|
||||
d = &domainStat{bySource: make(map[Source]*sourceStat)}
|
||||
@@ -586,16 +664,32 @@ func (s *Stats) record(domain string, source Source, clientIP string, update fun
|
||||
d.byClient = make(map[string]*clientStat)
|
||||
}
|
||||
dc = d.byClient[clientIP]
|
||||
if dc == nil {
|
||||
if dc == nil && len(d.byClient) < maxClientsPerDomain {
|
||||
dc = &clientStat{}
|
||||
d.byClient[clientIP] = dc
|
||||
}
|
||||
}
|
||||
update(d, src, dc)
|
||||
d.lastSeen = time.Now()
|
||||
src.lastSeen = time.Now()
|
||||
now := time.Now()
|
||||
d.lastSeen = now
|
||||
src.lastSeen = now
|
||||
if dc != nil {
|
||||
dc.lastSeen = time.Now()
|
||||
dc.lastSeen = now
|
||||
}
|
||||
|
||||
if clientIP != "" {
|
||||
cc := s.clients[clientIP]
|
||||
if cc == nil {
|
||||
if len(s.clients) >= maxTotalClients {
|
||||
s.evictOldestClientLocked()
|
||||
}
|
||||
cc = &clientStat{}
|
||||
s.clients[clientIP] = cc
|
||||
}
|
||||
cc.conns += conns
|
||||
cc.bytesUp += up
|
||||
cc.bytesDown += down
|
||||
cc.lastSeen = now
|
||||
}
|
||||
}
|
||||
|
||||
@@ -615,6 +709,22 @@ func (s *Stats) evictOldestLocked() {
|
||||
}
|
||||
}
|
||||
|
||||
// evictOldestClientLocked drops the single least-recently-seen client to keep
|
||||
// the cumulative client table bounded.
|
||||
func (s *Stats) evictOldestClientLocked() {
|
||||
var oldestKey string
|
||||
var oldest time.Time
|
||||
for ip, c := range s.clients {
|
||||
if oldestKey == "" || c.lastSeen.Before(oldest) {
|
||||
oldestKey = ip
|
||||
oldest = c.lastSeen
|
||||
}
|
||||
}
|
||||
if oldestKey != "" {
|
||||
delete(s.clients, oldestKey)
|
||||
}
|
||||
}
|
||||
|
||||
// countingConn wraps a client connection and feeds byte counts into Stats.
|
||||
type countingConn struct {
|
||||
net.Conn
|
||||
@@ -675,7 +785,13 @@ func (c *countingConn) recordBytes(n int, up bool) {
|
||||
return
|
||||
}
|
||||
source := Source(c.kind)
|
||||
c.stats.record(domain, source, c.clientIP, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
var upBytes, downBytes uint64
|
||||
if up {
|
||||
upBytes = uint64(n)
|
||||
} else {
|
||||
downBytes = uint64(n)
|
||||
}
|
||||
c.stats.record(domain, source, c.clientIP, 0, upBytes, downBytes, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
if up {
|
||||
d.bytesUp += uint64(n)
|
||||
src.bytesUp += uint64(n)
|
||||
|
||||
@@ -139,13 +139,13 @@ func TestStatsSnapshotEvictsStaleDomains(t *testing.T) {
|
||||
|
||||
func TestStatsDomainCapIsEnforced(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
for i := 0; i < maxDomainEntries+50; i++ {
|
||||
s.record(fmt.Sprintf("host%d.example", i), SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {})
|
||||
for i := 0; i < maxTotalDomains+50; i++ {
|
||||
s.record(fmt.Sprintf("host%d.example", i), SourceHTTP, "", 1, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {})
|
||||
}
|
||||
s.mu.Lock()
|
||||
n := len(s.domains)
|
||||
s.mu.Unlock()
|
||||
if n > maxDomainEntries {
|
||||
if n > maxTotalDomains {
|
||||
t.Fatalf("expected domain cap enforced, got %d entries", n)
|
||||
}
|
||||
if n == 0 {
|
||||
@@ -169,15 +169,15 @@ func TestEvictOldestLocked(t *testing.T) {
|
||||
|
||||
func TestStatsSnapshotSortsByTotalBytes(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
s.record("small.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
s.record("small.example", SourceHTTP, "", 0, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.bytesUp, d.bytesDown = 1, 1
|
||||
src.bytesUp, src.bytesDown = 1, 1
|
||||
})
|
||||
s.record("big.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
s.record("big.example", SourceHTTP, "", 0, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.bytesUp, d.bytesDown = 100, 100
|
||||
src.bytesUp, src.bytesDown = 100, 100
|
||||
})
|
||||
s.record("mid.example", SourceHTTP, "", func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
s.record("mid.example", SourceHTTP, "", 0, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.bytesUp, d.bytesDown = 10, 10
|
||||
src.bytesUp, src.bytesDown = 10, 10
|
||||
})
|
||||
@@ -531,3 +531,75 @@ func TestStatsSnapshotClientsAggregate(t *testing.T) {
|
||||
t.Fatalf("unexpected second client: %+v", snap.Clients[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatsTotalsSurviveStaleness(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
old := time.Now().Add(-2 * staleDomainAge)
|
||||
s.mu.Lock()
|
||||
s.domains["old.example"] = &domainStat{
|
||||
conns: 5, bytesUp: 500, bytesDown: 500, lastSeen: old,
|
||||
bySource: map[Source]*sourceStat{
|
||||
SourceHTTP: {conns: 5, bytesUp: 500, bytesDown: 500, lastSeen: old},
|
||||
},
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
// the window view hides the stale domain, the totals keep it
|
||||
snap := s.Snapshot(DomainSortBytes, SourceAll, "")
|
||||
if len(snap.Domains) != 0 {
|
||||
t.Fatalf("expected stale domain hidden from window, got %+v", snap.Domains)
|
||||
}
|
||||
tot := s.Totals()
|
||||
if len(tot.Domains) != 1 || tot.Domains[0].Domain != "old.example" || tot.Domains[0].Conns != 5 {
|
||||
t.Fatalf("expected stale domain kept in totals, got %+v", tot.Domains)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatsTotalsClientsSurviveDomainEviction(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
s.record("evicted.example", SourceHTTP, "192.168.1.10", 2, 100, 200, func(d *domainStat, src *sourceStat, dc *clientStat) {
|
||||
d.conns += 2
|
||||
src.conns += 2
|
||||
})
|
||||
s.mu.Lock()
|
||||
delete(s.domains, "evicted.example") // simulate cap eviction
|
||||
s.mu.Unlock()
|
||||
|
||||
tot := s.Totals()
|
||||
if len(tot.Domains) != 0 {
|
||||
t.Fatalf("expected no domains after eviction, got %+v", tot.Domains)
|
||||
}
|
||||
if len(tot.Clients) != 1 || tot.Clients[0].IP != "192.168.1.10" ||
|
||||
tot.Clients[0].Conns != 2 || tot.Clients[0].BytesUp != 100 || tot.Clients[0].BytesDown != 200 {
|
||||
t.Fatalf("expected client totals to survive domain eviction, got %+v", tot.Clients)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatsClientCapIsEnforced(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
for i := 0; i < maxTotalClients+10; i++ {
|
||||
s.record(fmt.Sprintf("host%d.example", i), SourceHTTP, fmt.Sprintf("192.168.%d.%d", i/256, i%256), 1, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {})
|
||||
}
|
||||
s.mu.Lock()
|
||||
n := len(s.clients)
|
||||
s.mu.Unlock()
|
||||
if n > maxTotalClients {
|
||||
t.Fatalf("expected client cap enforced, got %d entries", n)
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatal("expected some clients to survive")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatsClientsPerDomainCap(t *testing.T) {
|
||||
s := newTestStats(t)
|
||||
for i := 0; i < maxClientsPerDomain+10; i++ {
|
||||
s.record("shared.example", SourceHTTP, fmt.Sprintf("192.0.2.%d", i), 1, 0, 0, func(d *domainStat, src *sourceStat, dc *clientStat) {})
|
||||
}
|
||||
s.mu.Lock()
|
||||
n := len(s.domains["shared.example"].byClient)
|
||||
s.mu.Unlock()
|
||||
if n > maxClientsPerDomain {
|
||||
t.Fatalf("expected per-domain client cap, got %d entries", n)
|
||||
}
|
||||
}
|
||||
|
||||
Vendored
-1233
File diff suppressed because one or more lines are too long
Vendored
+1233
File diff suppressed because one or more lines are too long
Vendored
+1
-1
@@ -5,7 +5,7 @@
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<meta name="color-scheme" content="light" />
|
||||
<title>Sower Admin</title>
|
||||
<script type="module" crossorigin src="/assets/index-5cU5DjAW.js"></script>
|
||||
<script type="module" crossorigin src="/assets/index-G9FsEZaz.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-C8oAZ1sn.css">
|
||||
</head>
|
||||
<body>
|
||||
|
||||
+8
-1
@@ -47,6 +47,11 @@ export interface TrafficSnapshot {
|
||||
clients: ClientStat[];
|
||||
}
|
||||
|
||||
export interface Totals {
|
||||
domains: DomainStat[];
|
||||
clients: ClientStat[];
|
||||
}
|
||||
|
||||
export interface HistorySample {
|
||||
at: string;
|
||||
bytesUp: number;
|
||||
@@ -138,7 +143,9 @@ export const api = {
|
||||
if (sort) params.push(`sort=${sort}`);
|
||||
if (source && source !== "all") params.push(`source=${source}`);
|
||||
if (client) params.push(`client=${encodeURIComponent(client)}`);
|
||||
return request<TrafficSnapshot>(params.length ? `/api/traffic?${params.join("&")}` : "/api/traffic");
|
||||
return request<TrafficSnapshot>(
|
||||
params.length ? `/api/traffic?${params.join("&")}` : "/api/traffic",
|
||||
);
|
||||
},
|
||||
history: () => request<History>("/api/history"),
|
||||
};
|
||||
|
||||
+43
-37
@@ -4,65 +4,71 @@
|
||||
// /api/stream; the browser auto-reconnects on drops. The connection carries
|
||||
// the active page's sort/source/client filters, so switching a filter
|
||||
// re-points the stream (connectLive reconnects only when params change).
|
||||
import type { HistorySample, Status, TrafficSnapshot } from '$lib/api'
|
||||
import type { HistorySample, Status, Totals, TrafficSnapshot } from "$lib/api";
|
||||
|
||||
export const live = $state({
|
||||
status: null as Status | null,
|
||||
traffic: null as TrafficSnapshot | null,
|
||||
history: [] as HistorySample[],
|
||||
totals: null as Totals | null,
|
||||
connected: false,
|
||||
})
|
||||
});
|
||||
|
||||
let es: EventSource | null = null
|
||||
let activeParams = ''
|
||||
let unauthorizedHandler: (() => void) | null = null
|
||||
let es: EventSource | null = null;
|
||||
let activeParams = "";
|
||||
let unauthorizedHandler: (() => void) | null = null;
|
||||
|
||||
// safeParse tolerates a malformed SSE payload without breaking the stream
|
||||
// handler; the previous value stays in place until a valid event arrives.
|
||||
function safeParse<T>(data: string): T | null {
|
||||
try {
|
||||
return JSON.parse(data) as T
|
||||
return JSON.parse(data) as T;
|
||||
} catch {
|
||||
return null
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function setUnauthorizedHandler(fn: (() => void) | null) {
|
||||
unauthorizedHandler = fn
|
||||
unauthorizedHandler = fn;
|
||||
}
|
||||
|
||||
// connectLive opens the SSE stream, or re-points it when the filter params
|
||||
// change. Calling it with the current params is a no-op.
|
||||
export function connectLive(params = '') {
|
||||
if (es && params === activeParams) return
|
||||
activeParams = params
|
||||
es?.close()
|
||||
es = new EventSource(`/api/stream${params ? `?${params}` : ''}`)
|
||||
es.onopen = () => (live.connected = true)
|
||||
es.onerror = () => (live.connected = false)
|
||||
es.addEventListener('status', (e) => {
|
||||
const parsed = safeParse<Status>(e.data)
|
||||
if (parsed) live.status = parsed
|
||||
})
|
||||
es.addEventListener('traffic', (e) => {
|
||||
const parsed = safeParse<TrafficSnapshot>(e.data)
|
||||
if (parsed) live.traffic = parsed
|
||||
})
|
||||
es.addEventListener('history', (e) => {
|
||||
const parsed = safeParse<{ samples: HistorySample[] }>(e.data)
|
||||
if (parsed) live.history = parsed.samples
|
||||
})
|
||||
es.addEventListener('auth', () => {
|
||||
unauthorizedHandler?.()
|
||||
})
|
||||
export function connectLive(params = "") {
|
||||
if (es && params === activeParams) return;
|
||||
activeParams = params;
|
||||
es?.close();
|
||||
es = new EventSource(`/api/stream${params ? `?${params}` : ""}`);
|
||||
es.onopen = () => (live.connected = true);
|
||||
es.onerror = () => (live.connected = false);
|
||||
es.addEventListener("status", (e) => {
|
||||
const parsed = safeParse<Status>(e.data);
|
||||
if (parsed) live.status = parsed;
|
||||
});
|
||||
es.addEventListener("traffic", (e) => {
|
||||
const parsed = safeParse<TrafficSnapshot>(e.data);
|
||||
if (parsed) live.traffic = parsed;
|
||||
});
|
||||
es.addEventListener("history", (e) => {
|
||||
const parsed = safeParse<{ samples: HistorySample[] }>(e.data);
|
||||
if (parsed) live.history = parsed.samples;
|
||||
});
|
||||
es.addEventListener("totals", (e) => {
|
||||
const parsed = safeParse<Totals>(e.data);
|
||||
if (parsed) live.totals = parsed;
|
||||
});
|
||||
es.addEventListener("auth", () => {
|
||||
unauthorizedHandler?.();
|
||||
});
|
||||
}
|
||||
|
||||
export function closeLive() {
|
||||
es?.close()
|
||||
es = null
|
||||
activeParams = ''
|
||||
live.status = null
|
||||
live.traffic = null
|
||||
live.history = []
|
||||
live.connected = false
|
||||
es?.close();
|
||||
es = null;
|
||||
activeParams = "";
|
||||
live.status = null;
|
||||
live.traffic = null;
|
||||
live.history = [];
|
||||
live.totals = null;
|
||||
live.connected = false;
|
||||
}
|
||||
|
||||
@@ -18,9 +18,17 @@
|
||||
{ value: 'conns', label: '连接数' },
|
||||
]
|
||||
|
||||
// The domain table is fed by the shared SSE stream, filtered by the
|
||||
// connection params below.
|
||||
const traffic = $derived(live.traffic)
|
||||
type View = 'live' | 'totals'
|
||||
let view = $state<View>('live')
|
||||
|
||||
// Live view is fed by the SSE traffic events; totals are the cumulative
|
||||
// since-start view pushed every 30s.
|
||||
const domains = $derived(
|
||||
view === 'totals' ? (live.totals?.domains ?? []) : (live.traffic?.domains ?? []),
|
||||
)
|
||||
const clients = $derived(
|
||||
view === 'totals' ? (live.totals?.clients ?? []) : (live.traffic?.clients ?? []),
|
||||
)
|
||||
let filter = $state('')
|
||||
let sort: SortMode = $state('bytes')
|
||||
let client = $state('')
|
||||
@@ -37,12 +45,15 @@
|
||||
const isDNSView = $derived(source === 'dns')
|
||||
|
||||
const visibleDomains = $derived.by(() => {
|
||||
const domains = traffic?.domains ?? []
|
||||
const q = filter.trim().toLowerCase()
|
||||
if (!q) return domains
|
||||
return domains.filter((d) => d.domain.toLowerCase().includes(q))
|
||||
})
|
||||
|
||||
// The totals view spans every source, so byte columns stay visible even
|
||||
// when the breadcrumb has narrowed the live view to DNS.
|
||||
const showBytes = $derived(view === 'totals' || !isDNSView)
|
||||
|
||||
// Re-point the SSE stream whenever a page filter changes.
|
||||
$effect(() => {
|
||||
const params = new URLSearchParams()
|
||||
@@ -55,7 +66,7 @@
|
||||
const headCell = 'sticky top-0 z-10 bg-card'
|
||||
</script>
|
||||
|
||||
{#if !traffic}
|
||||
{#if !live.traffic && !live.totals}
|
||||
<Loading />
|
||||
{:else}
|
||||
{#if !live.connected}
|
||||
@@ -76,50 +87,81 @@
|
||||
<div
|
||||
class="flex shrink-0 items-center gap-0.5 rounded-md border bg-muted/50 p-0.5"
|
||||
role="group"
|
||||
aria-label="排序方式"
|
||||
aria-label="数据范围"
|
||||
>
|
||||
{#each sortModes as m}
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-sm px-2.5 py-1 text-xs font-medium transition-colors {sort === m.value
|
||||
? 'bg-card text-foreground shadow-sm'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={sort === m.value}
|
||||
onclick={() => (sort = m.value)}
|
||||
>
|
||||
{m.label}
|
||||
</button>
|
||||
{/each}
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-sm px-2.5 py-1 text-xs font-medium transition-colors {view === 'live'
|
||||
? 'bg-card text-foreground shadow-sm'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={view === 'live'}
|
||||
onclick={() => (view = 'live')}
|
||||
>
|
||||
实时
|
||||
</button>
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-sm px-2.5 py-1 text-xs font-medium transition-colors {view === 'totals'
|
||||
? 'bg-card text-foreground shadow-sm'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={view === 'totals'}
|
||||
onclick={() => (view = 'totals')}
|
||||
>
|
||||
总计
|
||||
</button>
|
||||
</div>
|
||||
{#if view === 'live'}
|
||||
<div
|
||||
class="flex shrink-0 items-center gap-0.5 rounded-md border bg-muted/50 p-0.5"
|
||||
role="group"
|
||||
aria-label="排序方式"
|
||||
>
|
||||
{#each sortModes as m}
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-sm px-2.5 py-1 text-xs font-medium transition-colors {sort === m.value
|
||||
? 'bg-card text-foreground shadow-sm'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={sort === m.value}
|
||||
onclick={() => (sort = m.value)}
|
||||
>
|
||||
{m.label}
|
||||
</button>
|
||||
{/each}
|
||||
</div>
|
||||
{/if}
|
||||
<Badge variant="secondary" class="ml-auto shrink-0">
|
||||
{#if visibleDomains.length === (traffic.domains ?? []).length}
|
||||
{#if visibleDomains.length === domains.length}
|
||||
{formatCount(visibleDomains.length)} domains
|
||||
{:else}
|
||||
{formatCount(visibleDomains.length)} / {formatCount((traffic.domains ?? []).length)} domains
|
||||
{formatCount(visibleDomains.length)} / {formatCount(domains.length)} domains
|
||||
{/if}
|
||||
</Badge>
|
||||
</div>
|
||||
{#if (traffic.clients ?? []).length > 0}
|
||||
{#if clients.length > 0}
|
||||
<div class="mb-3 flex flex-wrap items-center gap-1.5" role="group" aria-label="客户端过滤">
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-full border px-3 py-1 text-xs font-medium transition-colors {!client
|
||||
class="rounded-full border px-3 py-1 text-xs font-medium transition-colors {view === 'live' && !client
|
||||
? 'border-primary/40 bg-primary/10 text-foreground'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={!client}
|
||||
aria-pressed={view === 'live' && !client}
|
||||
onclick={() => (client = '')}
|
||||
>
|
||||
全部客户端
|
||||
</button>
|
||||
{#each traffic.clients as c}
|
||||
{#each clients as c}
|
||||
<button
|
||||
type="button"
|
||||
class="rounded-full border px-3 py-1 font-mono text-xs transition-colors {client === c.ip
|
||||
class="rounded-full border px-3 py-1 font-mono text-xs transition-colors {view === 'live' && client === c.ip
|
||||
? 'border-primary/40 bg-primary/10 text-foreground'
|
||||
: 'text-muted-foreground hover:text-foreground'}"
|
||||
aria-pressed={client === c.ip}
|
||||
aria-pressed={view === 'live' && client === c.ip}
|
||||
title={`${formatCount(c.conns)} 次 · ↑${formatBytes(c.bytesUp)} ↓${formatBytes(c.bytesDown)}`}
|
||||
onclick={() => (client = c.ip)}
|
||||
onclick={() => {
|
||||
if (view === 'totals') view = 'live'
|
||||
client = c.ip
|
||||
}}
|
||||
>
|
||||
{c.ip}
|
||||
<span class="ms-1 text-muted-foreground">{formatCount(c.conns)}</span>
|
||||
@@ -127,10 +169,10 @@
|
||||
{/each}
|
||||
</div>
|
||||
{/if}
|
||||
{#if (traffic.domains ?? []).length === 0}
|
||||
{#if domains.length === 0}
|
||||
<div class="flex flex-col items-center gap-2 rounded-lg border border-dashed py-10 text-sm text-muted-foreground">
|
||||
<Inbox class="size-5" aria-hidden="true" />
|
||||
<p>暂无{sourceLabel}记录。</p>
|
||||
<p>{view === 'totals' ? '暂无总计记录。' : `暂无${sourceLabel}记录。`}</p>
|
||||
</div>
|
||||
{:else}
|
||||
<Card.Card class="max-h-[70vh] overflow-auto">
|
||||
@@ -139,7 +181,7 @@
|
||||
<Table.TableRow>
|
||||
<Table.TableHead class={headCell}>Domain</Table.TableHead>
|
||||
<Table.TableHead class={headCell + ' text-right'}>Requests</Table.TableHead>
|
||||
{#if !isDNSView}
|
||||
{#if showBytes}
|
||||
<Table.TableHead class={headCell + ' text-right'}>Up</Table.TableHead>
|
||||
<Table.TableHead class={headCell + ' text-right'}>Down</Table.TableHead>
|
||||
<Table.TableHead class={headCell + ' text-right'}>Total</Table.TableHead>
|
||||
@@ -151,7 +193,7 @@
|
||||
<Table.TableBody>
|
||||
{#if visibleDomains.length === 0}
|
||||
<Table.TableRow>
|
||||
<Table.TableCell colspan={isDNSView ? 4 : 7} class="py-8 text-center text-sm text-muted-foreground">
|
||||
<Table.TableCell colspan={showBytes ? 7 : 4} class="py-8 text-center text-sm text-muted-foreground">
|
||||
没有匹配“{filter}”的域名。
|
||||
</Table.TableCell>
|
||||
</Table.TableRow>
|
||||
@@ -160,7 +202,7 @@
|
||||
<Table.TableRow>
|
||||
<Table.TableCell class="max-w-[20rem] truncate font-mono text-xs">{d.domain}</Table.TableCell>
|
||||
<Table.TableCell class="text-right tabular-nums">{formatCount(d.conns)}</Table.TableCell>
|
||||
{#if !isDNSView}
|
||||
{#if showBytes}
|
||||
<Table.TableCell class="text-right tabular-nums">{formatBytes(d.bytesUp)}</Table.TableCell>
|
||||
<Table.TableCell class="text-right tabular-nums">{formatBytes(d.bytesDown)}</Table.TableCell>
|
||||
<Table.TableCell class="text-right font-medium tabular-nums">
|
||||
@@ -179,7 +221,13 @@
|
||||
</Table.Table>
|
||||
</Card.Card>
|
||||
{/if}
|
||||
<p class="mt-2 text-xs text-muted-foreground">
|
||||
按{activeSortLabel}排序的前 100 个{isDNSView ? 'DNS 查询域名' : sourceLabel + '域名'},实时推送。
|
||||
</p>
|
||||
{#if view === 'totals'}
|
||||
<p class="mt-2 text-xs text-muted-foreground">
|
||||
自启动以来按流量排序的前 {formatCount(domains.length)} 个域名与客户端,每 30 秒更新。
|
||||
</p>
|
||||
{:else}
|
||||
<p class="mt-2 text-xs text-muted-foreground">
|
||||
按{activeSortLabel}排序的前 100 个{isDNSView ? 'DNS 查询域名' : sourceLabel + '域名'},实时推送。
|
||||
</p>
|
||||
{/if}
|
||||
{/if}
|
||||
|
||||
Reference in New Issue
Block a user