diff --git a/CHANGELOG.md b/CHANGELOG.md index fe188a7..321c843 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ All notable changes to teploy-dash are recorded here. ## [Unreleased] +### Added + +- Connecting Observe links straight to its token form, filled in. +- Disk alerts: Dash checks each server's filesystems every 10 minutes and alerts at 90% full, critical at 95%, clearing below 85%. + ## [0.1.24] - 2026-11-05 ## [0.1.23] - 2026-11-05 diff --git a/cmd/teploy-dash/frontend/index.html b/cmd/teploy-dash/frontend/index.html index 62f372c..1cf74b7 100644 --- a/cmd/teploy-dash/frontend/index.html +++ b/cmd/teploy-dash/frontend/index.html @@ -3016,7 +3016,7 @@

Plan —

Connect Observe

- Mint a service token in Observe (Settings → Service Tokens; scopes sites:create and keys:mint). + Create a service token in Observe, then paste it here. Open the filled-in form in ObserveEnter the address below for a link to the filled-in form, or use Settings → Service Tokens with scopes sites:create and keys:mint. Dash never sees an Observe admin login, and nothing is written until you confirm this instance's identity. A confirmed Observe was found on the fleet — its address is prefilled.

diff --git a/cmd/teploy-dash/frontend/js/app.js b/cmd/teploy-dash/frontend/js/app.js index 48d57a4..b5abe68 100644 --- a/cmd/teploy-dash/frontend/js/app.js +++ b/cmd/teploy-dash/frontend/js/app.js @@ -4773,6 +4773,16 @@ document.addEventListener('alpine:init', () => { // Step 1 of connect: read the discovery doc so the owner confirms the // instance identity BEFORE any token is sent. + // Observe's create form, filled in. The admin still presses Create there. + observeTokenURL() { + const base = (this.previewDoc && this.previewDoc.dashboard_url) || this.connectForm.base_url; + try { + const u = new URL(base); + if (u.protocol !== 'http:' && u.protocol !== 'https:') return ''; + return u.origin + '/settings?tab=service-tokens&name=teploy-dash&scopes=sites:create,keys:mint'; + } catch { return ''; } + }, + async previewObserve() { this.previewDoc = null; this.previewError = ''; diff --git a/cmd/teploy-dash/main.go b/cmd/teploy-dash/main.go index 40cad69..737b938 100644 --- a/cmd/teploy-dash/main.go +++ b/cmd/teploy-dash/main.go @@ -381,6 +381,13 @@ func run() error { srv.RunSiblings(cleanupCtx) }() + // Disk space on every server. Same cancel/join. + cleanupWG.Add(1) + go func() { + defer cleanupWG.Done() + srv.RunDiskWatch(cleanupCtx) + }() + // Start the HTTP server on the pre-bound listener. A LATER failure // (accept error, TLS misconfiguration) funnels through the same shutdown // epilogue as a signal — it no longer bypasses runner joins and the diff --git a/internal/alert/alert.go b/internal/alert/alert.go index 34b33bc..6ab3894 100644 --- a/internal/alert/alert.go +++ b/internal/alert/alert.go @@ -41,6 +41,7 @@ const ( SourceRestoreTest = "restore-test" SourceBackup = "backup" SourceDomain = "domain" + SourceDisk = "disk" SourceOperation = "operation" SourceAudit = "audit" // SourceHeartbeat labels push/heartbeat monitors (a server backup job or diff --git a/internal/server/alerting.go b/internal/server/alerting.go index af103ac..a85334b 100644 --- a/internal/server/alerting.go +++ b/internal/server/alerting.go @@ -35,6 +35,7 @@ import ( "github.com/useteploy/teploy-dash/internal/monitor" "github.com/useteploy/teploy-dash/internal/operation" "github.com/useteploy/teploy-dash/internal/outbox" + "github.com/useteploy/teploy-dash/internal/remote" ) // alertHoldGrace is how long a hold outlives the condition that caused it @@ -60,6 +61,7 @@ type alertingState struct { audit *audit.Log heartbeat *heartbeat.Runner pushLimit *rateLimiter + diskRead func(context.Context, remote.ServerConn) (string, error) // a seam for tests; nil reads over SSH holdMu sync.Mutex lockFirst map[string]time.Time // first time a deploy lock was observed diff --git a/internal/server/diskalerts.go b/internal/server/diskalerts.go new file mode 100644 index 0000000..650561c --- /dev/null +++ b/internal/server/diskalerts.go @@ -0,0 +1,205 @@ +package server + +// diskalerts.go: a full disk stops backups, builds and databases, so Dash reads +// each server's filesystems and raises one incident per filesystem. A server +// that cannot be read proves nothing, so its rules are left as they are. + +import ( + "context" + "fmt" + "log" + "strconv" + "strings" + "sync" + "time" + + "github.com/useteploy/teploy-dash/internal/alert" + "github.com/useteploy/teploy-dash/internal/incident" + "github.com/useteploy/teploy-dash/internal/remote" + sshclient "github.com/useteploy/teploy-dash/internal/ssh" +) + +const ( + diskWarnPercent = 90 + diskCriticalPercent = 95 + // Between this and diskWarnPercent an open incident stays open, so a disk + // hovering at the threshold does not alert over and over. + diskRecoverPercent = 85 + diskSweepInterval = 10 * time.Minute + diskReadTimeout = 30 * time.Second + diskSweepWorkers = 4 + // Pseudo and read-only filesystems are always "full" or say nothing useful. + diskReadCommand = "df -kP -x tmpfs -x devtmpfs -x squashfs -x overlay -x iso9660 -x efivarfs" +) + +// Two reads, so a build that briefly fills a disk does not page anyone. +var diskPolicy = incident.Policy{FailAfter: 2, RecoverAfter: 1} + +type diskUsage struct { + Mount string + UsedPercent int + FreeKB int64 +} + +// parseDiskUsage reads `df -kP` output. A filesystem listed twice (a bind +// mount) counts once, under the first mount shown. +func parseDiskUsage(out string) ([]diskUsage, error) { + var disks []diskUsage + seen := map[string]bool{} + sawHeader := false + for _, line := range strings.Split(out, "\n") { + f := strings.Fields(line) + if len(f) == 0 { + continue + } + if f[0] == "Filesystem" { + sawHeader = true + continue + } + if len(f) < 6 { + return nil, fmt.Errorf("short df line %q", line) + } + if f[4] == "-" { + continue // a filesystem with no size: nothing to judge + } + pct, err := strconv.Atoi(strings.TrimSuffix(f[4], "%")) + if err != nil { + return nil, fmt.Errorf("df capacity in %q: %w", line, err) + } + free, err := strconv.ParseInt(f[3], 10, 64) + if err != nil { + return nil, fmt.Errorf("df free space in %q: %w", line, err) + } + if seen[f[0]] { + continue + } + seen[f[0]] = true + disks = append(disks, diskUsage{Mount: strings.Join(f[5:], " "), UsedPercent: pct, FreeKB: free}) + } + if !sawHeader { + return nil, fmt.Errorf("not df output: %q", strings.TrimSpace(out)) + } + return disks, nil +} + +func diskRule(server, mount string, percent int) incident.Rule { + severity := alert.SeverityWarning + if percent >= diskCriticalPercent { + severity = alert.SeverityCritical + } + return incident.Rule{Key: "disk:" + server + ":" + mount, Source: alert.SourceDisk, ID: "disk-" + server + "-" + mount, + Name: "Disk " + mount + " on " + server, Server: server, Target: server + ":" + mount, Severity: severity, + Path: "/", Policy: diskPolicy} +} + +func formatFreeKB(kb int64) string { + if gb := float64(kb) / (1 << 20); gb >= 1 { + return fmt.Sprintf("%.0f GB", gb) + } + return fmt.Sprintf("%d MB", kb>>10) +} + +// observeDisks applies one complete read of one server. +func (s *Server) observeDisks(server string, disks []diskUsage, now time.Time) { + t := s.alerts.tracker + if t == nil { + return + } + present := map[string]bool{} + for _, d := range disks { + rule := diskRule(server, d.Mount, d.UsedPercent) + present[rule.Key] = true + msg := fmt.Sprintf("%s on %s is %d%% full, %s free.", d.Mount, server, d.UsedPercent, formatFreeKB(d.FreeKB)) + switch { + case d.UsedPercent >= diskWarnPercent: + t.Observe(rule, false, now, msg) + case d.UsedPercent < diskRecoverPercent: + t.Observe(rule, true, now, msg) + } + } + for _, inc := range t.Open() { + if inc.Source == alert.SourceDisk && inc.Server == server && !present[inc.Key] { + t.Retire(inc.Key, "this filesystem is no longer on the server") + } + } +} + +// retireRemovedDiskServers closes incidents for servers no longer in the fleet. +func (s *Server) retireRemovedDiskServers(servers []remote.ServerConn) { + t := s.alerts.tracker + if t == nil { + return + } + known := map[string]bool{} + for _, srv := range servers { + known[srv.Name] = true + } + for _, inc := range t.Open() { + if inc.Source == alert.SourceDisk && !known[inc.Server] { + t.Retire(inc.Key, "this server was removed") + } + } +} + +func readServerDisks(ctx context.Context, srv remote.ServerConn) (string, error) { + c, err := sshclient.Connect(ctx, srv.Host, srv.User, srv.KeyPath) + if err != nil { + return "", err + } + defer c.Close() + return c.Run(ctx, diskReadCommand) +} + +func (s *Server) sweepDisks(ctx context.Context) { + servers, err := s.resolveServers(ctx) + if err != nil { + log.Printf("[disk] server discovery: %v", err) + return + } + read := s.alerts.diskRead + if read == nil { + read = readServerDisks + } + sem := make(chan struct{}, diskSweepWorkers) + var wg sync.WaitGroup + for _, srv := range servers { + wg.Add(1) + sem <- struct{}{} + go func(srv remote.ServerConn) { + defer wg.Done() + defer func() { <-sem }() + readCtx, cancel := context.WithTimeout(ctx, diskReadTimeout) + defer cancel() + out, err := read(readCtx, srv) + if err != nil { + log.Printf("[disk] reading %s: %v", srv.Name, err) + return + } + disks, err := parseDiskUsage(out) + if err != nil { + log.Printf("[disk] reading %s: %v", srv.Name, err) + return + } + s.observeDisks(srv.Name, disks, time.Now()) + }(srv) + } + wg.Wait() + s.retireRemovedDiskServers(servers) +} + +// RunDiskWatch sweeps until ctx ends. +func (s *Server) RunDiskWatch(ctx context.Context) { + if s.alerts.tracker == nil { + return + } + t := time.NewTicker(diskSweepInterval) + defer t.Stop() + for { + s.sweepDisks(ctx) + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} diff --git a/internal/server/diskalerts_test.go b/internal/server/diskalerts_test.go new file mode 100644 index 0000000..de19074 --- /dev/null +++ b/internal/server/diskalerts_test.go @@ -0,0 +1,135 @@ +package server + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/useteploy/teploy-dash/internal/alert" + "github.com/useteploy/teploy-dash/internal/remote" +) + +const dfHeader = "Filesystem 1024-blocks Used Available Capacity Mounted on\n" + +func TestParseDiskUsage(t *testing.T) { + disks, err := parseDiskUsage(dfHeader + + "/dev/mapper/root 689963008 620000000 34603008 95% /\n" + + "/dev/mapper/root 689963008 620000000 34603008 95% /var/lib/bind\n" + + "/dev/sdb1 1000 10 990 1% /mnt/my disk\n" + + "none - - - - /proc/odd\n") + if err != nil { + t.Fatal(err) + } + if len(disks) != 2 { + t.Fatalf("want a bind mount and a sizeless filesystem left out, got %+v", disks) + } + if disks[0].Mount != "/" || disks[0].UsedPercent != 95 || disks[0].FreeKB != 34603008 { + t.Fatalf("root: %+v", disks[0]) + } + if disks[1].Mount != "/mnt/my disk" { + t.Fatalf("a mount with a space: %+v", disks[1]) + } + for _, bad := range []string{"", "sh: df: not found", dfHeader + "/dev/sda1 1 2\n", dfHeader + "/dev/sda1 1 2 x 5% /\n"} { + if _, err := parseDiskUsage(bad); err == nil { + t.Errorf("%q must not parse", bad) + } + } +} + +func diskRead(s *Server, server string, pct int, at time.Time) { + s.observeDisks(server, []diskUsage{{Mount: "/", UsedPercent: pct, FreeKB: 33 << 20}}, at) +} + +func TestDiskIncident_Lifecycle(t *testing.T) { + s, sink := alertServer(t) + writeNotifications(t, `{"channels":[{"id":"hook","type":"webhook","url":"https://h.example"}]}`) + now := time.Now() + const key = "disk:infra:/" + + diskRead(s, "infra", 92, now) + if _, open := openIncidents(t, s)[key]; open { + t.Fatal("one read must not open an incident: a build can fill a disk briefly") + } + diskRead(s, "infra", 80, now.Add(time.Minute)) + diskRead(s, "infra", 92, now.Add(2*time.Minute)) + if _, open := openIncidents(t, s)[key]; open { + t.Fatal("a read below the threshold resets the count") + } + diskRead(s, "infra", 92, now.Add(3*time.Minute)) + msg, open := openIncidents(t, s)[key] + if !open || !strings.Contains(msg, "/ on infra is 92% full, 33 GB free") { + t.Fatalf("open=%v msg=%q", open, msg) + } + + diskRead(s, "infra", 96, now.Add(4*time.Minute)) + for _, inc := range s.alerts.tracker.Open() { + if inc.Key == key && inc.Severity != alert.SeverityCritical { + t.Fatalf("95%% and over is critical, got %q", inc.Severity) + } + } + + diskRead(s, "infra", 87, now.Add(5*time.Minute)) + if _, open := openIncidents(t, s)[key]; !open { + t.Fatal("between 85% and 90% an open incident stays open") + } + diskRead(s, "infra", 84, now.Add(6*time.Minute)) + if _, open := openIncidents(t, s)[key]; open { + t.Fatal("under 85% recovers") + } + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && len(sink.events()) == 0 { + time.Sleep(10 * time.Millisecond) + } + evs := sink.events() + if len(evs) == 0 || evs[0].Source != alert.SourceDisk { + t.Fatalf("want a disk alert sent to the channel, got %+v", evs) + } +} + +func TestDiskSweep_UnreadableServerProvesNothing(t *testing.T) { + c := &backupCLI{} + s := newBackupsServer(t, c) + s.alerts.tracker.EndAdoption() + now := time.Now() + const key = "disk:prod:/" + diskRead(s, "prod", 96, now) + diskRead(s, "prod", 96, now) + if _, open := openIncidents(t, s)[key]; !open { + t.Fatal("setup: incident not open") + } + s.alerts.diskRead = func(context.Context, remote.ServerConn) (string, error) { + return "", errors.New("connection refused") + } + s.sweepDisks(context.Background()) + if _, open := openIncidents(t, s)[key]; !open { + t.Fatal("an unreachable server must not close or retire its incident") + } + s.alerts.diskRead = func(context.Context, remote.ServerConn) (string, error) { return "garbage", nil } + s.sweepDisks(context.Background()) + if _, open := openIncidents(t, s)[key]; !open { + t.Fatal("an unparseable read must not close its incident") + } +} + +func TestDiskIncident_Retired(t *testing.T) { + s, _ := alertServer(t) + now := time.Now() + for i := 0; i < 2; i++ { + s.observeDisks("a", []diskUsage{{Mount: "/mnt/x", UsedPercent: 96}, {Mount: "/", UsedPercent: 10}}, now) + s.observeDisks("b", []diskUsage{{Mount: "/", UsedPercent: 96}}, now) + } + s.observeDisks("a", []diskUsage{{Mount: "/", UsedPercent: 10}}, now) + inc := openIncidents(t, s) + if _, open := inc["disk:a:/mnt/x"]; open { + t.Fatal("a filesystem gone from a complete read is retired") + } + if _, open := inc["disk:b:/"]; !open { + t.Fatal("another server's incident must stay") + } + s.retireRemovedDiskServers([]remote.ServerConn{{Name: "a"}}) + if _, open := openIncidents(t, s)["disk:b:/"]; open { + t.Fatal("a removed server's incident is retired") + } +}