diff --git a/Makefile b/Makefile index 57e30c2..2b5fb44 100644 --- a/Makefile +++ b/Makefile @@ -37,7 +37,7 @@ TAR_OWNER := $(shell tar --version 2>/dev/null | grep -q GNU && echo '--owner=0 # "go run pkg@version" builds each tool with the Go version of this module, # so a tool cannot be older than go.mod. -GOLANGCI_LINT := go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.13.2 +GOLANGCI_LINT := go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.15.0 GOVULNCHECK := go run golang.org/x/vuln/cmd/govulncheck@v1.8.0 build: ## Build bin/fly for this platform diff --git a/docs/monitoring-agent.md b/docs/monitoring-agent.md index 32a0951..d8a49ff 100644 --- a/docs/monitoring-agent.md +++ b/docs/monitoring-agent.md @@ -4,7 +4,7 @@ values to FlyWP. FlyWP shows them as charts and alerts, and can restart or update the agent without SSH. -The agent follows the FlyWP monitoring agent contract **v0.5.0**. The contract defines the requests, +The agent follows the FlyWP monitoring agent contract **v0.6.0**. The contract defines the requests, the fields and the rules on both sides. ## How it runs @@ -40,6 +40,7 @@ After a change, run `sudo systemctl restart fly-agent`. | `ran.json` | The commands that ran, so that no command runs two times. | | `counters.json` | The last counters, so that a restart does not lose a minute of traffic. | | `state.json` | The report interval, and the time of the last release check. | +| `traffic.json` | Where the next read of the proxy's traffic log starts: the file's inode, the byte offset, and a hash of the file's first bytes. | | `lock` | Stops a second agent. | Each file is written to a temporary file first, then renamed, so a crash never leaves half a file. @@ -59,6 +60,7 @@ average of the minute hides. | Pressure (PSI) | the share of time that tasks waited for CPU, memory or disk I/O, and its peak. Not sent when the kernel has no PSI. | | Disk activity | bytes and operations read and written on the physical disks, and the peaks each second | | Sites | the CPU, memory and disk use of each site (see below) | +| Traffic | the bytes the front proxy sent for each site name in the minute (see below) | With the samples, the agent sends the **status** of the server: restart needed, waiting updates (and security updates), OS, kernel, uptime, architecture, CPU count, Docker state (`running`, @@ -76,6 +78,22 @@ For each site, the agent sends: the lowest I/O priority. A folder that the server user cannot list is skipped, so the value can be lower than the real use (for example, the database files in `~/.fly` belong to the container user). +- **Own files:** the size of its `app` folder, from the same walk: the site's files without its + cache, backups and logs. `null` for a project with no `app` folder. + +### Traffic + +On a server where FlyWP set it up (White Label customer sites), the front proxy writes a second log +for the agent: `~/.fly/nginx/vhost/flywp-traffic.log`, one line for each request with the name of the site +(`$server_name`) and the bytes sent. Each sample adds the lines written since the sample before, by +name, at most 1000 names. + +- The first read starts at the end of the file. The offset is saved before the sample is buffered, + so a stop loses those bytes and never sends them two times. +- A new file (another inode, or other first bytes), or a file shorter than the offset, is read from + the start. +- When the agent has read all of a file larger than 64 MiB, it cuts the file to 0. +- Without the file, `traffic` is `null`. The agent reads Docker through its socket with two requests only: `GET /version` and `GET /containers/json`. It reads CPU and memory from cgroup v2. Without Docker or cgroup v2, it diff --git a/internal/agent/wire/wire.go b/internal/agent/wire/wire.go index 952f3cb..b40b2b7 100644 --- a/internal/agent/wire/wire.go +++ b/internal/agent/wire/wire.go @@ -1,5 +1,5 @@ // Package wire holds the JSON bodies of the FlyWP monitoring agent contract -// v0.5.0: the requests that the agent sends and the replies that it reads. +// v0.6.0: the requests that the agent sends and the replies that it reads. package wire import ( @@ -94,15 +94,31 @@ type Sample struct { // the agent cannot read Docker; an empty, non-nil slice ([]) means that // Docker runs and no project matches. Sites []Site `json:"sites"` + + // Traffic holds the bytes that the server's front proxy sent for each + // site name since the sample before, from its traffic log + // ~/.fly/nginx/vhost/flywp-traffic.log (contract v0.6.0). nil (JSON null) means that the server has no traffic log; an + // empty, non-nil slice means that no request came. + Traffic []Traffic `json:"traffic"` } // 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. +// null) means "not known". DiskUsedBytes and FilesBytes are set in one +// sample each hour, from the same walk: FilesBytes is the project's app +// folder, the site's own files (contract v0.6.0). type Site struct { Directory string `json:"directory"` CPUPercent *float64 `json:"cpu_percent"` MemoryUsedBytes *uint64 `json:"memory_used_bytes"` DiskUsedBytes *uint64 `json:"disk_used_bytes"` + FilesBytes *uint64 `json:"files_bytes"` +} + +// Traffic is what the front proxy sent for one site name: the name of its +// matched server block (nginx $server_name), and the bytes with the headers. +type Traffic struct { + Host string `json:"host"` + Bytes uint64 `json:"bytes"` } // MetricsReply is the reply to POST /agent/v1/metrics. diff --git a/internal/metrics/diskusage_other.go b/internal/metrics/diskusage_other.go index 569424c..f8a9135 100644 --- a/internal/metrics/diskusage_other.go +++ b/internal/metrics/diskusage_other.go @@ -5,11 +5,17 @@ package metrics import ( "context" "errors" + "io/fs" + "os" "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) +func openNoFollow(path string) (*os.File, error) { return os.OpenFile(path, os.O_RDWR, 0) } + +func inodeOf(fs.FileInfo) uint64 { return 0 } + +func diskUsage(context.Context, string) (folderUse, int, error) { + return folderUse{}, 0, errors.New("disk use is not measured on " + runtime.GOOS) } diff --git a/internal/metrics/diskusage_unix.go b/internal/metrics/diskusage_unix.go index 583128d..0989ffe 100644 --- a/internal/metrics/diskusage_unix.go +++ b/internal/metrics/diskusage_unix.go @@ -5,10 +5,25 @@ package metrics import ( "context" "io/fs" + "os" "path/filepath" + "strings" "syscall" ) +// openNoFollow opens a file to read and cut it, and refuses a symbolic link. +func openNoFollow(path string) (*os.File, error) { + return os.OpenFile(path, os.O_RDWR|syscall.O_NOFOLLOW, 0) +} + +// inodeOf returns the inode of an open file's info, or 0. +func inodeOf(info fs.FileInfo) uint64 { + if st, ok := info.Sys().(*syscall.Stat_t); ok { + return uint64(st.Ino) + } + return 0 +} + // fileID returns the inode of a path, or 0. func fileID(path string) uint64 { var st syscall.Stat_t @@ -19,13 +34,15 @@ func fileID(path string) uint64 { } // 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 +// allocated blocks, as du shows them (st_blocks × 512), and the part of it in +// the app folder of root, the site's own files (nil without one). 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) (use folderUse, skipped int, err error) { + var dev, own uint64 seen := map[[2]uint64]bool{} + app := filepath.Join(root, "app") err = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { if ctxErr := ctx.Err(); ctxErr != nil { @@ -69,9 +86,16 @@ func diskUsage(ctx context.Context, root string) (used uint64, skipped int, err seen[key] = true } - used += uint64(st.Blocks) * 512 + blocks := uint64(st.Blocks) * 512 + use.used += blocks + if path == app && d.IsDir() { + use.own = &own + } + if path == app || strings.HasPrefix(path, app+string(filepath.Separator)) { + own += blocks + } return nil }) - return used, skipped, err + return use, skipped, err } diff --git a/internal/metrics/diskusage_unix_test.go b/internal/metrics/diskusage_unix_test.go index 115af84..1170ee1 100644 --- a/internal/metrics/diskusage_unix_test.go +++ b/internal/metrics/diskusage_unix_test.go @@ -50,12 +50,33 @@ func TestDiskUsage(t *testing.T) { want += blocks(t, filepath.Join(root, p)) } + var wantOwn uint64 + for _, p := range []string{"app", "app/uploads", "app/index.php", "app/uploads/big.jpg"} { + wantOwn += 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) + t.Fatalf("diskUsage() = %d, %d skipped, %v", got.used, skipped, err) + } + if got.used != want { + t.Errorf("diskUsage() = %d, want %d: the blocks of each file one time, without the target of the symbolic link", got.used, want) + } + // The site's own files: the app folder (contract v0.6.0). + if got.own == nil || *got.own != wantOwn { + t.Errorf("diskUsage() own = %v, want %d: the app folder", got.own, wantOwn) + } +} + +func TestDiskUsageWithNoAppFolder(t *testing.T) { + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "docker-compose.yml"), []byte("services: {}"), 0o644); err != nil { + t.Fatal(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) + + got, _, err := diskUsage(context.Background(), root) + if err != nil || got.used == 0 || got.own != nil { + t.Errorf("diskUsage() = %+v, %v, want a size and no own files", got, err) } } @@ -80,8 +101,8 @@ func TestDiskUsageSkipsWhatItCannotRead(t *testing.T) { 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) + if skipped != 1 || got.used != blocks(t, root)+blocks(t, locked) { + t.Errorf("diskUsage() = %d with %d skipped, want the two folders and 1 skip", got.used, skipped) } } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 7f48fbe..276c0b9 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -1,7 +1,7 @@ // Package metrics measures a Linux server for the monitoring agent: CPU, // load, memory, swap, disk, network, pressure (PSI) and disk activity each // minute, with the peaks of the minute from a reading each 10 seconds, the -// use of each site, and the status of the server (contract v0.5.0). It needs +// use of each site, and the status of the server (contract v0.5.0; own files and traffic v0.6.0). It needs // no root. package metrics @@ -99,6 +99,8 @@ type Collector struct { // and walker measures the disk use of the sites. prevContainers *containerReadings walker *diskWalker + // traffic reads the front proxy's traffic log (contract v0.6.0). + traffic *trafficReader updatesAt time.Time updatesKnown bool @@ -121,6 +123,10 @@ func New(root, stateDir string, log *slog.Logger) *Collector { c.docker = dockerapi.New(c.file("var/run/docker.sock")) c.home = homeDir() c.walker = newDiskWalker() + c.traffic = &trafficReader{state: filepath.Join(stateDir, "traffic.json"), limit: trafficLimit} + if c.home != "" { + c.traffic.path = filepath.Join(c.home, ".fly", "nginx", "vhost", "flywp-traffic.log") + } // Take a reading now, so that the first sample has a CPU value for the // time since the start. The saved reading comes before it only when it is @@ -242,6 +248,9 @@ func (c *Collector) Sample(now time.Time) (wire.Sample, error) { // The sites come after each step that can fail: they take the result // of the disk walk, which must not go with a sample that is dropped. s.Sites = c.sites(context.Background(), cur, readAt) + // Last too: the read saves its offset, so its bytes go with this sample + // or with none. + s.Traffic = c.traffic.read(c.log) c.prev = &cur c.readings = nil diff --git a/internal/metrics/sites.go b/internal/metrics/sites.go index 9447742..484dfc7 100644 --- a/internal/metrics/sites.go +++ b/internal/metrics/sites.go @@ -58,7 +58,7 @@ func homeDir() string { } // sites measures each Docker Compose project in the home folder at the tick -// cur, read at readAt (contract v0.5.0). It returns nil when the agent cannot read Docker or +// cur, read at readAt (contract v0.5.0; own files and traffic v0.6.0). It returns nil when the agent cannot read Docker or // the cgroups: Docker does not run, the socket refuses the agent, or the // server has cgroup v1. It returns an empty, non-nil slice when no project // matches. @@ -148,7 +148,8 @@ func (c *Collector) sites(ctx context.Context, cur reading, readAt time.Time) [] site.MemoryUsedBytes = &p.mem } if d, ok := disk[name]; ok { - site.DiskUsedBytes = &d + site.DiskUsedBytes = &d.used + site.FilesBytes = d.own } out = append(out, site) } @@ -228,21 +229,28 @@ func statValue(key string) func([]byte) (uint64, error) { } } +// folderUse is the disk use of one project folder: all of it, and its app +// folder (the site's own files), nil when it has none (contract v0.6.0). +type folderUse struct { + used uint64 + own *uint64 +} + // diskWalker measures the disk use of the project folders in a goroutine of // its own, so that a large folder does not delay the samples. The collector // takes each result one time. type diskWalker struct { mu sync.Mutex running bool - last time.Time // the start of the last walk - results map[string]uint64 // by directory, not yet sent + last time.Time // the start of the last walk + results map[string]folderUse // by directory, not yet sent // skipWarned holds the directories whose skipped files were logged as a // warning. Only the walk goroutine uses it. skipWarned map[string]bool // done is closed when the running walk ends. Tests wait for it. done chan struct{} // walk measures one folder, and timeout limits it. Tests replace them. - walk func(ctx context.Context, root string) (uint64, int, error) + walk func(ctx context.Context, root string) (folderUse, int, error) timeout time.Duration } @@ -265,7 +273,7 @@ func (w *diskWalker) start(folders map[string]string, log *slog.Logger) { go func() { defer close(done) - results := map[string]uint64{} + results := map[string]folderUse{} for _, name := range slices.Sorted(func(yield func(string) bool) { for name := range folders { if !yield(name) { @@ -302,12 +310,12 @@ func (w *diskWalker) start(folders map[string]string, log *slog.Logger) { // for it after the timeout. A walk that hangs in a system call, for example on // a network mount that does not answer, stays behind; the other folders and // the next walks go on. -func (w *diskWalker) walkOne(root string) (uint64, int, error) { +func (w *diskWalker) walkOne(root string) (folderUse, int, error) { ctx, cancel := context.WithTimeout(context.Background(), w.timeout) defer cancel() type result struct { - used uint64 + used folderUse skipped int err error } @@ -325,12 +333,12 @@ func (w *diskWalker) walkOne(root string) (uint64, int, error) { case r := <-ch: return r.used, r.skipped, r.err case <-ctx.Done(): - return 0, 0, ctx.Err() + return folderUse{}, 0, ctx.Err() } } // take returns the results of the last walk that ended, one time. -func (w *diskWalker) take() map[string]uint64 { +func (w *diskWalker) take() map[string]folderUse { w.mu.Lock() defer w.mu.Unlock() r := w.results diff --git a/internal/metrics/sites_test.go b/internal/metrics/sites_test.go index ec95d1d..f8fa869 100644 --- a/internal/metrics/sites_test.go +++ b/internal/metrics/sites_test.go @@ -58,7 +58,7 @@ func sitesCollector(t *testing.T, srv *server, cs []fakeContainer) *Collector { c := srv.collector(t.TempDir()) c.docker = dockerapi.New(engine.Socket) c.home = filepath.Join(srv.root, "home/fly") - c.walker.walk = func(context.Context, string) (uint64, int, error) { return 4096, 0, nil } + c.walker.walk = func(context.Context, string) (folderUse, int, error) { return folderUse{used: 4096}, 0, nil } return c } @@ -220,9 +220,10 @@ func TestSitesDiskIsSentOneTime(t *testing.T) { srv.cgroup("a1", 1, 1, 0) c := sitesCollector(t, srv, []fakeContainer{{"a1", srv.home("example.com")}}) var walked []string - c.walker.walk = func(_ context.Context, root string) (uint64, int, error) { + c.walker.walk = func(_ context.Context, root string) (folderUse, int, error) { walked = append(walked, root) - return 8192, 0, nil + own := uint64(4096) + return folderUse{used: 8192, own: &own}, 0, nil } base := time.Now().Add(time.Second) @@ -245,6 +246,10 @@ func TestSitesDiskIsSentOneTime(t *testing.T) { if d := second.Sites[0].DiskUsedBytes; d == nil || *d != 8192 { t.Errorf("disk in the second sample = %v, want 8192", ptr(d)) } + // The site's own files come from the same walk (contract v0.6.0). + if f := second.Sites[0].FilesBytes; f == nil || *f != 4096 { + t.Errorf("own files in the second sample = %v, want 4096", ptr(f)) + } // One time only, and no new walk within one hour. third, err := c.Sample(base.Add(2 * time.Minute)) @@ -261,7 +266,7 @@ func TestSitesDiskWalkThatFails(t *testing.T) { srv := newServer(t) srv.cgroup("a1", 1, 1, 0) c := sitesCollector(t, srv, []fakeContainer{{"a1", srv.home("example.com")}}) - c.walker.walk = func(context.Context, string) (uint64, int, error) { return 0, 0, context.DeadlineExceeded } + c.walker.walk = func(context.Context, string) (folderUse, int, error) { return folderUse{}, 0, context.DeadlineExceeded } if _, err := c.Sample(time.Now().Add(time.Second)); err != nil { t.Fatal(err) @@ -353,11 +358,11 @@ func TestSitesDiskWalkThatHangs(t *testing.T) { hang := make(chan struct{}) t.Cleanup(func() { close(hang) }) c.walker.timeout = 50 * time.Millisecond - c.walker.walk = func(_ context.Context, root string) (uint64, int, error) { + c.walker.walk = func(_ context.Context, root string) (folderUse, int, error) { if strings.HasSuffix(root, "a.com") { <-hang } - return 4096, 0, nil + return folderUse{used: 4096}, 0, nil } if _, err := c.Sample(time.Now().Add(time.Second)); err != nil { @@ -385,7 +390,7 @@ func TestSitesWarnOneTimeForFilesThatCannotBeRead(t *testing.T) { c := sitesCollector(t, srv, []fakeContainer{{"a1", srv.home(".fly")}}) rec := &levels{} c.log = slog.New(rec) - c.walker.walk = func(context.Context, string) (uint64, int, error) { return 4096, 9, nil } + c.walker.walk = func(context.Context, string) (folderUse, int, error) { return folderUse{used: 4096}, 9, nil } for i := range 3 { c.walker.mu.Lock() diff --git a/internal/metrics/traffic.go b/internal/metrics/traffic.go new file mode 100644 index 0000000..9d7ad53 --- /dev/null +++ b/internal/metrics/traffic.go @@ -0,0 +1,158 @@ +package metrics + +import ( + "bufio" + "cmp" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "io/fs" + "log/slog" + "maps" + "os" + "slices" + "strconv" + "strings" + + "github.com/flywp/server-cli/internal/agent/wire" + "github.com/flywp/server-cli/internal/statefile" +) + +const ( + // trafficLimit is the size past which the agent cuts the traffic log, + // once it has read all of it (contract v0.6.0). + trafficLimit = 64 << 20 + // trafficMaxHosts is the most names in a sample: past it, the names with + // the fewest bytes are left out (contract v0.6.0). + trafficMaxHosts = 1000 + // trafficHead is the most bytes from the start of the log that the agent + // hashes to know the file again: Linux can give a new file the inode of + // one that was just deleted. + trafficHead = 256 +) + +// trafficReader reads the traffic log that the server's front proxy writes +// for the agent (contract v0.6.0, "Traffic and own files"): one line for each +// request, "$server_name $bytes_sent". The offset of the next line to read is +// kept in a state file, so a restart reads on from where it stopped. +type trafficReader struct { + path string // the log, under the home folder; "" without a home + state string // traffic.json + limit int64 // trafficLimit; tests set a smaller one +} + +// trafficOffset is where the next read starts, in the file with that inode +// and that hash of its first bytes (up to trafficHead, and not past Offset). +type trafficOffset struct { + Inode uint64 `json:"inode"` + Offset int64 `json:"offset"` + Head string `json:"head"` +} + +// read returns the bytes the proxy sent for each name since the last read, +// most first. It returns nil when the server has no traffic log, and an +// empty, non-nil slice when no request came. The new offset is saved before +// it returns: a stop before the sample is buffered loses those bytes, and +// never sends them two times. +func (r *trafficReader) read(log *slog.Logger) []wire.Traffic { + if r.path == "" { + return nil + } + // Never through a symbolic link, and only a regular file: the agent cuts + // this file, so a link in its place must not make it cut another one. + f, err := openNoFollow(r.path) + if err != nil { + if !errors.Is(err, fs.ErrNotExist) { + log.Warn("cannot read the traffic log", "error", err) + } + return nil + } + defer func() { _ = f.Close() }() + + info, err := f.Stat() + if err != nil || !info.Mode().IsRegular() { + log.Warn("cannot read the traffic log: not a regular file", "error", err) + return nil + } + size, inode := info.Size(), inodeOf(info) + + var saved trafficOffset + if err := statefile.Read(r.state, &saved); err != nil { + if !errors.Is(err, fs.ErrNotExist) { + log.Warn("ignoring the saved traffic offset", "error", err) + } + // A first read starts at the end: the lines before it can be of an + // earlier month, and one sample must not bring them all. + r.save(log, f, trafficOffset{Inode: inode, Offset: size}) + return []wire.Traffic{} + } + + offset := saved.Offset + if offset < 0 || saved.Inode != inode || size < offset || head(f, offset) != saved.Head { + // A new file, or one that was cut: from the start. The head tells a + // new file that got the inode of the old one. + offset = 0 + } + if _, err := f.Seek(offset, io.SeekStart); err != nil { + log.Warn("cannot read the traffic log", "error", err) + return nil + } + + sent := map[string]uint64{} + lines := bufio.NewReader(io.LimitReader(f, size-offset)) + for { + line, err := lines.ReadString('\n') + if err != nil { + // The end, or a line that is still being written: it waits for + // the next read. + break + } + offset += int64(len(line)) + host, bytes, ok := strings.Cut(strings.TrimSpace(line), " ") + n, err := strconv.ParseUint(bytes, 10, 64) + if !ok || err != nil || host == "" || len(host) > 255 { + continue + } + sent[host] += n + } + + // Read to the end of a large file: cut it, so that it stays small. The + // lines written between the read and the cut are lost. + if offset == size && size > r.limit { + if err := f.Truncate(0); err != nil { + log.Warn("cannot cut the traffic log", "error", err) + } else { + offset = 0 + } + } + r.save(log, f, trafficOffset{Inode: inode, Offset: offset}) + + hosts := slices.SortedFunc(maps.Keys(sent), func(a, b string) int { + return cmp.Or(cmp.Compare(sent[b], sent[a]), cmp.Compare(a, b)) + }) + out := make([]wire.Traffic, 0, min(len(hosts), trafficMaxHosts)) + for _, host := range hosts[:min(len(hosts), trafficMaxHosts)] { + out = append(out, wire.Traffic{Host: host, Bytes: sent[host]}) + } + return out +} + +// save writes the offset o in the file f, with the hash of f's head. +func (r *trafficReader) save(log *slog.Logger, f *os.File, o trafficOffset) { + o.Head = head(f, o.Offset) + if err := statefile.Write(r.state, o); err != nil { + log.Warn("saving the traffic offset", "error", err) + } +} + +// head returns the hash of the first bytes of f, up to trafficHead and not +// past offset, or "" when it cannot read them. +func head(f *os.File, offset int64) string { + buf := make([]byte, min(max(offset, 0), trafficHead)) + if _, err := f.ReadAt(buf, 0); err != nil { + return "" + } + sum := sha256.Sum256(buf) + return hex.EncodeToString(sum[:]) +} diff --git a/internal/metrics/traffic_test.go b/internal/metrics/traffic_test.go new file mode 100644 index 0000000..b1d75ae --- /dev/null +++ b/internal/metrics/traffic_test.go @@ -0,0 +1,167 @@ +package metrics + +import ( + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "github.com/flywp/server-cli/internal/agent/wire" +) + +// trafficLog returns a reader of a traffic log in a temporary folder, and a +// function that appends lines to the log. +func trafficLog(t *testing.T) (*trafficReader, func(string)) { + t.Helper() + dir := t.TempDir() + r := &trafficReader{path: filepath.Join(dir, "traffic.log"), state: filepath.Join(dir, "traffic.json"), limit: trafficLimit} + write := func(lines string) { + f, err := os.OpenFile(r.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o664) + if err != nil { + t.Fatal(err) + } + defer func() { _ = f.Close() }() + if _, err := f.WriteString(lines); err != nil { + t.Fatal(err) + } + } + return r, write +} + +var quiet = slog.New(slog.NewTextHandler(io.Discard, nil)) + +func TestTrafficWithNoLogIsNull(t *testing.T) { + r, _ := trafficLog(t) + if got := r.read(quiet); got != nil { + t.Errorf("read() = %v, want nil: no traffic log", got) + } + if got := (&trafficReader{}).read(quiet); got != nil { + t.Errorf("read() with no home = %v, want nil", got) + } +} + +func TestTrafficStartsAtTheEndThenAddsEachName(t *testing.T) { + r, write := trafficLog(t) + // Written before the agent read the log: of an earlier month, perhaps. + write("old.example.com 999999\n") + + if got := r.read(quiet); got == nil || len(got) != 0 { + t.Fatalf("first read = %v, want an empty list: it starts at the end", got) + } + + write("shop.example.com 100\n*.blog.example.com 40\nshop.example.com 250\nnot a line\nbad.example.com -3\n\nhalf.example.com 7") + want := []wire.Traffic{{Host: "shop.example.com", Bytes: 350}, {Host: "*.blog.example.com", Bytes: 40}} + if got := r.read(quiet); !reflect.DeepEqual(got, want) { + t.Errorf("read() = %v, want %v: the lines since, by name, most first, bad lines skipped", got, want) + } + + // The line that was still being written comes with the next read. + write("1\n") + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "half.example.com", Bytes: 71}}) { + t.Errorf("read() = %v, want the finished line", got) + } + if got := r.read(quiet); got == nil || len(got) != 0 { + t.Errorf("read() with no new line = %v, want an empty list", got) + } +} + +func TestTrafficReadsANewOrCutFileFromTheStart(t *testing.T) { + r, write := trafficLog(t) + write("a.example.com 1\n") + r.read(quiet) + write("a.example.com 10\n") + r.read(quiet) + + // Cut by the cron line: shorter than the offset. + if err := os.Truncate(r.path, 0); err != nil { + t.Fatal(err) + } + write("a.example.com 5\n") + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "a.example.com", Bytes: 5}}) { + t.Errorf("read() after a cut = %v, want the new lines from the start", got) + } + + // A new file in its place. + if err := os.Remove(r.path); err != nil { + t.Fatal(err) + } + write("a.example.com 300000\n") + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "a.example.com", Bytes: 300000}}) { + t.Errorf("read() of a new file = %v, want it from the start", got) + } +} + +func TestTrafficKnowsANewFileWithTheSameInode(t *testing.T) { + r, write := trafficLog(t) + write("") + r.read(quiet) + write("shop.example.com 7\n") + r.read(quiet) + + // Linux can give a new file the inode of one that was just deleted. The + // same file written again has the same inode on each system. + if err := os.WriteFile(r.path, []byte("a.shop.example.com 300000\n"), 0o664); err != nil { + t.Fatal(err) + } + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "a.shop.example.com", Bytes: 300000}}) { + t.Errorf("read() of a new file with the same inode = %v, want it from the start", got) + } +} + +func TestTrafficCutsALargeLogOnceItIsRead(t *testing.T) { + r, write := trafficLog(t) + r.limit = 64 + write("") + r.read(quiet) + write(strings.Repeat("big.example.com 2\n", 10)) + + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "big.example.com", Bytes: 20}}) { + t.Fatalf("read() = %v, want all of it", got) + } + if info, err := os.Stat(r.path); err != nil || info.Size() != 0 { + t.Errorf("log after the read = %v, %v, want cut to 0", info, err) + } + write("big.example.com 3\n") + if got := r.read(quiet); !reflect.DeepEqual(got, []wire.Traffic{{Host: "big.example.com", Bytes: 3}}) { + t.Errorf("read() after the cut = %v, want the new line", got) + } +} + +func TestTrafficSendsTheNamesWithTheMostBytes(t *testing.T) { + r, write := trafficLog(t) + write("") + r.read(quiet) + var lines strings.Builder + for i := range trafficMaxHosts + 5 { + fmt.Fprintf(&lines, "site%d.example.com %d\n", i, i+1) + } + write(lines.String()) + + got := r.read(quiet) + if len(got) != trafficMaxHosts || got[0].Bytes != trafficMaxHosts+5 || got[len(got)-1].Bytes != 6 { + t.Errorf("read() = %d names from %d to %d bytes, want the %d with the most", len(got), got[0].Bytes, got[len(got)-1].Bytes, trafficMaxHosts) + } +} + +func TestTrafficRefusesALinkInPlaceOfTheLog(t *testing.T) { + r, _ := trafficLog(t) + other := filepath.Join(t.TempDir(), "other") + if err := os.WriteFile(other, []byte(strings.Repeat("x 1\n", 100)), 0o644); err != nil { + t.Fatal(err) + } + if err := os.Symlink(other, r.path); err != nil { + t.Fatal(err) + } + r.limit = 1 + + if got := r.read(quiet); got != nil { + t.Errorf("read() through a link = %v, want nil", got) + } + if info, err := os.Stat(other); err != nil || info.Size() != 400 { + t.Errorf("the link's target = %v, %v, want it untouched", info, err) + } +}