mirror of
https://github.com/Dvorinka/Dash.git
synced 2026-10-28 07:41:27 +00:00
feat: system monitoring + push agent (merge plan M3)
systems + system_stats tables; agents push JSON every N seconds to POST /api/systems/ingest with a per-system bearer token. The server stores each sample, keeps a `latest` snapshot for cheap list/widget reads, prunes stats after 7d, and marks systems down when silent > max(3x interval, 30s) — firing system.up/down webhook events on transitions. cmd/dash-agent is a standalone Linux collector: /proc for cpu/mem/net/load/uptime, /sys hwmon temps, statfs disk, and container liveness via /var/run/docker.sock when present. No imports from the server module — the sample shape is duplicated deliberately so the agent stays a single static binary. Frontend: /systems card grid with usage bars, /systems/:id with recharts cpu/mem/net/load, temps + containers tables, and the agent command; `system` board widget; nav + palette + WidgetDialog picker wired. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed
1 parent
2f3e8a1879
commit
74c987c18a
16 files changed
+1541
-10
No files matched your search
@@ -0,0 +1,359 @@
|
||||
// Command dash-agent collects host metrics and pushes them to a Dash server.
|
||||
// Linux-first: /proc for cpu/mem/net/load/uptime, /sys hwmon for temps,
|
||||
// statfs for disk, and the Docker socket for container liveness when present.
|
||||
//
|
||||
// env DASH_URL=http://dash:8080 DASH_TOKEN=agent_xxx dash-agent
|
||||
//
|
||||
// Flags override env: -url, -token, -interval (seconds), -once.
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
func main() {
|
||||
url := flag.String("url", os.Getenv("DASH_URL"), "dash server base URL")
|
||||
token := flag.String("token", os.Getenv("DASH_TOKEN"), "system token")
|
||||
interval := flag.Int("interval", envInt("DASH_INTERVAL", 10), "push interval seconds")
|
||||
once := flag.Bool("once", false, "collect and print one sample, no push")
|
||||
flag.Parse()
|
||||
|
||||
c := &collector{}
|
||||
c.host, _ = os.Hostname()
|
||||
c.os = runtime.GOOS
|
||||
c.arch = runtime.GOARCH
|
||||
c.cpuModel, c.cores = cpuInfo()
|
||||
c.prevCPU = readCPUTicks()
|
||||
c.prevNet = readNetBytes()
|
||||
c.prevAt = time.Now()
|
||||
|
||||
if *once {
|
||||
// Rates need a delta window — idle briefly so cpu/net are real.
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
s := c.sample(*interval)
|
||||
out, _ := json.MarshalIndent(s, "", " ")
|
||||
fmt.Println(string(out))
|
||||
return
|
||||
}
|
||||
if *url == "" || *token == "" {
|
||||
log.Fatal("dash-agent: -url and -token (or DASH_URL/DASH_TOKEN) required")
|
||||
}
|
||||
log.Printf("dash-agent: pushing to %s every %ds", *url, *interval)
|
||||
for {
|
||||
s := c.sample(*interval)
|
||||
if err := push(*url, *token, s); err != nil {
|
||||
log.Printf("dash-agent: push: %v", err)
|
||||
}
|
||||
time.Sleep(time.Duration(*interval) * time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
func envInt(k string, def int) int {
|
||||
if v, err := strconv.Atoi(os.Getenv(k)); err == nil && v > 0 {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// sample mirrors api.StatSample — duplicated here so the agent stays a
|
||||
// standalone binary with zero imports from the server module.
|
||||
type sample struct {
|
||||
Host string `json:"host"`
|
||||
OS string `json:"os"`
|
||||
Arch string `json:"arch"`
|
||||
CPUModel string `json:"cpuModel"`
|
||||
Cores int `json:"cores"`
|
||||
IntervalS int `json:"intervalS"`
|
||||
UptimeS float64 `json:"uptimeS"`
|
||||
CPU float64 `json:"cpu"`
|
||||
MemTotal float64 `json:"memTotal"`
|
||||
MemUsed float64 `json:"memUsed"`
|
||||
SwapTotal float64 `json:"swapTotal"`
|
||||
SwapUsed float64 `json:"swapUsed"`
|
||||
DiskTotal float64 `json:"diskTotal"`
|
||||
DiskUsed float64 `json:"diskUsed"`
|
||||
NetRx float64 `json:"netRx"`
|
||||
NetTx float64 `json:"netTx"`
|
||||
Load1 float64 `json:"load1"`
|
||||
Load5 float64 `json:"load5"`
|
||||
Load15 float64 `json:"load15"`
|
||||
Temps map[string]float64 `json:"temps,omitempty"`
|
||||
Containers []containerStat `json:"containers,omitempty"`
|
||||
}
|
||||
|
||||
type containerStat struct {
|
||||
Name string `json:"name"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
|
||||
type collector struct {
|
||||
host, os, arch, cpuModel string
|
||||
cores int
|
||||
prevCPU cpuTicks
|
||||
prevNet [2]uint64 // rx, tx bytes
|
||||
prevAt time.Time
|
||||
}
|
||||
|
||||
type cpuTicks struct{ idle, total uint64 }
|
||||
|
||||
func (c *collector) sample(interval int) sample {
|
||||
s := sample{
|
||||
Host: c.host, OS: c.os, Arch: c.arch,
|
||||
CPUModel: c.cpuModel, Cores: c.cores, IntervalS: interval,
|
||||
}
|
||||
now := time.Now()
|
||||
dt := now.Sub(c.prevAt).Seconds()
|
||||
if dt <= 0 {
|
||||
dt = float64(interval)
|
||||
}
|
||||
|
||||
if cur := readCPUTicks(); cur.total > c.prevCPU.total {
|
||||
busy := float64(cur.total-cur.idle) - float64(c.prevCPU.total-c.prevCPU.idle)
|
||||
s.CPU = busy / float64(cur.total-c.prevCPU.total) * 100
|
||||
c.prevCPU = cur
|
||||
}
|
||||
s.MemTotal, s.MemUsed, s.SwapTotal, s.SwapUsed = readMem()
|
||||
s.DiskTotal, s.DiskUsed = diskUsage("/")
|
||||
if cur := readNetBytes(); cur[0] >= c.prevNet[0] {
|
||||
s.NetRx = float64(cur[0]-c.prevNet[0]) / dt
|
||||
s.NetTx = float64(cur[1]-c.prevNet[1]) / dt
|
||||
c.prevNet = cur
|
||||
}
|
||||
c.prevAt = now
|
||||
s.Load1, s.Load5, s.Load15 = readLoad()
|
||||
s.UptimeS = readUptime()
|
||||
s.Temps = readTemps()
|
||||
s.Containers = readContainers()
|
||||
return s
|
||||
}
|
||||
|
||||
func push(url, token string, s sample) error {
|
||||
body, _ := json.Marshal(s)
|
||||
req, _ := http.NewRequest(http.MethodPost,
|
||||
strings.TrimRight(url, "/")+"/api/systems/ingest", bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
res, err := (&http.Client{Timeout: 10 * time.Second}).Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer res.Body.Close()
|
||||
io.Copy(io.Discard, res.Body)
|
||||
if res.StatusCode >= 300 {
|
||||
return fmt.Errorf("server returned %d", res.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---------- /proc + /sys readers (Linux; all no-op-safe elsewhere) ----------
|
||||
|
||||
// readCPUTicks parses the aggregate cpu line: fields are user nice system
|
||||
// idle iowait irq softirq steal — everything but idle+iowait counts busy.
|
||||
func readCPUTicks() cpuTicks {
|
||||
b, err := os.ReadFile("/proc/stat")
|
||||
if err != nil {
|
||||
return cpuTicks{}
|
||||
}
|
||||
var t cpuTicks
|
||||
for _, line := range strings.Split(string(b), "\n") {
|
||||
if !strings.HasPrefix(line, "cpu ") {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(line)[1:]
|
||||
var vals []uint64
|
||||
for _, v := range f {
|
||||
n, _ := strconv.ParseUint(v, 10, 64)
|
||||
vals = append(vals, n)
|
||||
t.total += n
|
||||
}
|
||||
if len(vals) >= 5 {
|
||||
t.idle = vals[3] + vals[4]
|
||||
}
|
||||
break
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
func cpuInfo() (model string, cores int) {
|
||||
b, err := os.ReadFile("/proc/cpuinfo")
|
||||
if err != nil {
|
||||
return "", runtime.NumCPU()
|
||||
}
|
||||
for _, line := range strings.Split(string(b), "\n") {
|
||||
if strings.HasPrefix(line, "processor") {
|
||||
cores++
|
||||
}
|
||||
if model == "" && strings.HasPrefix(line, "model name") {
|
||||
model = strings.TrimSpace(strings.SplitN(line, ":", 2)[1])
|
||||
}
|
||||
}
|
||||
if cores == 0 {
|
||||
cores = runtime.NumCPU()
|
||||
}
|
||||
return model, cores
|
||||
}
|
||||
|
||||
// readMem returns total/used bytes for RAM and swap from /proc/meminfo.
|
||||
func readMem() (memTotal, memUsed, swapTotal, swapUsed float64) {
|
||||
b, err := os.ReadFile("/proc/meminfo")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
kb := map[string]float64{}
|
||||
for _, line := range strings.Split(string(b), "\n") {
|
||||
f := strings.Fields(line)
|
||||
if len(f) >= 2 {
|
||||
kb[strings.TrimSuffix(f[0], ":")], _ = strconv.ParseFloat(f[1], 64)
|
||||
}
|
||||
}
|
||||
memTotal = kb["MemTotal"] * 1024
|
||||
memUsed = (kb["MemTotal"] - kb["MemAvailable"]) * 1024
|
||||
swapTotal = kb["SwapTotal"] * 1024
|
||||
swapUsed = (kb["SwapTotal"] - kb["SwapFree"]) * 1024
|
||||
return
|
||||
}
|
||||
|
||||
func diskUsage(path string) (total, used float64) {
|
||||
var st syscall.Statfs_t
|
||||
if err := syscall.Statfs(path, &st); err != nil {
|
||||
return 0, 0
|
||||
}
|
||||
total = float64(st.Blocks) * float64(st.Bsize)
|
||||
used = float64(st.Blocks-st.Bfree) * float64(st.Bsize)
|
||||
return total, used
|
||||
}
|
||||
|
||||
// readNetBytes sums rx/tx across interfaces, skipping loopback.
|
||||
func readNetBytes() [2]uint64 {
|
||||
b, err := os.ReadFile("/proc/net/dev")
|
||||
if err != nil {
|
||||
return [2]uint64{}
|
||||
}
|
||||
var rx, tx uint64
|
||||
for _, line := range strings.Split(string(b), "\n") {
|
||||
i := strings.Index(line, ":")
|
||||
if i < 0 || strings.Contains(line[:i], "lo") {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(line[i+1:])
|
||||
if len(f) >= 9 {
|
||||
r, _ := strconv.ParseUint(f[0], 10, 64)
|
||||
t, _ := strconv.ParseUint(f[8], 10, 64)
|
||||
rx += r
|
||||
tx += t
|
||||
}
|
||||
}
|
||||
return [2]uint64{rx, tx}
|
||||
}
|
||||
|
||||
func readLoad() (l1, l5, l15 float64) {
|
||||
b, err := os.ReadFile("/proc/loadavg")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
f := strings.Fields(string(b))
|
||||
if len(f) >= 3 {
|
||||
l1, _ = strconv.ParseFloat(f[0], 64)
|
||||
l5, _ = strconv.ParseFloat(f[1], 64)
|
||||
l15, _ = strconv.ParseFloat(f[2], 64)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func readUptime() float64 {
|
||||
b, err := os.ReadFile("/proc/uptime")
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
f := strings.Fields(string(b))
|
||||
if len(f) > 0 {
|
||||
v, _ := strconv.ParseFloat(f[0], 64)
|
||||
return v
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// readTemps walks /sys/class/hwmon for temp*_input sensors.
|
||||
func readTemps() map[string]float64 {
|
||||
out := map[string]float64{}
|
||||
hwmons, _ := os.ReadDir("/sys/class/hwmon")
|
||||
for _, hw := range hwmons {
|
||||
dir := "/sys/class/hwmon/" + hw.Name()
|
||||
chip, _ := os.ReadFile(dir + "/name")
|
||||
entries, _ := os.ReadDir(dir)
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if !strings.HasPrefix(name, "temp") || !strings.HasSuffix(name, "_input") {
|
||||
continue
|
||||
}
|
||||
b, err := os.ReadFile(dir + "/" + name)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
millideg, err := strconv.ParseFloat(strings.TrimSpace(string(b)), 64)
|
||||
if err != nil || millideg <= 0 || millideg > 150000 {
|
||||
continue
|
||||
}
|
||||
label := strings.TrimSpace(string(chip))
|
||||
if lb, err := os.ReadFile(dir + "/" + strings.TrimSuffix(name, "_input") + "_label"); err == nil {
|
||||
if l := strings.TrimSpace(string(lb)); l != "" {
|
||||
label = l
|
||||
}
|
||||
}
|
||||
if _, dup := out[label]; !dup {
|
||||
out[label] = millideg / 1000
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// readContainers lists Docker containers via the unix socket when present.
|
||||
func readContainers() []containerStat {
|
||||
if _, err := os.Stat("/var/run/docker.sock"); err != nil {
|
||||
return nil
|
||||
}
|
||||
client := &http.Client{
|
||||
Timeout: 3 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", "/var/run/docker.sock")
|
||||
},
|
||||
},
|
||||
}
|
||||
res, err := client.Get("http://d/containers/json?all=1")
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
defer res.Body.Close()
|
||||
var raw []struct {
|
||||
Names []string `json:"Names"`
|
||||
State string `json:"State"`
|
||||
}
|
||||
if json.NewDecoder(res.Body).Decode(&raw) != nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]containerStat, 0, len(raw))
|
||||
for _, r := range raw {
|
||||
name := ""
|
||||
if len(r.Names) > 0 {
|
||||
name = strings.TrimPrefix(r.Names[0], "/")
|
||||
}
|
||||
out = append(out, containerStat{Name: name, State: r.State})
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -41,6 +41,7 @@ func newServer(logger *zap.Logger, db *sql.DB, iconsDir string) *Server {
|
||||
s.lookup = domain.NewLookup().Run
|
||||
widget.Register(&monitorWidget{db: db})
|
||||
widget.Register(&domainWidget{db: db})
|
||||
widget.Register(&systemWidget{db: db})
|
||||
return s
|
||||
}
|
||||
|
||||
@@ -92,6 +93,14 @@ func (s *Server) routes() *gin.Engine {
|
||||
v1.GET("/domains/:id/checks", s.domainChecks)
|
||||
v1.POST("/notify/test", s.notifyTest)
|
||||
|
||||
v1.GET("/systems", s.listSystemsH)
|
||||
v1.POST("/systems", s.createSystem)
|
||||
v1.GET("/systems/:id", s.getSystem)
|
||||
v1.PATCH("/systems/:id", s.patchSystem)
|
||||
v1.DELETE("/systems/:id", s.deleteSystem)
|
||||
v1.GET("/systems/:id/stats", s.systemStats)
|
||||
v1.POST("/systems/ingest", s.ingestSystem)
|
||||
|
||||
v1.GET("/status", s.getStatus)
|
||||
v1.GET("/widgets/types", s.widgetTypes)
|
||||
v1.GET("/widgets/:id/data", s.widgetData)
|
||||
|
||||
@@ -539,6 +539,102 @@ func TestNotifyWebhook(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSystemIngest(t *testing.T) {
|
||||
srv := testServer(t)
|
||||
defer srv.Close()
|
||||
|
||||
// No token -> 401.
|
||||
res, _ := do(t, "POST", srv.URL+"/api/systems/ingest", `{"cpu":1}`)
|
||||
if res.StatusCode != http.StatusUnauthorized {
|
||||
t.Fatalf("expected 401, got %d", res.StatusCode)
|
||||
}
|
||||
|
||||
res, sys := do(t, "POST", srv.URL+"/api/systems", `{"name":"nas"}`)
|
||||
if res.StatusCode != http.StatusCreated {
|
||||
t.Fatalf("create: %v", sys)
|
||||
}
|
||||
id, token := sys["id"].(string), sys["token"].(string)
|
||||
if token == "" {
|
||||
t.Fatal("no agent token issued")
|
||||
}
|
||||
|
||||
// Bad token -> 401.
|
||||
req, _ := http.NewRequest("POST", srv.URL+"/api/systems/ingest",
|
||||
strings.NewReader(`{"cpu":50}`))
|
||||
req.Header.Set("Authorization", "Bearer wrong")
|
||||
if r, err := http.DefaultClient.Do(req); err != nil || r.StatusCode != http.StatusUnauthorized {
|
||||
t.Fatalf("bad token: %v %v", r.StatusCode, err)
|
||||
}
|
||||
|
||||
// Real push -> system goes up, latest + stats stored.
|
||||
push := func() (*http.Response, map[string]any) {
|
||||
t.Helper()
|
||||
req, _ := http.NewRequest("POST", srv.URL+"/api/systems/ingest",
|
||||
strings.NewReader(`{"host":"nas.local","os":"linux","arch":"amd64",
|
||||
"cpuModel":"Test CPU","cores":4,"cpu":42.5,"memTotal":8000,"memUsed":4000,
|
||||
"diskTotal":100000,"diskUsed":50000,"netRx":100,"netTx":50,"load1":1.5}`))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
res, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("push: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
var out map[string]any
|
||||
_ = json.NewDecoder(res.Body).Decode(&out)
|
||||
return res, out
|
||||
}
|
||||
if res, out := push(); res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("ingest: %v", out)
|
||||
}
|
||||
|
||||
res, sys = do(t, "GET", srv.URL+"/api/systems/"+id, "")
|
||||
if res.StatusCode != http.StatusOK || sys["status"] != "up" {
|
||||
t.Fatalf("get: %v", sys)
|
||||
}
|
||||
if sys["host"] != "nas.local" || sys["cores"].(float64) != 4 {
|
||||
t.Fatalf("host fields not stored: %v", sys)
|
||||
}
|
||||
latest := sys["latest"].(map[string]any)
|
||||
if latest["cpu"].(float64) != 42.5 {
|
||||
t.Fatalf("latest not stored: %v", latest)
|
||||
}
|
||||
|
||||
res, list := do(t, "GET", srv.URL+"/api/systems/"+id+"/stats?hours=1", "")
|
||||
if res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("stats: %v", list)
|
||||
}
|
||||
// stats returns an array — decode separately.
|
||||
req2, _ := http.NewRequest("GET", srv.URL+"/api/systems/"+id+"/stats?hours=1", nil)
|
||||
r2, _ := http.DefaultClient.Do(req2)
|
||||
var arr []map[string]any
|
||||
_ = json.NewDecoder(r2.Body).Decode(&arr)
|
||||
r2.Body.Close()
|
||||
if len(arr) != 1 || arr[0]["cpu"].(float64) != 42.5 {
|
||||
t.Fatalf("stats rows: %v", arr)
|
||||
}
|
||||
|
||||
// Widget fetcher serves the latest sample.
|
||||
_, a := do(t, "POST", srv.URL+"/api/sections", `{"name":"A"}`)
|
||||
_, w := do(t, "POST", srv.URL+"/api/items", `{"sectionId":"`+a["id"].(string)+`","kind":"widget","name":"S","config":{"type":"system","systemId":"`+id+`"}}`)
|
||||
res, data := do(t, "GET", srv.URL+"/api/widgets/"+w["id"].(string)+"/data", "")
|
||||
inner, _ := data["data"].(map[string]any)
|
||||
if res.StatusCode != http.StatusOK || inner["name"] != "nas" || inner["cpu"].(float64) != 42.5 {
|
||||
t.Fatalf("system widget: %d %v", res.StatusCode, data)
|
||||
}
|
||||
|
||||
// Rename + token rotation.
|
||||
res, sys = do(t, "PATCH", srv.URL+"/api/systems/"+id, `{"name":"nas2","rotateToken":true}`)
|
||||
if sys["name"] != "nas2" || sys["token"] == token {
|
||||
t.Fatalf("patch: %v", sys)
|
||||
}
|
||||
|
||||
res, _ = do(t, "DELETE", srv.URL+"/api/systems/"+id, "")
|
||||
if res.StatusCode != http.StatusNoContent {
|
||||
t.Fatalf("delete: %d", res.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMidpoint(t *testing.T) {
|
||||
f := func(v float64) *float64 { return &v }
|
||||
if got := midpoint(nil, nil); got != 1024 {
|
||||
|
||||
@@ -411,6 +411,10 @@ func (s *Server) runScheduler(done <-chan struct{}) {
|
||||
if _, err := s.db.Exec(`DELETE FROM heartbeats WHERE checked_at < datetime('now','-30 days')`); err != nil {
|
||||
s.log.Error("heartbeat prune", zap.Error(err))
|
||||
}
|
||||
// System samples are high-rate; a week of raw ticks is enough.
|
||||
if _, err := s.db.Exec(`DELETE FROM system_stats WHERE ts < datetime('now','-7 days')`); err != nil {
|
||||
s.log.Error("system stats prune", zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -453,6 +457,7 @@ func (s *Server) sweep() {
|
||||
}
|
||||
|
||||
s.sweepDomains()
|
||||
s.sweepSystems()
|
||||
}
|
||||
|
||||
// runCheck executes one monitor check (with retries) and persists the result.
|
||||
|
||||
@@ -0,0 +1,343 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/tdvorak/dash/internal/widget"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// ---------- model ----------
|
||||
|
||||
// StatSample is one agent payload. Numeric fields are optional-ish: the
|
||||
// agent may omit what a platform cannot provide.
|
||||
type StatSample struct {
|
||||
Host string `json:"host"`
|
||||
OS string `json:"os"`
|
||||
Arch string `json:"arch"`
|
||||
CPUModel string `json:"cpuModel"`
|
||||
Cores int `json:"cores"`
|
||||
IntervalS int `json:"intervalS"`
|
||||
UptimeS float64 `json:"uptimeS"`
|
||||
CPU float64 `json:"cpu"` // percent
|
||||
MemTotal float64 `json:"memTotal"` // bytes
|
||||
MemUsed float64 `json:"memUsed"`
|
||||
SwapTotal float64 `json:"swapTotal"`
|
||||
SwapUsed float64 `json:"swapUsed"`
|
||||
DiskTotal float64 `json:"diskTotal"` // bytes, root fs
|
||||
DiskUsed float64 `json:"diskUsed"`
|
||||
NetRx float64 `json:"netRx"` // bytes/sec
|
||||
NetTx float64 `json:"netTx"`
|
||||
Load1 float64 `json:"load1"`
|
||||
Load5 float64 `json:"load5"`
|
||||
Load15 float64 `json:"load15"`
|
||||
Temps map[string]float64 `json:"temps"`
|
||||
Containers []ContainerStat `json:"containers"`
|
||||
}
|
||||
|
||||
// ContainerStat is one Docker container's liveness; per-container CPU/mem
|
||||
// needs cgroup sampling — deferred (v1 ships name+state only).
|
||||
type ContainerStat struct {
|
||||
Name string `json:"name"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
|
||||
// System is one monitored host; `latest` is the newest StatSample.
|
||||
type System struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Token string `json:"token"`
|
||||
Host string `json:"host"`
|
||||
OS string `json:"os"`
|
||||
Arch string `json:"arch"`
|
||||
CPUModel string `json:"cpuModel"`
|
||||
Cores int `json:"cores"`
|
||||
IntervalS int `json:"intervalS"`
|
||||
Status string `json:"status"` // pending | up | down
|
||||
LastSeenAt *string `json:"lastSeenAt"`
|
||||
Latest *StatSample `json:"latest"`
|
||||
CreatedAt string `json:"createdAt"`
|
||||
}
|
||||
|
||||
const systemCols = `id, name, token, host, os, arch, cpu_model, cores,
|
||||
interval_s, status, last_seen_at, latest, created_at`
|
||||
|
||||
func scanSystem(row interface{ Scan(...any) error }) (*System, error) {
|
||||
var sys System
|
||||
var latest string
|
||||
err := row.Scan(&sys.ID, &sys.Name, &sys.Token, &sys.Host, &sys.OS,
|
||||
&sys.Arch, &sys.CPUModel, &sys.Cores, &sys.IntervalS, &sys.Status,
|
||||
&sys.LastSeenAt, &latest, &sys.CreatedAt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if latest != "" && latest != "{}" {
|
||||
var s StatSample
|
||||
if json.Unmarshal([]byte(latest), &s) == nil {
|
||||
sys.Latest = &s
|
||||
}
|
||||
}
|
||||
return &sys, nil
|
||||
}
|
||||
|
||||
func (s *Server) loadSystem(id string) (*System, error) {
|
||||
return scanSystem(s.db.QueryRow(`SELECT `+systemCols+` FROM systems WHERE id = ?`, id))
|
||||
}
|
||||
|
||||
func (s *Server) loadSystemByToken(token string) (*System, error) {
|
||||
return scanSystem(s.db.QueryRow(
|
||||
`SELECT `+systemCols+` FROM systems WHERE token = ?`, token))
|
||||
}
|
||||
|
||||
// ---------- handlers ----------
|
||||
|
||||
func (s *Server) listSystemsH(c *gin.Context) {
|
||||
rows, err := s.db.Query(`SELECT ` + systemCols + ` FROM systems ORDER BY name`)
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []System{}
|
||||
for rows.Next() {
|
||||
sys, err := scanSystem(rows)
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
out = append(out, *sys)
|
||||
}
|
||||
c.JSON(http.StatusOK, out)
|
||||
}
|
||||
|
||||
func (s *Server) createSystem(c *gin.Context) {
|
||||
var in struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&in); err != nil || in.Name == "" {
|
||||
fail(c, http.StatusBadRequest, "name required")
|
||||
return
|
||||
}
|
||||
id := newID("sys")
|
||||
if _, err := s.db.Exec(`INSERT INTO systems (id, name, token) VALUES (?,?,?)`,
|
||||
id, in.Name, newID("agent")); err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
sys, err := s.loadSystem(id)
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusCreated, sys)
|
||||
}
|
||||
|
||||
func (s *Server) getSystem(c *gin.Context) {
|
||||
sys, err := s.loadSystem(c.Param("id"))
|
||||
if notFound(err) {
|
||||
fail(c, http.StatusNotFound, "system not found")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, sys)
|
||||
}
|
||||
|
||||
func (s *Server) patchSystem(c *gin.Context) {
|
||||
id := c.Param("id")
|
||||
if _, err := s.loadSystem(id); notFound(err) {
|
||||
fail(c, http.StatusNotFound, "system not found")
|
||||
return
|
||||
}
|
||||
var in struct {
|
||||
Name *string `json:"name"`
|
||||
Rotate *bool `json:"rotateToken"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&in); err != nil {
|
||||
fail(c, http.StatusBadRequest, "invalid body")
|
||||
return
|
||||
}
|
||||
if in.Name != nil && *in.Name != "" {
|
||||
_, _ = s.db.Exec(`UPDATE systems SET name = ? WHERE id = ?`, *in.Name, id)
|
||||
}
|
||||
if in.Rotate != nil && *in.Rotate {
|
||||
_, _ = s.db.Exec(`UPDATE systems SET token = ? WHERE id = ?`, newID("agent"), id)
|
||||
}
|
||||
sys, err := s.loadSystem(id)
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, sys)
|
||||
}
|
||||
|
||||
func (s *Server) deleteSystem(c *gin.Context) {
|
||||
res, err := s.db.Exec(`DELETE FROM systems WHERE id = ?`, c.Param("id"))
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if n, _ := res.RowsAffected(); n == 0 {
|
||||
fail(c, http.StatusNotFound, "system not found")
|
||||
return
|
||||
}
|
||||
c.Status(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// systemStats returns stored samples newest-last for charting.
|
||||
func (s *Server) systemStats(c *gin.Context) {
|
||||
if _, err := s.loadSystem(c.Param("id")); notFound(err) {
|
||||
fail(c, http.StatusNotFound, "system not found")
|
||||
return
|
||||
}
|
||||
hours, _ := strconv.Atoi(c.DefaultQuery("hours", "24"))
|
||||
if hours <= 0 || hours > 24*7 {
|
||||
hours = 24
|
||||
}
|
||||
rows, err := s.db.Query(`SELECT ts, payload FROM system_stats
|
||||
WHERE system_id = ? AND ts >= datetime('now', ?) ORDER BY ts`,
|
||||
c.Param("id"), "-"+strconv.Itoa(hours)+" hours")
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
type sample struct {
|
||||
TS string `json:"ts"`
|
||||
StatSample
|
||||
}
|
||||
out := []sample{}
|
||||
for rows.Next() {
|
||||
var ts, payload string
|
||||
if err := rows.Scan(&ts, &payload); err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
var sm sample
|
||||
sm.TS = ts
|
||||
if json.Unmarshal([]byte(payload), &sm.StatSample) == nil {
|
||||
out = append(out, sm)
|
||||
}
|
||||
}
|
||||
c.JSON(http.StatusOK, out)
|
||||
}
|
||||
|
||||
// ingestSystem receives one agent push. Auth: Authorization: Bearer <token>.
|
||||
func (s *Server) ingestSystem(c *gin.Context) {
|
||||
token := strings.TrimPrefix(c.GetHeader("Authorization"), "Bearer ")
|
||||
if token == "" {
|
||||
token = c.GetHeader("X-Dash-Token")
|
||||
}
|
||||
if token == "" {
|
||||
fail(c, http.StatusUnauthorized, "token required")
|
||||
return
|
||||
}
|
||||
sys, err := s.loadSystemByToken(token)
|
||||
if err != nil {
|
||||
fail(c, http.StatusUnauthorized, "unknown token")
|
||||
return
|
||||
}
|
||||
var sample StatSample
|
||||
if err := c.ShouldBindJSON(&sample); err != nil {
|
||||
fail(c, http.StatusBadRequest, "invalid body")
|
||||
return
|
||||
}
|
||||
if sample.IntervalS <= 0 {
|
||||
sample.IntervalS = 10
|
||||
}
|
||||
payload, _ := json.Marshal(&sample)
|
||||
prev := sys.Status
|
||||
_, err = s.db.Exec(`UPDATE systems SET
|
||||
host = ?, os = ?, arch = ?, cpu_model = ?, cores = ?, interval_s = ?,
|
||||
status = 'up', last_seen_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
|
||||
latest = ?
|
||||
WHERE id = ?`,
|
||||
sample.Host, sample.OS, sample.Arch, sample.CPUModel, sample.Cores,
|
||||
sample.IntervalS, string(payload), sys.ID)
|
||||
if err != nil {
|
||||
fail(c, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if _, err := s.db.Exec(`INSERT INTO system_stats (system_id, payload) VALUES (?,?)`,
|
||||
sys.ID, string(payload)); err != nil {
|
||||
s.log.Error("system stats insert", zap.Error(err))
|
||||
}
|
||||
if prev != "up" {
|
||||
s.notify("system.up", sys.Name,
|
||||
fmt.Sprintf("System %q is back online (%s)", sys.Name, sample.Host))
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
// sweepSystems marks agents down when silent for > 3 intervals (min 30s).
|
||||
func (s *Server) sweepSystems() {
|
||||
rows, err := s.db.Query(`UPDATE systems SET status = 'down'
|
||||
WHERE status = 'up' AND last_seen_at IS NOT NULL
|
||||
AND datetime(last_seen_at, '+' || MAX(3 * interval_s, 30) || ' seconds') <= datetime('now')
|
||||
RETURNING id, name, host`)
|
||||
if err != nil {
|
||||
s.log.Error("system sweep", zap.Error(err))
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var id, name, host string
|
||||
if rows.Scan(&id, &name, &host) == nil {
|
||||
s.notify("system.down", name,
|
||||
fmt.Sprintf("System %q stopped reporting (%s)", name, host))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- widget fetcher ----------
|
||||
|
||||
// systemWidget serves board tiles bound to a system via config.systemId.
|
||||
type systemWidget struct{ db *sql.DB }
|
||||
|
||||
func (w *systemWidget) Meta() widget.Type {
|
||||
return widget.Type{
|
||||
Type: "system", Name: "System",
|
||||
Description: "Live CPU/memory/disk of a monitored host",
|
||||
Fields: []widget.Field{
|
||||
{Key: "systemId", Label: "System", Required: true},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (w *systemWidget) Fetch(_ context.Context, cfg json.RawMessage) (any, error) {
|
||||
var c struct {
|
||||
SystemID string `json:"systemId"`
|
||||
}
|
||||
if err := json.Unmarshal(cfg, &c); err != nil || c.SystemID == "" {
|
||||
return nil, errors.New("config.systemId required")
|
||||
}
|
||||
var name, status, latest string
|
||||
var lastSeen *string
|
||||
err := w.db.QueryRow(`SELECT name, status, last_seen_at, latest
|
||||
FROM systems WHERE id = ?`, c.SystemID).Scan(&name, &status, &lastSeen, &latest)
|
||||
if err != nil {
|
||||
return nil, errors.New("system not found")
|
||||
}
|
||||
var sm StatSample
|
||||
_ = json.Unmarshal([]byte(latest), &sm)
|
||||
return map[string]any{
|
||||
"name": name,
|
||||
"status": status,
|
||||
"lastSeen": lastSeen,
|
||||
"cpu": sm.CPU,
|
||||
"memTotal": sm.MemTotal,
|
||||
"memUsed": sm.MemUsed,
|
||||
"diskTotal": sm.DiskTotal,
|
||||
"diskUsed": sm.DiskUsed,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
-- +goose Up
|
||||
CREATE TABLE systems (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
token TEXT NOT NULL UNIQUE,
|
||||
host TEXT NOT NULL DEFAULT '',
|
||||
os TEXT NOT NULL DEFAULT '',
|
||||
arch TEXT NOT NULL DEFAULT '',
|
||||
cpu_model TEXT NOT NULL DEFAULT '',
|
||||
cores INTEGER NOT NULL DEFAULT 0,
|
||||
interval_s INTEGER NOT NULL DEFAULT 10,
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
last_seen_at TEXT,
|
||||
latest TEXT NOT NULL DEFAULT '{}',
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
);
|
||||
|
||||
-- Whole agent payload per tick; no sub-field queries, so one JSON column
|
||||
-- instead of a wide row that drifts with every agent version.
|
||||
CREATE TABLE system_stats (
|
||||
system_id TEXT NOT NULL REFERENCES systems(id) ON DELETE CASCADE,
|
||||
ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||
payload TEXT NOT NULL,
|
||||
PRIMARY KEY (system_id, ts)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_system_stats_ts ON system_stats(ts);
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE IF EXISTS system_stats;
|
||||
DROP TABLE IF EXISTS systems;
|
||||
Reference in new issue
Block a user