Merge pull request #308 from zouyee/main

support psi

Makefile

@@ -26,4 +26,4 @@ cgutil:
 proto:
 	protobuild --quiet ${PACKAGES}
 	# Keep them Go-idiomatic and backward-compatible with the gogo/protobuf era.
-	go-fix-acronym -w -a '(Cpu|Tcp|Rss)' $(shell find cgroup1/stats/ cgroup2/stats/ -name '*.pb.go')
+	go-fix-acronym -w -a '(Cpu|Tcp|Rss|Psi)' $(shell find cgroup1/stats/ cgroup2/stats/ -name '*.pb.go')

cgroup2/manager.go

@@ -559,6 +559,7 @@ func (c *Manager) Stat() (*stats.Metrics, error) {
 		NrPeriods:     out["nr_periods"],
 		NrThrottled:   out["nr_throttled"],
 		ThrottledUsec: out["throttled_usec"],
+		PSI:           getStatPSIFromFile(filepath.Join(c.path, "cpu.pressure")),
 	}
 	metrics.Memory = &stats.MemoryStat{
 		Anon:                  out["anon"],
@@ -598,6 +599,7 @@ func (c *Manager) Stat() (*stats.Metrics, error) {
 		SwapUsage:             getStatFileContentUint64(filepath.Join(c.path, "memory.swap.current")),
 		SwapLimit:             getStatFileContentUint64(filepath.Join(c.path, "memory.swap.max")),
 		SwapMaxUsage:          getStatFileContentUint64(filepath.Join(c.path, "memory.swap.peak")),
+		PSI:                   getStatPSIFromFile(filepath.Join(c.path, "memory.pressure")),
 	}
 	if len(memoryEvents) > 0 {
 		metrics.MemoryEvents = &stats.MemoryEvents{
@@ -608,7 +610,10 @@ func (c *Manager) Stat() (*stats.Metrics, error) {
 			OomKill: memoryEvents["oom_kill"],
 		}
 	}
-	metrics.Io = &stats.IOStat{Usage: readIoStats(c.path)}
+	metrics.Io = &stats.IOStat{
+		Usage: readIoStats(c.path),
+		PSI:   getStatPSIFromFile(filepath.Join(c.path, "io.pressure")),
+	}
 	metrics.Rdma = &stats.RdmaStat{
 		Current: rdmaStats(filepath.Join(c.path, "rdma.current")),
 		Limit:   rdmaStats(filepath.Join(c.path, "rdma.max")),

cgroup2/manager_test.go

@@ -296,6 +296,39 @@ func TestCgroupType(t *testing.T) {
 	require.Equal(t, cgType, Threaded)
 }
 
+func TestCgroupv2PSIStats(t *testing.T) {
+	checkCgroupMode(t)
+	group := "/psi-test-cg"
+	groupPath := fmt.Sprintf("%s-%d", group, os.Getpid())
+	res := Resources{}
+	c, err := NewManager(defaultCgroup2Path, groupPath, &res)
+	require.NoError(t, err, "failed to init new cgroup manager")
+	t.Cleanup(func() {
+		os.Remove(c.path)
+	})
+
+	stats, err := c.Stat()
+	require.NoError(t, err, "failed to get cgroup stats")
+	if stats.CPU.PSI == nil || stats.Memory.PSI == nil || stats.Io.PSI == nil {
+		t.Error("expected psi not nil but got nil")
+	}
+}
+
+func TestSystemdCgroupPSIController(t *testing.T) {
+	checkCgroupMode(t)
+	group := fmt.Sprintf("testing-psi-%d.scope", os.Getpid())
+	pid := os.Getpid()
+	res := Resources{}
+	c, err := NewSystemd("", group, pid, &res)
+	require.NoError(t, err, "failed to init new cgroup systemd manager")
+
+	stats, err := c.Stat()
+	require.NoError(t, err, "failed to get cgroup stats")
+	if stats.CPU.PSI == nil || stats.Memory.PSI == nil || stats.Io.PSI == nil {
+		t.Error("expected psi not nil but got nil")
+	}
+}
+
 func BenchmarkStat(b *testing.B) {
 	checkCgroupMode(b)
 	group := "/stat-test-cg"

cgroup2/stats/metrics.pb.txt

@@ -60,6 +60,56 @@ file {
       json_name: "memoryEvents"
     }
   }
+  message_type {
+    name: "PSIData"
+    field {
+      name: "avg10"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_DOUBLE
+      json_name: "avg10"
+    }
+    field {
+      name: "avg60"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_DOUBLE
+      json_name: "avg60"
+    }
+    field {
+      name: "avg300"
+      number: 3
+      label: LABEL_OPTIONAL
+      type: TYPE_DOUBLE
+      json_name: "avg300"
+    }
+    field {
+      name: "total"
+      number: 4
+      label: LABEL_OPTIONAL
+      type: TYPE_UINT64
+      json_name: "total"
+    }
+  }
+  message_type {
+    name: "PSIStats"
+    field {
+      name: "some"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".io.containerd.cgroups.v2.PSIData"
+      json_name: "some"
+    }
+    field {
+      name: "full"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".io.containerd.cgroups.v2.PSIData"
+      json_name: "full"
+    }
+  }
   message_type {
     name: "PidsStat"
     field {
@@ -121,6 +171,14 @@ file {
       type: TYPE_UINT64
       json_name: "throttledUsec"
     }
+    field {
+      name: "psi"
+      number: 7
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".io.containerd.cgroups.v2.PSIStats"
+      json_name: "psi"
+    }
   }
   message_type {
     name: "MemoryStat"
@@ -383,6 +441,14 @@ file {
       type: TYPE_UINT64
       json_name: "swapMaxUsage"
     }
+    field {
+      name: "psi"
+      number: 38
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".io.containerd.cgroups.v2.PSIStats"
+      json_name: "psi"
+    }
   }
   message_type {
     name: "MemoryEvents"
@@ -475,6 +541,14 @@ file {
       type_name: ".io.containerd.cgroups.v2.IOEntry"
       json_name: "usage"
     }
+    field {
+      name: "psi"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".io.containerd.cgroups.v2.PSIStats"
+      json_name: "psi"
+    }
   }
   message_type {
     name: "IOEntry"

cgroup2/stats/metrics.proto

@@ -14,6 +14,18 @@ message Metrics {
 	MemoryEvents memory_events = 8;
 }
 
+message PSIData {
+	double avg10 = 1;
+	double avg60 = 2;
+	double avg300 = 3;
+	uint64 total = 4;
+}
+
+message PSIStats {
+	PSIData some = 1;
+	PSIData full = 2;
+}
+
 message PidsStat {
 	uint64 current = 1;
 	uint64 limit = 2;
@@ -26,6 +38,7 @@ message CPUStat {
 	uint64 nr_periods = 4;
 	uint64 nr_throttled = 5;
 	uint64 throttled_usec = 6;
+	PSIStats psi = 7;
 }
 
 message MemoryStat {
@@ -66,6 +79,7 @@ message MemoryStat {
 	uint64 swap_limit = 35;
 	uint64 max_usage = 36;
 	uint64 swap_max_usage = 37;
+	PSIStats psi = 38;
 }
 
 message MemoryEvents {
@@ -89,6 +103,7 @@ message RdmaEntry {
 
 message IOStat {
 	repeated IOEntry usage = 1;
+	PSIStats psi = 2;
 }
 
 message IOEntry {

cgroup2/utils.go

@@ -479,6 +479,74 @@ func getHugePageSizeFromFilenames(fileNames []string) ([]string, error) {
 	return pageSizes, warn
 }
 
+func getStatPSIFromFile(path string) *stats.PSIStats {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil
+	}
+	defer f.Close()
+
+	psistats := &stats.PSIStats{}
+	sc := bufio.NewScanner(f)
+	for sc.Scan() {
+		parts := strings.Fields(sc.Text())
+		var pv *stats.PSIData
+		switch parts[0] {
+		case "some":
+			psistats.Some = &stats.PSIData{}
+			pv = psistats.Some
+		case "full":
+			psistats.Full = &stats.PSIData{}
+			pv = psistats.Full
+		}
+		if pv != nil {
+			err = parsePSIData(parts[1:], pv)
+			if err != nil {
+				logrus.Errorf("failed to read file %s: %v", path, err)
+				return nil
+			}
+		}
+	}
+
+	if err := sc.Err(); err != nil {
+		logrus.Errorf("unable to parse PSI data: %v", err)
+		return nil
+	}
+	return psistats
+}
+
+func parsePSIData(psi []string, data *stats.PSIData) error {
+	for _, f := range psi {
+		kv := strings.SplitN(f, "=", 2)
+		if len(kv) != 2 {
+			return fmt.Errorf("invalid PSI data: %q", f)
+		}
+		var pv *float64
+		switch kv[0] {
+		case "avg10":
+			pv = &data.Avg10
+		case "avg60":
+			pv = &data.Avg60
+		case "avg300":
+			pv = &data.Avg300
+		case "total":
+			v, err := strconv.ParseUint(kv[1], 10, 64)
+			if err != nil {
+				return fmt.Errorf("invalid %s PSI value: %w", kv[0], err)
+			}
+			data.Total = v
+		}
+		if pv != nil {
+			v, err := strconv.ParseFloat(kv[1], 64)
+			if err != nil {
+				return fmt.Errorf("invalid %s PSI value: %w", kv[0], err)
+			}
+			*pv = v
+		}
+	}
+	return nil
+}
+
 func getSubreaper() (int, error) {
 	var i uintptr
 	if err := unix.Prctl(unix.PR_GET_CHILD_SUBREAPER, uintptr(unsafe.Pointer(&i)), 0, 0, 0); err != nil {

cgroup2/utils_test.go

@@ -17,9 +17,13 @@
 package cgroup2
 
 import (
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
+	"github.com/containerd/cgroups/v3/cgroup2/stats"
+
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/stretchr/testify/assert"
 )
@@ -41,6 +45,36 @@ func TestParseCgroupFromReader(t *testing.T) {
 	}
 }
 
+func TestParseStatCPUPSI(t *testing.T) {
+	const examplePSIData = `some avg10=1.71 avg60=2.36 avg300=2.57 total=230548833
+full avg10=1.00 avg60=1.01 avg300=1.00 total=157622356`
+
+	fakeCgroupDir := t.TempDir()
+	statPath := filepath.Join(fakeCgroupDir, "cpu.pressure")
+
+	if err := os.WriteFile(statPath, []byte(examplePSIData), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	st := getStatPSIFromFile(filepath.Join(fakeCgroupDir, "cpu.pressure"))
+	expected := stats.PSIStats{
+		Some: &stats.PSIData{
+			Avg10:  1.71,
+			Avg60:  2.36,
+			Avg300: 2.57,
+			Total:  230548833,
+		},
+		Full: &stats.PSIData{
+			Avg10:  1.00,
+			Avg60:  1.01,
+			Avg300: 1.00,
+			Total:  157622356,
+		},
+	}
+	assert.Equal(t, &st.Some, &expected.Some)
+	assert.Equal(t, &st.Full, &expected.Full)
+}
+
 func TestToResources(t *testing.T) {
 	var (
 		quota  int64  = 8000