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