DRA kubelet: add dra_resource_claims_in_use gauge vector

The new metric informs admins whether DRA in general (special "driver_name: <any>"
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

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

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)