fix(agent): measure first disk I/O sample of an interval from the full baseline (#2420)

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

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

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