Record IO MiB read/write to client resource usage records

fleetspeak/src/client/internal/monitoring/resource_usage_fetcher.go

@@ -39,6 +39,12 @@ type ResourceUsage struct {
 
 	// Resident set size for a process, in bytes.
 	ResidentMemory int64
+
+	// Client IO data read so far, in bytes.
+	IORead int64
+
+	// Client IO data written so far, in bytes.
+	IOWrite int64
 }
 
 // ResourceUsageFetcher obtains resource-usage data for a process from the OS.
@@ -48,7 +54,8 @@ type ResourceUsageFetcher struct{}
 // information from exec.Cmd.ProcessState. NOTE that this is only possible
 // after the process has finished and has been waited for, and will most
 // probably panic otherwise.
-// This function doesn't fill in ResourceUsage.ResidentMemory.
+// This function doesn't fill in ResourceUsage.ResidentMemory, ResourceUsage.IORead
+// and ResourceUsage.IOWrite.
 func (f ResourceUsageFetcher) ResourceUsageFromFinishedCmd(cmd *exec.Cmd) *ResourceUsage {
 	return &ResourceUsage{
 		Timestamp:       time.Now(),
@@ -78,11 +85,17 @@ func (f ResourceUsageFetcher) ResourceUsageForPID(pid int) (*ResourceUsage, erro
 		return nil, err
 	}
 
+	ioCounters, err := process.IOCounters()
+	if err != nil {
+		return nil, err
+	}
 	return &ResourceUsage{
 		Timestamp:       timestamp,
 		UserCPUMillis:   times.User * 1e3,
 		SystemCPUMillis: times.System * 1e3,
 		ResidentMemory:  int64(memoryInfo.RSS),
+		IORead:          int64(ioCounters.ReadBytes),
+		IOWrite:         int64(ioCounters.WriteBytes),
 	}, nil
 }
 

fleetspeak/src/client/internal/monitoring/resource_usage_fetcher_unix.go

@@ -25,6 +25,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/shirou/gopsutil/process"
 )
 
 // TODO: Support monitoring on other platforms.
@@ -42,6 +44,12 @@ type ResourceUsage struct {
 
 	// Resident set size for a process.
 	ResidentMemory int64
+
+	// Client IO data read so far, in bytes.
+	IORead int64
+
+	// Client IO data written so far, in bytes.
+	IOWrite int64
 }
 
 // ResourceUsageFetcher obtains resource-usage data for a process from the OS.
@@ -125,11 +133,22 @@ func (f ResourceUsageFetcher) ResourceUsageForPID(pid int) (*ResourceUsage, erro
 		return nil, fmt.Errorf("error while parsing resident from %s: %v", statmFilename, err)
 	}
 
+	process, err := process.NewProcess(int32(pid))
+	if err != nil {
+		return nil, err
+	}
+
+	ioCounters, err := process.IOCounters()
+	if err != nil {
+		return nil, err
+	}
 	return &ResourceUsage{
 		Timestamp:       timestamp,
 		UserCPUMillis:   float64((utime + cutime) * 10), // Assume rate of 100 ticks/second
 		SystemCPUMillis: float64((stime + cstime) * 10), // Assume rate of 100 ticks/second
 		ResidentMemory:  resident * pageSize,
+		IORead:          int64(ioCounters.ReadBytes),
+		IOWrite:         int64(ioCounters.WriteBytes),
 	}, nil
 }
 

fleetspeak/src/client/internal/monitoring/resource_usage_monitor.go

@@ -63,6 +63,10 @@ func AggregateResourceUsage(prevRU *ResourceUsage, currRU *ResourceUsage, numRUC
 		}
 		aggRU.MeanResidentMemory = float64(currRU.ResidentMemory) / float64(numRUCalls)
 		aggRU.MaxResidentMemory = currRU.ResidentMemory
+		aggRU.MeanIoRead = float64(currRU.IORead) / float64(numRUCalls)
+		aggRU.MaxIoRead = currRU.IORead
+		aggRU.MeanIoWrite = float64(currRU.IOWrite) / float64(numRUCalls)
+		aggRU.MaxIoWrite = currRU.IOWrite
 		return nil
 	}
 
@@ -80,6 +84,10 @@ func AggregateResourceUsage(prevRU *ResourceUsage, currRU *ResourceUsage, numRUC
 		return nil
 	}
 
+	if err := aggregateIOResourceUsage(currRU, numRUCalls, aggRU); err != nil {
+		return err
+	}
+
 	return aggregateMemoryResourceUsage(currRU, numRUCalls, aggRU)
 }
 
@@ -119,6 +127,18 @@ func aggregateMemoryResourceUsage(currRU *ResourceUsage, numRUCalls int, aggRU *
 	return nil
 }
 
+func aggregateIOResourceUsage(currRU *ResourceUsage, numRUCalls int, aggRU *mpb.AggregatedResourceUsage) error {
+	aggRU.MeanIoRead += float64(currRU.IORead) / float64(numRUCalls)
+	if currRU.IORead > aggRU.MaxIoRead {
+		aggRU.MaxIoRead = currRU.IORead
+	}
+	aggRU.MeanIoWrite += float64(currRU.IOWrite) / float64(numRUCalls)
+	if currRU.IOWrite > aggRU.MaxIoWrite {
+		aggRU.MaxIoWrite = currRU.IOWrite
+	}
+	return nil
+}
+
 // AggregateResourceUsageForFinishedCmd computes resource-usage for a finished process, given
 // resource-usage before and after the process ran.
 func AggregateResourceUsageForFinishedCmd(initialRU, finalRU *ResourceUsage) (*mpb.AggregatedResourceUsage, error) {
@@ -132,13 +152,21 @@ func AggregateResourceUsageForFinishedCmd(initialRU, finalRU *ResourceUsage) (*m
 		return nil, err
 	}
 
-	// If this field is untouched, we have not aggregated memory resource usage
+	// If those field are untouched, we have not aggregated memory and IO resources usage
 	// for this process yet. We fill it in with what we have.
 	// TODO
 	if aggRU.MaxResidentMemory == 0 {
 		aggRU.MeanResidentMemory = float64(initialRU.ResidentMemory)
 		aggRU.MaxResidentMemory = initialRU.ResidentMemory
 	}
+	if aggRU.MaxIoRead == 0 {
+		aggRU.MeanIoRead = float64(initialRU.IORead)
+		aggRU.MaxIoRead = initialRU.IORead
+	}
+	if aggRU.MaxIoWrite == 0 {
+		aggRU.MeanIoWrite = float64(initialRU.IOWrite)
+		aggRU.MaxIoWrite = initialRU.IOWrite
+	}
 
 	return &aggRU, nil
 }

fleetspeak/src/common/proto/fleetspeak_monitoring/resource.proto

@@ -14,6 +14,10 @@ message AggregatedResourceUsage {
   double max_system_cpu_rate = 4;
   double mean_resident_memory = 5;
   int64 max_resident_memory = 6;
+  double mean_io_read = 7;
+  int64 max_io_read = 8;
+  double mean_io_write = 9;
+  int64 max_io_write = 10;
 }
 
 // A fleetspeak.Message with message type "ResourceUsage" is sent regularly by

fleetspeak/src/server/dbtesting/clientstore_suite.go

@@ -360,6 +360,8 @@ func fetchResourceUsageRecordsTest(t *testing.T, ds db.Store) {
 	}
 
 	meanRAM, maxRAM := 190, 200
+	meanIoReadData, maxIoReadData := 300, 310
+	meanIoWriteData, maxIoWriteData := 400, 410
 	rud := mpb.ResourceUsageData{
 		Scope:             "test-scope",
 		Pid:               1234,
@@ -373,6 +375,10 @@ func fetchResourceUsageRecordsTest(t *testing.T, ds db.Store) {
 			MaxSystemCpuRate:   80.0,
 			MeanResidentMemory: float64(meanRAM) * 1024 * 1024,
 			MaxResidentMemory:  int64(maxRAM) * 1024 * 1024,
+			MeanIoRead:         float64(meanIoReadData) * 1024 * 1024,
+			MaxIoRead:          int64(maxIoReadData) * 1024 * 1024,
+			MeanIoWrite:        float64(meanIoWriteData) * 1024 * 1024,
+			MaxIoWrite:         int64(maxIoWriteData) * 1024 * 1024,
 		},
 	}
 
@@ -414,6 +420,10 @@ func fetchResourceUsageRecordsTest(t *testing.T, ds db.Store) {
 		MaxSystemCpuRate:      80.0,
 		MeanResidentMemoryMib: int32(meanRAM),
 		MaxResidentMemoryMib:  int32(maxRAM),
+		MeanIoReadMib:         int32(meanIoReadData),
+		MaxIoReadMib:          int32(maxIoReadData),
+		MeanIoWriteMib:        int32(meanIoWriteData),
+		MaxIoWriteMib:         int32(maxIoWriteData),
 	}
 
 	if got, want := record, expected; !proto.Equal(got, want) {