kubernetes
/
apiserver
mirror of https://github.com/kubernetes/apiserver.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #131845 from serathius/unify-metrics 

Unify references to group resource in metrics

Kubernetes-commit: b5608aea94cfb54fea3a63e1d74235759d036c51
This commit is contained in:
Kubernetes Publisher 2025-05-22 07:32:35 -07:00
parent 08959fe41f a559cb8be1
commit 7937de32ff
23 changed files with 293 additions and 288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/endpoints/handlers/create.go
View File

@ -91,7 +91,7 @@ func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Int
return return
} }
body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource().String(), requestmetrics.Create) body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.Create)
if err != nil { if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error())) span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req) scope.err(err, w, req)

4
pkg/endpoints/handlers/delete.go
View File

@ -85,7 +85,7 @@ func DeleteResource(r rest.GracefulDeleter, allowsOptions bool, scope *RequestSc
options := &metav1.DeleteOptions{} options := &metav1.DeleteOptions{}
if allowsOptions { if allowsOptions {
body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource().String(), requestmetrics.Delete) body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.Delete)
if err != nil { if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error())) span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req) scope.err(err, w, req)
@ -260,7 +260,7 @@ func DeleteCollection(r rest.CollectionDeleter, checkBody bool, scope *RequestSc
options := &metav1.DeleteOptions{} options := &metav1.DeleteOptions{}
if checkBody { if checkBody {
body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource().String(), requestmetrics.DeleteCollection) body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.DeleteCollection)
if err != nil { if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error())) span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req) scope.err(err, w, req)

7
pkg/endpoints/handlers/metrics/metrics.go
View File

@ -20,6 +20,7 @@ import (
"context" "context"
"sync" "sync"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/component-base/metrics" "k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/legacyregistry" "k8s.io/component-base/metrics/legacyregistry"
) )
@ -45,7 +46,7 @@ var (
Buckets: metrics.LinearBuckets(50000, 100000, 31), Buckets: metrics.LinearBuckets(50000, 100000, 31),
StabilityLevel: metrics.ALPHA, StabilityLevel: metrics.ALPHA,
}, },
[]string{"resource", "verb"}, []string{"group", "resource", "verb"},
) )
) )
@ -58,6 +59,6 @@ func Register() {
}) })
} }
func RecordRequestBodySize(ctx context.Context, resource string, verb RequestBodyVerb, size int) { func RecordRequestBodySize(ctx context.Context, groupResource schema.GroupResource, verb RequestBodyVerb, size int) {
RequestBodySizes.WithContext(ctx).WithLabelValues(resource, string(verb)).Observe(float64(size)) RequestBodySizes.WithContext(ctx).WithLabelValues(groupResource.Group, groupResource.Resource, string(verb)).Observe(float64(size))
} }

2
pkg/endpoints/handlers/patch.go
View File

@ -107,7 +107,7 @@ func PatchResource(r rest.Patcher, scope *RequestScope, admit admission.Interfac
return return
} }
patchBytes, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource().String(), requestmetrics.Patch) patchBytes, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.Patch)
if err != nil { if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(patchBytes)), attribute.String("err", err.Error())) span.AddEvent("limitedReadBody failed", attribute.Int("len", len(patchBytes)), attribute.String("err", err.Error()))
scope.err(err, w, req) scope.err(err, w, req)

8
pkg/endpoints/handlers/response.go
View File

@ -144,7 +144,7 @@ func (e *watchEmbeddedEncoder) embeddedIdentifier() runtime.Identifier {
// NOTE: watchEncoder is NOT thread-safe. // NOTE: watchEncoder is NOT thread-safe.
type watchEncoder struct { type watchEncoder struct {
ctx context.Context ctx context.Context
kind schema.GroupVersionKind groupVersionResource schema.GroupVersionResource
embeddedEncoder runtime.Encoder embeddedEncoder runtime.Encoder
encoder runtime.Encoder encoder runtime.Encoder
framer io.Writer framer io.Writer
@ -158,10 +158,10 @@ type watchEncoder struct {
identifiers map[watch.EventType]runtime.Identifier identifiers map[watch.EventType]runtime.Identifier
} }
func newWatchEncoder(ctx context.Context, kind schema.GroupVersionKind, embeddedEncoder runtime.Encoder, encoder runtime.Encoder, framer io.Writer, watchListTransformerFn watchListTransformerFunction) *watchEncoder { func newWatchEncoder(ctx context.Context, gvr schema.GroupVersionResource, embeddedEncoder runtime.Encoder, encoder runtime.Encoder, framer io.Writer, watchListTransformerFn watchListTransformerFunction) *watchEncoder {
return &watchEncoder{ return &watchEncoder{
ctx: ctx, ctx: ctx,
kind: kind, groupVersionResource: gvr,
embeddedEncoder: embeddedEncoder, embeddedEncoder: embeddedEncoder,
encoder: encoder, encoder: encoder,
framer: framer, framer: framer,
@ -203,7 +203,7 @@ func (e *watchEncoder) doEncode(obj runtime.Object, event watch.Event, w io.Writ
Type: string(event.Type), Type: string(event.Type),
Object: runtime.RawExtension{Raw: e.buffer.Bytes()}, Object: runtime.RawExtension{Raw: e.buffer.Bytes()},
} }
metrics.WatchEventsSizes.WithContext(e.ctx).WithLabelValues(e.kind.Group, e.kind.Version, e.kind.Kind).Observe(float64(len(outEvent.Object.Raw))) metrics.WatchEventsSizes.WithContext(e.ctx).WithLabelValues(e.groupVersionResource.Group, e.groupVersionResource.Version, e.groupVersionResource.Resource).Observe(float64(len(outEvent.Object.Raw)))
defer e.eventBuffer.Reset() defer e.eventBuffer.Reset()
if err := e.encoder.Encode(outEvent, e.eventBuffer); err != nil { if err := e.encoder.Encode(outEvent, e.eventBuffer); err != nil {

4
pkg/endpoints/handlers/rest.go
View File

@ -388,11 +388,11 @@ func limitedReadBody(req *http.Request, limit int64) ([]byte, error) {
return data, nil return data, nil
} }
func limitedReadBodyWithRecordMetric(ctx context.Context, req *http.Request, limit int64, resourceGroup string, verb requestmetrics.RequestBodyVerb) ([]byte, error) { func limitedReadBodyWithRecordMetric(ctx context.Context, req *http.Request, limit int64, groupResource schema.GroupResource, verb requestmetrics.RequestBodyVerb) ([]byte, error) {
readBody, err := limitedReadBody(req, limit) readBody, err := limitedReadBody(req, limit)
if err == nil { if err == nil {
// only record if we've read successfully // only record if we've read successfully
requestmetrics.RecordRequestBodySize(ctx, resourceGroup, verb, len(readBody)) requestmetrics.RecordRequestBodySize(ctx, groupResource, verb, len(readBody))
} }
return readBody, err return readBody, err
} }

70
pkg/endpoints/handlers/rest_test.go
View File

@ -136,40 +136,40 @@ func TestLimitedReadBody(t *testing.T) {
expectedMetrics: ` expectedMetrics: `
# HELP apiserver_request_body_size_bytes [ALPHA] Apiserver request body size in bytes broken out by resource and verb. # HELP apiserver_request_body_size_bytes [ALPHA] Apiserver request body size in bytes broken out by resource and verb.
# TYPE apiserver_request_body_size_bytes histogram # TYPE apiserver_request_body_size_bytes histogram
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="50000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="50000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="150000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="150000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="250000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="250000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="350000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="350000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="450000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="450000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="550000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="550000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="650000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="650000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="750000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="750000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="850000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="850000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="950000"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="950000"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.05e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.05e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.15e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.15e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.25e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.25e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.35e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.35e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.45e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.45e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.55e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.55e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.65e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.65e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.75e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.75e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.85e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.85e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="1.95e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="1.95e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.05e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.05e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.15e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.15e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.25e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.25e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.35e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.35e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.45e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.45e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.55e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.55e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.65e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.65e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.75e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.75e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.85e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.85e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="2.95e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="2.95e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="3.05e+06"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="3.05e+06"} 1
apiserver_request_body_size_bytes_bucket{resource="resource.group",verb="create",le="+Inf"} 1 apiserver_request_body_size_bytes_bucket{group="group",resource="resource",verb="create",le="+Inf"} 1
apiserver_request_body_size_bytes_sum{resource="resource.group",verb="create"} 4 apiserver_request_body_size_bytes_sum{group="group",resource="resource",verb="create"} 4
apiserver_request_body_size_bytes_count{resource="resource.group",verb="create"} 1 apiserver_request_body_size_bytes_count{group="group",resource="resource",verb="create"} 1
`, `,
expectedErr: false, expectedErr: false,
}, },
@ -186,7 +186,7 @@ func TestLimitedReadBody(t *testing.T) {
if err != nil { if err != nil {
t.Errorf("err not expected: got %v", err) t.Errorf("err not expected: got %v", err)
} }
_, err = limitedReadBodyWithRecordMetric(context.Background(), req, tc.limit, "resource.group", metrics.Create) _, err = limitedReadBodyWithRecordMetric(context.Background(), req, tc.limit, schema.GroupResource{Group: "group", Resource: "resource"}, metrics.Create)
if tc.expectedErr { if tc.expectedErr {
if err == nil { if err == nil {
t.Errorf("err expected: got nil") t.Errorf("err expected: got nil")

2
pkg/endpoints/handlers/update.go
View File

@ -75,7 +75,7 @@ func UpdateResource(r rest.Updater, scope *RequestScope, admit admission.Interfa
return return
} }
body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource().String(), requestmetrics.Update) body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.Update)
if err != nil { if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error())) span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req) scope.err(err, w, req)

10
pkg/endpoints/handlers/watch.go
View File

@ -246,8 +246,8 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
flusher.Flush() flusher.Flush()
kind := s.Scope.Kind gvr := s.Scope.Resource
watchEncoder := newWatchEncoder(req.Context(), kind, s.EmbeddedEncoder, s.Encoder, framer, s.watchListTransformerFn) watchEncoder := newWatchEncoder(req.Context(), gvr, s.EmbeddedEncoder, s.Encoder, framer, s.watchListTransformerFn)
ch := s.Watching.ResultChan() ch := s.Watching.ResultChan()
done := req.Context().Done() done := req.Context().Done()
@ -271,7 +271,7 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) {
// End of results. // End of results.
return return
} }
metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Inc() metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(gvr.Group, gvr.Version, gvr.Resource).Inc()
isWatchListLatencyRecordingRequired := shouldRecordWatchListLatency(event) isWatchListLatencyRecordingRequired := shouldRecordWatchListLatency(event)
if err := watchEncoder.Encode(event); err != nil { if err := watchEncoder.Encode(event); err != nil {
@ -315,8 +315,8 @@ func (s *WatchServer) HandleWS(ws *websocket.Conn) {
framer := newWebsocketFramer(ws, s.UseTextFraming) framer := newWebsocketFramer(ws, s.UseTextFraming)
kind := s.Scope.Kind gvr := s.Scope.Resource
watchEncoder := newWatchEncoder(context.TODO(), kind, s.EmbeddedEncoder, s.Encoder, framer, s.watchListTransformerFn) watchEncoder := newWatchEncoder(context.TODO(), gvr, s.EmbeddedEncoder, s.Encoder, framer, s.watchListTransformerFn)
ch := s.Watching.ResultChan() ch := s.Watching.ResultChan()
for { for {

8
pkg/endpoints/metrics/metrics.go
View File

@ -176,7 +176,7 @@ var (
Help: "Number of events sent in watch clients", Help: "Number of events sent in watch clients",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"group", "version", "kind"}, []string{"group", "version", "resource"},
) )
WatchEventsSizes = compbasemetrics.NewHistogramVec( WatchEventsSizes = compbasemetrics.NewHistogramVec(
&compbasemetrics.HistogramOpts{ &compbasemetrics.HistogramOpts{
@ -186,7 +186,7 @@ var (
Buckets: compbasemetrics.ExponentialBuckets(1024, 2.0, 8), // 1K, 2K, 4K, 8K, ..., 128K. Buckets: compbasemetrics.ExponentialBuckets(1024, 2.0, 8), // 1K, 2K, 4K, 8K, ..., 128K.
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"group", "version", "kind"}, []string{"group", "version", "resource"},
) )
// Because of volatility of the base metric this is pre-aggregated one. Instead of reporting current usage all the time // Because of volatility of the base metric this is pre-aggregated one. Instead of reporting current usage all the time
// it reports maximal usage during the last second. // it reports maximal usage during the last second.
@ -226,7 +226,7 @@ var (
Help: "Counter of apiserver self-requests broken out for each verb, API resource and subresource.", Help: "Counter of apiserver self-requests broken out for each verb, API resource and subresource.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"verb", "resource", "subresource"}, []string{"verb", "group", "resource", "subresource"},
) )
requestFilterDuration = compbasemetrics.NewHistogramVec( requestFilterDuration = compbasemetrics.NewHistogramVec(
@ -604,7 +604,7 @@ func MonitorRequest(req *http.Request, verb, group, version, resource, subresour
// MonitorRequest happens after authentication, so we can trust the username given by the request // MonitorRequest happens after authentication, so we can trust the username given by the request
info, ok := request.UserFrom(req.Context()) info, ok := request.UserFrom(req.Context())
if ok && info.GetName() == user.APIServerUser { if ok && info.GetName() == user.APIServerUser {
apiSelfRequestCounter.WithContext(req.Context()).WithLabelValues(reportedVerb, resource, subresource).Inc() apiSelfRequestCounter.WithContext(req.Context()).WithLabelValues(reportedVerb, group, resource, subresource).Inc()
} }
if deprecated { if deprecated {
deprecatedRequestGauge.WithContext(req.Context()).WithLabelValues(group, version, resource, subresource, removedRelease).Set(1) deprecatedRequestGauge.WithContext(req.Context()).WithLabelValues(group, version, resource, subresource, removedRelease).Set(1)

2
pkg/registry/generic/registry/store.go
View File

@ -1669,7 +1669,7 @@ func (e *Store) startObservingCount(period time.Duration, objectCountTracker flo
count = -1 count = -1
} }
metrics.UpdateObjectCount(resourceName, count) metrics.UpdateObjectCount(e.DefaultQualifiedResource, count)
if objectCountTracker != nil { if objectCountTracker != nil {
objectCountTracker.Set(resourceName, count) objectCountTracker.Set(resourceName, count)
} }

4
pkg/storage/cacher/cache_watcher.go
View File

@ -176,7 +176,7 @@ func (c *cacheWatcher) add(event *watchCacheEvent, timer *time.Timer) bool {
// This means that we couldn't send event to that watcher. // This means that we couldn't send event to that watcher.
// Since we don't want to block on it infinitely, // Since we don't want to block on it infinitely,
// we simply terminate it. // we simply terminate it.
metrics.TerminatedWatchersCounter.WithLabelValues(c.groupResource.String()).Inc() metrics.TerminatedWatchersCounter.WithLabelValues(c.groupResource.Group, c.groupResource.Resource).Inc()
// This means that we couldn't send event to that watcher. // This means that we couldn't send event to that watcher.
// Since we don't want to block on it infinitely, we simply terminate it. // Since we don't want to block on it infinitely, we simply terminate it.
@ -503,7 +503,7 @@ func (c *cacheWatcher) processInterval(ctx context.Context, cacheInterval *watch
} }
if initEventCount > 0 { if initEventCount > 0 {
metrics.InitCounter.WithLabelValues(c.groupResource.String()).Add(float64(initEventCount)) metrics.InitCounter.WithLabelValues(c.groupResource.Group, c.groupResource.Resource).Add(float64(initEventCount))
} }
processingTime := time.Since(startTime) processingTime := time.Since(startTime)
if processingTime > initProcessThreshold { if processingTime > initProcessThreshold {

8
pkg/storage/cacher/cacher.go
View File

@ -466,8 +466,8 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
func (c *Cacher) startCaching(stopChannel <-chan struct{}) { func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
c.watchCache.SetOnReplace(func() { c.watchCache.SetOnReplace(func() {
c.ready.setReady() c.ready.setReady()
klog.V(1).Infof("cacher (%v): initialized", c.groupResource.String()) klog.V(1).InfoS("cacher initialized", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
metrics.WatchCacheInitializations.WithLabelValues(c.groupResource.String()).Inc() metrics.WatchCacheInitializations.WithLabelValues(c.groupResource.Group, c.groupResource.Resource).Inc()
}) })
var err error var err error
defer func() { defer func() {
@ -812,7 +812,7 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
return err return err
} }
} }
metrics.RecordListCacheMetrics(c.resourcePrefix, indexUsed, len(resp.Items), listVal.Len()) metrics.RecordListCacheMetrics(c.groupResource, indexUsed, len(resp.Items), listVal.Len())
return nil return nil
} }
@ -891,7 +891,7 @@ func (c *Cacher) dispatchEvents() {
c.dispatchEvent(&event) c.dispatchEvent(&event)
} }
lastProcessedResourceVersion = event.ResourceVersion lastProcessedResourceVersion = event.ResourceVersion
metrics.EventsCounter.WithLabelValues(c.groupResource.String()).Inc() metrics.EventsCounter.WithLabelValues(c.groupResource.Group, c.groupResource.Resource).Inc()
case <-bookmarkTimer.C(): case <-bookmarkTimer.C():
bookmarkTimer.Reset(wait.Jitter(time.Second, 0.25)) bookmarkTimer.Reset(wait.Jitter(time.Second, 0.25))
bookmarkEvent := &watchCacheEvent{ bookmarkEvent := &watchCacheEvent{

6
pkg/storage/cacher/cacher_whitebox_test.go
View File

@ -492,7 +492,7 @@ func TestConsistentReadFallback(t *testing.T) {
expectMetric: ` expectMetric: `
# HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache. # HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache.
# TYPE apiserver_watch_cache_consistent_read_total counter # TYPE apiserver_watch_cache_consistent_read_total counter
apiserver_watch_cache_consistent_read_total{fallback="false", resource="pods", success="true"} 1 apiserver_watch_cache_consistent_read_total{fallback="false", group="", resource="pods", success="true"} 1
`, `,
}, },
{ {
@ -506,7 +506,7 @@ apiserver_watch_cache_consistent_read_total{fallback="false", resource="pods", s
expectMetric: ` expectMetric: `
# HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache. # HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache.
# TYPE apiserver_watch_cache_consistent_read_total counter # TYPE apiserver_watch_cache_consistent_read_total counter
apiserver_watch_cache_consistent_read_total{fallback="true", resource="pods", success="true"} 1 apiserver_watch_cache_consistent_read_total{fallback="true", group="", resource="pods", success="true"} 1
`, `,
}, },
{ {
@ -521,7 +521,7 @@ apiserver_watch_cache_consistent_read_total{fallback="true", resource="pods", su
expectMetric: ` expectMetric: `
# HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache. # HELP apiserver_watch_cache_consistent_read_total [ALPHA] Counter for consistent reads from cache.
# TYPE apiserver_watch_cache_consistent_read_total counter # TYPE apiserver_watch_cache_consistent_read_total counter
apiserver_watch_cache_consistent_read_total{fallback="true", resource="pods", success="false"} 1 apiserver_watch_cache_consistent_read_total{fallback="true", group="", resource="pods", success="false"} 1
`, `,
}, },
{ {

27
pkg/storage/cacher/delegator.go
View File

@ -31,6 +31,7 @@ import (
"k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch" "k8s.io/apimachinery/pkg/watch"
"k8s.io/apiserver/pkg/audit" "k8s.io/apiserver/pkg/audit"
@ -64,7 +65,7 @@ func NewCacheDelegator(cacher *Cacher, storage storage.Interface) *CacheDelegato
stopCh: make(chan struct{}), stopCh: make(chan struct{}),
} }
if ConsistencyCheckerEnabled { if ConsistencyCheckerEnabled {
d.checker = newConsistencyChecker(cacher.resourcePrefix, cacher.newListFunc, cacher, storage) d.checker = newConsistencyChecker(cacher.resourcePrefix, cacher.groupResource, cacher.newListFunc, cacher, storage)
d.wg.Add(1) d.wg.Add(1)
go func() { go func() {
defer d.wg.Done() defer d.wg.Done()
@ -231,12 +232,12 @@ func (c *CacheDelegator) GetList(ctx context.Context, key string, opts storage.L
if err != nil { if err != nil {
success = "false" success = "false"
} }
metrics.ConsistentReadTotal.WithLabelValues(c.cacher.resourcePrefix, success, fallback).Add(1) metrics.ConsistentReadTotal.WithLabelValues(c.cacher.groupResource.Group, c.cacher.groupResource.Resource, success, fallback).Add(1)
} }
return err return err
} }
if result.ConsistentRead { if result.ConsistentRead {
metrics.ConsistentReadTotal.WithLabelValues(c.cacher.resourcePrefix, success, fallback).Add(1) metrics.ConsistentReadTotal.WithLabelValues(c.cacher.groupResource.Group, c.cacher.groupResource.Resource, success, fallback).Add(1)
} }
return nil return nil
} }
@ -286,8 +287,9 @@ func (c *CacheDelegator) Stop() {
c.wg.Wait() c.wg.Wait()
} }
func newConsistencyChecker(resourcePrefix string, newListFunc func() runtime.Object, cacher getListerReady, etcd getLister) *consistencyChecker { func newConsistencyChecker(resourcePrefix string, groupResource schema.GroupResource, newListFunc func() runtime.Object, cacher getListerReady, etcd getLister) *consistencyChecker {
return &consistencyChecker{ return &consistencyChecker{
groupResource: groupResource,
resourcePrefix: resourcePrefix, resourcePrefix: resourcePrefix,
newListFunc: newListFunc, newListFunc: newListFunc,
cacher: cacher, cacher: cacher,
@ -297,6 +299,7 @@ func newConsistencyChecker(resourcePrefix string, newListFunc func() runtime.Obj
type consistencyChecker struct { type consistencyChecker struct {
resourcePrefix string resourcePrefix string
groupResource schema.GroupResource
newListFunc func() runtime.Object newListFunc func() runtime.Object
cacher getListerReady cacher getListerReady
@ -318,25 +321,25 @@ func (c consistencyChecker) startChecking(stopCh <-chan struct{}) {
return false, nil return false, nil
}) })
if err != nil { if err != nil {
klog.InfoS("Cache consistency check exiting", "resource", c.resourcePrefix, "err", err) klog.InfoS("Cache consistency check exiting", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "err", err)
} }
} }
func (c *consistencyChecker) check(ctx context.Context) { func (c *consistencyChecker) check(ctx context.Context) {
digests, err := c.calculateDigests(ctx) digests, err := c.calculateDigests(ctx)
if err != nil { if err != nil {
klog.ErrorS(err, "Cache consistency check error", "resource", c.resourcePrefix) klog.ErrorS(err, "Cache consistency check error", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
metrics.StorageConsistencyCheckTotal.WithLabelValues(c.resourcePrefix, "error").Inc() metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "error").Inc()
return return
} }
if digests.CacheDigest == digests.EtcdDigest { if digests.CacheDigest == digests.EtcdDigest {
klog.V(3).InfoS("Cache consistency check passed", "resource", c.resourcePrefix, "resourceVersion", digests.ResourceVersion, "digest", digests.CacheDigest) klog.V(3).InfoS("Cache consistency check passed", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "digest", digests.CacheDigest)
metrics.StorageConsistencyCheckTotal.WithLabelValues(c.resourcePrefix, "success").Inc() metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "success").Inc()
} else { } else {
klog.ErrorS(nil, "Cache consistency check failed", "resource", c.resourcePrefix, "resourceVersion", digests.ResourceVersion, "etcdDigest", digests.EtcdDigest, "cacheDigest", digests.CacheDigest) klog.ErrorS(nil, "Cache consistency check failed", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "etcdDigest", digests.EtcdDigest, "cacheDigest", digests.CacheDigest)
metrics.StorageConsistencyCheckTotal.WithLabelValues(c.resourcePrefix, "failure").Inc() metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "failure").Inc()
// Panic on internal consistency checking enabled only by environment variable. R // Panic on internal consistency checking enabled only by environment variable. R
panic(fmt.Sprintf("Cache consistency check failed, resource: %q, resourceVersion: %q, etcdDigest: %q, cacheDigest: %q", c.resourcePrefix, digests.ResourceVersion, digests.EtcdDigest, digests.CacheDigest)) panic(fmt.Sprintf("Cache consistency check failed, group: %q, resource: %q, resourceVersion: %q, etcdDigest: %q, cacheDigest: %q", c.groupResource.Group, c.groupResource.Resource, digests.ResourceVersion, digests.EtcdDigest, digests.CacheDigest))
} }
} }

4
pkg/storage/cacher/delegator_test.go
View File

@ -22,6 +22,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apiserver/pkg/apis/example" "k8s.io/apiserver/pkg/apis/example"
"k8s.io/apiserver/pkg/storage" "k8s.io/apiserver/pkg/storage"
) )
@ -34,7 +35,6 @@ func TestCalculateDigest(t *testing.T) {
cacherReady bool cacherReady bool
cacherItems []example.Pod cacherItems []example.Pod
etcdItems []example.Pod etcdItems []example.Pod
resourcePrefix string
expectListKey string expectListKey string
expectDigest storageDigest expectDigest storageDigest
@ -178,7 +178,7 @@ func TestCalculateDigest(t *testing.T) {
}, },
}, },
} }
checker := newConsistencyChecker(tc.resourcePrefix, newListFunc, cacher, etcd) checker := newConsistencyChecker("", schema.GroupResource{}, newListFunc, cacher, etcd)
digest, err := checker.calculateDigests(context.Background()) digest, err := checker.calculateDigests(context.Background())
if (err != nil) != tc.expectErr { if (err != nil) != tc.expectErr {
t.Fatalf("Expect error: %v, got: %v", tc.expectErr, err) t.Fatalf("Expect error: %v, got: %v", tc.expectErr, err)

51
pkg/storage/cacher/metrics/metrics.go
View File

@ -19,6 +19,7 @@ package metrics
import ( import (
"sync" "sync"
"k8s.io/apimachinery/pkg/runtime/schema"
compbasemetrics "k8s.io/component-base/metrics" compbasemetrics "k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/legacyregistry" "k8s.io/component-base/metrics/legacyregistry"
) )
@ -44,7 +45,7 @@ var (
Help: "Number of LIST requests served from watch cache", Help: "Number of LIST requests served from watch cache",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource_prefix", "index"}, []string{"group", "resource", "index"},
) )
listCacheNumFetched = compbasemetrics.NewCounterVec( listCacheNumFetched = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -53,7 +54,7 @@ var (
Help: "Number of objects read from watch cache in the course of serving a LIST request", Help: "Number of objects read from watch cache in the course of serving a LIST request",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource_prefix", "index"}, []string{"group", "resource", "index"},
) )
listCacheNumReturned = compbasemetrics.NewCounterVec( listCacheNumReturned = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -62,7 +63,7 @@ var (
Help: "Number of objects returned for a LIST request from watch cache", Help: "Number of objects returned for a LIST request from watch cache",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource_prefix"}, []string{"group", "resource"},
) )
InitCounter = compbasemetrics.NewCounterVec( InitCounter = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -71,7 +72,7 @@ var (
Help: "Counter of init events processed in watch cache broken by resource type.", Help: "Counter of init events processed in watch cache broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
EventsReceivedCounter = compbasemetrics.NewCounterVec( EventsReceivedCounter = compbasemetrics.NewCounterVec(
@ -82,7 +83,7 @@ var (
Help: "Counter of events received in watch cache broken by resource type.", Help: "Counter of events received in watch cache broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
EventsCounter = compbasemetrics.NewCounterVec( EventsCounter = compbasemetrics.NewCounterVec(
@ -93,7 +94,7 @@ var (
Help: "Counter of events dispatched in watch cache broken by resource type.", Help: "Counter of events dispatched in watch cache broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
TerminatedWatchersCounter = compbasemetrics.NewCounterVec( TerminatedWatchersCounter = compbasemetrics.NewCounterVec(
@ -103,7 +104,7 @@ var (
Help: "Counter of watchers closed due to unresponsiveness broken by resource type.", Help: "Counter of watchers closed due to unresponsiveness broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
watchCacheResourceVersion = compbasemetrics.NewGaugeVec( watchCacheResourceVersion = compbasemetrics.NewGaugeVec(
@ -114,7 +115,7 @@ var (
Help: "Current resource version of watch cache broken by resource type.", Help: "Current resource version of watch cache broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
watchCacheCapacityIncreaseTotal = compbasemetrics.NewCounterVec( watchCacheCapacityIncreaseTotal = compbasemetrics.NewCounterVec(
@ -124,7 +125,7 @@ var (
Help: "Total number of watch cache capacity increase events broken by resource type.", Help: "Total number of watch cache capacity increase events broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
watchCacheCapacityDecreaseTotal = compbasemetrics.NewCounterVec( watchCacheCapacityDecreaseTotal = compbasemetrics.NewCounterVec(
@ -134,7 +135,7 @@ var (
Help: "Total number of watch cache capacity decrease events broken by resource type.", Help: "Total number of watch cache capacity decrease events broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
WatchCacheCapacity = compbasemetrics.NewGaugeVec( WatchCacheCapacity = compbasemetrics.NewGaugeVec(
@ -144,7 +145,7 @@ var (
Help: "Total capacity of watch cache broken by resource type.", Help: "Total capacity of watch cache broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
WatchCacheInitializations = compbasemetrics.NewCounterVec( WatchCacheInitializations = compbasemetrics.NewCounterVec(
@ -155,7 +156,7 @@ var (
Help: "Counter of watch cache initializations broken by resource type.", Help: "Counter of watch cache initializations broken by resource type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
WatchCacheReadWait = compbasemetrics.NewHistogramVec( WatchCacheReadWait = compbasemetrics.NewHistogramVec(
@ -166,7 +167,7 @@ var (
Help: "Histogram of time spent waiting for a watch cache to become fresh.", Help: "Histogram of time spent waiting for a watch cache to become fresh.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
Buckets: []float64{0.005, 0.025, 0.05, 0.1, 0.2, 0.4, 0.6, 0.8, 1.0, 1.25, 1.5, 2, 3}, Buckets: []float64{0.005, 0.025, 0.05, 0.1, 0.2, 0.4, 0.6, 0.8, 1.0, 1.25, 1.5, 2, 3},
}, []string{"resource"}) }, []string{"group", "resource"})
ConsistentReadTotal = compbasemetrics.NewCounterVec( ConsistentReadTotal = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -175,7 +176,7 @@ var (
Name: "consistent_read_total", Name: "consistent_read_total",
Help: "Counter for consistent reads from cache.", Help: "Counter for consistent reads from cache.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, []string{"resource", "success", "fallback"}) }, []string{"group", "resource", "success", "fallback"})
StorageConsistencyCheckTotal = compbasemetrics.NewCounterVec( StorageConsistencyCheckTotal = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -183,7 +184,7 @@ var (
Name: "storage_consistency_checks_total", Name: "storage_consistency_checks_total",
Help: "Counter for status of consistency checks between etcd and watch cache", Help: "Counter for status of consistency checks between etcd and watch cache",
StabilityLevel: compbasemetrics.INTERNAL, StabilityLevel: compbasemetrics.INTERNAL,
}, []string{"resource", "status"}) }, []string{"group", "resource", "status"})
) )
var registerMetrics sync.Once var registerMetrics sync.Once
@ -211,23 +212,23 @@ func Register() {
} }
// RecordListCacheMetrics notes various metrics of the cost to serve a LIST request // RecordListCacheMetrics notes various metrics of the cost to serve a LIST request
func RecordListCacheMetrics(resourcePrefix, indexName string, numFetched, numReturned int) { func RecordListCacheMetrics(groupResource schema.GroupResource, indexName string, numFetched, numReturned int) {
listCacheCount.WithLabelValues(resourcePrefix, indexName).Inc() listCacheCount.WithLabelValues(groupResource.Group, groupResource.Resource, indexName).Inc()
listCacheNumFetched.WithLabelValues(resourcePrefix, indexName).Add(float64(numFetched)) listCacheNumFetched.WithLabelValues(groupResource.Group, groupResource.Resource, indexName).Add(float64(numFetched))
listCacheNumReturned.WithLabelValues(resourcePrefix).Add(float64(numReturned)) listCacheNumReturned.WithLabelValues(groupResource.Group, groupResource.Resource).Add(float64(numReturned))
} }
// RecordResourceVersion sets the current resource version for a given resource type. // RecordResourceVersion sets the current resource version for a given resource type.
func RecordResourceVersion(resourcePrefix string, resourceVersion uint64) { func RecordResourceVersion(groupResource schema.GroupResource, resourceVersion uint64) {
watchCacheResourceVersion.WithLabelValues(resourcePrefix).Set(float64(resourceVersion)) watchCacheResourceVersion.WithLabelValues(groupResource.Group, groupResource.Resource).Set(float64(resourceVersion))
} }
// RecordsWatchCacheCapacityChange record watchCache capacity resize(increase or decrease) operations. // RecordsWatchCacheCapacityChange record watchCache capacity resize(increase or decrease) operations.
func RecordsWatchCacheCapacityChange(objType string, old, new int) { func RecordsWatchCacheCapacityChange(groupResource schema.GroupResource, old, new int) {
WatchCacheCapacity.WithLabelValues(objType).Set(float64(new)) WatchCacheCapacity.WithLabelValues(groupResource.Group, groupResource.Resource).Set(float64(new))
if old < new { if old < new {
watchCacheCapacityIncreaseTotal.WithLabelValues(objType).Inc() watchCacheCapacityIncreaseTotal.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
return return
} }
watchCacheCapacityDecreaseTotal.WithLabelValues(objType).Inc() watchCacheCapacityDecreaseTotal.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
} }

14
pkg/storage/cacher/watch_cache.go
View File

@ -190,7 +190,7 @@ func newWatchCache(
if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) { if utilfeature.DefaultFeatureGate.Enabled(features.ListFromCacheSnapshot) {
wc.snapshots = newStoreSnapshotter() wc.snapshots = newStoreSnapshotter()
} }
metrics.WatchCacheCapacity.WithLabelValues(groupResource.String()).Set(float64(wc.capacity)) metrics.WatchCacheCapacity.WithLabelValues(groupResource.Group, groupResource.Resource).Set(float64(wc.capacity))
wc.cond = sync.NewCond(wc.RLocker()) wc.cond = sync.NewCond(wc.RLocker())
wc.indexValidator = wc.isIndexValidLocked wc.indexValidator = wc.isIndexValidLocked
@ -272,7 +272,7 @@ func (w *watchCache) objectToVersionedRuntimeObject(obj interface{}) (runtime.Ob
// processEvent is safe as long as there is at most one call to it in flight // processEvent is safe as long as there is at most one call to it in flight
// at any point in time. // at any point in time.
func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(*storeElement) error) error { func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(*storeElement) error) error {
metrics.EventsReceivedCounter.WithLabelValues(w.groupResource.String()).Inc() metrics.EventsReceivedCounter.WithLabelValues(w.groupResource.Group, w.groupResource.Resource).Inc()
key, err := w.keyFunc(event.Object) key, err := w.keyFunc(event.Object)
if err != nil { if err != nil {
@ -343,7 +343,7 @@ func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, upd
if w.eventHandler != nil { if w.eventHandler != nil {
w.eventHandler(wcEvent) w.eventHandler(wcEvent)
} }
metrics.RecordResourceVersion(w.groupResource.String(), resourceVersion) metrics.RecordResourceVersion(w.groupResource, resourceVersion)
return nil return nil
} }
@ -397,7 +397,7 @@ func (w *watchCache) doCacheResizeLocked(capacity int) {
newCache[i%capacity] = w.cache[i%w.capacity] newCache[i%capacity] = w.cache[i%w.capacity]
} }
w.cache = newCache w.cache = newCache
metrics.RecordsWatchCacheCapacityChange(w.groupResource.String(), w.capacity, capacity) metrics.RecordsWatchCacheCapacityChange(w.groupResource, w.capacity, capacity)
w.capacity = capacity w.capacity = capacity
} }
@ -426,7 +426,7 @@ func (w *watchCache) UpdateResourceVersion(resourceVersion string) {
} }
w.eventHandler(wcEvent) w.eventHandler(wcEvent)
} }
metrics.RecordResourceVersion(w.groupResource.String(), rv) metrics.RecordResourceVersion(w.groupResource, rv)
} }
// List returns list of pointers to <storeElement> objects. // List returns list of pointers to <storeElement> objects.
@ -441,7 +441,7 @@ func (w *watchCache) waitUntilFreshAndBlock(ctx context.Context, resourceVersion
startTime := w.clock.Now() startTime := w.clock.Now()
defer func() { defer func() {
if resourceVersion > 0 { if resourceVersion > 0 {
metrics.WatchCacheReadWait.WithContext(ctx).WithLabelValues(w.groupResource.String()).Observe(w.clock.Since(startTime).Seconds()) metrics.WatchCacheReadWait.WithContext(ctx).WithLabelValues(w.groupResource.Group, w.groupResource.Resource).Observe(w.clock.Since(startTime).Seconds())
} }
}() }()
@ -708,7 +708,7 @@ func (w *watchCache) Replace(objs []interface{}, resourceVersion string) error {
} }
w.cond.Broadcast() w.cond.Broadcast()
metrics.RecordResourceVersion(w.groupResource.String(), version) metrics.RecordResourceVersion(w.groupResource, version)
klog.V(3).Infof("Replaced watchCache (rev: %v) ", resourceVersion) klog.V(3).Infof("Replaced watchCache (rev: %v) ", resourceVersion)
return nil return nil
} }

32
pkg/storage/cacher/watch_cache_test.go
View File

@ -1268,22 +1268,22 @@ func TestHistogramCacheReadWait(t *testing.T) {
want: ` want: `
# HELP apiserver_watch_cache_read_wait_seconds [ALPHA] Histogram of time spent waiting for a watch cache to become fresh. # HELP apiserver_watch_cache_read_wait_seconds [ALPHA] Histogram of time spent waiting for a watch cache to become fresh.
# TYPE apiserver_watch_cache_read_wait_seconds histogram # TYPE apiserver_watch_cache_read_wait_seconds histogram
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.005"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.005"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.025"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.025"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.05"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.05"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.1"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.1"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.2"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.2"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.4"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.4"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.6"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.6"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="0.8"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="0.8"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="1"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="1"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="1.25"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="1.25"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="1.5"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="1.5"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="2"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="2"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="3"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="3"} 1
apiserver_watch_cache_read_wait_seconds_bucket{resource="pods",le="+Inf"} 1 apiserver_watch_cache_read_wait_seconds_bucket{group="",resource="pods",le="+Inf"} 1
apiserver_watch_cache_read_wait_seconds_sum{resource="pods"} 0 apiserver_watch_cache_read_wait_seconds_sum{group="",resource="pods"} 0
apiserver_watch_cache_read_wait_seconds_count{resource="pods"} 1 apiserver_watch_cache_read_wait_seconds_count{group="",resource="pods"} 1
`, `,
}, },
{ {

56
pkg/storage/etcd3/metrics/metrics.go
View File

@ -22,6 +22,7 @@ import (
"sync" "sync"
"time" "time"
"k8s.io/apimachinery/pkg/runtime/schema"
compbasemetrics "k8s.io/component-base/metrics" compbasemetrics "k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/legacyregistry" "k8s.io/component-base/metrics/legacyregistry"
"k8s.io/klog/v2" "k8s.io/klog/v2"
@ -48,7 +49,7 @@ var (
4, 5, 6, 8, 10, 15, 20, 30, 45, 60}, 4, 5, 6, 8, 10, 15, 20, 30, 45, 60},
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"operation", "type"}, []string{"operation", "group", "resource"},
) )
etcdRequestCounts = compbasemetrics.NewCounterVec( etcdRequestCounts = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -56,7 +57,7 @@ var (
Help: "Etcd request counts for each operation and object type.", Help: "Etcd request counts for each operation and object type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"operation", "type"}, []string{"operation", "group", "resource"},
) )
etcdRequestErrorCounts = compbasemetrics.NewCounterVec( etcdRequestErrorCounts = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -64,7 +65,7 @@ var (
Help: "Etcd failed request counts for each operation and object type.", Help: "Etcd failed request counts for each operation and object type.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"operation", "type"}, []string{"operation", "group", "resource"},
) )
objectCounts = compbasemetrics.NewGaugeVec( objectCounts = compbasemetrics.NewGaugeVec(
&compbasemetrics.GaugeOpts{ &compbasemetrics.GaugeOpts{
@ -93,7 +94,7 @@ var (
Help: "Number of etcd events received split by kind.", Help: "Number of etcd events received split by kind.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
etcdBookmarkCounts = compbasemetrics.NewGaugeVec( etcdBookmarkCounts = compbasemetrics.NewGaugeVec(
&compbasemetrics.GaugeOpts{ &compbasemetrics.GaugeOpts{
@ -101,7 +102,7 @@ var (
Help: "Number of etcd bookmarks (progress notify events) split by kind.", Help: "Number of etcd bookmarks (progress notify events) split by kind.",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
etcdLeaseObjectCounts = compbasemetrics.NewHistogramVec( etcdLeaseObjectCounts = compbasemetrics.NewHistogramVec(
&compbasemetrics.HistogramOpts{ &compbasemetrics.HistogramOpts{
@ -118,7 +119,7 @@ var (
Help: "Number of LIST requests served from storage", Help: "Number of LIST requests served from storage",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
listStorageNumFetched = compbasemetrics.NewCounterVec( listStorageNumFetched = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -126,7 +127,7 @@ var (
Help: "Number of objects read from storage in the course of serving a LIST request", Help: "Number of objects read from storage in the course of serving a LIST request",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
listStorageNumSelectorEvals = compbasemetrics.NewCounterVec( listStorageNumSelectorEvals = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -134,7 +135,7 @@ var (
Help: "Number of objects tested in the course of serving a LIST request from storage", Help: "Number of objects tested in the course of serving a LIST request from storage",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
listStorageNumReturned = compbasemetrics.NewCounterVec( listStorageNumReturned = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -142,7 +143,7 @@ var (
Help: "Number of objects returned for a LIST request from storage", Help: "Number of objects returned for a LIST request from storage",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
decodeErrorCounts = compbasemetrics.NewCounterVec( decodeErrorCounts = compbasemetrics.NewCounterVec(
&compbasemetrics.CounterOpts{ &compbasemetrics.CounterOpts{
@ -151,7 +152,7 @@ var (
Help: "Number of stored object decode errors split by object type", Help: "Number of stored object decode errors split by object type",
StabilityLevel: compbasemetrics.ALPHA, StabilityLevel: compbasemetrics.ALPHA,
}, },
[]string{"resource"}, []string{"group", "resource"},
) )
) )
@ -179,34 +180,33 @@ func Register() {
} }
// UpdateObjectCount sets the apiserver_storage_object_counts metric. // UpdateObjectCount sets the apiserver_storage_object_counts metric.
func UpdateObjectCount(resourcePrefix string, count int64) { func UpdateObjectCount(groupResource schema.GroupResource, count int64) {
objectCounts.WithLabelValues(resourcePrefix).Set(float64(count)) objectCounts.WithLabelValues(groupResource.String()).Set(float64(count))
} }
// RecordEtcdRequest updates and sets the etcd_request_duration_seconds, // RecordEtcdRequest updates and sets the etcd_request_duration_seconds,
// etcd_request_total, etcd_request_errors_total metrics. // etcd_request_total, etcd_request_errors_total metrics.
func RecordEtcdRequest(verb, resource string, err error, startTime time.Time) { func RecordEtcdRequest(verb string, groupResource schema.GroupResource, err error, startTime time.Time) {
v := []string{verb, resource} etcdRequestLatency.WithLabelValues(verb, groupResource.Group, groupResource.Resource).Observe(sinceInSeconds(startTime))
etcdRequestLatency.WithLabelValues(v...).Observe(sinceInSeconds(startTime)) etcdRequestCounts.WithLabelValues(verb, groupResource.Group, groupResource.Resource).Inc()
etcdRequestCounts.WithLabelValues(v...).Inc()
if err != nil { if err != nil {
etcdRequestErrorCounts.WithLabelValues(v...).Inc() etcdRequestErrorCounts.WithLabelValues(verb, groupResource.Group, groupResource.Resource).Inc()
} }
} }
// RecordEtcdEvent updated the etcd_events_received_total metric. // RecordEtcdEvent updated the etcd_events_received_total metric.
func RecordEtcdEvent(resource string) { func RecordEtcdEvent(groupResource schema.GroupResource) {
etcdEventsReceivedCounts.WithLabelValues(resource).Inc() etcdEventsReceivedCounts.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
} }
// RecordEtcdBookmark updates the etcd_bookmark_counts metric. // RecordEtcdBookmark updates the etcd_bookmark_counts metric.
func RecordEtcdBookmark(resource string) { func RecordEtcdBookmark(groupResource schema.GroupResource) {
etcdBookmarkCounts.WithLabelValues(resource).Inc() etcdBookmarkCounts.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
} }
// RecordDecodeError sets the storage_decode_errors metrics. // RecordDecodeError sets the storage_decode_errors metrics.
func RecordDecodeError(resource string) { func RecordDecodeError(groupResource schema.GroupResource) {
decodeErrorCounts.WithLabelValues(resource).Inc() decodeErrorCounts.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
} }
// Reset resets the etcd_request_duration_seconds metric. // Reset resets the etcd_request_duration_seconds metric.
@ -240,11 +240,11 @@ func UpdateLeaseObjectCount(count int64) {
} }
// RecordListEtcd3Metrics notes various metrics of the cost to serve a LIST request // RecordListEtcd3Metrics notes various metrics of the cost to serve a LIST request
func RecordStorageListMetrics(resource string, numFetched, numEvald, numReturned int) { func RecordStorageListMetrics(groupResource schema.GroupResource, numFetched, numEvald, numReturned int) {
listStorageCount.WithLabelValues(resource).Inc() listStorageCount.WithLabelValues(groupResource.Group, groupResource.Resource).Inc()
listStorageNumFetched.WithLabelValues(resource).Add(float64(numFetched)) listStorageNumFetched.WithLabelValues(groupResource.Group, groupResource.Resource).Add(float64(numFetched))
listStorageNumSelectorEvals.WithLabelValues(resource).Add(float64(numEvald)) listStorageNumSelectorEvals.WithLabelValues(groupResource.Group, groupResource.Resource).Add(float64(numEvald))
listStorageNumReturned.WithLabelValues(resource).Add(float64(numReturned)) listStorageNumReturned.WithLabelValues(groupResource.Group, groupResource.Resource).Add(float64(numReturned))
} }
type Monitor interface { type Monitor interface {

136
pkg/storage/etcd3/metrics/metrics_test.go
View File

@ -23,6 +23,7 @@ import (
"testing" "testing"
"time" "time"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/component-base/metrics" "k8s.io/component-base/metrics"
"k8s.io/component-base/metrics/testutil" "k8s.io/component-base/metrics/testutil"
) )
@ -31,20 +32,19 @@ func TestRecordDecodeError(t *testing.T) {
registry := metrics.NewKubeRegistry() registry := metrics.NewKubeRegistry()
defer registry.Reset() defer registry.Reset()
registry.Register(decodeErrorCounts) registry.Register(decodeErrorCounts)
resourceName := "pods"
testedMetrics := "apiserver_storage_decode_errors_total" testedMetrics := "apiserver_storage_decode_errors_total"
testCases := []struct { testCases := []struct {
desc string desc string
resource string resource schema.GroupResource
want string want string
}{ }{
{ {
desc: "test success", desc: "test success",
resource: resourceName, resource: schema.GroupResource{Resource: "pods"},
want: ` want: `
# HELP apiserver_storage_decode_errors_total [ALPHA] Number of stored object decode errors split by object type # HELP apiserver_storage_decode_errors_total [ALPHA] Number of stored object decode errors split by object type
# TYPE apiserver_storage_decode_errors_total counter # TYPE apiserver_storage_decode_errors_total counter
apiserver_storage_decode_errors_total{resource="pods"} 1 apiserver_storage_decode_errors_total{group="",resource="pods"} 1
`, `,
}, },
} }
@ -82,7 +82,7 @@ func TestRecordEtcdRequest(t *testing.T) {
testCases := []struct { testCases := []struct {
desc string desc string
operation string operation string
typ string groupResource schema.GroupResource
err error err error
startTime time.Time startTime time.Time
want string want string
@ -90,82 +90,82 @@ func TestRecordEtcdRequest(t *testing.T) {
{ {
desc: "success_request", desc: "success_request",
operation: "foo", operation: "foo",
typ: "bar", groupResource: schema.GroupResource{Group: "bar", Resource: "baz"},
err: nil, err: nil,
startTime: time.Unix(0, 0), // 0.3s startTime: time.Unix(0, 0), // 0.3s
want: `# HELP etcd_request_duration_seconds [ALPHA] Etcd request latency in seconds for each operation and object type. want: `# HELP etcd_request_duration_seconds [ALPHA] Etcd request latency in seconds for each operation and object type.
# TYPE etcd_request_duration_seconds histogram # TYPE etcd_request_duration_seconds histogram
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.005"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.005"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.025"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.025"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.05"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.05"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.1"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.1"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.2"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.2"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.4"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.4"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.6"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.6"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.8"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.8"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1.25"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1.25"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1.5"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1.5"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="2"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="2"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="3"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="3"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="4"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="4"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="5"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="5"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="6"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="6"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="8"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="8"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="10"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="10"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="15"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="15"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="20"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="20"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="30"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="30"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="45"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="45"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="60"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="60"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="+Inf"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="+Inf"} 1
etcd_request_duration_seconds_sum{operation="foo",type="bar"} 0.3 etcd_request_duration_seconds_sum{group="bar",operation="foo",resource="baz"} 0.3
etcd_request_duration_seconds_count{operation="foo",type="bar"} 1 etcd_request_duration_seconds_count{group="bar",operation="foo",resource="baz"} 1
# HELP etcd_requests_total [ALPHA] Etcd request counts for each operation and object type. # HELP etcd_requests_total [ALPHA] Etcd request counts for each operation and object type.
# TYPE etcd_requests_total counter # TYPE etcd_requests_total counter
etcd_requests_total{operation="foo",type="bar"} 1 etcd_requests_total{group="bar",operation="foo",resource="baz"} 1
`, `,
}, },
{ {
desc: "failed_request", desc: "failed_request",
operation: "foo", operation: "foo",
typ: "bar", groupResource: schema.GroupResource{Group: "bar", Resource: "baz"},
err: errors.New("some error"), err: errors.New("some error"),
startTime: time.Unix(0, 0), // 0.3s startTime: time.Unix(0, 0), // 0.3s
want: `# HELP etcd_request_duration_seconds [ALPHA] Etcd request latency in seconds for each operation and object type. want: `# HELP etcd_request_duration_seconds [ALPHA] Etcd request latency in seconds for each operation and object type.
# TYPE etcd_request_duration_seconds histogram # TYPE etcd_request_duration_seconds histogram
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.005"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.005"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.025"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.025"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.05"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.05"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.1"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.1"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.2"} 0 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.2"} 0
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.4"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.4"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.6"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.6"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="0.8"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="0.8"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1.25"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1.25"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="1.5"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="1.5"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="2"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="2"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="3"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="3"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="4"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="4"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="5"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="5"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="6"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="6"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="8"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="8"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="10"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="10"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="15"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="15"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="20"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="20"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="30"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="30"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="45"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="45"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="60"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="60"} 1
etcd_request_duration_seconds_bucket{operation="foo",type="bar",le="+Inf"} 1 etcd_request_duration_seconds_bucket{group="bar",operation="foo",resource="baz",le="+Inf"} 1
etcd_request_duration_seconds_sum{operation="foo",type="bar"} 0.3 etcd_request_duration_seconds_sum{group="bar",operation="foo",resource="baz"} 0.3
etcd_request_duration_seconds_count{operation="foo",type="bar"} 1 etcd_request_duration_seconds_count{group="bar",operation="foo",resource="baz"} 1
# HELP etcd_requests_total [ALPHA] Etcd request counts for each operation and object type. # HELP etcd_requests_total [ALPHA] Etcd request counts for each operation and object type.
# TYPE etcd_requests_total counter # TYPE etcd_requests_total counter
etcd_requests_total{operation="foo",type="bar"} 1 etcd_requests_total{group="bar",operation="foo",resource="baz"} 1
# HELP etcd_request_errors_total [ALPHA] Etcd failed request counts for each operation and object type. # HELP etcd_request_errors_total [ALPHA] Etcd failed request counts for each operation and object type.
# TYPE etcd_request_errors_total counter # TYPE etcd_request_errors_total counter
etcd_request_errors_total{operation="foo",type="bar"} 1 etcd_request_errors_total{group="bar",operation="foo",resource="baz"} 1
`, `,
}, },
} }
@ -173,7 +173,7 @@ etcd_request_errors_total{operation="foo",type="bar"} 1
for _, test := range testCases { for _, test := range testCases {
t.Run(test.desc, func(t *testing.T) { t.Run(test.desc, func(t *testing.T) {
defer registry.Reset() defer registry.Reset()
RecordEtcdRequest(test.operation, test.typ, test.err, test.startTime) RecordEtcdRequest(test.operation, test.groupResource, test.err, test.startTime)
if err := testutil.GatherAndCompare(registry, strings.NewReader(test.want), testedMetricsName...); err != nil { if err := testutil.GatherAndCompare(registry, strings.NewReader(test.want), testedMetricsName...); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -233,26 +233,26 @@ func TestUpdateObjectCount(t *testing.T) {
testCases := []struct { testCases := []struct {
desc string desc string
resource string resource schema.GroupResource
count int64 count int64
want string want string
}{ }{
{ {
desc: "successful fetch", desc: "successful fetch",
resource: "foo", resource: schema.GroupResource{Group: "foo", Resource: "bar"},
count: 10, count: 10,
want: `# HELP apiserver_storage_objects [STABLE] Number of stored objects at the time of last check split by kind. In case of a fetching error, the value will be -1. want: `# HELP apiserver_storage_objects [STABLE] Number of stored objects at the time of last check split by kind. In case of a fetching error, the value will be -1.
# TYPE apiserver_storage_objects gauge # TYPE apiserver_storage_objects gauge
apiserver_storage_objects{resource="foo"} 10 apiserver_storage_objects{resource="bar.foo"} 10
`, `,
}, },
{ {
desc: "failed fetch", desc: "failed fetch",
resource: "bar", resource: schema.GroupResource{Group: "foo", Resource: "bar"},
count: -1, count: -1,
want: `# HELP apiserver_storage_objects [STABLE] Number of stored objects at the time of last check split by kind. In case of a fetching error, the value will be -1. want: `# HELP apiserver_storage_objects [STABLE] Number of stored objects at the time of last check split by kind. In case of a fetching error, the value will be -1.
# TYPE apiserver_storage_objects gauge # TYPE apiserver_storage_objects gauge
apiserver_storage_objects{resource="bar"} -1 apiserver_storage_objects{resource="bar.foo"} -1
`, `,
}, },
} }

44
pkg/storage/etcd3/store.go
View File

@ -81,7 +81,6 @@ type store struct {
transformer value.Transformer transformer value.Transformer
pathPrefix string pathPrefix string
groupResource schema.GroupResource groupResource schema.GroupResource
groupResourceString string
watcher *watcher watcher *watcher
leaseManager *leaseManager leaseManager *leaseManager
decoder Decoder decoder Decoder
@ -185,7 +184,6 @@ func newStore(c *kubernetes.Client, codec runtime.Codec, newFunc, newListFunc fu
transformer: transformer, transformer: transformer,
pathPrefix: pathPrefix, pathPrefix: pathPrefix,
groupResource: groupResource, groupResource: groupResource,
groupResourceString: groupResource.String(),
watcher: w, watcher: w,
leaseManager: newDefaultLeaseManager(c.Client, leaseManagerConfig), leaseManager: newDefaultLeaseManager(c.Client, leaseManagerConfig),
decoder: decoder, decoder: decoder,
@ -217,7 +215,7 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
} }
startTime := time.Now() startTime := time.Now()
getResp, err := s.client.Kubernetes.Get(ctx, preparedKey, kubernetes.GetOptions{}) getResp, err := s.client.Kubernetes.Get(ctx, preparedKey, kubernetes.GetOptions{})
metrics.RecordEtcdRequest("get", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("get", s.groupResource, err, startTime)
if err != nil { if err != nil {
return err return err
} }
@ -239,7 +237,7 @@ func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, ou
err = s.decoder.Decode(data, out, getResp.KV.ModRevision) err = s.decoder.Decode(data, out, getResp.KV.ModRevision)
if err != nil { if err != nil {
recordDecodeError(s.groupResourceString, preparedKey) recordDecodeError(s.groupResource, preparedKey)
return err return err
} }
return nil return nil
@ -255,7 +253,8 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)), attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)),
attribute.String("key", key), attribute.String("key", key),
attribute.String("type", getTypeName(obj)), attribute.String("type", getTypeName(obj)),
attribute.String("resource", s.groupResourceString), attribute.String("group", s.groupResource.Group),
attribute.String("resource", s.groupResource.Resource),
) )
defer span.End(500 * time.Millisecond) defer span.End(500 * time.Millisecond)
if version, err := s.versioner.ObjectResourceVersion(obj); err == nil && version != 0 { if version, err := s.versioner.ObjectResourceVersion(obj); err == nil && version != 0 {
@ -289,7 +288,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
startTime := time.Now() startTime := time.Now()
txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, preparedKey, newData, 0, kubernetes.PutOptions{LeaseID: lease}) txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, preparedKey, newData, 0, kubernetes.PutOptions{LeaseID: lease})
metrics.RecordEtcdRequest("create", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("create", s.groupResource, err, startTime)
if err != nil { if err != nil {
span.AddEvent("Txn call failed", attribute.String("err", err.Error())) span.AddEvent("Txn call failed", attribute.String("err", err.Error()))
return err return err
@ -304,7 +303,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
err = s.decoder.Decode(data, out, txnResp.Revision) err = s.decoder.Decode(data, out, txnResp.Revision)
if err != nil { if err != nil {
span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error())) span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
recordDecodeError(s.groupResourceString, preparedKey) recordDecodeError(s.groupResource, preparedKey)
return err return err
} }
span.AddEvent("decode succeeded", attribute.Int("len", len(data))) span.AddEvent("decode succeeded", attribute.Int("len", len(data)))
@ -408,7 +407,7 @@ func (s *store) conditionalDelete(
txnResp, err := s.client.Kubernetes.OptimisticDelete(ctx, key, origState.rev, kubernetes.DeleteOptions{ txnResp, err := s.client.Kubernetes.OptimisticDelete(ctx, key, origState.rev, kubernetes.DeleteOptions{
GetOnFailure: true, GetOnFailure: true,
}) })
metrics.RecordEtcdRequest("delete", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("delete", s.groupResource, err, startTime)
if err != nil { if err != nil {
return err return err
} }
@ -425,7 +424,7 @@ func (s *store) conditionalDelete(
if !skipTransformDecode { if !skipTransformDecode {
err = s.decoder.Decode(origState.data, out, txnResp.Revision) err = s.decoder.Decode(origState.data, out, txnResp.Revision)
if err != nil { if err != nil {
recordDecodeError(s.groupResourceString, key) recordDecodeError(s.groupResource, key)
return err return err
} }
} }
@ -445,7 +444,8 @@ func (s *store) GuaranteedUpdate(
attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)), attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)),
attribute.String("key", key), attribute.String("key", key),
attribute.String("type", getTypeName(destination)), attribute.String("type", getTypeName(destination)),
attribute.String("resource", s.groupResourceString)) attribute.String("group", s.groupResource.Group),
attribute.String("resource", s.groupResource.Resource))
defer span.End(500 * time.Millisecond) defer span.End(500 * time.Millisecond)
v, err := conversion.EnforcePtr(destination) v, err := conversion.EnforcePtr(destination)
@ -542,7 +542,7 @@ func (s *store) GuaranteedUpdate(
if !origState.stale { if !origState.stale {
err = s.decoder.Decode(origState.data, destination, origState.rev) err = s.decoder.Decode(origState.data, destination, origState.rev)
if err != nil { if err != nil {
recordDecodeError(s.groupResourceString, preparedKey) recordDecodeError(s.groupResource, preparedKey)
return err return err
} }
return nil return nil
@ -571,7 +571,7 @@ func (s *store) GuaranteedUpdate(
GetOnFailure: true, GetOnFailure: true,
LeaseID: lease, LeaseID: lease,
}) })
metrics.RecordEtcdRequest("update", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("update", s.groupResource, err, startTime)
if err != nil { if err != nil {
span.AddEvent("Txn call failed", attribute.String("err", err.Error())) span.AddEvent("Txn call failed", attribute.String("err", err.Error()))
return err return err
@ -592,7 +592,7 @@ func (s *store) GuaranteedUpdate(
err = s.decoder.Decode(data, destination, txnResp.Revision) err = s.decoder.Decode(data, destination, txnResp.Revision)
if err != nil { if err != nil {
span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error())) span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
recordDecodeError(s.groupResourceString, preparedKey) recordDecodeError(s.groupResource, preparedKey)
return err return err
} }
span.AddEvent("decode succeeded", attribute.Int("len", len(data))) span.AddEvent("decode succeeded", attribute.Int("len", len(data)))
@ -632,7 +632,7 @@ func (s *store) Count(key string) (int64, error) {
startTime := time.Now() startTime := time.Now()
count, err := s.client.Kubernetes.Count(context.Background(), preparedKey, kubernetes.CountOptions{}) count, err := s.client.Kubernetes.Count(context.Background(), preparedKey, kubernetes.CountOptions{})
metrics.RecordEtcdRequest("listWithCount", preparedKey, err, startTime) metrics.RecordEtcdRequest("listWithCount", s.groupResource, err, startTime)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@ -726,7 +726,7 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
// get them recorded even in error cases. // get them recorded even in error cases.
defer func() { defer func() {
numReturn := v.Len() numReturn := v.Len()
metrics.RecordStorageListMetrics(s.groupResourceString, numFetched, numEvald, numReturn) metrics.RecordStorageListMetrics(s.groupResource, numFetched, numEvald, numReturn)
}() }()
metricsOp := "get" metricsOp := "get"
@ -742,7 +742,7 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
Limit: limit, Limit: limit,
Continue: continueKey, Continue: continueKey,
}) })
metrics.RecordEtcdRequest(metricsOp, s.groupResourceString, err, startTime) metrics.RecordEtcdRequest(metricsOp, s.groupResource, err, startTime)
if err != nil { if err != nil {
return interpretListError(err, len(opts.Predicate.Continue) > 0, continueKey, keyPrefix) return interpretListError(err, len(opts.Predicate.Continue) > 0, continueKey, keyPrefix)
} }
@ -794,7 +794,7 @@ func (s *store) GetList(ctx context.Context, key string, opts storage.ListOption
obj, err := s.decoder.DecodeListItem(ctx, data, uint64(kv.ModRevision), newItemFunc) obj, err := s.decoder.DecodeListItem(ctx, data, uint64(kv.ModRevision), newItemFunc)
if err != nil { if err != nil {
recordDecodeError(s.groupResourceString, string(kv.Key)) recordDecodeError(s.groupResource, string(kv.Key))
if done := aggregator.Aggregate(string(kv.Key), err); done { if done := aggregator.Aggregate(string(kv.Key), err); done {
return aggregator.Err() return aggregator.Err()
} }
@ -927,7 +927,7 @@ func (s *store) getCurrentState(ctx context.Context, key string, v reflect.Value
return func() (*objState, error) { return func() (*objState, error) {
startTime := time.Now() startTime := time.Now()
getResp, err := s.client.Kubernetes.Get(ctx, key, kubernetes.GetOptions{}) getResp, err := s.client.Kubernetes.Get(ctx, key, kubernetes.GetOptions{})
metrics.RecordEtcdRequest("get", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("get", s.groupResource, err, startTime)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -979,7 +979,7 @@ func (s *store) getState(ctx context.Context, kv *mvccpb.KeyValue, key string, v
state.stale = stale state.stale = stale
if err := s.decoder.Decode(state.data, state.obj, state.rev); err != nil { if err := s.decoder.Decode(state.data, state.obj, state.rev); err != nil {
recordDecodeError(s.groupResourceString, key) recordDecodeError(s.groupResource, key)
return nil, err return nil, err
} }
} }
@ -1073,9 +1073,9 @@ func (s *store) prepareKey(key string) (string, error) {
} }
// recordDecodeError record decode error split by object type. // recordDecodeError record decode error split by object type.
func recordDecodeError(resource string, key string) { func recordDecodeError(groupResource schema.GroupResource, key string) {
metrics.RecordDecodeError(resource) metrics.RecordDecodeError(groupResource)
klog.V(4).Infof("Decoding %s \"%s\" failed", resource, key) klog.V(4).Infof("Decoding %s \"%s\" failed", groupResource, key)
} }
// getTypeName returns type name of an object for reporting purposes. // getTypeName returns type name of an object for reporting purposes.

6
pkg/storage/etcd3/watcher.go
View File

@ -287,7 +287,7 @@ func (wc *watchChan) sync() error {
for { for {
startTime := time.Now() startTime := time.Now()
getResp, err = wc.watcher.client.KV.Get(wc.ctx, preparedKey, opts...) getResp, err = wc.watcher.client.KV.Get(wc.ctx, preparedKey, opts...)
metrics.RecordEtcdRequest(metricsOp, wc.watcher.groupResource.String(), err, startTime) metrics.RecordEtcdRequest(metricsOp, wc.watcher.groupResource, err, startTime)
if err != nil { if err != nil {
return interpretListError(err, true, preparedKey, wc.key) return interpretListError(err, true, preparedKey, wc.key)
} }
@ -397,12 +397,12 @@ func (wc *watchChan) startWatching(watchClosedCh chan struct{}, initialEventsEnd
} }
if wres.IsProgressNotify() { if wres.IsProgressNotify() {
wc.queueEvent(progressNotifyEvent(wres.Header.GetRevision())) wc.queueEvent(progressNotifyEvent(wres.Header.GetRevision()))
metrics.RecordEtcdBookmark(wc.watcher.groupResource.String()) metrics.RecordEtcdBookmark(wc.watcher.groupResource)
continue continue
} }
for _, e := range wres.Events { for _, e := range wres.Events {
metrics.RecordEtcdEvent(wc.watcher.groupResource.String()) metrics.RecordEtcdEvent(wc.watcher.groupResource)
parsedEvent, err := parseEvent(e) parsedEvent, err := parseEvent(e)
if err != nil { if err != nil {
logWatchChannelErr(err) logWatchChannelErr(err)