apiserver/handlers/watch: encode initialEventsListBlueprint with watchEncoder (kubernetes#127587)

* apiserver/handlers/get: construct versionedList

* storage/cacher: document caching the serialization of bookmark events

* endpoints/handlers/response: add watchListTransformer

* endpoints/handlers/watch: wire watchListTransformer

Author: p0lyn0mial

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go

@@ -265,6 +265,16 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatc
 			if timeout == 0 && minRequestTimeout > 0 {
 				timeout = time.Duration(float64(minRequestTimeout) * (rand.Float64() + 1.0))
 			}
+
+			var emptyVersionedList runtime.Object
+			if isListWatchRequest(opts) {
+				emptyVersionedList, err = scope.Convertor.ConvertToVersion(r.NewList(), scope.Kind.GroupVersion())
+				if err != nil {
+					scope.err(errors.NewInternalError(err), w, req)
+					return
+				}
+			}
+
 			klog.V(3).InfoS("Starting watch", "path", req.URL.Path, "resourceVersion", opts.ResourceVersion, "labels", opts.LabelSelector, "fields", opts.FieldSelector, "timeout", timeout)
 			ctx, cancel := context.WithTimeout(ctx, timeout)
 			defer func() { cancel() }()
@@ -273,7 +283,7 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatc
 				scope.err(err, w, req)
 				return
 			}
-			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), emptyVersionedList)
 			if err != nil {
 				scope.err(err, w, req)
 				return

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response.go

@@ -18,6 +18,7 @@ package handlers
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -38,8 +39,9 @@ import (
 	"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"
 
-	klog "k8s.io/klog/v2"
+	"k8s.io/klog/v2"
 )
 
 // watchEmbeddedEncoder performs encoding of the embedded object.
@@ -147,22 +149,25 @@ type watchEncoder struct {
 	encoder         runtime.Encoder
 	framer          io.Writer
 
+	watchListTransformerFn watchListTransformerFunction
+
 	buffer      runtime.Splice
 	eventBuffer runtime.Splice
 
 	currentEmbeddedIdentifier 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) *watchEncoder {
+func newWatchEncoder(ctx context.Context, kind schema.GroupVersionKind, embeddedEncoder runtime.Encoder, encoder runtime.Encoder, framer io.Writer, watchListTransformerFn watchListTransformerFunction) *watchEncoder {
 	return &watchEncoder{
-		ctx:             ctx,
-		kind:            kind,
-		embeddedEncoder: embeddedEncoder,
-		encoder:         encoder,
-		framer:          framer,
-		buffer:          runtime.NewSpliceBuffer(),
-		eventBuffer:     runtime.NewSpliceBuffer(),
+		ctx:                    ctx,
+		kind:                   kind,
+		embeddedEncoder:        embeddedEncoder,
+		encoder:                encoder,
+		framer:                 framer,
+		watchListTransformerFn: watchListTransformerFn,
+		buffer:                 runtime.NewSpliceBuffer(),
+		eventBuffer:            runtime.NewSpliceBuffer(),
 	}
 }
 
@@ -174,6 +179,12 @@ func (e *watchEncoder) Encode(event watch.Event) error {
 	encodeFunc := func(obj runtime.Object, w io.Writer) error {
 		return e.doEncode(obj, event, w)
 	}
+	if event.Type == watch.Bookmark {
+		// Bookmark objects are small, and we don't yet support serialization for them.
+		// Additionally, we need to additionally transform them to support watch-list feature
+		event = e.watchListTransformerFn(event)
+		return encodeFunc(event.Object, e.framer)
+	}
 	if co, ok := event.Object.(runtime.CacheableObject); ok {
 		return co.CacheEncode(e.identifier(event.Type), encodeFunc, e.framer)
 	}
@@ -479,3 +490,94 @@ func asPartialObjectMetadataList(result runtime.Object, groupVersion schema.Grou
 		return nil, newNotAcceptableError(fmt.Sprintf("no PartialObjectMetadataList exists in group version %s", groupVersion))
 	}
 }
+
+// watchListTransformerFunction an optional function
+// applied to watchlist bookmark events that transforms
+// the embedded object before sending it to a client.
+type watchListTransformerFunction func(watch.Event) watch.Event
+
+// watchListTransformer performs transformation of
+// a special watchList bookmark event.
+//
+// The bookmark is annotated with InitialEventsListBlueprintAnnotationKey
+// and contains an empty, versioned list that we must encode in the requested format
+// (e.g., protobuf, JSON, CBOR) and then store as a base64-encoded string.
+type watchListTransformer struct {
+	initialEventsListBlueprint runtime.Object
+	targetGVK                  *schema.GroupVersionKind
+	negotiatedEncoder          runtime.Encoder
+	buffer                     runtime.Splice
+}
+
+// createWatchListTransformerIfRequested returns a transformer function for watchlist bookmark event.
+func newWatchListTransformer(initialEventsListBlueprint runtime.Object, targetGVK *schema.GroupVersionKind, negotiatedEncoder runtime.Encoder) *watchListTransformer {
+	return &watchListTransformer{
+		initialEventsListBlueprint: initialEventsListBlueprint,
+		targetGVK:                  targetGVK,
+		negotiatedEncoder:          negotiatedEncoder,
+		buffer:                     runtime.NewSpliceBuffer(),
+	}
+}
+
+func (e *watchListTransformer) transform(event watch.Event) watch.Event {
+	if e.initialEventsListBlueprint == nil {
+		return event
+	}
+	hasAnnotation, err := storage.HasInitialEventsEndBookmarkAnnotation(event.Object)
+	if err != nil {
+		return newWatchEventErrorFor(err)
+	}
+	if !hasAnnotation {
+		return event
+	}
+
+	if err = e.encodeInitialEventsListBlueprint(event.Object); err != nil {
+		return newWatchEventErrorFor(err)
+	}
+
+	return event
+}
+
+func (e *watchListTransformer) encodeInitialEventsListBlueprint(object runtime.Object) error {
+	initialEventsListBlueprint, err := e.transformInitialEventsListBlueprint()
+	if err != nil {
+		return err
+	}
+
+	defer e.buffer.Reset()
+	if err = e.negotiatedEncoder.Encode(initialEventsListBlueprint, e.buffer); err != nil {
+		return err
+	}
+	encodedInitialEventsListBlueprint := e.buffer.Bytes()
+
+	// the storage layer creates a deep copy of the obj before modifying it.
+	// since the object has the annotation, we can modify it directly.
+	objectMeta, err := meta.Accessor(object)
+	if err != nil {
+		return err
+	}
+	annotations := objectMeta.GetAnnotations()
+	annotations[metav1.InitialEventsListBlueprintAnnotationKey] = base64.StdEncoding.EncodeToString(encodedInitialEventsListBlueprint)
+	objectMeta.SetAnnotations(annotations)
+
+	return nil
+}
+
+func (e *watchListTransformer) transformInitialEventsListBlueprint() (runtime.Object, error) {
+	if e.targetGVK != nil && e.targetGVK.Kind == "PartialObjectMetadata" {
+		return asPartialObjectMetadataList(e.initialEventsListBlueprint, e.targetGVK.GroupVersion())
+	}
+	return e.initialEventsListBlueprint, nil
+}
+
+func newWatchEventErrorFor(err error) watch.Event {
+	return watch.Event{
+		Type: watch.Error,
+		Object: &metav1.Status{
+			Status:  metav1.StatusFailure,
+			Message: err.Error(),
+			Reason:  metav1.StatusReasonInternalError,
+			Code:    http.StatusInternalServerError,
+		},
+	}
+}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/response_test.go

@@ -17,21 +17,28 @@ limitations under the License.
 package handlers
 
 import (
+	"bytes"
 	"context"
+	"encoding/base64"
 	"fmt"
 	"io"
 	"net/http"
 	"reflect"
 	"testing"
 	"time"
 
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	runtimejson "k8s.io/apimachinery/pkg/runtime/serializer/json"
+	"k8s.io/apimachinery/pkg/watch"
 	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 )
 
 var _ runtime.CacheableObject = &mockCacheableObject{}
@@ -222,3 +229,118 @@ func TestWatchEncoderIdentifier(t *testing.T) {
 		t.Error("  - watchEncoder::typeIdentifier to capture all relevant fields in identifier")
 	}
 }
+
+func TestWatchListEncoder(t *testing.T) {
+	makePartialObjectMetadataListWithoutKind := func(rv string) *metav1.PartialObjectMetadataList {
+		return &metav1.PartialObjectMetadataList{
+			// do not set the type info to match
+			// newWatchListTransformer
+			ListMeta: metav1.ListMeta{ResourceVersion: rv},
+		}
+	}
+	makePodListWithKind := func(rv string) *v1.PodList {
+		return &v1.PodList{
+			TypeMeta: metav1.TypeMeta{
+				// set the type info so
+				// that it differs from
+				// PartialObjectMetadataList
+				Kind: "PodList",
+			},
+			ListMeta: metav1.ListMeta{
+				ResourceVersion: rv,
+			},
+		}
+	}
+	makeBookmarkEventFor := func(pod *v1.Pod) watch.Event {
+		return watch.Event{
+			Type:   watch.Bookmark,
+			Object: pod,
+		}
+	}
+	makePod := func(name string) *v1.Pod {
+		return &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        name,
+				Namespace:   "ns",
+				Annotations: map[string]string{},
+			},
+		}
+	}
+	makePodWithInitialEventsAnnotation := func(name string) *v1.Pod {
+		p := makePod(name)
+		p.Annotations[metav1.InitialEventsAnnotationKey] = "true"
+		return p
+	}
+
+	scenarios := []struct {
+		name              string
+		negotiatedEncoder runtime.Serializer
+		targetGVK         *schema.GroupVersionKind
+
+		actualEvent   watch.Event
+		listBlueprint runtime.Object
+
+		expectedBase64ListBlueprint string
+	}{
+		{
+			name:              "pass through, an obj without the annotation received",
+			actualEvent:       makeBookmarkEventFor(makePod("1")),
+			negotiatedEncoder: newJSONSerializer(),
+		},
+		{
+			name:                        "encodes the initialEventsListBlueprint if an obj with the annotation is passed",
+			actualEvent:                 makeBookmarkEventFor(makePodWithInitialEventsAnnotation("1")),
+			listBlueprint:               makePodListWithKind("100"),
+			expectedBase64ListBlueprint: encodeObjectToBase64String(makePodListWithKind("100"), t),
+			negotiatedEncoder:           newJSONSerializer(),
+		},
+		{
+			name:                        "encodes the initialEventsListBlueprint as PartialObjectMetadata when requested",
+			targetGVK:                   &schema.GroupVersionKind{Group: "meta.k8s.io", Version: "v1", Kind: "PartialObjectMetadata"},
+			actualEvent:                 makeBookmarkEventFor(makePodWithInitialEventsAnnotation("2")),
+			listBlueprint:               makePodListWithKind("101"),
+			expectedBase64ListBlueprint: encodeObjectToBase64String(makePartialObjectMetadataListWithoutKind("101"), t),
+			negotiatedEncoder:           newJSONSerializer(),
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			target := newWatchListTransformer(scenario.listBlueprint, scenario.targetGVK, scenario.negotiatedEncoder)
+			transformedEvent := target.transform(scenario.actualEvent)
+
+			actualObjectMeta, err := meta.Accessor(transformedEvent.Object)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			base64ListBlueprint, ok := actualObjectMeta.GetAnnotations()[metav1.InitialEventsListBlueprintAnnotationKey]
+			if !ok && len(scenario.expectedBase64ListBlueprint) != 0 {
+				t.Fatalf("the encoded obj doesn't have %q", metav1.InitialEventsListBlueprintAnnotationKey)
+			}
+			if base64ListBlueprint != scenario.expectedBase64ListBlueprint {
+				t.Fatalf("unexpected base64ListBlueprint = %s, expected = %s", base64ListBlueprint, scenario.expectedBase64ListBlueprint)
+			}
+		})
+	}
+}
+
+func encodeObjectToBase64String(obj runtime.Object, t *testing.T) string {
+	e := newJSONSerializer()
+
+	var buf bytes.Buffer
+	err := e.Encode(obj, &buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return base64.StdEncoding.EncodeToString(buf.Bytes())
+}
+
+func newJSONSerializer() runtime.Serializer {
+	return runtimejson.NewSerializerWithOptions(
+		runtimejson.DefaultMetaFactory,
+		clientgoscheme.Scheme,
+		clientgoscheme.Scheme,
+		runtimejson.SerializerOptions{},
+	)
+}