update to go 1.23 and latest k8s version (1.32.5)

Signed-off-by: Muhammad Adil Ghaffar <muhammad.adil.ghaffar@est.tech>

vendor/k8s.io/client-go/rest/request.go

@@ -19,6 +19,7 @@ package rest
 import (
 	"bytes"
 	"context"
+	"encoding/base64"
 	"encoding/hex"
 	"fmt"
 	"io"
@@ -37,12 +38,15 @@ import (
 	"golang.org/x/net/http2"
 
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/conversion"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer/streaming"
 	"k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/watch"
+	clientfeatures "k8s.io/client-go/features"
 	restclientwatch "k8s.io/client-go/rest/watch"
 	"k8s.io/client-go/tools/metrics"
 	"k8s.io/client-go/util/flowcontrol"
@@ -96,6 +100,9 @@ func defaultRequestRetryFn(maxRetries int) WithRetry {
 type Request struct {
 	c *RESTClient
 
+	contentConfig     ClientContentConfig
+	contentTypeNotSet bool
+
 	warningHandler WarningHandler
 
 	rateLimiter flowcontrol.RateLimiter
@@ -120,7 +127,7 @@ type Request struct {
 	// output
 	err error
 
-	// only one of body / bodyBytes may be set. requests using body are not retriable.
+	// only one of body / bodyBytes may be set. requests using body are not retryable.
 	body      io.Reader
 	bodyBytes []byte
 
@@ -149,6 +156,11 @@ func NewRequest(c *RESTClient) *Request {
 		timeout = c.Client.Timeout
 	}
 
+	// A request needs to know whether the content type was explicitly configured or selected by
+	// default in order to support the per-request Protobuf override used by clients generated
+	// with --prefers-protobuf.
+	contentConfig, contentTypeDefaulted := c.content.GetClientContentConfig()
+
 	r := &Request{
 		c:              c,
 		rateLimiter:    c.rateLimiter,
@@ -158,14 +170,12 @@ func NewRequest(c *RESTClient) *Request {
 		maxRetries:     10,
 		retryFn:        defaultRequestRetryFn,
 		warningHandler: c.warningHandler,
+
+		contentConfig:     contentConfig,
+		contentTypeNotSet: contentTypeDefaulted,
 	}
 
-	switch {
-	case len(c.content.AcceptContentTypes) > 0:
-		r.SetHeader("Accept", c.content.AcceptContentTypes)
-	case len(c.content.ContentType) > 0:
-		r.SetHeader("Accept", c.content.ContentType+", */*")
-	}
+	r.setAcceptHeader()
 	return r
 }
 
@@ -174,11 +184,36 @@ func NewRequestWithClient(base *url.URL, versionedAPIPath string, content Client
 	return NewRequest(&RESTClient{
 		base:             base,
 		versionedAPIPath: versionedAPIPath,
-		content:          content,
+		content:          requestClientContentConfigProvider{base: content},
 		Client:           client,
 	})
 }
 
+func (r *Request) UseProtobufAsDefaultIfPreferred(prefersProtobuf bool) *Request {
+	if prefersProtobuf {
+		return r.UseProtobufAsDefault()
+	}
+	return r
+}
+
+func (r *Request) UseProtobufAsDefault() *Request {
+	if r.contentTypeNotSet && len(r.contentConfig.AcceptContentTypes) == 0 {
+		r.contentConfig.AcceptContentTypes = "application/vnd.kubernetes.protobuf,application/json"
+		r.contentConfig.ContentType = "application/vnd.kubernetes.protobuf"
+		r.setAcceptHeader()
+	}
+	return r
+}
+
+func (r *Request) setAcceptHeader() {
+	switch {
+	case len(r.contentConfig.AcceptContentTypes) > 0:
+		r.SetHeader("Accept", r.contentConfig.AcceptContentTypes)
+	case len(r.contentConfig.ContentType) > 0:
+		r.SetHeader("Accept", r.contentConfig.ContentType+", */*")
+	}
+}
+
 // Verb sets the verb this request will use.
 func (r *Request) Verb(verb string) *Request {
 	r.verb = verb
@@ -367,7 +402,7 @@ func (r *Request) Param(paramName, s string) *Request {
 // VersionedParams will not write query parameters that have omitempty set and are empty. If a
 // parameter has already been set it is appended to (Params and VersionedParams are additive).
 func (r *Request) VersionedParams(obj runtime.Object, codec runtime.ParameterCodec) *Request {
-	return r.SpecificallyVersionedParams(obj, codec, r.c.content.GroupVersion)
+	return r.SpecificallyVersionedParams(obj, codec, r.contentConfig.GroupVersion)
 }
 
 func (r *Request) SpecificallyVersionedParams(obj runtime.Object, codec runtime.ParameterCodec, version schema.GroupVersion) *Request {
@@ -447,11 +482,9 @@ func (r *Request) Body(obj interface{}) *Request {
 			r.err = err
 			return r
 		}
-		glogBody("Request Body", data)
 		r.body = nil
 		r.bodyBytes = data
 	case []byte:
-		glogBody("Request Body", t)
 		r.body = nil
 		r.bodyBytes = t
 	case io.Reader:
@@ -462,7 +495,7 @@ func (r *Request) Body(obj interface{}) *Request {
 		if reflect.ValueOf(t).IsNil() {
 			return r
 		}
-		encoder, err := r.c.content.Negotiator.Encoder(r.c.content.ContentType, nil)
+		encoder, err := r.contentConfig.Negotiator.Encoder(r.contentConfig.ContentType, nil)
 		if err != nil {
 			r.err = err
 			return r
@@ -472,10 +505,9 @@ func (r *Request) Body(obj interface{}) *Request {
 			r.err = err
 			return r
 		}
-		glogBody("Request Body", data)
 		r.body = nil
 		r.bodyBytes = data
-		r.SetHeader("Content-Type", r.c.content.ContentType)
+		r.SetHeader("Content-Type", r.contentConfig.ContentType)
 	default:
 		r.err = fmt.Errorf("unknown type used for body: %+v", obj)
 	}
@@ -701,10 +733,19 @@ func (b *throttledLogger) Infof(message string, args ...interface{}) {
 // Watch attempts to begin watching the requested location.
 // Returns a watch.Interface, or an error.
 func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
+	w, _, e := r.watchInternal(ctx)
+	return w, e
+}
+
+func (r *Request) watchInternal(ctx context.Context) (watch.Interface, runtime.Decoder, error) {
+	if r.body == nil {
+		logBody(ctx, 2, "Request Body", r.bodyBytes)
+	}
+
 	// We specifically don't want to rate limit watches, so we
 	// don't use r.rateLimiter here.
 	if r.err != nil {
-		return nil, r.err
+		return nil, nil, r.err
 	}
 
 	client := r.c.Client
@@ -724,12 +765,12 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
 	url := r.URL().String()
 	for {
 		if err := retry.Before(ctx, r); err != nil {
-			return nil, retry.WrapPreviousError(err)
+			return nil, nil, retry.WrapPreviousError(err)
 		}
 
 		req, err := r.newHTTPRequest(ctx)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 
 		resp, err := client.Do(req)
@@ -749,34 +790,194 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
 				// the server must have sent us an error in 'err'
 				return true, nil
 			}
-			if result := r.transformResponse(resp, req); result.err != nil {
-				return true, result.err
+			result := r.transformResponse(ctx, resp, req)
+			if err := result.Error(); err != nil {
+				return true, err
 			}
 			return true, fmt.Errorf("for request %s, got status: %v", url, resp.StatusCode)
 		}()
 		if done {
 			if isErrRetryableFunc(req, err) {
-				return watch.NewEmptyWatch(), nil
+				return watch.NewEmptyWatch(), nil, nil
 			}
 			if err == nil {
 				// if the server sent us an HTTP Response object,
 				// we need to return the error object from that.
 				err = transformErr
 			}
-			return nil, retry.WrapPreviousError(err)
+			return nil, nil, retry.WrapPreviousError(err)
 		}
 	}
 }
 
-func (r *Request) newStreamWatcher(resp *http.Response) (watch.Interface, error) {
+type WatchListResult struct {
+	// err holds any errors we might have received
+	// during streaming.
+	err error
+
+	// items hold the collected data
+	items []runtime.Object
+
+	// initialEventsEndBookmarkRV holds the resource version
+	// extracted from the bookmark event that marks
+	// the end of the stream.
+	initialEventsEndBookmarkRV string
+
+	// negotiatedObjectDecoder knows how to decode
+	// the initialEventsListBlueprint
+	negotiatedObjectDecoder runtime.Decoder
+
+	// base64EncodedInitialEventsListBlueprint contains an empty,
+	// versioned list encoded in the requested format
+	// (e.g., protobuf, JSON, CBOR) and stored as a base64-encoded string
+	base64EncodedInitialEventsListBlueprint string
+}
+
+// Into stores the result into obj. The passed obj parameter must be a pointer to a list type.
+//
+// Note:
+//
+// Special attention should be given to the type *unstructured.Unstructured,
+// which represents a list type but does not have an "Items" field.
+// Users who directly use RESTClient may store the response in such an object.
+// This particular case is not handled by the current implementation of this function,
+// but may be considered for future updates.
+func (r WatchListResult) Into(obj runtime.Object) error {
+	if r.err != nil {
+		return r.err
+	}
+
+	listItemsPtr, err := meta.GetItemsPtr(obj)
+	if err != nil {
+		return err
+	}
+	listVal, err := conversion.EnforcePtr(listItemsPtr)
+	if err != nil {
+		return err
+	}
+	if listVal.Kind() != reflect.Slice {
+		return fmt.Errorf("need a pointer to slice, got %v", listVal.Kind())
+	}
+
+	encodedInitialEventsListBlueprint, err := base64.StdEncoding.DecodeString(r.base64EncodedInitialEventsListBlueprint)
+	if err != nil {
+		return fmt.Errorf("failed to decode the received blueprint list, err %w", err)
+	}
+
+	err = runtime.DecodeInto(r.negotiatedObjectDecoder, encodedInitialEventsListBlueprint, obj)
+	if err != nil {
+		return err
+	}
+
+	if len(r.items) == 0 {
+		listVal.Set(reflect.MakeSlice(listVal.Type(), 0, 0))
+	} else {
+		listVal.Set(reflect.MakeSlice(listVal.Type(), len(r.items), len(r.items)))
+		for i, o := range r.items {
+			if listVal.Type().Elem() != reflect.TypeOf(o).Elem() {
+				return fmt.Errorf("received object type = %v at index = %d, doesn't match the list item type = %v", reflect.TypeOf(o).Elem(), i, listVal.Type().Elem())
+			}
+			listVal.Index(i).Set(reflect.ValueOf(o).Elem())
+		}
+	}
+
+	listMeta, err := meta.ListAccessor(obj)
+	if err != nil {
+		return err
+	}
+	listMeta.SetResourceVersion(r.initialEventsEndBookmarkRV)
+	return nil
+}
+
+// WatchList establishes a stream to get a consistent snapshot of data
+// from the server as described in https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list#proposal
+//
+// Note that the watchlist requires properly setting the ListOptions
+// otherwise it just establishes a regular watch with the server.
+// Check the documentation https://kubernetes.io/docs/reference/using-api/api-concepts/#streaming-lists
+// to see what parameters are currently required.
+func (r *Request) WatchList(ctx context.Context) WatchListResult {
+	if r.body == nil {
+		logBody(ctx, 2, "Request Body", r.bodyBytes)
+	}
+
+	if !clientfeatures.FeatureGates().Enabled(clientfeatures.WatchListClient) {
+		return WatchListResult{err: fmt.Errorf("%q feature gate is not enabled", clientfeatures.WatchListClient)}
+	}
+	// TODO(#115478): consider validating request parameters (i.e sendInitialEvents).
+	//  Most users use the generated client, which handles the proper setting of parameters.
+	//  We don't have validation for other methods (e.g., the Watch)
+	//  thus, for symmetry, we haven't added additional checks for the WatchList method.
+	w, d, err := r.watchInternal(ctx)
+	if err != nil {
+		return WatchListResult{err: err}
+	}
+	return r.handleWatchList(ctx, w, d)
+}
+
+// handleWatchList holds the actual logic for easier unit testing.
+// Note that this function will close the passed watch.
+func (r *Request) handleWatchList(ctx context.Context, w watch.Interface, negotiatedObjectDecoder runtime.Decoder) WatchListResult {
+	defer w.Stop()
+	var lastKey string
+	var items []runtime.Object
+
+	for {
+		select {
+		case <-ctx.Done():
+			return WatchListResult{err: ctx.Err()}
+		case event, ok := <-w.ResultChan():
+			if !ok {
+				return WatchListResult{err: fmt.Errorf("unexpected watch close")}
+			}
+			if event.Type == watch.Error {
+				return WatchListResult{err: errors.FromObject(event.Object)}
+			}
+			meta, err := meta.Accessor(event.Object)
+			if err != nil {
+				return WatchListResult{err: fmt.Errorf("failed to parse watch event: %#v", event)}
+			}
+
+			switch event.Type {
+			case watch.Added:
+				// the following check ensures that the response is ordered.
+				// earlier servers had a bug that caused them to not sort the output.
+				// in such cases, return an error which can trigger fallback logic.
+				key := objectKeyFromMeta(meta)
+				if len(lastKey) > 0 && lastKey > key {
+					return WatchListResult{err: fmt.Errorf("cannot add the obj (%#v) with the key = %s, as it violates the ordering guarantees provided by the watchlist feature in beta phase, lastInsertedKey was = %s", event.Object, key, lastKey)}
+				}
+				items = append(items, event.Object)
+				lastKey = key
+			case watch.Bookmark:
+				if meta.GetAnnotations()[metav1.InitialEventsAnnotationKey] == "true" {
+					base64EncodedInitialEventsListBlueprint := meta.GetAnnotations()[metav1.InitialEventsListBlueprintAnnotationKey]
+					if len(base64EncodedInitialEventsListBlueprint) == 0 {
+						return WatchListResult{err: fmt.Errorf("%q annotation is missing content", metav1.InitialEventsListBlueprintAnnotationKey)}
+					}
+					return WatchListResult{
+						items:                                   items,
+						initialEventsEndBookmarkRV:              meta.GetResourceVersion(),
+						negotiatedObjectDecoder:                 negotiatedObjectDecoder,
+						base64EncodedInitialEventsListBlueprint: base64EncodedInitialEventsListBlueprint,
+					}
+				}
+			default:
+				return WatchListResult{err: fmt.Errorf("unexpected watch event %#v, expected to only receive watch.Added and watch.Bookmark events", event)}
+			}
+		}
+	}
+}
+
+func (r *Request) newStreamWatcher(resp *http.Response) (watch.Interface, runtime.Decoder, error) {
 	contentType := resp.Header.Get("Content-Type")
 	mediaType, params, err := mime.ParseMediaType(contentType)
 	if err != nil {
 		klog.V(4).Infof("Unexpected content type from the server: %q: %v", contentType, err)
 	}
-	objectDecoder, streamingSerializer, framer, err := r.c.content.Negotiator.StreamDecoder(mediaType, params)
+	objectDecoder, streamingSerializer, framer, err := r.contentConfig.Negotiator.StreamDecoder(mediaType, params)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	handleWarnings(resp.Header, r.warningHandler)
@@ -789,7 +990,7 @@ func (r *Request) newStreamWatcher(resp *http.Response) (watch.Interface, error)
 		// use 500 to indicate that the cause of the error is unknown - other error codes
 		// are more specific to HTTP interactions, and set a reason
 		errors.NewClientErrorReporter(http.StatusInternalServerError, r.verb, "ClientWatchDecoding"),
-	), nil
+	), objectDecoder, nil
 }
 
 // updateRequestResultMetric increments the RequestResult metric counter,
@@ -829,6 +1030,10 @@ func sanitize(req *Request, resp *http.Response, err error) (string, string) {
 // Any non-2xx http status code causes an error.  If we get a non-2xx code, we try to convert the body into an APIStatus object.
 // If we can, we return that as an error.  Otherwise, we create an error that lists the http status and the content of the response.
 func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
+	if r.body == nil {
+		logBody(ctx, 2, "Request Body", r.bodyBytes)
+	}
+
 	if r.err != nil {
 		return nil, r.err
 	}
@@ -872,7 +1077,7 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
 				if retry.IsNextRetry(ctx, r, req, resp, err, neverRetryError) {
 					return false, nil
 				}
-				result := r.transformResponse(resp, req)
+				result := r.transformResponse(ctx, resp, req)
 				if err := result.Error(); err != nil {
 					return true, err
 				}
@@ -1004,7 +1209,7 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
 			return false
 		}
 		// For connection errors and apiserver shutdown errors retry.
-		if net.IsConnectionReset(err) || net.IsProbableEOF(err) {
+		if net.IsConnectionReset(err) || net.IsProbableEOF(err) || net.IsHTTP2ConnectionLost(err) {
 			return true
 		}
 		return false
@@ -1026,6 +1231,9 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
 		if req.ContentLength >= 0 && !(req.Body != nil && req.ContentLength == 0) {
 			metrics.RequestSize.Observe(ctx, r.verb, r.URL().Host, float64(req.ContentLength))
 		}
+		if resp != nil && resp.StatusCode == http.StatusUnsupportedMediaType {
+			r.c.content.UnsupportedMediaType(resp.Request.Header.Get("Content-Type"))
+		}
 		retry.After(ctx, r, resp, err)
 
 		done := func() bool {
@@ -1059,9 +1267,13 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
 //   - If the server responds with a status: *errors.StatusError or *errors.UnexpectedObjectError
 //   - http.Client.Do errors are returned directly.
 func (r *Request) Do(ctx context.Context) Result {
+	if r.body == nil {
+		logBody(ctx, 2, "Request Body", r.bodyBytes)
+	}
+
 	var result Result
 	err := r.request(ctx, func(req *http.Request, resp *http.Response) {
-		result = r.transformResponse(resp, req)
+		result = r.transformResponse(ctx, resp, req)
 	})
 	if err != nil {
 		return Result{err: err}
@@ -1074,10 +1286,14 @@ func (r *Request) Do(ctx context.Context) Result {
 
 // DoRaw executes the request but does not process the response body.
 func (r *Request) DoRaw(ctx context.Context) ([]byte, error) {
+	if r.body == nil {
+		logBody(ctx, 2, "Request Body", r.bodyBytes)
+	}
+
 	var result Result
 	err := r.request(ctx, func(req *http.Request, resp *http.Response) {
 		result.body, result.err = io.ReadAll(resp.Body)
-		glogBody("Response Body", result.body)
+		logBody(ctx, 2, "Response Body", result.body)
 		if resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent {
 			result.err = r.transformUnstructuredResponseError(resp, req, result.body)
 		}
@@ -1092,7 +1308,7 @@ func (r *Request) DoRaw(ctx context.Context) ([]byte, error) {
 }
 
 // transformResponse converts an API response into a structured API object
-func (r *Request) transformResponse(resp *http.Response, req *http.Request) Result {
+func (r *Request) transformResponse(ctx context.Context, resp *http.Response, req *http.Request) Result {
 	var body []byte
 	if resp.Body != nil {
 		data, err := io.ReadAll(resp.Body)
@@ -1121,13 +1337,14 @@ func (r *Request) transformResponse(resp *http.Response, req *http.Request) Resu
 		}
 	}
 
-	glogBody("Response Body", body)
+	// Call depth is tricky. This one is okay for Do and DoRaw.
+	logBody(ctx, 7, "Response Body", body)
 
 	// verify the content type is accurate
 	var decoder runtime.Decoder
 	contentType := resp.Header.Get("Content-Type")
 	if len(contentType) == 0 {
-		contentType = r.c.content.ContentType
+		contentType = r.contentConfig.ContentType
 	}
 	if len(contentType) > 0 {
 		var err error
@@ -1135,7 +1352,7 @@ func (r *Request) transformResponse(resp *http.Response, req *http.Request) Resu
 		if err != nil {
 			return Result{err: errors.NewInternalError(err)}
 		}
-		decoder, err = r.c.content.Negotiator.Decoder(mediaType, params)
+		decoder, err = r.contentConfig.Negotiator.Decoder(mediaType, params)
 		if err != nil {
 			// if we fail to negotiate a decoder, treat this as an unstructured error
 			switch {
@@ -1181,14 +1398,14 @@ func (r *Request) transformResponse(resp *http.Response, req *http.Request) Resu
 }
 
 // truncateBody decides if the body should be truncated, based on the glog Verbosity.
-func truncateBody(body string) string {
+func truncateBody(logger klog.Logger, body string) string {
 	max := 0
 	switch {
-	case bool(klog.V(10).Enabled()):
+	case bool(logger.V(10).Enabled()):
 		return body
-	case bool(klog.V(9).Enabled()):
+	case bool(logger.V(9).Enabled()):
 		max = 10240
-	case bool(klog.V(8).Enabled()):
+	case bool(logger.V(8).Enabled()):
 		max = 1024
 	}
 
@@ -1199,17 +1416,21 @@ func truncateBody(body string) string {
 	return body[:max] + fmt.Sprintf(" [truncated %d chars]", len(body)-max)
 }
 
-// glogBody logs a body output that could be either JSON or protobuf. It explicitly guards against
+// logBody logs a body output that could be either JSON or protobuf. It explicitly guards against
 // allocating a new string for the body output unless necessary. Uses a simple heuristic to determine
 // whether the body is printable.
-func glogBody(prefix string, body []byte) {
-	if klogV := klog.V(8); klogV.Enabled() {
+//
+// It needs to be called by all functions which send or receive the data.
+func logBody(ctx context.Context, callDepth int, prefix string, body []byte) {
+	logger := klog.FromContext(ctx)
+	if loggerV := logger.V(8); loggerV.Enabled() {
+		loggerV := loggerV.WithCallDepth(callDepth)
 		if bytes.IndexFunc(body, func(r rune) bool {
 			return r < 0x0a
 		}) != -1 {
-			klogV.Infof("%s:\n%s", prefix, truncateBody(hex.Dump(body)))
+			loggerV.Info(prefix, "body", truncateBody(logger, hex.Dump(body)))
 		} else {
-			klogV.Infof("%s: %s", prefix, truncateBody(string(body)))
+			loggerV.Info(prefix, "body", truncateBody(logger, string(body)))
 		}
 	}
 }
@@ -1258,7 +1479,7 @@ func (r *Request) newUnstructuredResponseError(body []byte, isTextResponse bool,
 	}
 	var groupResource schema.GroupResource
 	if len(r.resource) > 0 {
-		groupResource.Group = r.c.content.GroupVersion.Group
+		groupResource.Group = r.contentConfig.GroupVersion.Group
 		groupResource.Resource = r.resource
 	}
 	return errors.NewGenericServerResponse(
@@ -1470,3 +1691,10 @@ func ValidatePathSegmentName(name string, prefix bool) []string {
 	}
 	return IsValidPathSegmentName(name)
 }
+
+func objectKeyFromMeta(objMeta metav1.Object) string {
+	if len(objMeta.GetNamespace()) > 0 {
+		return fmt.Sprintf("%s/%s", objMeta.GetNamespace(), objMeta.GetName())
+	}
+	return objMeta.GetName()
+}