diff --git a/agent/agent.go b/agent/agent.go index b01fe4512..a859ea723 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -29,6 +29,7 @@ type Agent struct { fsNames []string // List of filesystem device names being monitored fsStats map[string]*system.FsStats // Keeps track of disk stats for each filesystem diskPrev map[uint16]map[string]prevDisk // Previous disk I/O counters per cache interval + diskBaseline map[string]prevDisk // Latest disk I/O counters of any interval, seeds a new interval diskUsageCacheDuration time.Duration // How long to cache disk usage (to avoid waking sleeping disks) lastDiskUsageUpdate time.Time // Last time disk usage was collected netInterfaces map[string]struct{} // Stores all valid network interfaces diff --git a/agent/disk.go b/agent/disk.go index 732872e07..52f018066 100644 --- a/agent/disk.go +++ b/agent/disk.go @@ -599,6 +599,7 @@ func (a *Agent) initializeDiskIoStats(diskIoCounters map[string]disk.IOCountersS stats.Time = now stats.TotalRead = d.ReadBytes stats.TotalWrite = d.WriteBytes + a.setDiskBaseline(device, prevDiskFromCounter(d, now)) // add to list of valid io device names a.fsNames = append(a.fsNames, device) } @@ -681,19 +682,9 @@ func (a *Agent) updateDiskIo(cacheTimeMs uint16, systemStats *system.Stats) { // Previous snapshot for this interval and device prev, hasPrev := a.diskPrev[cacheTimeMs][name] if !hasPrev { - // Seed from agent-level fsStats if present, else seed from current - prev = prevDisk{ - readBytes: stats.TotalRead, - writeBytes: stats.TotalWrite, - readTime: d.ReadTime, - writeTime: d.WriteTime, - ioTime: d.IoTime, - weightedIO: d.WeightedIO, - readCount: d.ReadCount, - writeCount: d.WriteCount, - at: stats.Time, - } - if prev.at.IsZero() { + // Seed from the latest counters of any interval, else seed from current + prev, hasPrev = a.diskBaseline[name] + if !hasPrev { prev = prevDiskFromCounter(d, now) } } @@ -752,6 +743,7 @@ func (a *Agent) updateDiskIo(cacheTimeMs uint16, systemStats *system.Stats) { } // Update global fsStats baseline for cross-interval correctness + a.setDiskBaseline(name, prevDiskFromCounter(d, now)) stats.Time = now stats.TotalRead = d.ReadBytes stats.TotalWrite = d.WriteBytes @@ -784,6 +776,15 @@ func (a *Agent) updateDiskIo(cacheTimeMs uint16, systemStats *system.Stats) { } } +// setDiskBaseline stores the latest counters of a device. A cache interval +// without its own snapshot measures its first sample from them. +func (a *Agent) setDiskBaseline(name string, d prevDisk) { + if a.diskBaseline == nil { + a.diskBaseline = make(map[string]prevDisk) + } + a.diskBaseline[name] = d +} + // ioTimeDelta returns the increase of a cumulative millisecond counter from // the disk I/O stats. Linux prints these fields of /proc/diskstats as 32-bit // unsigned ints, so they wrap to zero at 2^32. A busy disk reaches that in diff --git a/agent/disk_io_linux_test.go b/agent/disk_io_linux_test.go index 90e462a02..546b6f126 100644 --- a/agent/disk_io_linux_test.go +++ b/agent/disk_io_linux_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/henrygd/beszel/internal/entities/system" + "github.com/shirou/gopsutil/v4/disk" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -73,3 +74,52 @@ func TestUpdateDiskIoTimeCounterWrap(t *testing.T) { }) } } + +// The first sample of a cache interval has no snapshot of its own. It must +// measure the time counters from the same baseline as the byte counters. +func TestUpdateDiskIoFirstSampleOfInterval(t *testing.T) { + dir := t.TempDir() + t.Setenv("HOST_PROC", dir) + t.Setenv("HOST_SYS", dir) + t.Setenv("HOST_DEV", dir) + t.Setenv("HOST_RUN", dir) + writeDiskstats := func(line string) { + require.NoError(t, os.WriteFile(filepath.Join(dir, "diskstats"), []byte(line), 0o644)) + } + + writeDiskstats(" 8 0 sda 1000 0 20000 900 500 0 10000 700 0 400 0\n") + counters, err := disk.IOCounters("sda") + require.NoError(t, err) + + fs := &system.FsStats{Root: true} + a := &Agent{ + fsStats: map[string]*system.FsStats{"sda": fs}, + diskPrev: map[uint16]map[string]prevDisk{}, + } + a.initializeDiskIoStats(counters) + + // updateDiskIo skips samples less than 100ms apart. + time.Sleep(150 * time.Millisecond) + + // Deltas: read 300ms / 10 ops, write 400ms / 20 ops, io time 1200ms, weighted io 3000ms. + writeDiskstats(" 8 0 sda 1010 0 21200 1200 520 0 10400 1100 0 1600 3000\n") + var stats system.Stats + a.updateDiskIo(60000, &stats) + + require.NotZero(t, fs.DiskReadBytes, "bytes are measured from the baseline") + for i := range 3 { + assert.NotZero(t, fs.DiskIoStats[i], "DiskIoStats[%d]", i) + } + assert.InDelta(t, 30, fs.DiskIoStats[3], 0.01, "r_await") + assert.InDelta(t, 20, fs.DiskIoStats[4], 0.01, "w_await") + assert.NotZero(t, fs.DiskIoStats[5], "weighted io") + + // A second interval starts from the latest counters, not from the ones at start. + time.Sleep(150 * time.Millisecond) + // Deltas: read 100ms / 10 ops, write 100ms / 20 ops. + writeDiskstats(" 8 0 sda 1020 0 22400 1300 540 0 10800 1200 0 1800 3500\n") + a.updateDiskIo(1000, &stats) + + assert.InDelta(t, 10, fs.DiskIoStats[3], 0.01, "r_await") + assert.InDelta(t, 5, fs.DiskIoStats[4], 0.01, "w_await") +}