Merge pull request #137202 from serathius/watchlist-trace

Mirror List and WatchList tracing

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
+}

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()
 				}
 			}
 		}

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
 }

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",
 					},
 				},
 				{