From 75432a14108455f87c65aaac63509d354c25655e Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Mon, 19 May 2025 08:30:08 +0200 Subject: [PATCH 1/2] component-base metrics: allow usage of Prometheus types in tests Test code using the testutil.Metrics type already depended on the Prometheus types, but couldn't reference them by name. This is necessary for example when using Gomega (to cast from `any` in a matcher) or when defining a `var sample *Sample` which gets set later. --- .../component-base/metrics/testutil/metrics.go | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/staging/src/k8s.io/component-base/metrics/testutil/metrics.go b/staging/src/k8s.io/component-base/metrics/testutil/metrics.go index 05d15b08d75..e1ed9243809 100644 --- a/staging/src/k8s.io/component-base/metrics/testutil/metrics.go +++ b/staging/src/k8s.io/component-base/metrics/testutil/metrics.go @@ -38,8 +38,20 @@ var ( QuantileLabel model.LabelName = model.QuantileLabel ) -// Metrics is generic metrics for other specific metrics -type Metrics map[string]model.Samples +// Metrics is generic metrics for other specific metrics. +// This directly exposes Prometheus types. Test code may +// use those via type aliases provided by this package +// instead of imorting the Prometheus packages (https://github.com/kubernetes/kubernetes/issues/89267). +type Metrics map[string]Samples +type Samples = model.Samples +type Sample = model.Sample +type Metric = model.Metric +type LabelValue = model.LabelValue +type LabelName = model.LabelName +type SampleHistogram = model.SampleHistogram +type FloatString = model.FloatString +type HistogramBuckets = model.HistogramBuckets +type HistogramBucket = model.HistogramBucket // Equal returns true if all metrics are the same as the arguments. func (m *Metrics) Equal(o Metrics) bool { From 6d6a749c62e7796002e72a6fb854d90207b0ea7b Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 7 May 2025 13:30:06 +0200 Subject: [PATCH 2/2] DRA kubelet: add dra_resource_claims_in_use gauge vector The new metric informs admins whether DRA in general (special "driver_name: " label) and/or specific DRA drivers (other label values) are in use on nodes. This is useful to know because removing a driver is only safe if it is not in use. If a driver gets removed while it has prepared a ResourceClaim, unpreparing that ResourceClaim and stopping pods is blocked. The implementation of the metric uses read locking of the claim info cache. It retrieves "claims in use" and turns those into the metric. The same code is also used to log changes in the claim info cache with a diff. This hooks into a write update of the claim info cache and uses contextual logging. The unit tests check that metrics get calculated. The e2e_node test checks that kubelet really exports the metrics data. While at it, some bugs in the claiminfo_test.go get fixed: the way how the cache got populated in the test did not match the code anymore. --- pkg/kubelet/cm/container_manager_linux.go | 4 +- pkg/kubelet/cm/dra/claiminfo.go | 169 ++++++++++++++++++++-- pkg/kubelet/cm/dra/claiminfo_test.go | 148 +++++++++++++++++-- pkg/kubelet/cm/dra/manager.go | 15 +- pkg/kubelet/cm/dra/manager_test.go | 18 ++- pkg/kubelet/cm/dra/metrics_test.go | 35 +++++ pkg/kubelet/metrics/metrics.go | 17 ++- test/e2e_node/dra_test.go | 39 +++++ 8 files changed, 405 insertions(+), 40 deletions(-) create mode 100644 pkg/kubelet/cm/dra/metrics_test.go diff --git a/pkg/kubelet/cm/container_manager_linux.go b/pkg/kubelet/cm/container_manager_linux.go index cd8b8f46adc..5b890abd0b0 100644 --- a/pkg/kubelet/cm/container_manager_linux.go +++ b/pkg/kubelet/cm/container_manager_linux.go @@ -62,6 +62,7 @@ import ( kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/events" "k8s.io/kubernetes/pkg/kubelet/lifecycle" + "k8s.io/kubernetes/pkg/kubelet/metrics" "k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache" "k8s.io/kubernetes/pkg/kubelet/stats/pidlimit" "k8s.io/kubernetes/pkg/kubelet/status" @@ -310,10 +311,11 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I // Initialize DRA manager if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DynamicResourceAllocation) { klog.InfoS("Creating Dynamic Resource Allocation (DRA) manager") - cm.draManager, err = dra.NewManager(kubeClient, nodeConfig.KubeletRootDir) + cm.draManager, err = dra.NewManager(klog.TODO(), kubeClient, nodeConfig.KubeletRootDir) if err != nil { return nil, err } + metrics.Register(cm.draManager.NewMetricsCollector()) } cm.kubeClient = kubeClient diff --git a/pkg/kubelet/cm/dra/claiminfo.go b/pkg/kubelet/cm/dra/claiminfo.go index 75aeb0e0707..19d4327e5f9 100644 --- a/pkg/kubelet/cm/dra/claiminfo.go +++ b/pkg/kubelet/cm/dra/claiminfo.go @@ -20,13 +20,19 @@ import ( "errors" "fmt" "slices" + "strings" "sync" + "github.com/go-logr/logr" + resourceapi "k8s.io/api/resource/v1beta1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/component-base/metrics" + "k8s.io/klog/v2" "k8s.io/kubernetes/pkg/kubelet/cm/dra/state" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + kubeletmetrics "k8s.io/kubernetes/pkg/kubelet/metrics" ) // ClaimInfo holds information required @@ -39,6 +45,8 @@ type ClaimInfo struct { // claimInfoCache is a cache of processed resource claims keyed by namespace/claimname. type claimInfoCache struct { + logger klog.Logger + sync.RWMutex checkpointer state.Checkpointer claimInfo map[string]*ClaimInfo @@ -111,8 +119,25 @@ func (info *ClaimInfo) isPrepared() bool { return info.prepared } +// cdiDevicesAsList returns a list of CDIDevices from the provided claim info. +// When the request name is non-empty, only devices relevant for that request +// are returned. +func (info *ClaimInfo) cdiDevicesAsList(requestName string) []kubecontainer.CDIDevice { + var cdiDevices []kubecontainer.CDIDevice + for _, driverData := range info.DriverState { + for _, device := range driverData.Devices { + if requestName == "" || len(device.RequestNames) == 0 || slices.Contains(device.RequestNames, requestName) { + for _, cdiDeviceID := range device.CDIDeviceIDs { + cdiDevices = append(cdiDevices, kubecontainer.CDIDevice{Name: cdiDeviceID}) + } + } + } + } + return cdiDevices +} + // newClaimInfoCache creates a new claim info cache object, pre-populated from a checkpoint (if present). -func newClaimInfoCache(stateDir, checkpointName string) (*claimInfoCache, error) { +func newClaimInfoCache(logger klog.Logger, stateDir, checkpointName string) (*claimInfoCache, error) { checkpointer, err := state.NewCheckpointer(stateDir, checkpointName) if err != nil { return nil, fmt.Errorf("could not initialize checkpoint manager, please drain node and remove DRA state file, err: %w", err) @@ -124,6 +149,7 @@ func newClaimInfoCache(stateDir, checkpointName string) (*claimInfoCache, error) } cache := &claimInfoCache{ + logger: logger, checkpointer: checkpointer, claimInfo: make(map[string]*ClaimInfo), } @@ -142,9 +168,31 @@ func newClaimInfoCache(stateDir, checkpointName string) (*claimInfoCache, error) } // withLock runs a function while holding the claimInfoCache lock. +// It logs changes. func (cache *claimInfoCache) withLock(f func() error) error { cache.Lock() defer cache.Unlock() + + if loggerV := cache.logger.V(5); loggerV.Enabled() { + claimsInUseBefore := cache.claimsInUse() + defer func() { + claimsInUseAfter := cache.claimsInUse() + delta := claimsInUseDelta(claimsInUseBefore, claimsInUseAfter) + + changed := false + for _, inUse := range delta { + if inUse.Delta != 0 { + changed = true + break + } + } + + if changed { + cache.logger.V(5).Info("ResourceClaim usage changed", "claimsInUse", delta) + } + }() + } + return f() } @@ -203,19 +251,112 @@ func (cache *claimInfoCache) syncToCheckpoint() error { return cache.checkpointer.Store(checkpoint) } -// cdiDevicesAsList returns a list of CDIDevices from the provided claim info. -// When the request name is non-empty, only devices relevant for that request -// are returned. -func (info *ClaimInfo) cdiDevicesAsList(requestName string) []kubecontainer.CDIDevice { - var cdiDevices []kubecontainer.CDIDevice - for _, driverData := range info.DriverState { - for _, device := range driverData.Devices { - if requestName == "" || len(device.RequestNames) == 0 || slices.Contains(device.RequestNames, requestName) { - for _, cdiDeviceID := range device.CDIDeviceIDs { - cdiDevices = append(cdiDevices, kubecontainer.CDIDevice{Name: cdiDeviceID}) - } - } +// claimsInUse computes the the current counter vector for DRAResourceClaimsInUse. +// It returns a map of driver name to number of claims which have been prepared using +// the driver. The [kubeletmetrics.DRAResourceClaimsInUseAnyDriver] key stands for +// all prepared claims. +// +// Must be called while the rlock is held. +func (cache *claimInfoCache) claimsInUse() map[string]int { + counts := make(map[string]int) + total := 0 + for _, claimInfo := range cache.claimInfo { + if !claimInfo.isPrepared() { + continue + } + total++ + + for driverName := range claimInfo.DriverState { + counts[driverName]++ } } - return cdiDevices + counts[kubeletmetrics.DRAResourceClaimsInUseAnyDriver] = total + return counts +} + +// claimsInUseDelta compares two maps returned by claimsInUse. +// The type can be used as value in structured logging. +func claimsInUseDelta(before, after map[string]int) ClaimsInUseDelta { + var delta ClaimsInUseDelta + for driverName, count := range before { + if _, stillSet := after[driverName]; !stillSet { + delta = append(delta, ClaimsInUse{DriverName: driverName, Count: 0, Delta: -count}) + } + } + for driverName, count := range after { + delta = append(delta, ClaimsInUse{DriverName: driverName, Count: count, Delta: count - before[driverName]}) + } + return delta +} + +// ClaimsInUseDelta provides String (for text logging) and MarshalLog (for structured logging). +type ClaimsInUseDelta []ClaimsInUse + +var _ fmt.Stringer = ClaimsInUseDelta{} +var _ logr.Marshaler = ClaimsInUseDelta{} + +func (d ClaimsInUseDelta) String() string { + d = d.sort() + var buffer strings.Builder + for i, inUse := range d { + if i > 0 { + buffer.WriteByte('\n') + } + buffer.WriteString(fmt.Sprintf("%s: %d (%+d)", inUse.DriverName, inUse.Count, inUse.Delta)) + } + return buffer.String() +} + +func (d ClaimsInUseDelta) MarshalLog() any { + d = d.sort() + return []ClaimsInUse(d) +} + +// sort returns a sorted copy of the slice. +func (d ClaimsInUseDelta) sort() ClaimsInUseDelta { + d = slices.Clone(d) + slices.SortFunc(d, func(a, b ClaimsInUse) int { + return strings.Compare(a.DriverName, b.DriverName) + }) + return d +} + +type ClaimsInUse struct { + DriverName string + Count int + Delta int +} + +// claimInfoCollector provides metrics for a claimInfoCache. +type claimInfoCollector struct { + metrics.BaseStableCollector + cache *claimInfoCache +} + +var _ metrics.StableCollector = &claimInfoCollector{} + +// DescribeWithStability implements the metrics.StableCollector interface. +func (collector *claimInfoCollector) DescribeWithStability(ch chan<- *metrics.Desc) { + ch <- kubeletmetrics.DRAResourceClaimsInUseDesc +} + +// CollectWithStability implements the metrics.StableCollector interface. +func (collector *claimInfoCollector) CollectWithStability(ch chan<- metrics.Metric) { + var claimsInUse map[string]int + _ = collector.cache.withRLock(func() error { + claimsInUse = collector.cache.claimsInUse() + return nil + }) + + // Only currently known drivers are listed. If a driver had active + // claims in the past, no longer does and then gets uninstalled, it no + // longer shows up. This avoids the memory leak problem in a normal + // GaugeVec which could grow over time unless obsolete drivers are + // actively deleted. + // + // The empty driver name provides the overall count of all active + // ResourceClaims regardless of the driver. + for driverName, count := range claimsInUse { + ch <- metrics.NewLazyConstMetric(kubeletmetrics.DRAResourceClaimsInUseDesc, metrics.GaugeValue, float64(count), driverName) + } } diff --git a/pkg/kubelet/cm/dra/claiminfo_test.go b/pkg/kubelet/cm/dra/claiminfo_test.go index 4f80c24794f..ccb2991c671 100644 --- a/pkg/kubelet/cm/dra/claiminfo_test.go +++ b/pkg/kubelet/cm/dra/claiminfo_test.go @@ -30,6 +30,8 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/kubernetes/pkg/kubelet/cm/dra/state" + "k8s.io/kubernetes/test/utils/ktesting" + "k8s.io/kubernetes/test/utils/ktesting/initoption" ) // ClaimInfo test cases @@ -436,7 +438,8 @@ func TestNewClaimInfoCache(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - result, err := newClaimInfoCache(test.stateDir, test.checkpointName) + tCtx := ktesting.Init(t) + result, err := newClaimInfoCache(tCtx.Logger(), test.stateDir, test.checkpointName) if test.wantErr { assert.Error(t, err) return @@ -494,7 +497,8 @@ func TestClaimInfoCacheWithLock(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - cache, err := newClaimInfoCache(t.TempDir(), "test-checkpoint") + tCtx := ktesting.Init(t) + cache, err := newClaimInfoCache(tCtx.Logger(), t.TempDir(), "test-checkpoint") require.NoError(t, err) assert.NotNil(t, cache) err = cache.withLock(test.funcGen(cache)) @@ -554,7 +558,8 @@ func TestClaimInfoCacheWithRLock(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - cache, err := newClaimInfoCache(t.TempDir(), "test-checkpoint") + tCtx := ktesting.Init(t) + cache, err := newClaimInfoCache(tCtx.Logger(), t.TempDir(), "test-checkpoint") require.NoError(t, err) assert.NotNil(t, cache) err = cache.withRLock(test.funcGen(cache)) @@ -569,8 +574,11 @@ func TestClaimInfoCacheWithRLock(t *testing.T) { func TestClaimInfoCacheAdd(t *testing.T) { for _, test := range []struct { - description string - claimInfo *ClaimInfo + description string + initialClaimInfo []*ClaimInfo + claimInfo *ClaimInfo + expectMetrics string + expectLog string }{ { description: "claimInfo successfully added", @@ -580,14 +588,90 @@ func TestClaimInfoCacheAdd(t *testing.T) { Namespace: namespace, }, }, + expectMetrics: `# HELP dra_resource_claims_in_use [ALPHA] The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name. +# TYPE dra_resource_claims_in_use gauge +dra_resource_claims_in_use{driver_name=""} 0 +`, + }, + { + description: "prepared claimInfo", + claimInfo: &ClaimInfo{ + ClaimInfoState: state.ClaimInfoState{ + ClaimName: claimName, + Namespace: namespace, + DriverState: map[string]state.DriverState{ + "test-driver": {}, + "other-test-driver": {}, + }, + }, + prepared: true, + }, + expectMetrics: `# HELP dra_resource_claims_in_use [ALPHA] The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name. +# TYPE dra_resource_claims_in_use gauge +dra_resource_claims_in_use{driver_name=""} 1 +dra_resource_claims_in_use{driver_name="other-test-driver"} 1 +dra_resource_claims_in_use{driver_name="test-driver"} 1 +`, + expectLog: `INFO ResourceClaim usage changed claimsInUse=< + : 1 (+1) + other-test-driver: 1 (+1) + test-driver: 1 (+1) + > +`, + }, + { + description: "add more prepared claimInfo", + initialClaimInfo: []*ClaimInfo{{ + ClaimInfoState: state.ClaimInfoState{ + ClaimName: claimName + "-old", + Namespace: namespace, + DriverState: map[string]state.DriverState{ + "test-driver": {}, + }, + }, + prepared: true, + }}, + claimInfo: &ClaimInfo{ + ClaimInfoState: state.ClaimInfoState{ + ClaimName: claimName, + Namespace: namespace, + DriverState: map[string]state.DriverState{ + "test-driver": {}, + "other-test-driver": {}, + }, + }, + prepared: true, + }, + expectMetrics: `# HELP dra_resource_claims_in_use [ALPHA] The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name. +# TYPE dra_resource_claims_in_use gauge +dra_resource_claims_in_use{driver_name=""} 2 +dra_resource_claims_in_use{driver_name="other-test-driver"} 1 +dra_resource_claims_in_use{driver_name="test-driver"} 2 +`, + expectLog: `INFO ResourceClaim usage changed claimsInUse=< + : 2 (+1) + other-test-driver: 1 (+1) + test-driver: 2 (+1) + > +`, }, } { t.Run(test.description, func(t *testing.T) { - cache, err := newClaimInfoCache(t.TempDir(), "test-checkpoint") + tCtx := ktesting.Init(t, initoption.BufferLogs(true)) + cache, err := newClaimInfoCache(tCtx.Logger(), t.TempDir(), "test-checkpoint") + for _, claimInfo := range test.initialClaimInfo { + cache.add(claimInfo) + } require.NoError(t, err) assert.NotNil(t, cache) - cache.add(test.claimInfo) + _ = cache.withLock(func() error { + cache.add(test.claimInfo) + return nil + }) assert.True(t, cache.contains(test.claimInfo.ClaimName, test.claimInfo.Namespace)) + testClaimsInUseMetric(tCtx, cache, test.expectMetrics) + logOutput := tCtx.Logger().GetSink().(ktesting.Underlier).GetBuffer() + assert.Equal(t, test.expectLog, logOutput.String()) }) } } @@ -682,28 +766,69 @@ func TestClaimInfoCacheDelete(t *testing.T) { for _, test := range []struct { description string claimInfoCache *claimInfoCache + expectMetrics string + expectLog string }{ { description: "item in cache", claimInfoCache: &claimInfoCache{ claimInfo: map[string]*ClaimInfo{ - claimName + namespace: { + namespace + "/" + claimName: { ClaimInfoState: state.ClaimInfoState{ ClaimName: claimName, Namespace: namespace, + DriverState: map[string]state.DriverState{ + "test-driver": {}, + }, }, + prepared: true, + }, + namespace + "/" + claimName + "-old": { + ClaimInfoState: state.ClaimInfoState{ + ClaimName: claimName, + Namespace: namespace, + DriverState: map[string]state.DriverState{ + "test-driver": {}, + "other-test-driver": {}, + }, + }, + prepared: true, }, }, }, + expectMetrics: `# HELP dra_resource_claims_in_use [ALPHA] The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name. +# TYPE dra_resource_claims_in_use gauge +dra_resource_claims_in_use{driver_name=""} 1 +dra_resource_claims_in_use{driver_name="test-driver"} 1 +dra_resource_claims_in_use{driver_name="other-test-driver"} 1 +`, + expectLog: `INFO ResourceClaim usage changed claimsInUse=< + : 1 (-1) + other-test-driver: 1 (+0) + test-driver: 1 (-1) + > +`, }, { description: "item not in cache", claimInfoCache: &claimInfoCache{}, + expectMetrics: `# HELP dra_resource_claims_in_use [ALPHA] The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name. +# TYPE dra_resource_claims_in_use gauge +dra_resource_claims_in_use{driver_name=""} 0 +`, }, } { t.Run(test.description, func(t *testing.T) { - test.claimInfoCache.delete(claimName, namespace) + tCtx := ktesting.Init(t, initoption.BufferLogs(true)) + test.claimInfoCache.logger = tCtx.Logger() + _ = test.claimInfoCache.withLock(func() error { + test.claimInfoCache.delete(claimName, namespace) + return nil + }) assert.False(t, test.claimInfoCache.contains(claimName, namespace)) + testClaimsInUseMetric(tCtx, test.claimInfoCache, test.expectMetrics) + logOutput := tCtx.Logger().GetSink().(ktesting.Underlier).GetBuffer() + assert.Equal(t, test.expectLog, logOutput.String()) }) } } @@ -718,7 +843,7 @@ func TestClaimInfoCacheHasPodReference(t *testing.T) { description: "uid is referenced", claimInfoCache: &claimInfoCache{ claimInfo: map[string]*ClaimInfo{ - claimName + namespace: { + namespace + "/" + claimName: { ClaimInfoState: state.ClaimInfoState{ ClaimName: claimName, Namespace: namespace, @@ -754,7 +879,8 @@ func TestSyncToCheckpoint(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - cache, err := newClaimInfoCache(test.stateDir, test.checkpointName) + tCtx := ktesting.Init(t) + cache, err := newClaimInfoCache(tCtx.Logger(), test.stateDir, test.checkpointName) require.NoError(t, err) err = cache.syncToCheckpoint() if test.wantErr { diff --git a/pkg/kubelet/cm/dra/manager.go b/pkg/kubelet/cm/dra/manager.go index 40ae6ee0414..e9bebbc0423 100644 --- a/pkg/kubelet/cm/dra/manager.go +++ b/pkg/kubelet/cm/dra/manager.go @@ -29,6 +29,7 @@ import ( "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" clientset "k8s.io/client-go/kubernetes" + "k8s.io/component-base/metrics" "k8s.io/dynamic-resource-allocation/resourceclaim" "k8s.io/klog/v2" drapb "k8s.io/kubelet/pkg/apis/dra/v1beta1" @@ -36,7 +37,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet/cm/dra/state" "k8s.io/kubernetes/pkg/kubelet/config" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" - "k8s.io/kubernetes/pkg/kubelet/metrics" + kubeletmetrics "k8s.io/kubernetes/pkg/kubelet/metrics" "k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache" ) @@ -101,8 +102,8 @@ type Manager struct { // - Don't include the namespace, it can be inferred from the context. // - Avoid repeated "failed to ...: failed to ..." when wrapping errors. // - Avoid wrapping when it does not provide relevant additional information to keep the user-visible error short. -func NewManager(kubeClient clientset.Interface, stateFileDirectory string) (*Manager, error) { - claimInfoCache, err := newClaimInfoCache(stateFileDirectory, draManagerStateFileName) +func NewManager(logger klog.Logger, kubeClient clientset.Interface, stateFileDirectory string) (*Manager, error) { + claimInfoCache, err := newClaimInfoCache(logger, stateFileDirectory, draManagerStateFileName) if err != nil { return nil, fmt.Errorf("create ResourceClaim cache: %w", err) } @@ -122,6 +123,10 @@ func NewManager(kubeClient clientset.Interface, stateFileDirectory string) (*Man return manager, nil } +func (m *Manager) NewMetricsCollector() metrics.StableCollector { + return &claimInfoCollector{cache: m.cache} +} + // GetWatcherHandler must be called after Start, it indirectly depends // on parameters which only get passed to Start, for example the context. func (m *Manager) GetWatcherHandler() cache.PluginHandler { @@ -199,7 +204,7 @@ func (m *Manager) reconcileLoop(ctx context.Context) { func (m *Manager) PrepareResources(ctx context.Context, pod *v1.Pod) error { startTime := time.Now() err := m.prepareResources(ctx, pod) - metrics.DRAOperationsDuration.WithLabelValues("PrepareResources", strconv.FormatBool(err == nil)).Observe(time.Since(startTime).Seconds()) + kubeletmetrics.DRAOperationsDuration.WithLabelValues("PrepareResources", strconv.FormatBool(err == nil)).Observe(time.Since(startTime).Seconds()) if err != nil { return fmt.Errorf("prepare dynamic resources: %w", err) } @@ -481,7 +486,7 @@ func (m *Manager) GetResources(pod *v1.Pod, container *v1.Container) (*Container func (m *Manager) UnprepareResources(ctx context.Context, pod *v1.Pod) error { startTime := time.Now() err := m.unprepareResourcesForPod(ctx, pod) - metrics.DRAOperationsDuration.WithLabelValues("UnprepareResources", strconv.FormatBool(err == nil)).Observe(time.Since(startTime).Seconds()) + kubeletmetrics.DRAOperationsDuration.WithLabelValues("UnprepareResources", strconv.FormatBool(err == nil)).Observe(time.Since(startTime).Seconds()) if err != nil { return fmt.Errorf("unprepare dynamic resources: %w", err) } diff --git a/pkg/kubelet/cm/dra/manager_test.go b/pkg/kubelet/cm/dra/manager_test.go index 2663d744f8a..6577b73d979 100644 --- a/pkg/kubelet/cm/dra/manager_test.go +++ b/pkg/kubelet/cm/dra/manager_test.go @@ -198,7 +198,8 @@ func TestNewManagerImpl(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - manager, err := NewManager(kubeClient, test.stateFileDirectory) + tCtx := ktesting.Init(t) + manager, err := NewManager(tCtx.Logger(), kubeClient, test.stateFileDirectory) if test.wantErr { assert.Error(t, err) return @@ -362,7 +363,8 @@ func TestGetResources(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - manager, err := NewManager(kubeClient, t.TempDir()) + tCtx := ktesting.Init(t) + manager, err := NewManager(tCtx.Logger(), kubeClient, t.TempDir()) require.NoError(t, err) if test.claimInfo != nil { @@ -558,7 +560,7 @@ func TestPrepareResources(t *testing.T) { t.Run(test.description, func(t *testing.T) { tCtx := ktesting.Init(t) - manager, err := NewManager(fakeKubeClient, t.TempDir()) + manager, err := NewManager(tCtx.Logger(), fakeKubeClient, t.TempDir()) require.NoError(t, err, "create DRA manager") manager.initDRAPluginManager(tCtx, getFakeNode, time.Second /* very short wiping delay for testing */) @@ -714,7 +716,7 @@ func TestUnprepareResources(t *testing.T) { } defer draServerInfo.teardownFn() - manager, err := NewManager(fakeKubeClient, t.TempDir()) + manager, err := NewManager(tCtx.Logger(), fakeKubeClient, t.TempDir()) require.NoError(t, err, "create DRA manager") manager.initDRAPluginManager(tCtx, getFakeNode, time.Second /* very short wiping delay for testing */) @@ -758,8 +760,9 @@ func TestUnprepareResources(t *testing.T) { } func TestPodMightNeedToUnprepareResources(t *testing.T) { + tCtx := ktesting.Init(t) fakeKubeClient := fake.NewSimpleClientset() - manager, err := NewManager(fakeKubeClient, t.TempDir()) + manager, err := NewManager(tCtx.Logger(), fakeKubeClient, t.TempDir()) require.NoError(t, err, "create DRA manager") claimInfo := &ClaimInfo{ @@ -833,7 +836,8 @@ func TestGetContainerClaimInfos(t *testing.T) { }, } { t.Run(test.description, func(t *testing.T) { - manager, err := NewManager(nil, t.TempDir()) + tCtx := ktesting.Init(t) + manager, err := NewManager(tCtx.Logger(), nil, t.TempDir()) require.NoError(t, err, "create DRA manager") if test.claimInfo != nil { @@ -871,7 +875,7 @@ func TestParallelPrepareUnprepareResources(t *testing.T) { // Create fake Kube client and DRA manager fakeKubeClient := fake.NewSimpleClientset() - manager, err := NewManager(fakeKubeClient, t.TempDir()) + manager, err := NewManager(tCtx.Logger(), fakeKubeClient, t.TempDir()) require.NoError(t, err, "create DRA manager") manager.initDRAPluginManager(tCtx, getFakeNode, time.Second /* very short wiping delay for testing */) diff --git a/pkg/kubelet/cm/dra/metrics_test.go b/pkg/kubelet/cm/dra/metrics_test.go new file mode 100644 index 00000000000..808303330e3 --- /dev/null +++ b/pkg/kubelet/cm/dra/metrics_test.go @@ -0,0 +1,35 @@ +/* +Copyright 2025 The Kubernetes 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 dra + +import ( + "strings" + + "k8s.io/component-base/metrics/testutil" + "k8s.io/kubernetes/test/utils/ktesting" +) + +func testClaimsInUseMetric(tCtx ktesting.TContext, claimInfoCache *claimInfoCache, expectedMetric string) { + tCtx.Helper() + // Must simulate registration which calls Create, otherwise collection crashes. + collector := &claimInfoCollector{cache: claimInfoCache} + collector.Create(nil, collector) + err := testutil.CollectAndCompare(collector, strings.NewReader(expectedMetric), "dra_resource_claims_in_use") + if err != nil { + tCtx.Error(err) + } +} diff --git a/pkg/kubelet/metrics/metrics.go b/pkg/kubelet/metrics/metrics.go index 7a04387ab6f..6f1cd2803af 100644 --- a/pkg/kubelet/metrics/metrics.go +++ b/pkg/kubelet/metrics/metrics.go @@ -160,6 +160,9 @@ const ( ImageVolumeRequestedTotalKey = "image_volume_requested_total" ImageVolumeMountedSucceedTotalKey = "image_volume_mounted_succeed_total" ImageVolumeMountedErrorsTotalKey = "image_volume_mounted_errors_total" + + // Special label for [DRAResourceClaimsInUseDesc] which counts ResourceClaims regardless of the driver. + DRAResourceClaimsInUseAnyDriver = "" ) type imageSizeBucket struct { @@ -1028,6 +1031,14 @@ var ( []string{"driver_name", "method_name", "grpc_status_code"}, ) + DRAResourceClaimsInUseDesc = metrics.NewDesc(DRASubsystem+"_resource_claims_in_use", + "The number of ResourceClaims that are currently in use on the node, by driver name (driver_name label value) and across all drivers (special value for driver_name). Note that the sum of all by-driver counts is not the total number of in-use ResourceClaims because the same ResourceClaim might use devices from different drivers. Instead, use the count for the driver_name.", + []string{"driver_name"}, + nil, + metrics.ALPHA, + "", + ) + // AdmissionRejectionsTotal tracks the number of failed admission times, currently, just record it for pod additions AdmissionRejectionsTotal = metrics.NewCounterVec( &metrics.CounterOpts{ @@ -1164,8 +1175,10 @@ func Register(collectors ...metrics.StableCollector) { legacyregistry.MustRegister(CgroupVersion) if utilfeature.DefaultFeatureGate.Enabled(features.DynamicResourceAllocation) { - legacyregistry.MustRegister(DRAOperationsDuration) - legacyregistry.MustRegister(DRAGRPCOperationsDuration) + legacyregistry.MustRegister( + DRAOperationsDuration, + DRAGRPCOperationsDuration, + ) } legacyregistry.MustRegister(AdmissionRejectionsTotal) diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index 74177a9a028..9b5051f6fa7 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -36,6 +36,7 @@ import ( "github.com/onsi/ginkgo/v2" "github.com/onsi/gomega" + "github.com/onsi/gomega/gstruct" "github.com/onsi/gomega/types" v1 "k8s.io/api/core/v1" @@ -45,6 +46,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes" + "k8s.io/component-base/metrics/testutil" "k8s.io/klog/v2" admissionapi "k8s.io/pod-security-admission/api" "k8s.io/utils/ptr" @@ -479,6 +481,43 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami gomega.Eventually(kubeletPlugin2.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded) }) + ginkgo.It("must provide metrics", func(ctx context.Context) { + kubeletPlugin1, kubeletPlugin2 := start(ctx) + + pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod", false, []string{kubeletPlugin1Name, kubeletPlugin2Name}) + + ginkgo.By("wait for pod to succeed") + err := e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod) + framework.ExpectNoError(err) + gomega.Expect(kubeletPlugin1.GetGRPCCalls()).Should(testdrivergomega.NodePrepareResourcesSucceeded, "Plugin 1 should have prepared resources.") + gomega.Expect(kubeletPlugin2.GetGRPCCalls()).Should(testdrivergomega.NodePrepareResourcesSucceeded, "Plugin 2 should have prepared resources.") + driverName := func(element any) string { + el := element.(*testutil.Sample) + return string(el.Metric[testutil.LabelName("driver_name")]) + } + + gomega.Expect(getKubeletMetrics(ctx)).Should(gstruct.MatchKeys(gstruct.IgnoreExtras, gstruct.Keys{ + "dra_resource_claims_in_use": gstruct.MatchAllElements(driverName, gstruct.Elements{ + "": timelessSample(1), + kubeletPlugin1Name: timelessSample(1), + kubeletPlugin2Name: timelessSample(1), + }), + }), "metrics while pod is running") + + ginkgo.By("delete pod") + err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}) + framework.ExpectNoError(err) + err = e2epod.WaitForPodNotFoundInNamespace(ctx, f.ClientSet, pod.Name, pod.Namespace, f.Timeouts.PodDelete) + framework.ExpectNoError(err) + gomega.Expect(kubeletPlugin1.GetGRPCCalls()).Should(testdrivergomega.NodeUnprepareResourcesSucceeded, "Plugin 2 should have unprepared resources.") + gomega.Expect(kubeletPlugin2.GetGRPCCalls()).Should(testdrivergomega.NodeUnprepareResourcesSucceeded, "Plugin 2 should have unprepared resources.") + gomega.Expect(getKubeletMetrics(ctx)).Should(gstruct.MatchKeys(gstruct.IgnoreExtras, gstruct.Keys{ + "dra_resource_claims_in_use": gstruct.MatchAllElements(driverName, gstruct.Elements{ + "": timelessSample(0), + }), + }), "metrics while pod is running") + }) + ginkgo.It("must run pod if NodePrepareResources fails for one plugin and then succeeds", func(ctx context.Context) { _, kubeletPlugin2 := start(ctx)