[+] refactor psutil OS metric fetchers (#1269)

* [-] stop using deprecated `psutil.TimesStat.Total()`

* [+] make `GetGoPsutilDiskPG()` work for Windows and Darwin

* [+] improve `GetGoPsutilDiskPG()` and add tests

* [+] improve `GetGoPsutilCPU`

* [*] check for Windows abs path in `GetGoPsutilDiskPG()`

Author: pashagolub

internal/reaper/file.go

@@ -30,10 +30,17 @@ const (
 	metricPsutilMem         = "psutil_mem"
 )
 
-const (
-	sqlPgDirs = `select current_setting('data_directory') as dd, current_setting('log_directory') as ld`
-	sqlTsDirs = `select spcname::text as name, pg_catalog.pg_tablespace_location(oid) as location from pg_catalog.pg_tablespace where not spcname like any(array[E'pg\\_%'])`
-)
+const sqlPgDirs = `select name, path from 
+(values 
+	('data_directory', current_setting('data_directory')),
+	('pg_wal', current_setting('data_directory')||'/pg_wal'),
+	('log_directory', case 
+        when current_setting('log_directory') ~ '^(\w:)?\/.+' then current_setting('log_directory') 
+        else current_setting('data_directory') || '/' || current_setting('log_directory') 
+    end)) as d(name, path)
+union all
+select spcname::text, pg_catalog.pg_tablespace_location(oid)
+from pg_catalog.pg_tablespace where spcname !~ 'pg_.+'`
 
 var directlyFetchableOSMetrics = []string{metricPsutilCPU, metricPsutilDisk, metricPsutilDiskIoTotal, metricPsutilMem, metricCPULoad}
 
@@ -42,7 +49,7 @@ func IsDirectlyFetchableMetric(md *sources.SourceConn, metric string) bool {
 }
 
 func (r *Reaper) FetchStatsDirectlyFromOS(ctx context.Context, md *sources.SourceConn, metricName string) (*metrics.MeasurementEnvelope, error) {
-	var data, dataDirs, dataTblspDirs metrics.Measurements
+	var data, pgDirs metrics.Measurements
 	var err error
 
 	switch metricName {
@@ -51,13 +58,10 @@ func (r *Reaper) FetchStatsDirectlyFromOS(ctx context.Context, md *sources.Sourc
 	case metricPsutilCPU:
 		data, err = GetGoPsutilCPU(md.GetMetricInterval(metricName))
 	case metricPsutilDisk:
-		if dataDirs, err = QueryMeasurements(ctx, md, sqlPgDirs); err != nil {
-			return nil, err
-		}
-		if dataTblspDirs, err = QueryMeasurements(ctx, md, sqlTsDirs); err != nil {
+		if pgDirs, err = QueryMeasurements(ctx, md, sqlPgDirs); err != nil {
 			return nil, err
 		}
-		data, err = GetGoPsutilDiskPG(dataDirs, dataTblspDirs)
+		data, err = GetGoPsutilDiskPG(pgDirs)
 	case metricPsutilDiskIoTotal:
 		data, err = GetGoPsutilDiskTotals()
 	case metricPsutilMem:

internal/reaper/psutil.go

@@ -3,8 +3,6 @@ package reaper
 import (
 	"math"
 	"os"
-	"path"
-	"strings"
 	"sync"
 	"time"
 
@@ -20,31 +18,39 @@ var prevCPULoadTimeStatsLock sync.RWMutex
 var prevCPULoadTimeStats cpu.TimesStat
 var prevCPULoadTimestamp time.Time
 
+func init() {
+	// initialize the cache with current stats so the first call returns a meaningful delta
+	if probe, err := cpu.Times(false); err == nil {
+		prevCPULoadTimeStats = probe[0]
+		prevCPULoadTimestamp = time.Now()
+	}
+}
+
+// cpuTotal returns the total number of seconds across all CPU states.
+// Guest and GuestNice are intentionally excluded because on Linux they are
+// already counted within User and Nice respectively (/proc/stat semantics),
+// so including them would double-count and skew percentage calculations.
+func cpuTotal(c cpu.TimesStat) float64 {
+	return c.User + c.System + c.Idle + c.Nice + c.Iowait + c.Irq +
+		c.Softirq + c.Steal
+}
+
 func goPsutilCalcCPUUtilization(probe0, probe1 cpu.TimesStat) float64 {
-	return 100 - (100.0 * (probe1.Idle - probe0.Idle + probe1.Iowait - probe0.Iowait + probe1.Steal - probe0.Steal) / (probe1.Total() - probe0.Total()))
+	return 100 - (100.0 * (probe1.Idle - probe0.Idle + probe1.Iowait - probe0.Iowait + probe1.Steal - probe0.Steal) / (cpuTotal(probe1) - cpuTotal(probe0)))
 }
 
-// Simulates "psutil" metric output. Assumes the result from last call as input, otherwise uses a 1s measurement
-func GetGoPsutilCPU(interval float64) ([]map[string]any, error) {
+// GetGoPsutilCPU simulates "psutil" metric output. Assumes the result from last call as input
+func GetGoPsutilCPU(interval float64) (metrics.Measurements, error) {
 	prevCPULoadTimeStatsLock.RLock()
 	prevTime := prevCPULoadTimestamp
 	prevTimeStat := prevCPULoadTimeStats
 	prevCPULoadTimeStatsLock.RUnlock()
 
-	if prevTime.IsZero() || (time.Now().UnixNano()-prevTime.UnixNano()) < 1e9 { // give "short" stats on first run, based on a 1s probe
-		probe0, err := cpu.Times(false)
-		if err != nil {
-			return nil, err
-		}
-		prevTimeStat = probe0[0]
-		time.Sleep(1e9)
-	}
-
 	curCallStats, err := cpu.Times(false)
 	if err != nil {
 		return nil, err
 	}
-	if prevTime.IsZero() || time.Now().UnixNano()-prevTime.UnixNano() < 1e9 || time.Now().Unix()-prevTime.Unix() >= int64(interval) {
+	if time.Since(prevTime) >= time.Duration(float64(time.Second)*interval) {
 		prevCPULoadTimeStatsLock.Lock() // update the cache
 		prevCPULoadTimeStats = curCallStats[0]
 		prevCPULoadTimestamp = time.Now()
@@ -67,17 +73,18 @@ func GetGoPsutilCPU(interval float64) ([]map[string]any, error) {
 	retMap["load_1m"] = math.Round(100*la.Load1) / 100
 	retMap["load_5m_norm"] = math.Round(100*la.Load5/float64(cpus)) / 100
 	retMap["load_5m"] = math.Round(100*la.Load5) / 100
-	retMap["user"] = math.Round(10000.0*(curCallStats[0].User-prevTimeStat.User)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-	retMap["system"] = math.Round(10000.0*(curCallStats[0].System-prevTimeStat.System)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-	retMap["idle"] = math.Round(10000.0*(curCallStats[0].Idle-prevTimeStat.Idle)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-	retMap["iowait"] = math.Round(10000.0*(curCallStats[0].Iowait-prevTimeStat.Iowait)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-	retMap["irqs"] = math.Round(10000.0*(curCallStats[0].Irq-prevTimeStat.Irq+curCallStats[0].Softirq-prevTimeStat.Softirq)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-	retMap["other"] = math.Round(10000.0*(curCallStats[0].Steal-prevTimeStat.Steal+curCallStats[0].Guest-prevTimeStat.Guest+curCallStats[0].GuestNice-prevTimeStat.GuestNice)/(curCallStats[0].Total()-prevTimeStat.Total())) / 100
-
-	return []map[string]any{retMap}, nil
+	totalDiff := cpuTotal(curCallStats[0]) - cpuTotal(prevTimeStat)
+	retMap["user"] = math.Round(10000.0*(curCallStats[0].User-prevTimeStat.User)/totalDiff) / 100
+	retMap["system"] = math.Round(10000.0*(curCallStats[0].System-prevTimeStat.System)/totalDiff) / 100
+	retMap["idle"] = math.Round(10000.0*(curCallStats[0].Idle-prevTimeStat.Idle)/totalDiff) / 100
+	retMap["iowait"] = math.Round(10000.0*(curCallStats[0].Iowait-prevTimeStat.Iowait)/totalDiff) / 100
+	retMap["irqs"] = math.Round(10000.0*(curCallStats[0].Irq-prevTimeStat.Irq+curCallStats[0].Softirq-prevTimeStat.Softirq)/totalDiff) / 100
+	retMap["other"] = math.Round(10000.0*(curCallStats[0].Steal-prevTimeStat.Steal+curCallStats[0].Guest-prevTimeStat.Guest+curCallStats[0].GuestNice-prevTimeStat.GuestNice)/totalDiff) / 100
+
+	return metrics.Measurements{retMap}, nil
 }
 
-func GetGoPsutilMem() ([]map[string]any, error) {
+func GetGoPsutilMem() (metrics.Measurements, error) {
 	vm, err := mem.VirtualMemory()
 	if err != nil {
 		return nil, err
@@ -95,10 +102,10 @@ func GetGoPsutilMem() ([]map[string]any, error) {
 	retMap["swap_free"] = int64(vm.SwapFree)
 	retMap["swap_percent"] = math.Round(100*float64(vm.SwapCached)/float64(vm.SwapTotal)) / 100
 
-	return []map[string]any{retMap}, nil
+	return metrics.Measurements{retMap}, nil
 }
 
-func GetGoPsutilDiskTotals() ([]map[string]any, error) {
+func GetGoPsutilDiskTotals() (metrics.Measurements, error) {
 	d, err := disk.IOCounters()
 	if err != nil {
 		return nil, err
@@ -119,10 +126,10 @@ func GetGoPsutilDiskTotals() ([]map[string]any, error) {
 	retMap["read_count"] = reads
 	retMap["write_count"] = writes
 
-	return []map[string]any{retMap}, nil
+	return metrics.Measurements{retMap}, nil
 }
 
-func GetLoadAvgLocal() ([]map[string]any, error) {
+func GetLoadAvgLocal() (metrics.Measurements, error) {
 	la, err := load.Avg()
 	if err != nil {
 		return nil, err
@@ -133,122 +140,52 @@ func GetLoadAvgLocal() ([]map[string]any, error) {
 	row["load_5min"] = la.Load5
 	row["load_15min"] = la.Load15
 
-	return []map[string]any{row}, nil
+	return metrics.Measurements{row}, nil
 }
 
 func CheckFolderExistsAndReadable(path string) bool {
+	if path == "" {
+		return false
+	}
 	_, err := os.ReadDir(path)
 	return err == nil
 }
 
-func GetGoPsutilDiskPG(DataDirs, TblspaceDirs []map[string]any) ([]map[string]any, error) {
-	var ddDevice, ldDevice, walDevice uint64
-
-	data := DataDirs
-
-	dataDirPath := data[0]["dd"].(string)
-	ddUsage, err := disk.Usage(dataDirPath)
-	if err != nil {
-		return nil, err
-	}
-
-	retRows := make([]map[string]any, 0)
+func GetGoPsutilDiskPG(pgDirs metrics.Measurements) (metrics.Measurements, error) {
+	usageCache := make(map[uint64]*disk.UsageStat)
+	retRows := make(metrics.Measurements, 0)
 	epochNs := time.Now().UnixNano()
-	dd := metrics.NewMeasurement(epochNs)
-	dd["tag_dir_or_tablespace"] = "data_directory"
-	dd["tag_path"] = dataDirPath
-	dd["total"] = float64(ddUsage.Total)
-	dd["used"] = float64(ddUsage.Used)
-	dd["free"] = float64(ddUsage.Free)
-	dd["percent"] = math.Round(100*ddUsage.UsedPercent) / 100
-	retRows = append(retRows, dd)
-
-	ddDevice, err = GetPathUnderlyingDeviceID(dataDirPath)
-	if err != nil {
-		return nil, err
-	}
 
-	logDirPath := data[0]["ld"].(string)
-	if !strings.HasPrefix(logDirPath, "/") {
-		logDirPath = path.Join(dataDirPath, logDirPath)
-	}
-	if len(logDirPath) > 0 && CheckFolderExistsAndReadable(logDirPath) { // syslog etc considered out of scope
-		ldDevice, err = GetPathUnderlyingDeviceID(logDirPath)
-		if err != nil {
-			return nil, err
-		}
-		if ldDevice != ddDevice { // no point to report same data in case of single folder configuration
-			ld := metrics.NewMeasurement(epochNs)
-			ldUsage, err := disk.Usage(logDirPath)
-			if err != nil {
-				return nil, err
-			}
+	for _, row := range pgDirs {
+		path := row["path"].(string)
+		name := row["name"].(string)
 
-			ld["tag_dir_or_tablespace"] = "log_directory"
-			ld["tag_path"] = logDirPath
-			ld["total"] = float64(ldUsage.Total)
-			ld["used"] = float64(ldUsage.Used)
-			ld["free"] = float64(ldUsage.Free)
-			ld["percent"] = math.Round(100*ldUsage.UsedPercent) / 100
-			retRows = append(retRows, ld)
+		if !CheckFolderExistsAndReadable(path) { // syslog etc considered out of scope
+			continue
 		}
-	}
-
-	var walDirPath string
-	if CheckFolderExistsAndReadable(path.Join(dataDirPath, "pg_wal")) {
-		walDirPath = path.Join(dataDirPath, "pg_wal")
-	}
 
-	if len(walDirPath) > 0 {
-		walDevice, err = GetPathUnderlyingDeviceID(walDirPath)
+		devID, err := GetPathUnderlyingDeviceID(path)
 		if err != nil {
 			return nil, err
 		}
 
-		if walDevice != ddDevice || walDevice != ldDevice { // no point to report same data in case of single folder configuration
-			walUsage, err := disk.Usage(walDirPath)
+		usage, ok := usageCache[devID]
+		if !ok {
+			usage, err = disk.Usage(path)
 			if err != nil {
 				return nil, err
 			}
-
-			wd := metrics.NewMeasurement(epochNs)
-			wd["tag_dir_or_tablespace"] = "pg_wal"
-			wd["tag_path"] = walDirPath
-			wd["total"] = float64(walUsage.Total)
-			wd["used"] = float64(walUsage.Used)
-			wd["free"] = float64(walUsage.Free)
-			wd["percent"] = math.Round(100*walUsage.UsedPercent) / 100
-			retRows = append(retRows, wd)
+			usageCache[devID] = usage
 		}
-	}
 
-	data = TblspaceDirs
-	if len(data) > 0 {
-		for _, row := range data {
-			tsPath := row["location"].(string)
-			tsName := row["name"].(string)
-
-			tsDevice, err := GetPathUnderlyingDeviceID(tsPath)
-			if err != nil {
-				return nil, err
-			}
-
-			if tsDevice == ddDevice || tsDevice == ldDevice || tsDevice == walDevice {
-				continue
-			}
-			tsUsage, err := disk.Usage(tsPath)
-			if err != nil {
-				return nil, err
-			}
-			ts := metrics.NewMeasurement(epochNs)
-			ts["tag_dir_or_tablespace"] = tsName
-			ts["tag_path"] = tsPath
-			ts["total"] = float64(tsUsage.Total)
-			ts["used"] = float64(tsUsage.Used)
-			ts["free"] = float64(tsUsage.Free)
-			ts["percent"] = math.Round(100*tsUsage.UsedPercent) / 100
-			retRows = append(retRows, ts)
-		}
+		m := metrics.NewMeasurement(epochNs)
+		m["tag_dir_or_tablespace"] = name
+		m["tag_path"] = path
+		m["total"] = float64(usage.Total)
+		m["used"] = float64(usage.Used)
+		m["free"] = float64(usage.Free)
+		m["percent"] = math.Round(100*usage.UsedPercent) / 100
+		retRows = append(retRows, m)
 	}
 
 	return retRows, nil

internal/reaper/psutil_darwin.go

@@ -1,9 +1,11 @@
 package reaper
 
-import "errors"
+import "hash/fnv"
 
-var ErrNotImplemented = errors.New("not implemented")
-
-func GetPathUnderlyingDeviceID(path string) (uint64, error) {
-	return 0, ErrNotImplemented
+// GetPathUnderlyingDeviceID on Darwin falls back to an FNV hash of the path.
+// Identical paths still get the same ID, so deduplication works correctly.
+func GetPathUnderlyingDeviceID(p string) (uint64, error) {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(p))
+	return h.Sum64(), nil
 }