From 71c25c3a552539de503e910c570c6a194647e46a Mon Sep 17 00:00:00 2001 From: Maciej Borsz Date: Thu, 20 Nov 2025 12:02:37 +0000 Subject: [PATCH] Fix apiserver_watch_events_sizes metric. This metric got broken by caching efforts -- the value is counted only once. --- .../pkg/endpoints/handlers/response.go | 2 - .../apiserver/pkg/endpoints/handlers/watch.go | 36 ++++- .../pkg/endpoints/handlers/watch_test.go | 146 ++++++++++++++++++ 3 files changed, 180 insertions(+), 4 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response.go index c34c6edd535..f36bb403909 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response.go @@ -36,7 +36,6 @@ import ( "k8s.io/apimachinery/pkg/watch" "k8s.io/apiserver/pkg/endpoints/handlers/negotiation" "k8s.io/apiserver/pkg/endpoints/handlers/responsewriters" - "k8s.io/apiserver/pkg/endpoints/metrics" endpointsrequest "k8s.io/apiserver/pkg/endpoints/request" "k8s.io/apiserver/pkg/storage" "k8s.io/apiserver/pkg/util/apihelpers" @@ -193,7 +192,6 @@ func (e *watchEncoder) doEncode(obj runtime.Object, event watch.Event, w io.Writ Type: string(event.Type), Object: runtime.RawExtension{Raw: e.buffer.Bytes()}, } - 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() if err := e.encoder.Encode(outEvent, e.eventBuffer); err != nil { diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go index 3f150a4f27f..822e5628894 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go @@ -21,6 +21,7 @@ import ( "fmt" "io" "net/http" + "sync/atomic" "time" "golang.org/x/net/websocket" @@ -37,6 +38,7 @@ import ( "k8s.io/apiserver/pkg/features" "k8s.io/apiserver/pkg/storage" utilfeature "k8s.io/apiserver/pkg/util/feature" + compbasemetrics "k8s.io/component-base/metrics" ) // timeoutFactory abstracts watch timeout logic for testing @@ -202,6 +204,29 @@ type WatchServer struct { metricsScope string } +// watchEventMetricsRecorder allows the caller to count bytes written and report the size of the event. +// It is thread-safe, as long as underlying io.Writer is thread-safe. +// Once all Write calls for a given watch event have finished, RecordEvent must be called. +type watchEventMetricsRecorder struct { + writer io.Writer + countMetric compbasemetrics.CounterMetric + sizeMetric compbasemetrics.ObserverMetric + byteCount atomic.Int64 +} + +// Write implements io.Writer. +func (c *watchEventMetricsRecorder) Write(p []byte) (n int, err error) { + n, err = c.writer.Write(p) + c.byteCount.Add(int64(n)) + return +} + +// Record reports the metrics and resets the byte count. +func (c *watchEventMetricsRecorder) RecordEvent() { + c.countMetric.Inc() + c.sizeMetric.Observe(float64(c.byteCount.Swap(0))) +} + // HandleHTTP serves a series of encoded events via HTTP with Transfer-Encoding: chunked. // or over a websocket connection. func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { @@ -239,7 +264,14 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { flusher.Flush() gvr := s.Scope.Resource - watchEncoder := newWatchEncoder(req.Context(), gvr, s.EmbeddedEncoder, s.Encoder, framer) + + recorder := &watchEventMetricsRecorder{ + writer: framer, + countMetric: metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(gvr.Group, gvr.Version, gvr.Resource), + sizeMetric: metrics.WatchEventsSizes.WithContext(req.Context()).WithLabelValues(gvr.Group, gvr.Version, gvr.Resource), + } + + watchEncoder := newWatchEncoder(req.Context(), gvr, s.EmbeddedEncoder, s.Encoder, recorder) ch := s.Watching.ResultChan() done := req.Context().Done() @@ -263,7 +295,6 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { // End of results. return } - metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(gvr.Group, gvr.Version, gvr.Resource).Inc() isWatchListLatencyRecordingRequired := shouldRecordWatchListLatency(event) if err := watchEncoder.Encode(event); err != nil { @@ -271,6 +302,7 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { // client disconnect. return } + recorder.RecordEvent() if len(ch) == 0 { flusher.Flush() diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch_test.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch_test.go index e9277c250c7..1e5784e322c 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch_test.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch_test.go @@ -17,6 +17,7 @@ limitations under the License. package handlers import ( + "bytes" "context" "encoding/json" "fmt" @@ -24,6 +25,8 @@ import ( "net/http" "net/http/httptest" "net/url" + "strings" + "sync" "testing" "time" @@ -36,9 +39,12 @@ import ( "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/apiserver/pkg/endpoints/handlers/responsewriters" + "k8s.io/apiserver/pkg/endpoints/metrics" endpointstesting "k8s.io/apiserver/pkg/endpoints/testing" "k8s.io/client-go/dynamic" restclient "k8s.io/client-go/rest" + "k8s.io/component-base/metrics/legacyregistry" + metricstestutil "k8s.io/component-base/metrics/testutil" ) // Fake API versions, similar to api/latest.go @@ -341,3 +347,143 @@ func serveWatch(watcher watch.Interface, watchServer *WatchServer, preServeErr e watchServer.HandleHTTP(w, req) } } + +type fakeCachingObject struct { + obj runtime.Object + + once sync.Once + raw []byte + err error +} + +func (f *fakeCachingObject) CacheEncode(_ runtime.Identifier, encode func(runtime.Object, io.Writer) error, w io.Writer) error { + f.once.Do(func() { + buffer := bytes.NewBuffer(nil) + f.err = encode(f.obj, buffer) + f.raw = buffer.Bytes() + }) + + if f.err != nil { + return f.err + } + + _, err := w.Write(f.raw) + return err +} + +func (f *fakeCachingObject) GetObject() runtime.Object { + return f.obj +} + +func (f *fakeCachingObject) GetObjectKind() schema.ObjectKind { + return f.obj.GetObjectKind() +} + +func (f *fakeCachingObject) DeepCopyObject() runtime.Object { + return &fakeCachingObject{obj: f.obj.DeepCopyObject()} +} + +var _ runtime.CacheableObject = &fakeCachingObject{} +var _ runtime.Object = &fakeCachingObject{} + +func TestWatchEventSizes(t *testing.T) { + metrics.Register() + gvr := schema.GroupVersionResource{Group: "group", Version: "version", Resource: "resource"} + testCases := []struct { + name string + event watch.Event + wantSize int64 + }{ + { + name: "regular object", + event: watch.Event{ + Type: watch.Added, + Object: &endpointstesting.Simple{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}, + }, + wantSize: 2 * 98, + }, + { + name: "cached object", + event: watch.Event{ + Type: watch.Added, + Object: &fakeCachingObject{obj: &runtime.Unknown{ + Raw: []byte(`{"kind":"Simple","apiVersion":"v1","metadata":{"name":"foo"}}`), + ContentType: runtime.ContentTypeJSON, + }}, + }, + wantSize: 2 * 88, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + + metrics.WatchEventsSizes.Reset() + metrics.WatchEvents.Reset() + + watcher := watch.NewFake() + timeoutCh := make(chan time.Time) + doneCh := make(chan struct{}) + + info, ok := runtime.SerializerInfoForMediaType(codecs.SupportedMediaTypes(), runtime.ContentTypeJSON) + require.True(t, ok) + require.NotNil(t, info.StreamSerializer) + serializer := info.StreamSerializer + + watchServer := &WatchServer{ + Scope: &RequestScope{ + Resource: gvr, + }, + Watching: watcher, + MediaType: "application/json", + Framer: serializer.Framer, + Encoder: testCodecV2, + EmbeddedEncoder: testCodecV2, + TimeoutFactory: &fakeTimeoutFactory{timeoutCh: timeoutCh, done: doneCh}, + } + + s := httptest.NewServer(serveWatch(watcher, watchServer, nil)) + defer s.Close() + + // Setup a client + dest, _ := url.Parse(s.URL) + dest.Path = "/" + namedGroupPrefix + "/" + testGroupV2.Group + "/" + testGroupV2.Version + "/simple" + dest.RawQuery = "watch=true" + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, dest.String(), nil) + client := http.Client{} + resp, err := client.Do(req) + require.NoError(t, err) + defer apitesting.Close(t, resp.Body) + + // Send object twice so that in case of caching the cached version is used. + watcher.Action(tc.event.Type, tc.event.Object) + watcher.Action(tc.event.Type, tc.event.Object) + + close(timeoutCh) + <-doneCh + + expected := fmt.Sprintf(`# HELP apiserver_watch_events_sizes [ALPHA] Watch event size distribution in bytes +# TYPE apiserver_watch_events_sizes histogram +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="1024"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="2048"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="4096"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="8192"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="16384"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="32768"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="65536"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="131072"} 2 +apiserver_watch_events_sizes_bucket{group="group",resource="resource",version="version",le="+Inf"} 2 +apiserver_watch_events_sizes_sum{group="group",resource="resource",version="version"} %d +apiserver_watch_events_sizes_count{group="group",resource="resource",version="version"} 2 + +# HELP apiserver_watch_events_total [ALPHA] Number of events sent in watch clients +# TYPE apiserver_watch_events_total counter +apiserver_watch_events_total{group="group",resource="resource",version="version"} 2 +`, tc.wantSize) + + err = metricstestutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expected), "apiserver_watch_events_sizes", "apiserver_watch_events_total") + require.NoError(t, err) + }) + } +}