From 8ec951bf107b2365862607e1751e7ca98835350f Mon Sep 17 00:00:00 2001 From: Marek Siarkowicz Date: Mon, 23 Feb 2026 14:45:38 +0100 Subject: [PATCH] Mirror List and WatchList tracing --- .../apiserver/pkg/endpoints/handlers/get.go | 21 ++- .../apiserver/pkg/endpoints/handlers/watch.go | 25 +++- .../apiserver/pkg/storage/cacher/cacher.go | 5 + .../apiserver/tracing/tracing_test.go | 125 +++++++++++++++++- 4 files changed, 162 insertions(+), 14 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go index 8ee1245c189..134e163eeb3 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go @@ -25,10 +25,7 @@ import ( "strings" "time" - "go.opentelemetry.io/otel/attribute" - "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/meta" metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" metainternalversionscheme "k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme" metainternalversionvalidation "k8s.io/apimachinery/pkg/apis/meta/internalversion/validation" @@ -259,6 +256,14 @@ func listOpts(req *http.Request, scope *RequestScope) (metainternalversion.ListO } func handleWatch(ctx context.Context, rw rest.Watcher, scope *RequestScope, req *http.Request, w http.ResponseWriter, opts metainternalversion.ListOptions, outputMediaType negotiation.MediaTypeOptions, minRequestTimeout time.Duration) error { + var span *tracing.Span + var onWatchListComplete WatchListCompleteHook + if isListWatchRequest(opts) { + ctx, span = tracing.Start(ctx, "WatchList", traceFields(req)...) + onWatchListComplete = func() { span.End(500 * time.Millisecond) } + req = req.WithContext(ctx) + } + if rw == nil { return errors.NewMethodNotSupported(scope.Resource.GroupResource(), "watch") } @@ -278,7 +283,7 @@ func handleWatch(ctx context.Context, rw rest.Watcher, scope *RequestScope, req if err != nil { return err } - handler, err := serveWatchHandler(watcher, scope, outputMediaType, req, w, timeout, metrics.CleanListScope(ctx, &opts)) + handler, err := serveWatchHandler(watcher, scope, outputMediaType, req, w, timeout, metrics.CleanListScope(ctx, &opts), onWatchListComplete) if err != nil { return err } @@ -311,13 +316,15 @@ func handleList(ctx context.Context, r rest.Lister, scope *RequestScope, req *ht defer span.End(500 * time.Millisecond) req = req.WithContext(ctx) - span.AddEvent("About to List from storage") result, err := r.List(ctx, &opts) if err != nil { return err } - span.AddEvent("Listing from storage done") - defer span.AddEvent("Writing http response done", attribute.Int("count", meta.LenList(result))) transformResponseObject(ctx, scope, req, w, http.StatusOK, outputMediaType, result) return nil } + +// isListWatchRequest is mirrored in staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go +func isListWatchRequest(opts metainternalversion.ListOptions) bool { + return opts.SendInitialEvents != nil && *opts.SendInitialEvents && opts.AllowWatchBookmarks +} 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 3dee9d83a91..4ac97d9f2f9 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go @@ -24,6 +24,7 @@ import ( "sync/atomic" "time" + "go.opentelemetry.io/otel/attribute" "golang.org/x/net/websocket" "k8s.io/apimachinery/pkg/api/errors" @@ -41,6 +42,7 @@ import ( "k8s.io/apiserver/pkg/storage" utilfeature "k8s.io/apiserver/pkg/util/feature" compbasemetrics "k8s.io/component-base/metrics" + "k8s.io/component-base/tracing" "k8s.io/klog/v2" ) @@ -67,7 +69,7 @@ func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) { // serveWatchHandler returns a handle to serve a watch response. // TODO: the functionality in this method and in WatchServer.Serve is not cleanly decoupled. -func serveWatchHandler(watcher watch.Interface, scope *RequestScope, mediaTypeOptions negotiation.MediaTypeOptions, req *http.Request, w http.ResponseWriter, timeout time.Duration, metricsScope string) (http.Handler, error) { +func serveWatchHandler(watcher watch.Interface, scope *RequestScope, mediaTypeOptions negotiation.MediaTypeOptions, req *http.Request, w http.ResponseWriter, timeout time.Duration, metricsScope string, completeHook WatchListCompleteHook) (http.Handler, error) { options, err := optionsForTransform(mediaTypeOptions, req) if err != nil { return nil, err @@ -174,7 +176,8 @@ func serveWatchHandler(watcher watch.Interface, scope *RequestScope, mediaTypeOp TimeoutFactory: &realTimeoutFactory{timeout}, ServerShuttingDownCh: serverShuttingDownCh, - metricsScope: metricsScope, + metricsScope: metricsScope, + watchListCompleteHook: completeHook, } if wsstream.IsWebSocketRequest(req) { @@ -204,9 +207,12 @@ type WatchServer struct { TimeoutFactory TimeoutFactory ServerShuttingDownCh <-chan struct{} - metricsScope string + metricsScope string + watchListCompleteHook WatchListCompleteHook } +type WatchListCompleteHook func() + // 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. @@ -233,6 +239,15 @@ func (c *watchEventMetricsRecorder) RecordEvent() { // 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) { + ctx := req.Context() + ctx, span := tracing.Start(ctx, "WatchServer.HandleHTTP", + attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)), + attribute.String("method", req.Method), + attribute.String("url", req.URL.Path), + attribute.String("protocol", req.Proto), + attribute.String("mediaType", s.MediaType), + attribute.String("encoder", string(s.Encoder.Identifier()))) + req = req.WithContext(ctx) defer func() { if s.MemoryAllocator != nil { runtime.AllocatorPool.Put(s.MemoryAllocator) @@ -278,6 +293,7 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { ch := s.Watching.ResultChan() done := req.Context().Done() + span.AddEvent("About to start writing response") for { select { case <-s.ServerShuttingDownCh: @@ -321,6 +337,9 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { auditID := audit.GetAuditIDTruncated(req.Context()) klog.V(3).InfoS("WatchList initial events sent", "path", req.URL.Path, "auditID", auditID, "initLatency", initLatency) httplog.AddKeyValue(req.Context(), "watchlist_init_latency", initLatency) + span.AddEvent("Writing initial events done") + span.End(5 * time.Second) + s.watchListCompleteHook() } } } diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go index 1fa74d7abd0..c2993872371 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go @@ -506,6 +506,10 @@ type namespacedName struct { } func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) { + ctx, span := tracing.Start(ctx, "cacher.Watch", + attribute.String("audit-id", audit.GetAuditIDTruncated(ctx)), + attribute.Stringer("type", c.groupResource)) + defer span.End(500 * time.Millisecond) key, err := c.prepareKey(key, opts.Recursive) if err != nil { return nil, err @@ -1239,6 +1243,7 @@ func (c *Cacher) getBookmarkAfterResourceVersionLockedFunc(parsedResourceVersion } } +// isListWatchRequest is mirrored in staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go func isListWatchRequest(opts storage.ListOptions) bool { return opts.SendInitialEvents != nil && *opts.SendInitialEvents && opts.Predicate.AllowWatchBookmarks } diff --git a/test/integration/apiserver/tracing/tracing_test.go b/test/integration/apiserver/tracing/tracing_test.go index 11a07f673e9..96d9b4a870d 100644 --- a/test/integration/apiserver/tracing/tracing_test.go +++ b/test/integration/apiserver/tracing/tracing_test.go @@ -43,6 +43,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/strategicpatch" + "k8s.io/apimachinery/pkg/watch" kmsv2mock "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/testing/v2" client "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -604,7 +605,109 @@ endpoint: %s`, listener.Addr().String())), os.FileMode(0755)); err != nil { }, }, { - desc: "list nodes", + desc: "WatchList nodes", + apiCall: func(ctx context.Context) error { + sendInitialEvents := true + w, err := clientSet.CoreV1().Nodes().Watch(ctx, metav1.ListOptions{SendInitialEvents: &sendInitialEvents, AllowWatchBookmarks: true, ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan}) + if err != nil { + return err + } + defer w.Stop() + for e := range w.ResultChan() { + switch e.Type { + case watch.Bookmark: + return nil + case watch.Error: + return fmt.Errorf("watch error: %v", e.Object) + } + } + return nil + }, + expectedTrace: []*spanExpectation{ + { + name: "GET /api/v1/nodes", + attributes: map[string]func(*commonv1.AnyValue) bool{ + "user_agent.original": func(v *commonv1.AnyValue) bool { + return strings.HasPrefix(v.GetStringValue(), "tracing.test") + }, + "url.path": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "/api/v1/nodes" + }, + "http.request.method": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "GET" + }, + "audit-id": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() != "" + }, + }, + }, + { + name: "WatchList", + attributes: map[string]func(*commonv1.AnyValue) bool{ + "url": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "/api/v1/nodes" + }, + "user-agent": func(v *commonv1.AnyValue) bool { + return strings.HasPrefix(v.GetStringValue(), "tracing.test") + }, + "audit-id": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() != "" + }, + "client": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "127.0.0.1" + }, + "accept": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "application/vnd.kubernetes.protobuf, */*" + }, + "protocol": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "HTTP/2.0" + }, + }, + events: []string{}, + }, + { + name: "cacher.Watch", + attributes: map[string]func(*commonv1.AnyValue) bool{ + "audit-id": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() != "" + }, + "type": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "nodes" + }, + }, + events: []string{ + "watchCache locked acquired", + "watchCache fresh enough", + }, + }, + { + name: "WatchServer.HandleHTTP", + attributes: map[string]func(*commonv1.AnyValue) bool{ + "url": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "/api/v1/nodes" + }, + "audit-id": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() != "" + }, + "protocol": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "HTTP/2.0" + }, + "method": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "GET" + }, + "mediaType": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "application/vnd.kubernetes.protobuf;stream=watch" + }, + "encoder": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "{\"encodeGV\":\"v1\",\"encoder\":\"raw-protobuf\",\"name\":\"versioning\"}" + }, + }, + events: []string{}, + }, + }, + }, + { + desc: "List nodes", apiCall: func(ctx context.Context) error { _, err = clientSet.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) return err @@ -649,10 +752,24 @@ endpoint: %s`, listener.Addr().String())), os.FileMode(0755)); err != nil { return v.GetStringValue() == "HTTP/2.0" }, }, + events: []string{}, + }, + { + name: "cacher.GetList", + attributes: map[string]func(*commonv1.AnyValue) bool{ + "audit-id": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() != "" + }, + "type": func(v *commonv1.AnyValue) bool { + return v.GetStringValue() == "nodes" + }, + }, events: []string{ - "About to List from storage", - "Listing from storage done", - "Writing http response done", + "Ready", + "watchCache locked acquired", + "watchCache fresh enough", + "Listed items from cache", + "Filtered items", }, }, {