feat(agent): send the CPU, memory and disk use of each site, and conform to contract v0.5.0

README.md

@@ -2,7 +2,7 @@
 
 Easy CLI tool for servers managed by FlyWP.
 
-Conforms to the FlyWP monitoring agent contract v0.4.0.
+Conforms to the FlyWP monitoring agent contract v0.5.0.
 
 ## Installation
 
@@ -103,7 +103,7 @@ All arguments after the WP-CLI command (or after the command for `fly exec`) go
 
 ### Monitoring agent
 
-`fly agent run` is the FlyWP monitoring agent. It runs all the time under systemd (`fly-agent.service`, as the server user, not root), and FlyWP installs it. Each minute it measures CPU, load, memory, swap, disk and network traffic, the pressure (PSI) and the disk activity. It reads the server each 10 seconds, so each minute also has its peaks. It sends the values and the server status (restart needed, waiting updates, OS, kernel, uptime, CPU count, Docker state and version) to FlyWP. It keeps unsent data on disk for up to 24 hours. FlyWP can update and restart the agent through it, without SSH. The agent does not need Docker. When Docker runs, the agent reads its socket with two requests only: `GET /version` and `GET /containers/json`.
+`fly agent run` is the FlyWP monitoring agent. It runs all the time under systemd (`fly-agent.service`, as the server user, not root), and FlyWP installs it. Each minute it measures CPU, load, memory, swap, disk and network traffic, the pressure (PSI) and the disk activity. It reads the server each 10 seconds, so each minute also has its peaks. It also measures the CPU, the memory and the disk use of each site: each Docker Compose project in the home folder of the server user. It measures the disk use at most one time each hour, at the lowest I/O priority. It sends the values and the server status (restart needed, waiting updates, OS, kernel, uptime, CPU count, Docker state and version) to FlyWP. It keeps unsent data on disk for up to 24 hours. FlyWP can update and restart the agent through it, without SSH. The agent does not need Docker. When Docker runs, the agent reads its socket with two requests only: `GET /version` and `GET /containers/json`.
 
 It reads `FLY_AGENT_URL` (https), `FLY_AGENT_TOKEN` and `FLY_AGENT_SERVER_ID` from `/etc/fly/agent.env`, and keeps its state in `STATE_DIRECTORY` (`/var/lib/fly-agent`).
 

internal/agent/clean.go

@@ -19,6 +19,8 @@ const (
 	maxStatusTextLen = 255
 	maxArchLen       = 16
 	maxCPUCount      = 4096
+	maxSites         = 1000
+	maxDirectoryLen  = 255
 	maxEventNameLen  = 64
 	maxErrorLen      = 2000
 )
@@ -46,9 +48,34 @@ func cleanSample(s wire.Sample) wire.Sample {
 	} {
 		*v = clampIntPtr(*v)
 	}
+	s.Sites = cleanSites(s.Sites)
 	return s
 }
 
+// cleanSites keeps at most maxSites items. It drops an item whose directory
+// is empty or too long: a cut directory would match an other site. A nil
+// slice stays nil, and an empty slice stays empty: they mean different things.
+func cleanSites(sites []wire.Site) []wire.Site {
+	if sites == nil {
+		return nil
+	}
+
+	out := make([]wire.Site, 0, min(len(sites), maxSites))
+	for _, site := range sites {
+		if len(out) == maxSites {
+			break
+		}
+		if n := utf8.RuneCountInString(site.Directory); n == 0 || n > maxDirectoryLen {
+			continue
+		}
+		site.CPUPercent = clampPtr(site.CPUPercent, 0, 100)
+		site.MemoryUsedBytes = clampIntPtr(site.MemoryUsedBytes)
+		site.DiskUsedBytes = clampIntPtr(site.DiskUsedBytes)
+		out = append(out, site)
+	}
+	return out
+}
+
 func cleanStatus(s wire.Status) wire.Status {
 	s.OS = truncate(s.OS, maxStatusTextLen)
 	s.Kernel = truncate(s.Kernel, maxStatusTextLen)

internal/agent/config.go

@@ -1,6 +1,6 @@
 // Package agent is the FlyWP monitoring agent: the long-running mode of fly
 // that "fly agent run" starts. It follows the FlyWP monitoring agent
-// contract v0.4.0.
+// contract v0.5.0.
 package agent
 
 import (

internal/agent/outbox_test.go

@@ -242,6 +242,57 @@ func TestCleanDockerStatus(t *testing.T) {
 	}
 }
 
+func TestCleanSites(t *testing.T) {
+	if s := cleanSample(wire.Sample{}); s.Sites != nil {
+		t.Errorf("sites = %v, want nil to stay nil (not known)", s.Sites)
+	}
+	s := cleanSample(wire.Sample{Sites: []wire.Site{}})
+	data, err := json.Marshal(s)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(data), `"sites":[]`) {
+		t.Errorf("sample JSON = %s, want an empty list to stay empty (no project)", data)
+	}
+
+	cpu, big := 250.0, uint64(math.MaxUint64)
+	many := []wire.Site{{Directory: ""}, {Directory: strings.Repeat("d", 256)}, {Directory: "example.com", CPUPercent: &cpu, MemoryUsedBytes: &big}}
+	for i := range 1200 {
+		many = append(many, wire.Site{Directory: fmt.Sprintf("site%d.com", i)})
+	}
+	s = cleanSample(wire.Sample{Sites: many})
+	if len(s.Sites) != maxSites {
+		t.Errorf("%d sites, want at most %d", len(s.Sites), maxSites)
+	}
+	if got := s.Sites[0]; got.Directory != "example.com" || *got.CPUPercent != 100 || *got.MemoryUsedBytes != math.MaxInt64 {
+		t.Errorf("first site = %+v, want example.com with its values in range, after the empty and the long directory", got)
+	}
+	if cpu != 250 {
+		t.Error("cleanSample() changed the value of the caller")
+	}
+}
+
+func TestOutboxKeepsSitesNullAndEmptyApart(t *testing.T) {
+	dir := t.TempDir()
+	o := loadOutbox(dir, slog.New(slog.DiscardHandler))
+	o.addSample(cleanSample(wire.Sample{Sites: nil}))
+	o.addSample(cleanSample(wire.Sample{Sites: []wire.Site{}}))
+
+	o = loadOutbox(dir, slog.New(slog.DiscardHandler))
+	if len(o.samples) != 2 {
+		t.Fatalf("samples = %d, want 2", len(o.samples))
+	}
+	for i, want := range []string{`"sites":null`, `"sites":[]`} {
+		data, err := json.Marshal(o.samples[i])
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !strings.Contains(string(data), want) {
+			t.Errorf("sample %d JSON = %s, want %s after the queue on disk", i, data, want)
+		}
+	}
+}
+
 func TestCleanEventDropsACommandIDThatIsNotAULID(t *testing.T) {
 	if e := cleanEvent(wire.Event{CommandID: "not-a-ulid"}); e.CommandID != "" {
 		t.Errorf("command_id = %q, want it removed", e.CommandID)

internal/agent/wire/wire.go

@@ -1,5 +1,5 @@
 // Package wire holds the JSON bodies of the FlyWP monitoring agent contract
-// v0.4.0: the requests that the agent sends and the replies that it reads.
+// v0.5.0: the requests that the agent sends and the replies that it reads.
 package wire
 
 import (
@@ -88,6 +88,21 @@ type Sample struct {
 	DiskWriteMaxBytesPerSecond *uint64 `json:"disk_write_max_bytes_per_second"`
 	DiskReadMaxOpsPerSecond    *uint64 `json:"disk_read_max_ops_per_second"`
 	DiskWriteMaxOpsPerSecond   *uint64 `json:"disk_write_max_ops_per_second"`
+
+	// Sites holds one item for each Docker Compose project in the home
+	// folder of the server user (contract v0.5.0). nil (JSON null) means that
+	// the agent cannot read Docker; an empty, non-nil slice ([]) means that
+	// Docker runs and no project matches.
+	Sites []Site `json:"sites"`
+}
+
+// Site is the use of one Docker Compose project in the minute. nil (JSON
+// null) means "not known". DiskUsedBytes is set in one sample each hour.
+type Site struct {
+	Directory       string   `json:"directory"`
+	CPUPercent      *float64 `json:"cpu_percent"`
+	MemoryUsedBytes *uint64  `json:"memory_used_bytes"`
+	DiskUsedBytes   *uint64  `json:"disk_used_bytes"`
 }
 
 // MetricsReply is the reply to POST /agent/v1/metrics.

internal/metrics/diskusage_other.go

@@ -0,0 +1,15 @@
+//go:build !unix
+
+package metrics
+
+import (
+	"context"
+	"errors"
+	"runtime"
+)
+
+func fileID(string) uint64 { return 0 }
+
+func diskUsage(context.Context, string) (uint64, int, error) {
+	return 0, 0, errors.New("disk use is not measured on " + runtime.GOOS)
+}

internal/metrics/diskusage_unix.go

@@ -0,0 +1,77 @@
+//go:build unix
+
+package metrics
+
+import (
+	"context"
+	"io/fs"
+	"path/filepath"
+	"syscall"
+)
+
+// fileID returns the inode of a path, or 0.
+func fileID(path string) uint64 {
+	var st syscall.Stat_t
+	if err := syscall.Stat(path, &st); err != nil {
+		return 0
+	}
+	return uint64(st.Ino)
+}
+
+// diskUsage returns the space that the files under root take on the disk: the
+// allocated blocks, as du shows them (st_blocks × 512). It does not follow a
+// symbolic link, does not go into an other file system, and counts a file
+// with more than one hard link one time. It skips what it cannot read, and
+// returns the number of those skips. It stops when ctx is done.
+func diskUsage(ctx context.Context, root string) (used uint64, skipped int, err error) {
+	var dev uint64
+	seen := map[[2]uint64]bool{}
+
+	err = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return ctxErr
+		}
+		if err != nil {
+			if path == root {
+				return err
+			}
+			// A folder that cannot be read is skipped. Its own blocks
+			// were counted before.
+			skipped++
+			return nil
+		}
+
+		info, err := d.Info()
+		if err != nil {
+			skipped++
+			return nil
+		}
+		st, ok := info.Sys().(*syscall.Stat_t)
+		if !ok {
+			skipped++
+			return nil
+		}
+
+		if path == root {
+			dev = uint64(st.Dev)
+		} else if uint64(st.Dev) != dev {
+			// A mount point of an other file system.
+			if d.IsDir() {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		if !d.IsDir() && st.Nlink > 1 {
+			key := [2]uint64{uint64(st.Dev), uint64(st.Ino)}
+			if seen[key] {
+				return nil
+			}
+			seen[key] = true
+		}
+
+		used += uint64(st.Blocks) * 512
+		return nil
+	})
+
+	return used, skipped, err
+}

internal/metrics/diskusage_unix_test.go

@@ -0,0 +1,97 @@
+//go:build unix
+
+package metrics
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"testing"
+)
+
+// blocks returns the allocated bytes of a path, as du counts them.
+func blocks(t *testing.T, path string) uint64 {
+	t.Helper()
+	var st syscall.Stat_t
+	if err := syscall.Lstat(path, &st); err != nil {
+		t.Fatal(err)
+	}
+	return uint64(st.Blocks) * 512
+}
+
+func TestDiskUsage(t *testing.T) {
+	root := filepath.Join(t.TempDir(), "example.com")
+	outside := t.TempDir()
+	for name, size := range map[string]int{"wp-config.php": 3000, "app/index.php": 100, "app/uploads/big.jpg": 200000} {
+		path := filepath.Join(root, name)
+		if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(path, []byte(strings.Repeat("x", size)), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// A hard link counts one time. A symbolic link to a big file outside
+	// the folder is not followed.
+	if err := os.Link(filepath.Join(root, "app/uploads/big.jpg"), filepath.Join(root, "big-link.jpg")); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(outside, "huge"), []byte(strings.Repeat("y", 1<<20)), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Symlink(filepath.Join(outside, "huge"), filepath.Join(root, "huge-link")); err != nil {
+		t.Fatal(err)
+	}
+
+	var want uint64
+	for _, p := range []string{"", "app", "app/uploads", "wp-config.php", "app/index.php", "app/uploads/big.jpg", "huge-link"} {
+		want += blocks(t, filepath.Join(root, p))
+	}
+
+	got, skipped, err := diskUsage(context.Background(), root)
+	if err != nil || skipped != 0 {
+		t.Fatalf("diskUsage() = %d, %d skipped, %v", got, skipped, err)
+	}
+	if got != want {
+		t.Errorf("diskUsage() = %d, want %d: the blocks of each file one time, without the target of the symbolic link", got, want)
+	}
+}
+
+func TestDiskUsageSkipsWhatItCannotRead(t *testing.T) {
+	if os.Getuid() == 0 {
+		t.Skip("root can read each folder")
+	}
+	root := t.TempDir()
+	locked := filepath.Join(root, "locked")
+	if err := os.MkdirAll(locked, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(locked, "secret"), []byte(strings.Repeat("s", 100000)), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Chmod(locked, 0); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
+
+	got, skipped, err := diskUsage(context.Background(), root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if skipped != 1 || got != blocks(t, root)+blocks(t, locked) {
+		t.Errorf("diskUsage() = %d with %d skipped, want the two folders and 1 skip", got, skipped)
+	}
+}
+
+func TestDiskUsageStops(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, _, err := diskUsage(ctx, t.TempDir()); err == nil {
+		t.Error("diskUsage() = nil error, want the error of the context")
+	}
+	if _, _, err := diskUsage(context.Background(), filepath.Join(t.TempDir(), "gone")); err == nil {
+		t.Error("diskUsage() = nil error, want an error for a folder that does not exist")
+	}
+}