feat(alerts): alert when a server's disk fills up

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to teploy-dash are recorded here.
 
 ## [Unreleased]
 
+### Added
+
+- 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

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

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

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

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:
+		}
+	}
+}

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")
+	}
+}