Merge pull request #131641 from pohly/dra-kubelet-in-use-metric

DRA kubelet: add dra_resource_claims_in_use gauge vector

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
 

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)
+	}
 }

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 <any> 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 <any> driver_name.
+# TYPE dra_resource_claims_in_use gauge
+dra_resource_claims_in_use{driver_name="<any>"} 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 <any> 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 <any> driver_name.
+# TYPE dra_resource_claims_in_use gauge
+dra_resource_claims_in_use{driver_name="<any>"} 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=<
+	<any>: 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 <any> 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 <any> driver_name.
+# TYPE dra_resource_claims_in_use gauge
+dra_resource_claims_in_use{driver_name="<any>"} 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=<
+	<any>: 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 <any> 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 <any> driver_name.
+# TYPE dra_resource_claims_in_use gauge
+dra_resource_claims_in_use{driver_name="<any>"} 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=<
+	<any>: 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 <any> 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 <any> driver_name.
+# TYPE dra_resource_claims_in_use gauge
+dra_resource_claims_in_use{driver_name="<any>"} 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 {

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)
 	}

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 */)
 

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)
+	}
+}

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 = "<any>"
 )
 
 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 <any> 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 <any> 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)

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 {

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{
+					"<any>":            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{
+					"<any>": 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)