[pull] main from containerd:main

docs/runtime-v2.md

@@ -559,7 +559,7 @@ func copy(wg *sync.WaitGroup, r io.Reader, pri journal.Priority, vars map[string
 
 #### Unsupported rpcs
 
-If a shim does not or cannot implement an rpc call, it MUST return a `github.com/containerd/containerd/errdefs.ErrNotImplemented` error.
+If a shim does not or cannot implement an rpc call, it MUST return a `github.com/containerd/errdefs.ErrNotImplemented` error.
 
 #### Debugging and Shim Logs
 

internal/cri/server/container_stats_list.go

@@ -428,13 +428,13 @@ func (c *criService) linuxContainerMetrics(
 		return containerStats{}, fmt.Errorf("unexpected metrics type: %T from %s", taskMetricsAny, reflect.TypeOf(taskMetricsAny).Elem().PkgPath())
 	}
 
-	cpuStats, err := c.cpuContainerStats(meta.ID, false /* isSandbox */, metrics, protobuf.FromTimestamp(stats.Timestamp))
+	cpuStats, err := c.cpuContainerStats(metrics, protobuf.FromTimestamp(stats.Timestamp))
 	if err != nil {
 		return containerStats{}, fmt.Errorf("failed to obtain cpu stats: %w", err)
 	}
 	cs.Cpu = cpuStats
 
-	memoryStats, err := c.memoryContainerStats(meta.ID, metrics, protobuf.FromTimestamp(stats.Timestamp))
+	memoryStats, err := c.memoryContainerStats(metrics, protobuf.FromTimestamp(stats.Timestamp))
 	if err != nil {
 		return containerStats{}, fmt.Errorf("failed to obtain memory stats: %w", err)
 	}
@@ -527,7 +527,7 @@ func convertCg2PSIToCRI(psi *cg2.PSIStats) *runtime.PsiStats {
 	return result
 }
 
-func (c *criService) cpuContainerStats(ID string, isSandbox bool, stats cgroupMetrics, timestamp time.Time) (*runtime.CpuUsage, error) {
+func (c *criService) cpuContainerStats(stats cgroupMetrics, timestamp time.Time) (*runtime.CpuUsage, error) {
 	switch {
 	case stats.v1 != nil:
 		metrics := stats.v1
@@ -553,7 +553,7 @@ func (c *criService) cpuContainerStats(ID string, isSandbox bool, stats cgroupMe
 	return nil, nil
 }
 
-func (c *criService) memoryContainerStats(ID string, stats cgroupMetrics, timestamp time.Time) (*runtime.MemoryUsage, error) {
+func (c *criService) memoryContainerStats(stats cgroupMetrics, timestamp time.Time) (*runtime.MemoryUsage, error) {
 	switch {
 	case stats.v1 != nil:
 		metrics := stats.v1

internal/cri/server/container_stats_list_test.go

@@ -341,7 +341,7 @@ func TestContainerMetricsMemory(t *testing.T) {
 		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
-			got, err := c.memoryContainerStats("ID", test.metrics, timestamp)
+			got, err := c.memoryContainerStats(test.metrics, timestamp)
 			assert.NoError(t, err)
 			assert.Equal(t, test.expected, got)
 		})

internal/cri/server/podsandbox/sandbox_stats_linux.go

@@ -0,0 +1,81 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package podsandbox
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/containerd/cgroups/v3"
+	"github.com/containerd/cgroups/v3/cgroup1"
+	cgroupsv2 "github.com/containerd/cgroups/v3/cgroup2"
+	"github.com/containerd/errdefs"
+	"github.com/containerd/typeurl/v2"
+
+	"github.com/containerd/containerd/api/types"
+	"github.com/containerd/containerd/v2/pkg/protobuf"
+)
+
+// Metrics returns the parent cgroup stats for the pod sandbox, encoded as a
+// *types.Metric (cgroupv1: *cgroup1/stats.Metrics, cgroupv2: *cgroup2/stats.Metrics).
+func (c *Controller) Metrics(ctx context.Context, sandboxID string) (*types.Metric, error) {
+	sb := c.store.Get(sandboxID)
+	if sb == nil {
+		return nil, fmt.Errorf("sandbox %q not found: %w", sandboxID, errdefs.ErrNotFound)
+	}
+
+	cgroupPath := sb.Metadata.Config.GetLinux().GetCgroupParent()
+	if cgroupPath == "" {
+		return nil, fmt.Errorf("sandbox %q has no cgroup parent", sandboxID)
+	}
+
+	var data any
+	switch cgroups.Mode() {
+	case cgroups.Unified:
+		cg, err := cgroupsv2.Load(cgroupPath)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load sandbox cgroup %q: %w", cgroupPath, err)
+		}
+		stats, err := cg.StatFiltered(cgroupsv2.StatCPU | cgroupsv2.StatMemory)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read sandbox cgroup %q: %w", cgroupPath, err)
+		}
+		data = stats
+	default:
+		control, err := cgroup1.Load(cgroup1.StaticPath(cgroupPath))
+		if err != nil {
+			return nil, fmt.Errorf("failed to load sandbox cgroup %q: %w", cgroupPath, err)
+		}
+		stats, err := control.Stat(cgroup1.IgnoreNotExist)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read sandbox cgroup %q: %w", cgroupPath, err)
+		}
+		data = stats
+	}
+
+	pbAny, err := typeurl.MarshalAnyToProto(data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal sandbox metrics: %w", err)
+	}
+
+	return &types.Metric{
+		Timestamp: protobuf.ToTimestamp(time.Now()),
+		ID:        sandboxID,
+		Data:      pbAny,
+	}, nil
+}

internal/cri/server/podsandbox/sandbox_stats.go → internal/cri/server/podsandbox/sandbox_stats_other.go

@@ -1,3 +1,5 @@
+//go:build !linux
+
 /*
    Copyright The containerd Authors.
 
@@ -19,11 +21,11 @@ package podsandbox
 import (
 	"context"
 
-	"github.com/containerd/containerd/api/types"
 	"github.com/containerd/errdefs"
+
+	"github.com/containerd/containerd/api/types"
 )
 
-// TODO(dcantah): Implement metrics to be used for SandboxStats rpc.
 func (c *Controller) Metrics(ctx context.Context, sandboxID string) (*types.Metric, error) {
 	return nil, errdefs.ErrNotImplemented
 }

internal/cri/server/sandbox_stats_linux.go

@@ -21,15 +21,16 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/containerd/cgroups/v3"
-	"github.com/containerd/cgroups/v3/cgroup1"
-	cgroupsv2 "github.com/containerd/cgroups/v3/cgroup2"
+	cg1 "github.com/containerd/cgroups/v3/cgroup1/stats"
+	cg2 "github.com/containerd/cgroups/v3/cgroup2/stats"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
+	"github.com/containerd/typeurl/v2"
 	"github.com/containernetworking/plugins/pkg/ns"
 	"github.com/vishvananda/netlink"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 
+	"github.com/containerd/containerd/api/types"
 	sandboxstore "github.com/containerd/containerd/v2/internal/cri/store/sandbox"
 )
 
@@ -42,10 +43,18 @@ func (c *criService) podSandboxStats(
 		return nil, fmt.Errorf("failed to get pod sandbox stats since sandbox container %q is not in ready state: %w", meta.ID, errdefs.ErrUnavailable)
 	}
 
-	stats, err := cgroupMetricsForSandbox(sandbox)
+	ctrl, err := c.sandboxService.SandboxController(sandbox.Sandboxer)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get controller for sandbox %s: %w", sandbox.ID, err)
+	}
+	metric, err := ctrl.Metrics(ctx, sandbox.ID)
 	if err != nil {
 		return nil, fmt.Errorf("failed getting metrics for sandbox %s: %w", sandbox.ID, err)
 	}
+	stats, err := decodeSandboxCgroupMetrics(metric)
+	if err != nil {
+		return nil, fmt.Errorf("failed decoding metrics for sandbox %s: %w", sandbox.ID, err)
+	}
 
 	podSandboxStats := &runtime.PodSandboxStats{
 		Linux: &runtime.LinuxPodSandboxStats{},
@@ -59,7 +68,7 @@ func (c *criService) podSandboxStats(
 
 	timestamp := time.Now()
 
-	cpuStats, err := c.cpuContainerStats(meta.ID, true /* isSandbox */, *stats, timestamp)
+	cpuStats, err := c.cpuContainerStats(*stats, timestamp)
 	if err != nil {
 		return nil, fmt.Errorf("failed to obtain cpu stats: %w", err)
 	}
@@ -74,7 +83,7 @@ func (c *criService) podSandboxStats(
 	}
 	podSandboxStats.Linux.Cpu = cpuStats
 
-	memoryStats, err := c.memoryContainerStats(meta.ID, *stats, timestamp)
+	memoryStats, err := c.memoryContainerStats(*stats, timestamp)
 	if err != nil {
 		return nil, fmt.Errorf("failed to obtain memory stats: %w", err)
 	}
@@ -191,32 +200,20 @@ func getAllContainerNetIO(ctx context.Context, netNsPath string) ([]interfaceSta
 	return allStats, err
 }
 
-func cgroupMetricsForSandbox(sandbox sandboxstore.Sandbox) (*cgroupMetrics, error) {
-	cgroupPath := sandbox.Config.GetLinux().GetCgroupParent()
-	if cgroupPath == "" {
-		return nil, fmt.Errorf("failed to get cgroup metrics for sandbox %v because cgroupPath is empty", sandbox.ID)
+func decodeSandboxCgroupMetrics(m *types.Metric) (*cgroupMetrics, error) {
+	if m == nil || m.Data == nil {
+		return nil, fmt.Errorf("sandbox metric is empty")
 	}
-
-	switch cgroups.Mode() {
-	case cgroups.Unified:
-		cg, err := cgroupsv2.Load(cgroupPath)
-		if err != nil {
-			return nil, fmt.Errorf("failed to load sandbox cgroup: %v: %w", cgroupPath, err)
-		}
-		stats, err := cg.StatFiltered(cgroupsv2.StatCPU | cgroupsv2.StatMemory)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get stats for cgroup: %v: %w", cgroupPath, err)
-		}
-		return &cgroupMetrics{v2: stats}, nil
+	a, err := typeurl.UnmarshalAny(m.Data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract sandbox metric: %w", err)
+	}
+	switch v := a.(type) {
+	case *cg1.Metrics:
+		return &cgroupMetrics{v1: v}, nil
+	case *cg2.Metrics:
+		return &cgroupMetrics{v2: v}, nil
 	default:
-		control, err := cgroup1.Load(cgroup1.StaticPath(cgroupPath))
-		if err != nil {
-			return nil, fmt.Errorf("failed to load sandbox cgroup %v: %w", cgroupPath, err)
-		}
-		stats, err := control.Stat(cgroup1.IgnoreNotExist)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get stats for cgroup %v: %w", cgroupPath, err)
-		}
-		return &cgroupMetrics{v1: stats}, nil
+		return nil, fmt.Errorf("unexpected sandbox metric type %T", v)
 	}
 }

internal/cri/server/service.go

@@ -286,10 +286,12 @@ func (c *criService) Run(ready func()) error {
 	// Start the background stats collector for UsageNanoCores calculation
 	log.L.Info("Start stats collector")
 	if c.statsCollector != nil {
+		// TODO: Find a better way to inject service dependencies.
 		c.statsCollector.SetDependencies(
 			c.client.TaskService(),
 			c.containerStore.List,
 			c.sandboxStore.List,
+			c.sandboxService.SandboxController,
 		)
 		c.statsCollector.Start()
 	}

internal/cri/server/stats_collector.go

@@ -23,15 +23,13 @@ import (
 	"sync"
 	"time"
 
-	"github.com/containerd/cgroups/v3"
-	"github.com/containerd/cgroups/v3/cgroup1"
 	cg1 "github.com/containerd/cgroups/v3/cgroup1/stats"
-	cgroupsv2 "github.com/containerd/cgroups/v3/cgroup2"
 	cg2 "github.com/containerd/cgroups/v3/cgroup2/stats"
 	"github.com/containerd/log"
 	"github.com/containerd/typeurl/v2"
 
 	"github.com/containerd/containerd/api/services/tasks/v1"
+	"github.com/containerd/containerd/v2/core/sandbox"
 	criconfig "github.com/containerd/containerd/v2/internal/cri/config"
 	containerstore "github.com/containerd/containerd/v2/internal/cri/store/container"
 	sandboxstore "github.com/containerd/containerd/v2/internal/cri/store/sandbox"
@@ -73,6 +71,8 @@ type StatsCollector struct {
 	listContainers func() []containerstore.Container
 	// listSandboxes returns the list of sandboxes
 	listSandboxes func() []sandboxstore.Sandbox
+	// sandboxController resolves a sandbox.Controller for a given sandboxer name
+	sandboxController func(sandboxer string) (sandbox.Controller, error)
 }
 
 // NewStatsCollector creates a new StatsCollector.
@@ -113,10 +113,12 @@ func (c *StatsCollector) SetDependencies(
 	taskService tasks.TasksClient,
 	listContainers func() []containerstore.Container,
 	listSandboxes func() []sandboxstore.Sandbox,
+	sandboxController func(sandboxer string) (sandbox.Controller, error),
 ) {
 	c.taskService = taskService
 	c.listContainers = listContainers
 	c.listSandboxes = listSandboxes
+	c.sandboxController = sandboxController
 }
 
 // Start begins the background stats collection loop.
@@ -192,67 +194,39 @@ func (c *StatsCollector) collectContainerStats(ctx context.Context) {
 	}
 }
 
-// collectSandboxStats fetches metrics for all sandboxes.
-// On Linux, sandbox/pod stats come from the parent cgroup (which includes all
-// containers in the pod), not from the pause container's task metrics.
-// This matches how sandbox_stats_linux.go retrieves pod-level CPU stats.
+// collectSandboxStats fetches sandbox-level CPU samples via the sandbox
+// controller. The controller reads the pod's parent cgroup (which includes all
+// containers in the pod) and returns the result as a *types.Metric; the
+// collector only decodes it via extractCPUUsage.
 func (c *StatsCollector) collectSandboxStats(ctx context.Context) {
 	sandboxes := c.listSandboxes()
 	if len(sandboxes) == 0 {
 		return
 	}
 
-	timestamp := time.Now()
 	for _, sb := range sandboxes {
-		// Get the parent cgroup path for the pod
-		cgroupPath := sb.Config.GetLinux().GetCgroupParent()
-		if cgroupPath == "" {
-			continue
-		}
-
-		usageCoreNanoSeconds, ok := c.getCgroupCPUUsage(ctx, cgroupPath)
-		if !ok {
-			continue
-		}
-		c.addSample(sb.ID, timestamp, usageCoreNanoSeconds)
-	}
-}
-
-// getCgroupCPUUsage reads CPU usage from a cgroup path.
-// Supports both cgroupv1 and cgroupv2.
-func (c *StatsCollector) getCgroupCPUUsage(ctx context.Context, cgroupPath string) (uint64, bool) {
-	switch cgroups.Mode() {
-	case cgroups.Unified:
-		cg, err := cgroupsv2.Load(cgroupPath)
-		if err != nil {
-			log.G(ctx).WithError(err).Debugf("StatsCollector: failed to load cgroupv2: %s", cgroupPath)
-			return 0, false
-		}
-		stats, err := cg.StatFiltered(cgroupsv2.StatCPU)
+		ctrl, err := c.sandboxController(sb.Sandboxer)
 		if err != nil {
-			log.G(ctx).WithError(err).Debugf("StatsCollector: failed to get cgroupv2 stats: %s", cgroupPath)
-			return 0, false
-		}
-		if stats.CPU != nil {
-			// cgroupv2 reports in microseconds, convert to nanoseconds
-			return stats.CPU.UsageUsec * 1000, true
+			log.G(ctx).WithError(err).Debugf("StatsCollector: no controller for sandbox %s", sb.ID)
+			continue
 		}
-	default:
-		control, err := cgroup1.Load(cgroup1.StaticPath(cgroupPath))
+		metric, err := ctrl.Metrics(ctx, sb.ID)
 		if err != nil {
-			log.G(ctx).WithError(err).Debugf("StatsCollector: failed to load cgroupv1: %s", cgroupPath)
-			return 0, false
+			log.G(ctx).WithError(err).Debugf("StatsCollector: failed to fetch sandbox %s metrics", sb.ID)
+			continue
 		}
-		stats, err := control.Stat(cgroup1.IgnoreNotExist)
-		if err != nil {
-			log.G(ctx).WithError(err).Debugf("StatsCollector: failed to get cgroupv1 stats: %s", cgroupPath)
-			return 0, false
+		if metric == nil {
+			continue
 		}
-		if stats.CPU != nil && stats.CPU.Usage != nil {
-			return stats.CPU.Usage.Total, true
+		usageCoreNanoSeconds, ok := extractCPUUsage(metric.Data)
+		if !ok {
+			continue
 		}
+		// Sample at the metric's own timestamp, captured by the controller at
+		// cgroup read time. This keeps the delta-time in the UsageNanoCores
+		// rate accurate and free of per-sandbox collection latency.
+		c.addSample(sb.ID, metric.GetTimestamp().AsTime(), usageCoreNanoSeconds)
 	}
-	return 0, false
 }
 
 // addSample adds a CPU sample for the given container/sandbox ID.