From 7bc354e15d5ddb3436782a75da966a77edf95264 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stanislav=20L=C3=A1zni=C4=8Dka?= Date: Thu, 31 Jul 2025 12:51:11 +0200 Subject: [PATCH] kubelet: add metrics for in-memory stored image-pull-related records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Stanislav Láznička --- .../images/pullmanager/benchmarks_test.go | 12 +- .../images/pullmanager/mem_pullrecords.go | 14 +- .../pullmanager/mem_pullrecords_test.go | 16 +- pkg/kubelet/images/pullmanager/metrics.go | 19 +- .../images/pullmanager/metrics_test.go | 162 +++++++++++++----- 5 files changed, 171 insertions(+), 52 deletions(-) diff --git a/pkg/kubelet/images/pullmanager/benchmarks_test.go b/pkg/kubelet/images/pullmanager/benchmarks_test.go index 2a453e8a2bf..6eb8a32957e 100644 --- a/pkg/kubelet/images/pullmanager/benchmarks_test.go +++ b/pkg/kubelet/images/pullmanager/benchmarks_test.go @@ -303,8 +303,16 @@ func setupInMemRecordsAccessor(t testing.TB, cacheSize int, authoritative bool) fsAccessor := setupFSRecordsAccessor(t) memcacheAccessor := NewCachedPullRecordsAccessor(fsAccessor, int32(cacheSize), int32(cacheSize), int32(runtime.NumCPU())) - memcacheAccessor.intents.authoritative.Store(authoritative) - memcacheAccessor.pulledRecords.authoritative.Store(authoritative) + gotMeteredAccessor, ok := memcacheAccessor.(*meteringRecordsAccessor) + if !ok { + t.Fatalf("the tested accessor must be a metered records accessor, got %T", memcacheAccessor) + } + inMemAccessor, ok := gotMeteredAccessor.sizeExposedPullRecordsAccessor.(*cachedPullRecordsAccessor) + if !ok { + t.Fatalf("the metered accessor's delegate is not an inMemAccesor: %T", gotMeteredAccessor.sizeExposedPullRecordsAccessor) + } + inMemAccessor.intents.authoritative.Store(authoritative) + inMemAccessor.pulledRecords.authoritative.Store(authoritative) return memcacheAccessor } diff --git a/pkg/kubelet/images/pullmanager/mem_pullrecords.go b/pkg/kubelet/images/pullmanager/mem_pullrecords.go index 796a367b856..3ed9f7fb120 100644 --- a/pkg/kubelet/images/pullmanager/mem_pullrecords.go +++ b/pkg/kubelet/images/pullmanager/mem_pullrecords.go @@ -98,7 +98,7 @@ type cachedPullRecordsAccessor struct { pulledRecords *lruCache[string, kubeletconfiginternal.ImagePulledRecord] } -func NewCachedPullRecordsAccessor(delegate PullRecordsAccessor, intentsCacheSize, pulledRecordsCacheSize, stripedLocksSize int32) *cachedPullRecordsAccessor { +func NewCachedPullRecordsAccessor(delegate PullRecordsAccessor, intentsCacheSize, pulledRecordsCacheSize, stripedLocksSize int32) PullRecordsAccessor { intentsCacheSize = min(intentsCacheSize, 1024) pulledRecordsCacheSize = min(pulledRecordsCacheSize, 2000) @@ -119,7 +119,7 @@ func NewCachedPullRecordsAccessor(delegate PullRecordsAccessor, intentsCacheSize if err != nil { klog.InfoS("there was an error initializing the image pulled records cache, the cache will work in a non-authoritative mode until the pulled records are listed successfully", "error", err) } - return c + return NewMeteringRecordsAccessor(c, inMemIntentsPercent, inMemPulledRecordsPercent) } func (c *cachedPullRecordsAccessor) ListImagePullIntents() ([]*kubeletconfiginternal.ImagePullIntent, error) { @@ -245,6 +245,16 @@ func (c *cachedPullRecordsAccessor) DeleteImagePulledRecord(imageRef string) err return nil } +func (f *cachedPullRecordsAccessor) intentsSize() (uint, error) { + intentsUsage := f.intents.Len() * 100 / f.intents.maxSize + return uint(intentsUsage), nil +} + +func (f *cachedPullRecordsAccessor) pulledRecordsSize() (uint, error) { + pulledRecordsUsage := f.pulledRecords.Len() * 100 / f.pulledRecords.maxSize + return uint(pulledRecordsUsage), nil +} + func cacheRefreshingList[K comparable, V any]( cache *lruCache[K, V], delegateLocks *StripedLockSet, diff --git a/pkg/kubelet/images/pullmanager/mem_pullrecords_test.go b/pkg/kubelet/images/pullmanager/mem_pullrecords_test.go index 4cee939a515..48fc411afbd 100644 --- a/pkg/kubelet/images/pullmanager/mem_pullrecords_test.go +++ b/pkg/kubelet/images/pullmanager/mem_pullrecords_test.go @@ -297,9 +297,17 @@ func TestNewCachedPullRecordsAccessor(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { gotAccessor := NewCachedPullRecordsAccessor(tt.delegate, 50, 100, 10) + gotMeteredAccessor, ok := gotAccessor.(*meteringRecordsAccessor) + if !ok { + t.Fatalf("the tested accessor must be a metered records accessor, got %T", gotAccessor) + } + inMemAccessor, ok := gotMeteredAccessor.sizeExposedPullRecordsAccessor.(*cachedPullRecordsAccessor) + if !ok { + t.Fatalf("the metered accessor's delegate is not an inMemAccesor: %T", gotMeteredAccessor.sizeExposedPullRecordsAccessor) + } expectedCachedIntents := tt.wantCacheIntents - if err := cmpRecordsMapAndCache(expectedCachedIntents, gotAccessor.intents); err != nil { + if err := cmpRecordsMapAndCache(expectedCachedIntents, inMemAccessor.intents); err != nil { t.Errorf("NewCachedPullRecordsAccessor cache does not match: %v", err) } @@ -312,7 +320,7 @@ func TestNewCachedPullRecordsAccessor(t *testing.T) { t.Errorf("NewCachedPullRecordsAccessor().ListImagePullIntents() errors don't match = %v, want %v", intentsErr, tt.wantIntentsError) } expectedPulledRecords := tt.wantCachePulledRecords - if err := cmpRecordsMapAndCache(expectedPulledRecords, gotAccessor.pulledRecords); err != nil { + if err := cmpRecordsMapAndCache(expectedPulledRecords, inMemAccessor.pulledRecords); err != nil { t.Errorf("NewCachedPullRecordsAccessor cache does not match: %v", err) } @@ -325,11 +333,11 @@ func TestNewCachedPullRecordsAccessor(t *testing.T) { t.Errorf("NewCachedPullRecordsAccessor().ListImagePullIntents() errors don't match = %v, want %v", pulledRecordsErr, tt.wantPulledRecordsError) } - if gotIntentsAuthoritative := gotAccessor.intents.authoritative.Load(); gotIntentsAuthoritative != tt.wantIntentsAuthoritative { + if gotIntentsAuthoritative := inMemAccessor.intents.authoritative.Load(); gotIntentsAuthoritative != tt.wantIntentsAuthoritative { t.Errorf("NewCachedPullRecordsAccessor().intents.authoritative = %v, want %v", gotIntentsAuthoritative, tt.wantIntentsAuthoritative) } - if gotPulledRecordsAuthoritative := gotAccessor.pulledRecords.authoritative.Load(); gotPulledRecordsAuthoritative != tt.wantPulledRecordsAuthoritative { + if gotPulledRecordsAuthoritative := inMemAccessor.pulledRecords.authoritative.Load(); gotPulledRecordsAuthoritative != tt.wantPulledRecordsAuthoritative { t.Errorf("NewCachedPullRecordsAccessor().pulledRecords.authoritative = %v, want %v", gotPulledRecordsAuthoritative, tt.wantPulledRecordsAuthoritative) } }) diff --git a/pkg/kubelet/images/pullmanager/metrics.go b/pkg/kubelet/images/pullmanager/metrics.go index 9482cf84dde..7be9812210f 100644 --- a/pkg/kubelet/images/pullmanager/metrics.go +++ b/pkg/kubelet/images/pullmanager/metrics.go @@ -52,7 +52,22 @@ var ( StabilityLevel: metrics.ALPHA, }, ) - + inMemIntentsPercent = metrics.NewGauge( + &metrics.GaugeOpts{ + Subsystem: imageManagerSubsystem, + Name: "inmemory_pullintents_usage_percent", + Help: "The ImagePullIntents in-memory cache usage in percent.", + StabilityLevel: metrics.ALPHA, + }, + ) + inMemPulledRecordsPercent = metrics.NewGauge( + &metrics.GaugeOpts{ + Subsystem: imageManagerSubsystem, + Name: "inmemory_pulledrecords_usage_percent", + Help: "The ImagePulledRecords in-memory cache usage in percent.", + StabilityLevel: metrics.ALPHA, + }, + ) mustPullChecksTotal = metrics.NewCounterVec( &metrics.CounterOpts{ Subsystem: imageManagerSubsystem, @@ -67,6 +82,8 @@ var ( func init() { legacyregistry.MustRegister(fsPullIntentsSize) legacyregistry.MustRegister(fsPulledRecordsSize) + legacyregistry.MustRegister(inMemIntentsPercent) + legacyregistry.MustRegister(inMemPulledRecordsPercent) legacyregistry.MustRegister(mustPullChecksTotal) } diff --git a/pkg/kubelet/images/pullmanager/metrics_test.go b/pkg/kubelet/images/pullmanager/metrics_test.go index 7b5c38f78c6..02b01ba81fd 100644 --- a/pkg/kubelet/images/pullmanager/metrics_test.go +++ b/pkg/kubelet/images/pullmanager/metrics_test.go @@ -44,85 +44,130 @@ func TestFSPullRecordsMetrics(t *testing.T) { t.Fatal(err) } - cmpIntents(t, 0) - require.NoError(t, fsAccessor.WriteImagePullIntent("test-image:latest")) - cmpIntents(t, 1) + const pullIntentsCacheSize, pulledRecordsCacheSize, stripedSetLocksSize int32 = 5, 10, 1 + inMemoryAccessor := NewCachedPullRecordsAccessor(fsAccessor, pullIntentsCacheSize, pulledRecordsCacheSize, stripedSetLocksSize) + + cmpFSIntents(t, 0) + cmpMemIntents(t, 0) + require.NoError(t, inMemoryAccessor.WriteImagePullIntent("test-image:latest")) + cmpFSIntents(t, 1) + cmpMemIntents(t, 20) // Test that writing the same record does not increase the count - require.NoError(t, fsAccessor.WriteImagePullIntent("test-image:latest")) - require.NoError(t, fsAccessor.WriteImagePullIntent("test-image:latest")) - cmpIntents(t, 1) + require.NoError(t, inMemoryAccessor.WriteImagePullIntent("test-image:latest")) + require.NoError(t, inMemoryAccessor.WriteImagePullIntent("test-image:latest")) + cmpFSIntents(t, 1) + cmpMemIntents(t, 20) // Test adding more records - require.NoError(t, fsAccessor.WriteImagePullIntent("test-image:v1")) - require.NoError(t, fsAccessor.WriteImagePullIntent("test-image:v1.1")) - cmpIntents(t, 3) + require.NoError(t, inMemoryAccessor.WriteImagePullIntent("test-image:v1")) + require.NoError(t, inMemoryAccessor.WriteImagePullIntent("test-image:v1.1")) + cmpFSIntents(t, 3) + cmpMemIntents(t, 60) - cmpPulledRecords(t, 0) - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + cmpFSPulledRecords(t, 0) + cmpMemPulledRecords(t, 0) + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-latest-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - cmpPulledRecords(t, 1) + cmpFSPulledRecords(t, 1) + cmpMemPulledRecords(t, 10) // Test that writing the same record does not increase the count - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-latest-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-latest-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - cmpPulledRecords(t, 1) + cmpFSPulledRecords(t, 1) + cmpMemPulledRecords(t, 10) // Test adding more records - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-v1-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-v1.1-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - require.NoError(t, fsAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ ImageRef: "test-image-v1.2-ref", LastUpdatedTime: metav1.NewTime(time.Now()), })) - cmpPulledRecords(t, 4) + cmpFSPulledRecords(t, 4) + cmpMemPulledRecords(t, 40) - cmpIntents(t, 3) // double-check that intents count is not affected + // double-check that intents count is not affected + cmpFSIntents(t, 3) + cmpMemIntents(t, 60) // Test deletions - require.NoError(t, fsAccessor.DeleteImagePullIntent("test-image:latest")) - cmpIntents(t, 2) + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image:latest")) + cmpFSIntents(t, 2) + cmpMemIntents(t, 40) - require.NoError(t, fsAccessor.DeleteImagePullIntent("test-image:latest")) - require.NoError(t, fsAccessor.DeleteImagePullIntent("test-image:latest")) - cmpIntents(t, 2) + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image:latest")) + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image:latest")) + cmpFSIntents(t, 2) + cmpMemIntents(t, 40) - require.NoError(t, fsAccessor.DeleteImagePullIntent("test-image:v1")) - require.NoError(t, fsAccessor.DeleteImagePullIntent("test-image:v1.1")) - cmpIntents(t, 0) + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image:v1")) + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image:v1.1")) + cmpFSIntents(t, 0) + cmpMemIntents(t, 0) - cmpPulledRecords(t, 4) // double-check that pulled records count is not affected + // double-check that pulled records count is not affected + cmpFSPulledRecords(t, 4) + cmpMemPulledRecords(t, 40) // Test image pulled record deletions - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) - cmpPulledRecords(t, 3) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) + cmpFSPulledRecords(t, 3) + cmpMemPulledRecords(t, 30) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) - cmpPulledRecords(t, 3) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1.1-ref")) + cmpFSPulledRecords(t, 3) + cmpMemPulledRecords(t, 30) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1.2-ref")) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1-ref")) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-latest-ref")) - cmpPulledRecords(t, 0) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1.2-ref")) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1-ref")) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-latest-ref")) + cmpFSPulledRecords(t, 0) + cmpMemPulledRecords(t, 0) + + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v1-ref")) + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-latest-ref")) + cmpFSPulledRecords(t, 0) + cmpMemPulledRecords(t, 0) + + // test exceeding memory cache sizes + for i := range 20 { + require.NoError(t, inMemoryAccessor.WriteImagePullIntent(fmt.Sprintf("test-image-%d:latest", i))) + require.NoError(t, inMemoryAccessor.WriteImagePulledRecord(&kubeletconfig.ImagePulledRecord{ + ImageRef: fmt.Sprintf("test-image-v%d-ref", i), + LastUpdatedTime: metav1.NewTime(time.Now()), + })) + } + cmpFSIntents(t, 20) + cmpFSPulledRecords(t, 20) + cmpMemIntents(t, 100) + cmpMemPulledRecords(t, 100) + + // test removing some of the latest records from the cache + require.NoError(t, inMemoryAccessor.DeleteImagePullIntent("test-image-19:latest")) + cmpFSIntents(t, 19) + cmpMemIntents(t, 80) + + require.NoError(t, inMemoryAccessor.DeleteImagePulledRecord("test-image-v19-ref")) + cmpFSPulledRecords(t, 19) + cmpMemPulledRecords(t, 90) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-v1-ref")) - require.NoError(t, fsAccessor.DeleteImagePulledRecord("test-image-latest-ref")) - cmpPulledRecords(t, 0) } func TestMustAttemptPullMetrics(t *testing.T) { @@ -192,7 +237,7 @@ func TestMustAttemptPullMetrics(t *testing.T) { cmpMustAttemptPullMetrics(t, expectedMetrics) } -func cmpIntents(t *testing.T, expected uint) { +func cmpFSIntents(t *testing.T, expected uint) { t.Helper() const metricFormat = ` # HELP kubelet_imagemanager_ondisk_pullintents [ALPHA] Number of ImagePullIntents stored on disk. @@ -207,7 +252,7 @@ kubelet_imagemanager_ondisk_pullintents %d t.Errorf("failed to gather metrics: %v", err) } } -func cmpPulledRecords(t *testing.T, expected uint) { +func cmpFSPulledRecords(t *testing.T, expected uint) { t.Helper() const metricFormat = ` # HELP kubelet_imagemanager_ondisk_pulledrecords [ALPHA] Number of ImagePulledRecords stored on disk. @@ -223,6 +268,37 @@ kubelet_imagemanager_ondisk_pulledrecords %d } } +func cmpMemIntents(t *testing.T, expected uint) { + t.Helper() + const metricFormat = ` +# HELP kubelet_imagemanager_inmemory_pullintents_usage_percent [ALPHA] The ImagePullIntents in-memory cache usage in percent. +# TYPE kubelet_imagemanager_inmemory_pullintents_usage_percent gauge +kubelet_imagemanager_inmemory_pullintents_usage_percent %d +` + + err := metricstestutil.GatherAndCompare( + legacyregistry.DefaultGatherer, strings.NewReader(fmt.Sprintf(metricFormat, expected)), "kubelet_imagemanager_inmemory_pullintents_usage_percent", + ) + if err != nil { + t.Errorf("failed to gather metrics: %v", err) + } +} +func cmpMemPulledRecords(t *testing.T, expected uint) { + t.Helper() + const metricFormat = ` +# HELP kubelet_imagemanager_inmemory_pulledrecords_usage_percent [ALPHA] The ImagePulledRecords in-memory cache usage in percent. +# TYPE kubelet_imagemanager_inmemory_pulledrecords_usage_percent gauge +kubelet_imagemanager_inmemory_pulledrecords_usage_percent %d +` + + err := metricstestutil.GatherAndCompare( + legacyregistry.DefaultGatherer, strings.NewReader(fmt.Sprintf(metricFormat, expected)), "kubelet_imagemanager_inmemory_pulledrecords_usage_percent", + ) + if err != nil { + t.Errorf("failed to gather metrics: %v", err) + } +} + func cmpMustAttemptPullMetrics(t *testing.T, labelMap map[string]int) { t.Helper() const metricFormat = `