Allow multiple Pinnipeds to work on same cluster

Yes, this is a huge commit.

The middleware allows you to customize the API groups of all of the
*.pinniped.dev API groups.

Some notes about other small things in this commit:
- We removed the internal/client package in favor of pkg/conciergeclient. The
  two packages do basically the same thing. I don't think we use the former
  anymore.
- We re-enabled cluster-scoped owner assertions in the integration tests.
  This code was added in internal/ownerref. See a0546942 for when this
  assertion was removed.
- Note: the middlware code is in charge of restoring the GV of a request object,
  so we should never need to write mutations that do that.
- We updated the supervisor secret generation to no longer manually set an owner
  reference to the deployment since the middleware code now does this. I think we
  still need some way to make an initial event for the secret generator
  controller, which involves knowing the namespace and the name of the generated
  secret, so I still wired the deployment through. We could use a namespace/name
  tuple here, but I was lazy.

Signed-off-by: Andrew Keesler <akeesler@vmware.com>
Co-authored-by: Ryan Richard <richardry@vmware.com>

internal/kubeclient/copied.go

@@ -0,0 +1,70 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"net/url"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	restclient "k8s.io/client-go/rest"
+
+	"go.pinniped.dev/internal/plog"
+)
+
+// defaultServerUrlFor was copied from k8s.io/client-go/rest/url_utils.go.
+//nolint: golint
+func defaultServerUrlFor(config *restclient.Config) (*url.URL, string, error) {
+	hasCA := len(config.CAFile) != 0 || len(config.CAData) != 0
+	hasCert := len(config.CertFile) != 0 || len(config.CertData) != 0
+	defaultTLS := hasCA || hasCert || config.Insecure
+	host := config.Host
+	if host == "" {
+		host = "localhost"
+	}
+
+	if config.GroupVersion != nil {
+		return restclient.DefaultServerURL(host, config.APIPath, *config.GroupVersion, defaultTLS)
+	}
+	return restclient.DefaultServerURL(host, config.APIPath, schema.GroupVersion{}, defaultTLS)
+}
+
+// truncateBody was copied from k8s.io/client-go/rest/request.go
+// ...except i changed klog invocations to analogous plog invocations
+//
+// truncateBody decides if the body should be truncated, based on the glog Verbosity.
+func truncateBody(body string) string {
+	max := 0
+	switch {
+	case plog.Enabled(plog.LevelAll):
+		return body
+	case plog.Enabled(plog.LevelTrace):
+		max = 10240
+	case plog.Enabled(plog.LevelDebug):
+		max = 1024
+	}
+
+	if len(body) <= max {
+		return body
+	}
+
+	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
+// 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 plog.Enabled(plog.LevelDebug) {
+		if bytes.IndexFunc(body, func(r rune) bool {
+			return r < 0x0a
+		}) != -1 {
+			plog.Debug(prefix, "body", truncateBody(hex.Dump(body)))
+		} else {
+			plog.Debug(prefix, "body", truncateBody(string(body)))
+		}
+	}
+}

internal/kubeclient/gvk.go

@@ -0,0 +1,79 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func maybeRestoreGVK(serializer runtime.Serializer, respData []byte, result *mutationResult) ([]byte, error) {
+	if !result.gvkChanged {
+		return respData, nil
+	}
+
+	// the body could be an API status, random trash or the actual object we want
+	unknown := &runtime.Unknown{}
+	_ = runtime.DecodeInto(serializer, respData, unknown) // we do not care about the error
+
+	doesNotNeedGVKFix := len(unknown.Raw) == 0 || unknown.GroupVersionKind() != result.newGVK
+
+	if doesNotNeedGVKFix {
+		return respData, nil
+	}
+
+	return restoreGVK(serializer, unknown, result.origGVK)
+}
+
+func restoreGVK(encoder runtime.Encoder, unknown *runtime.Unknown, gvk schema.GroupVersionKind) ([]byte, error) {
+	typeMeta := runtime.TypeMeta{}
+	typeMeta.APIVersion, typeMeta.Kind = gvk.ToAPIVersionAndKind()
+
+	newUnknown := &runtime.Unknown{}
+	*newUnknown = *unknown
+	newUnknown.TypeMeta = typeMeta
+
+	switch newUnknown.ContentType {
+	case runtime.ContentTypeJSON:
+		// json is messy if we want to avoid decoding the whole object
+		keysOnly := map[string]json.RawMessage{}
+
+		// get the keys.  this does not preserve order.
+		if err := json.Unmarshal(newUnknown.Raw, &keysOnly); err != nil {
+			return nil, fmt.Errorf("failed to unmarshall json keys: %w", err)
+		}
+
+		// turn the type meta into JSON bytes
+		typeMetaBytes, err := json.Marshal(typeMeta)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshall type meta: %w", err)
+		}
+
+		// overwrite the type meta keys with the new data
+		if err := json.Unmarshal(typeMetaBytes, &keysOnly); err != nil {
+			return nil, fmt.Errorf("failed to type meta keys: %w", err)
+		}
+
+		// marshall everything back to bytes
+		newRaw, err := json.Marshal(keysOnly)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshall new raw: %w", err)
+		}
+
+		// we could just return the bytes but it feels weird to not use the encoder
+		newUnknown.Raw = newRaw
+
+	case runtime.ContentTypeProtobuf:
+		// protobuf is easy because of the unknown wrapper
+		// newUnknown.Raw already contains the correct data we need
+
+	default:
+		return nil, fmt.Errorf("unknown content type: %s", newUnknown.ContentType) // this should never happen
+	}
+
+	return runtime.Encode(encoder, newUnknown)
+}

internal/kubeclient/gvk_test.go

@@ -0,0 +1,222 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func Test_maybeRestoreGVK(t *testing.T) {
+	type args struct {
+		unknown         *runtime.Unknown
+		origGVK, newGVK schema.GroupVersionKind
+	}
+	tests := []struct {
+		name        string
+		args        args
+		want        runtime.Object
+		wantChanged bool
+		wantErr     string
+	}{
+		{
+			name: "should update gvk via JSON",
+			args: args{
+				unknown: &runtime.Unknown{
+					TypeMeta: runtime.TypeMeta{
+						APIVersion: "new/v1",
+						Kind:       "Tree",
+					},
+					Raw:         []byte(`{"apiVersion":"new/v1","kind":"Tree","spec":{"pandas":"love"}}`),
+					ContentType: runtime.ContentTypeJSON,
+				},
+				origGVK: schema.GroupVersionKind{
+					Group:   "old",
+					Version: "v1",
+					Kind:    "Tree",
+				},
+				newGVK: schema.GroupVersionKind{
+					Group:   "new",
+					Version: "v1",
+					Kind:    "Tree",
+				},
+			},
+			want: &runtime.Unknown{
+				TypeMeta: runtime.TypeMeta{
+					APIVersion: "old/v1",
+					Kind:       "Tree",
+				},
+				Raw:         []byte(`{"apiVersion":"old/v1","kind":"Tree","spec":{"pandas":"love"}}`),
+				ContentType: runtime.ContentTypeJSON,
+			},
+			wantChanged: true,
+			wantErr:     "",
+		},
+		{
+			name: "should update gvk via protobuf",
+			args: args{
+				unknown: &runtime.Unknown{
+					TypeMeta: runtime.TypeMeta{
+						APIVersion: "new/v1",
+						Kind:       "Tree",
+					},
+					Raw:         []byte(`assumed to be valid and does not change`),
+					ContentType: runtime.ContentTypeProtobuf,
+				},
+				origGVK: schema.GroupVersionKind{
+					Group:   "original",
+					Version: "v1",
+					Kind:    "Tree",
+				},
+				newGVK: schema.GroupVersionKind{
+					Group:   "new",
+					Version: "v1",
+					Kind:    "Tree",
+				},
+			},
+			want: &runtime.Unknown{
+				TypeMeta: runtime.TypeMeta{
+					APIVersion: "original/v1",
+					Kind:       "Tree",
+				},
+				Raw:         []byte(`assumed to be valid and does not change`),
+				ContentType: runtime.ContentTypeProtobuf,
+			},
+			wantChanged: true,
+			wantErr:     "",
+		},
+		{
+			name: "should ignore because gvk is different",
+			args: args{
+				unknown: &runtime.Unknown{
+					TypeMeta: runtime.TypeMeta{
+						APIVersion: "new/v1",
+						Kind:       "Tree",
+					},
+				},
+				newGVK: schema.GroupVersionKind{
+					Group:   "new",
+					Version: "v1",
+					Kind:    "Forest",
+				},
+			},
+			want:        nil,
+			wantChanged: false,
+			wantErr:     "",
+		},
+		{
+			name: "empty raw is ignored",
+			args: args{
+				unknown: &runtime.Unknown{},
+			},
+			want:        nil,
+			wantChanged: false,
+			wantErr:     "",
+		},
+		{
+			name: "invalid content type errors",
+			args: args{
+				unknown: &runtime.Unknown{
+					TypeMeta: runtime.TypeMeta{
+						APIVersion: "walrus.tld/v1",
+						Kind:       "Seal",
+					},
+					Raw:         []byte(`data that should be ignored because we do not used YAML`),
+					ContentType: runtime.ContentTypeYAML,
+				},
+				origGVK: schema.GroupVersionKind{
+					Group:   "pinniped.dev",
+					Version: "v1",
+					Kind:    "Seal",
+				},
+				newGVK: schema.GroupVersionKind{
+					Group:   "walrus.tld",
+					Version: "v1",
+					Kind:    "Seal",
+				},
+			},
+			want:        nil,
+			wantChanged: false,
+			wantErr:     "unknown content type: application/yaml",
+		},
+		{
+			name: "invalid JSON should error",
+			args: args{
+				unknown: &runtime.Unknown{
+					TypeMeta: runtime.TypeMeta{
+						APIVersion: "ocean/v1",
+						Kind:       "Water",
+					},
+					Raw:         []byte(`lol not JSON`),
+					ContentType: runtime.ContentTypeJSON,
+				},
+				origGVK: schema.GroupVersionKind{
+					Group:   "dirt",
+					Version: "v1",
+					Kind:    "Land",
+				},
+				newGVK: schema.GroupVersionKind{
+					Group:   "ocean",
+					Version: "v1",
+					Kind:    "Water",
+				},
+			},
+			want:        nil,
+			wantChanged: false,
+			wantErr:     "failed to unmarshall json keys: invalid character 'l' looking for beginning of value",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			serializer := &testSerializer{unknown: tt.args.unknown}
+			respData := []byte(`original`)
+			result := &mutationResult{origGVK: tt.args.origGVK, newGVK: tt.args.newGVK, gvkChanged: tt.args.origGVK != tt.args.newGVK}
+
+			newRespData, err := maybeRestoreGVK(serializer, respData, result)
+
+			if len(tt.wantErr) > 0 {
+				require.EqualError(t, err, tt.wantErr)
+				require.Nil(t, newRespData)
+				require.Nil(t, serializer.obj)
+				return
+			}
+
+			require.NoError(t, err)
+
+			if tt.wantChanged {
+				require.Equal(t, []byte(`changed`), newRespData)
+			} else {
+				require.Equal(t, []byte(`original`), newRespData)
+			}
+
+			require.Equal(t, tt.want, serializer.obj)
+		})
+	}
+}
+
+type testSerializer struct {
+	unknown *runtime.Unknown
+	obj     runtime.Object
+}
+
+func (s *testSerializer) Encode(obj runtime.Object, w io.Writer) error {
+	s.obj = obj
+	_, err := w.Write([]byte(`changed`))
+	return err
+}
+
+func (s *testSerializer) Decode(_ []byte, _ *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
+	u := into.(*runtime.Unknown)
+	*u = *s.unknown
+	return u, nil, nil
+}
+
+func (s *testSerializer) Identifier() runtime.Identifier {
+	panic("not called")
+}

internal/kubeclient/kubeclient.go

@@ -6,7 +6,6 @@ package kubeclient
 import (
 	"fmt"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	kubescheme "k8s.io/client-go/kubernetes/scheme"
@@ -29,12 +28,6 @@ type Client struct {
 	JSONConfig, ProtoConfig *restclient.Config
 }
 
-// TODO expand this interface to address more complex use cases.
-type Middleware interface {
-	Handles(httpMethod string) bool
-	Mutate(obj metav1.Object) (mutated bool)
-}
-
 func New(opts ...Option) (*Client, error) {
 	c := &clientConfig{}
 
@@ -58,13 +51,13 @@ func New(opts ...Option) (*Client, error) {
 	protoKubeConfig := createProtoKubeConfig(c.config)
 
 	// Connect to the core Kubernetes API.
-	k8sClient, err := kubernetes.NewForConfig(configWithWrapper(protoKubeConfig, kubescheme.Codecs, c.middlewares))
+	k8sClient, err := kubernetes.NewForConfig(configWithWrapper(protoKubeConfig, kubescheme.Scheme, kubescheme.Codecs, c.middlewares))
 	if err != nil {
 		return nil, fmt.Errorf("could not initialize Kubernetes client: %w", err)
 	}
 
 	// Connect to the Kubernetes aggregation API.
-	aggregatorClient, err := aggregatorclient.NewForConfig(configWithWrapper(protoKubeConfig, aggregatorclientscheme.Codecs, c.middlewares))
+	aggregatorClient, err := aggregatorclient.NewForConfig(configWithWrapper(protoKubeConfig, aggregatorclientscheme.Scheme, aggregatorclientscheme.Codecs, c.middlewares))
 	if err != nil {
 		return nil, fmt.Errorf("could not initialize aggregation client: %w", err)
 	}
@@ -72,8 +65,7 @@ func New(opts ...Option) (*Client, error) {
 	// Connect to the pinniped concierge API.
 	// We cannot use protobuf encoding here because we are using CRDs
 	// (for which protobuf encoding is not yet supported).
-	// TODO we should try to add protobuf support to TokenCredentialRequests since it is an aggregated API
-	pinnipedConciergeClient, err := pinnipedconciergeclientset.NewForConfig(configWithWrapper(jsonKubeConfig, pinnipedconciergeclientsetscheme.Codecs, c.middlewares))
+	pinnipedConciergeClient, err := pinnipedconciergeclientset.NewForConfig(configWithWrapper(jsonKubeConfig, pinnipedconciergeclientsetscheme.Scheme, pinnipedconciergeclientsetscheme.Codecs, c.middlewares))
 	if err != nil {
 		return nil, fmt.Errorf("could not initialize pinniped client: %w", err)
 	}
@@ -81,7 +73,7 @@ func New(opts ...Option) (*Client, error) {
 	// Connect to the pinniped supervisor API.
 	// We cannot use protobuf encoding here because we are using CRDs
 	// (for which protobuf encoding is not yet supported).
-	pinnipedSupervisorClient, err := pinnipedsupervisorclientset.NewForConfig(configWithWrapper(jsonKubeConfig, pinnipedsupervisorclientsetscheme.Codecs, c.middlewares))
+	pinnipedSupervisorClient, err := pinnipedsupervisorclientset.NewForConfig(configWithWrapper(jsonKubeConfig, pinnipedsupervisorclientsetscheme.Scheme, pinnipedsupervisorclientsetscheme.Codecs, c.middlewares))
 	if err != nil {
 		return nil, fmt.Errorf("could not initialize pinniped client: %w", err)
 	}

internal/kubeclient/kubeclient_test.go

@@ -0,0 +1,775 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"runtime"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/transport"
+	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
+
+	loginv1alpha1 "go.pinniped.dev/generated/1.20/apis/concierge/login/v1alpha1"
+	configv1alpha1 "go.pinniped.dev/generated/1.20/apis/supervisor/config/v1alpha1"
+	"go.pinniped.dev/internal/testutil/fakekubeapi"
+)
+
+const (
+	someClusterName = "some cluster name"
+)
+
+var (
+	podGVK  = corev1.SchemeGroupVersion.WithKind("Pod")
+	goodPod = &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "good-pod",
+			Namespace: "good-namespace",
+		},
+	}
+
+	apiServiceGVK  = apiregistrationv1.SchemeGroupVersion.WithKind("APIService")
+	goodAPIService = &apiregistrationv1.APIService{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "good-api-service",
+		},
+	}
+
+	tokenCredentialRequestGVK  = loginv1alpha1.SchemeGroupVersion.WithKind("TokenCredentialRequest")
+	goodTokenCredentialRequest = &loginv1alpha1.TokenCredentialRequest{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "good-token-credential-request",
+			Namespace: "good-namespace",
+		},
+	}
+
+	federationDomainGVK  = configv1alpha1.SchemeGroupVersion.WithKind("FederationDomain")
+	goodFederationDomain = &configv1alpha1.FederationDomain{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "good-federation-domain",
+			Namespace: "good-namespace",
+		},
+	}
+
+	middlewareAnnotations = map[string]string{"some-annotation": "thing 1"}
+	middlewareLabels      = map[string]string{"some-label": "thing 2"}
+)
+
+// TestKubeclient tests a subset of kubeclient functionality (from the public interface down). We
+// intend for the following list of things to be tested with the integration tests:
+//   list (running in every informer cache)
+//   watch (running in every informer cache)
+//   discovery
+//   api errors
+func TestKubeclient(t *testing.T) {
+	// plog.ValidateAndSetLogLevelGlobally(plog.LevelDebug) // uncomment me to get some more debug logs
+
+	tests := []struct {
+		name                                    string
+		editRestConfig                          func(t *testing.T, restConfig *rest.Config)
+		middlewares                             func(t *testing.T) []*spyMiddleware
+		reallyRunTest                           func(t *testing.T, c *Client)
+		wantMiddlewareReqs, wantMiddlewareResps [][]Object
+	}{
+		{
+			name: "crud core api",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newAnnotationMiddleware(t), newLabelMiddleware(t)}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				pod, err := c.Kubernetes.
+					CoreV1().
+					Pods(goodPod.Namespace).
+					Create(context.Background(), goodPod, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodPod, pod)
+
+				// read
+				pod, err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Get(context.Background(), pod.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodPod, annotations(), labels()), pod)
+
+				// read when not found
+				_, err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Get(context.Background(), "this-pod-does-not-exist", metav1.GetOptions{})
+				require.EqualError(t, err, "the server could not find the requested resource (get pods this-pod-does-not-exist)")
+
+				// update
+				goodPodWithAnnotationsAndLabelsAndClusterName := with(goodPod, annotations(), labels(), clusterName()).(*corev1.Pod)
+				pod, err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Update(context.Background(), goodPodWithAnnotationsAndLabelsAndClusterName, metav1.UpdateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodPodWithAnnotationsAndLabelsAndClusterName, pod)
+
+				// delete
+				err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			},
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodPod, gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+					with(goodPod, annotations(), labels(), clusterName(), gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+				},
+				{
+					with(goodPod, annotations(), gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+					with(goodPod, annotations(), labels(), clusterName(), gvk(podGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(podGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodPod, annotations(), labels(), gvk(podGVK)),
+					with(goodPod, annotations(), labels(), gvk(podGVK)),
+					with(goodPod, annotations(), labels(), clusterName(), gvk(podGVK)),
+				},
+				{
+					with(goodPod, emptyAnnotations(), labels(), gvk(podGVK)),
+					with(goodPod, annotations(), labels(), gvk(podGVK)),
+					with(goodPod, annotations(), labels(), clusterName(), gvk(podGVK)),
+				},
+			},
+		},
+		{
+			name: "crud core api without middlewares",
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				pod, err := c.Kubernetes.
+					CoreV1().
+					Pods(goodPod.Namespace).
+					Create(context.Background(), goodPod, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodPod, pod)
+
+				// read
+				pod, err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Get(context.Background(), pod.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodPod), pod)
+
+				// update
+				pod, err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Update(context.Background(), goodPod, metav1.UpdateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodPod, pod)
+
+				// delete
+				err = c.Kubernetes.
+					CoreV1().
+					Pods(pod.Namespace).
+					Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			},
+		},
+		{
+			name: "crud aggregation api",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newAnnotationMiddleware(t), newLabelMiddleware(t)}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				apiService, err := c.Aggregation.
+					ApiregistrationV1().
+					APIServices().
+					Create(context.Background(), goodAPIService, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodAPIService, apiService)
+
+				// read
+				apiService, err = c.Aggregation.
+					ApiregistrationV1().
+					APIServices().
+					Get(context.Background(), apiService.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodAPIService, annotations(), labels()), apiService)
+
+				// update
+				goodAPIServiceWithAnnotationsAndLabelsAndClusterName := with(goodAPIService, annotations(), labels(), clusterName()).(*apiregistrationv1.APIService)
+				apiService, err = c.Aggregation.
+					ApiregistrationV1().
+					APIServices().
+					Update(context.Background(), goodAPIServiceWithAnnotationsAndLabelsAndClusterName, metav1.UpdateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodAPIServiceWithAnnotationsAndLabelsAndClusterName, apiService)
+
+				// delete
+				err = c.Aggregation.
+					ApiregistrationV1().
+					APIServices().
+					Delete(context.Background(), apiService.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			},
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodAPIService, gvk(apiServiceGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), clusterName(), gvk(apiServiceGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(apiServiceGVK)),
+				},
+				{
+					with(goodAPIService, annotations(), gvk(apiServiceGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), clusterName(), gvk(apiServiceGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(apiServiceGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodAPIService, annotations(), labels(), gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), clusterName(), gvk(apiServiceGVK)),
+				},
+				{
+					with(goodAPIService, emptyAnnotations(), labels(), gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), gvk(apiServiceGVK)),
+					with(goodAPIService, annotations(), labels(), clusterName(), gvk(apiServiceGVK)),
+				},
+			},
+		},
+		{
+			name: "crud concierge api",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newAnnotationMiddleware(t), newLabelMiddleware(t)}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				tokenCredentialRequest, err := c.PinnipedConcierge.
+					LoginV1alpha1().
+					TokenCredentialRequests(goodTokenCredentialRequest.Namespace).
+					Create(context.Background(), goodTokenCredentialRequest, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodTokenCredentialRequest, tokenCredentialRequest)
+
+				// read
+				tokenCredentialRequest, err = c.PinnipedConcierge.
+					LoginV1alpha1().
+					TokenCredentialRequests(tokenCredentialRequest.Namespace).
+					Get(context.Background(), tokenCredentialRequest.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodTokenCredentialRequest, annotations(), labels()), tokenCredentialRequest)
+
+				// update
+				goodTokenCredentialRequestWithAnnotationsAndLabelsAndClusterName := with(goodTokenCredentialRequest, annotations(), labels(), clusterName()).(*loginv1alpha1.TokenCredentialRequest)
+				tokenCredentialRequest, err = c.PinnipedConcierge.
+					LoginV1alpha1().
+					TokenCredentialRequests(tokenCredentialRequest.Namespace).
+					Update(context.Background(), goodTokenCredentialRequestWithAnnotationsAndLabelsAndClusterName, metav1.UpdateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodTokenCredentialRequestWithAnnotationsAndLabelsAndClusterName, tokenCredentialRequest)
+
+				// delete
+				err = c.PinnipedConcierge.
+					LoginV1alpha1().
+					TokenCredentialRequests(tokenCredentialRequest.Namespace).
+					Delete(context.Background(), tokenCredentialRequest.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			},
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodTokenCredentialRequest, gvk(tokenCredentialRequestGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), clusterName(), gvk(tokenCredentialRequestGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(tokenCredentialRequestGVK)),
+				},
+				{
+					with(goodTokenCredentialRequest, annotations(), gvk(tokenCredentialRequestGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), clusterName(), gvk(tokenCredentialRequestGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(tokenCredentialRequestGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodTokenCredentialRequest, annotations(), labels(), gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), clusterName(), gvk(tokenCredentialRequestGVK)),
+				},
+				{
+					with(goodTokenCredentialRequest, emptyAnnotations(), labels(), gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), gvk(tokenCredentialRequestGVK)),
+					with(goodTokenCredentialRequest, annotations(), labels(), clusterName(), gvk(tokenCredentialRequestGVK)),
+				},
+			},
+		},
+		{
+			name: "crud supervisor api",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newAnnotationMiddleware(t), newLabelMiddleware(t)}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				federationDomain, err := c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(goodFederationDomain.Namespace).
+					Create(context.Background(), goodFederationDomain, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodFederationDomain, federationDomain)
+
+				// read
+				federationDomain, err = c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(federationDomain.Namespace).
+					Get(context.Background(), federationDomain.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodFederationDomain, annotations(), labels()), federationDomain)
+
+				// update
+				goodFederationDomainWithAnnotationsAndLabelsAndClusterName := with(goodFederationDomain, annotations(), labels(), clusterName()).(*configv1alpha1.FederationDomain)
+				federationDomain, err = c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(federationDomain.Namespace).
+					Update(context.Background(), goodFederationDomainWithAnnotationsAndLabelsAndClusterName, metav1.UpdateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, goodFederationDomainWithAnnotationsAndLabelsAndClusterName, federationDomain)
+
+				// delete
+				err = c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(federationDomain.Namespace).
+					Delete(context.Background(), federationDomain.Name, metav1.DeleteOptions{})
+				require.NoError(t, err)
+			},
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), clusterName(), gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, annotations(), gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), clusterName(), gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodFederationDomain, annotations(), labels(), gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), clusterName(), gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, emptyAnnotations(), labels(), gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), gvk(federationDomainGVK)),
+					with(goodFederationDomain, annotations(), labels(), clusterName(), gvk(federationDomainGVK)),
+				},
+			},
+		},
+		{
+			name: "we don't call any middleware if there are no mutation funcs",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, false, false, false), newSimpleMiddleware(t, false, false, false)}
+			},
+			reallyRunTest:       createGetFederationDomainTest,
+			wantMiddlewareReqs:  [][]Object{nil, nil},
+			wantMiddlewareResps: [][]Object{nil, nil},
+		},
+		{
+			name: "we don't call any resp middleware if there was no req mutations done and there are no resp mutation funcs",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, true, false, false), newSimpleMiddleware(t, true, false, false)}
+			},
+			reallyRunTest: createGetFederationDomainTest,
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{nil, nil},
+		},
+		{
+			name: "we don't call any resp middleware if there are no resp mutation funcs even if there was req mutations done",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, true, true, false), newSimpleMiddleware(t, true, true, false)}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				// create
+				federationDomain, err := c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(goodFederationDomain.Namespace).
+					Create(context.Background(), goodFederationDomain, metav1.CreateOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodFederationDomain, clusterName()), federationDomain)
+
+				// read
+				federationDomain, err = c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(federationDomain.Namespace).
+					Get(context.Background(), federationDomain.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				require.Equal(t, with(goodFederationDomain, clusterName()), federationDomain)
+			},
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, clusterName(), gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{nil, nil},
+		},
+		{
+			name: "we still call resp middleware if there is a resp mutation func even if there were req mutation funcs",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, false, false, true), newSimpleMiddleware(t, false, false, true)}
+			},
+			reallyRunTest:      createGetFederationDomainTest,
+			wantMiddlewareReqs: [][]Object{nil, nil},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+				},
+			},
+		},
+		{
+			name: "we still call resp middleware if there is a resp mutation func even if there was no req mutation",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, true, false, true), newSimpleMiddleware(t, true, false, true)}
+			},
+			reallyRunTest: createGetFederationDomainTest,
+			wantMiddlewareReqs: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK)),
+				},
+			},
+			wantMiddlewareResps: [][]Object{
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+				},
+				{
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+					with(goodFederationDomain, gvk(federationDomainGVK)),
+				},
+			},
+		},
+		{
+			name: "mutating object meta on a get request is not allowed since that isn't pertinent to the api request",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{{
+					name: "non-pertinent mutater",
+					t:    t,
+					mutateReq: func(rt RoundTrip, obj Object) {
+						clusterName()(obj)
+					},
+				}}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				_, err := c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(goodFederationDomain.Namespace).
+					Get(context.Background(), goodFederationDomain.Name, metav1.GetOptions{})
+				require.Error(t, err)
+				require.Contains(t, err.Error(), "invalid object meta mutation")
+			},
+			wantMiddlewareReqs:  [][]Object{{with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK))}},
+			wantMiddlewareResps: [][]Object{nil},
+		},
+		{
+			name: "when the client gets errors from the api server",
+			middlewares: func(t *testing.T) []*spyMiddleware {
+				return []*spyMiddleware{newSimpleMiddleware(t, true, false, false)}
+			},
+			editRestConfig: func(t *testing.T, restConfig *rest.Config) {
+				restConfig.Dial = func(_ context.Context, _, _ string) (net.Conn, error) {
+					return nil, fmt.Errorf("some fake connection error")
+				}
+			},
+			reallyRunTest: func(t *testing.T, c *Client) {
+				_, err := c.PinnipedSupervisor.
+					ConfigV1alpha1().
+					FederationDomains(goodFederationDomain.Namespace).
+					Get(context.Background(), goodFederationDomain.Name, metav1.GetOptions{})
+				require.Error(t, err)
+				require.Contains(t, err.Error(), ": some fake connection error")
+			},
+			wantMiddlewareReqs:  [][]Object{{with(&metav1.PartialObjectMetadata{}, gvk(federationDomainGVK))}},
+			wantMiddlewareResps: [][]Object{nil},
+		},
+	}
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			server, restConfig := fakekubeapi.Start(t, nil)
+			defer server.Close()
+
+			if test.editRestConfig != nil {
+				test.editRestConfig(t, restConfig)
+			}
+
+			// our rt chain is:
+			//  kubeclient -> wantCloseResp -> http.DefaultTransport -> wantCloseResp -> kubeclient
+			restConfig.Wrap(wantCloseRespWrapper(t))
+
+			var middlewares []*spyMiddleware
+			if test.middlewares != nil {
+				middlewares = test.middlewares(t)
+			}
+
+			opts := []Option{WithConfig(restConfig)}
+			for _, middleware := range middlewares {
+				opts = append(opts, WithMiddleware(middleware))
+			}
+			client, err := New(opts...)
+			require.NoError(t, err)
+
+			test.reallyRunTest(t, client)
+
+			for i, spyMiddleware := range middlewares {
+				require.Equalf(t, test.wantMiddlewareReqs[i], spyMiddleware.reqObjs, "unexpected req obj in middleware %q (index %d)", spyMiddleware.name, i)
+				require.Equalf(t, test.wantMiddlewareResps[i], spyMiddleware.respObjs, "unexpected resp obj in middleware %q (index %d)", spyMiddleware.name, i)
+			}
+		})
+	}
+}
+
+type spyMiddleware struct {
+	name       string
+	t          *testing.T
+	mutateReq  func(RoundTrip, Object)
+	mutateResp func(RoundTrip, Object)
+	reqObjs    []Object
+	respObjs   []Object
+}
+
+func (s *spyMiddleware) Handle(_ context.Context, rt RoundTrip) {
+	s.t.Log(s.name, "handling", reqStr(rt, nil))
+
+	if s.mutateReq != nil {
+		rt.MutateRequest(func(obj Object) {
+			s.t.Log(s.name, "mutating request", reqStr(rt, obj))
+			s.reqObjs = append(s.reqObjs, obj.DeepCopyObject().(Object))
+			s.mutateReq(rt, obj)
+		})
+	}
+
+	if s.mutateResp != nil {
+		rt.MutateResponse(func(obj Object) {
+			s.t.Log(s.name, "mutating response", reqStr(rt, obj))
+			s.respObjs = append(s.respObjs, obj.DeepCopyObject().(Object))
+			s.mutateResp(rt, obj)
+		})
+	}
+}
+
+func reqStr(rt RoundTrip, obj Object) string {
+	b := strings.Builder{}
+	fmt.Fprintf(&b, "%s /%s", rt.Verb(), rt.Resource().GroupVersion())
+	if rt.NamespaceScoped() {
+		fmt.Fprintf(&b, "/namespaces/%s", rt.Namespace())
+	}
+	fmt.Fprintf(&b, "/%s", rt.Resource().Resource)
+	if obj != nil {
+		fmt.Fprintf(&b, "/%s", obj.GetName())
+	}
+	return b.String()
+}
+
+func newAnnotationMiddleware(t *testing.T) *spyMiddleware {
+	return &spyMiddleware{
+		name: "annotater",
+		t:    t,
+		mutateReq: func(rt RoundTrip, obj Object) {
+			if rt.Verb() == VerbCreate {
+				annotations()(obj)
+			}
+		},
+		mutateResp: func(rt RoundTrip, obj Object) {
+			if rt.Verb() == VerbCreate {
+				for key := range middlewareAnnotations {
+					delete(obj.GetAnnotations(), key)
+				}
+			}
+		},
+	}
+}
+
+func newLabelMiddleware(t *testing.T) *spyMiddleware {
+	return &spyMiddleware{
+		name: "labeler",
+		t:    t,
+		mutateReq: func(rt RoundTrip, obj Object) {
+			if rt.Verb() == VerbCreate {
+				labels()(obj)
+			}
+		},
+		mutateResp: func(rt RoundTrip, obj Object) {
+			if rt.Verb() == VerbCreate {
+				for key := range middlewareLabels {
+					delete(obj.GetLabels(), key)
+				}
+			}
+		},
+	}
+}
+
+func newSimpleMiddleware(t *testing.T, hasMutateReqFunc, mutatedReq, hasMutateRespFunc bool) *spyMiddleware {
+	m := &spyMiddleware{
+		name: "nop",
+		t:    t,
+	}
+	if hasMutateReqFunc {
+		m.mutateReq = func(rt RoundTrip, obj Object) {
+			if mutatedReq {
+				if rt.Verb() == VerbCreate {
+					obj.SetClusterName(someClusterName)
+				}
+			}
+		}
+	}
+	if hasMutateRespFunc {
+		m.mutateResp = func(rt RoundTrip, obj Object) {}
+	}
+	return m
+}
+
+type wantCloser struct {
+	io.ReadCloser
+	closeCount                      int
+	couldReadBytesJustBeforeClosing bool
+}
+
+func (wc *wantCloser) Close() error {
+	wc.closeCount++
+	n, _ := wc.ReadCloser.Read([]byte{0})
+	if n > 0 {
+		// there were still bytes left to be read
+		wc.couldReadBytesJustBeforeClosing = true
+	}
+	return wc.ReadCloser.Close()
+}
+
+// wantCloseRespWrapper returns a transport.WrapperFunc that validates that the http.Response
+// returned by the underlying http.RoundTripper is closed properly.
+func wantCloseRespWrapper(t *testing.T) transport.WrapperFunc {
+	_, file, line, ok := runtime.Caller(1)
+	if !ok {
+		file = "???"
+		line = 0
+	}
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return roundTripperFunc(func(req *http.Request) (bool, *http.Response, error) {
+			resp, err := rt.RoundTrip(req)
+			if err != nil {
+				// request failed, so there is no response body to watch for Close() calls on
+				return false, resp, err
+			}
+			wc := &wantCloser{ReadCloser: resp.Body}
+			t.Cleanup(func() {
+				require.False(t, wc.couldReadBytesJustBeforeClosing, "did not consume all response body bytes before closing %s:%d", file, line)
+				require.Equalf(t, wc.closeCount, 1, "did not close resp body at %s:%d", file, line)
+			})
+			resp.Body = wc
+			return false, resp, err
+		})
+	}
+}
+
+type withFunc func(obj Object)
+
+func with(obj Object, withFuncs ...withFunc) Object {
+	obj = obj.DeepCopyObject().(Object)
+	for _, withFunc := range withFuncs {
+		withFunc(obj)
+	}
+	return obj
+}
+
+func gvk(gvk schema.GroupVersionKind) withFunc {
+	return func(obj Object) {
+		obj.GetObjectKind().SetGroupVersionKind(gvk)
+	}
+}
+
+func annotations() withFunc {
+	return func(obj Object) {
+		obj.SetAnnotations(middlewareAnnotations)
+	}
+}
+
+func emptyAnnotations() withFunc {
+	return func(obj Object) {
+		obj.SetAnnotations(make(map[string]string))
+	}
+}
+
+func labels() withFunc {
+	return func(obj Object) {
+		obj.SetLabels(middlewareLabels)
+	}
+}
+
+func clusterName() withFunc {
+	return func(obj Object) {
+		obj.SetClusterName(someClusterName)
+	}
+}
+
+func createGetFederationDomainTest(t *testing.T, client *Client) {
+	t.Helper()
+
+	// create
+	federationDomain, err := client.PinnipedSupervisor.
+		ConfigV1alpha1().
+		FederationDomains(goodFederationDomain.Namespace).
+		Create(context.Background(), goodFederationDomain, metav1.CreateOptions{})
+	require.NoError(t, err)
+	require.Equal(t, goodFederationDomain, federationDomain)
+
+	// read
+	federationDomain, err = client.PinnipedSupervisor.
+		ConfigV1alpha1().
+		FederationDomains(federationDomain.Namespace).
+		Get(context.Background(), federationDomain.Name, metav1.GetOptions{})
+	require.NoError(t, err)
+	require.Equal(t, goodFederationDomain, federationDomain)
+}

internal/kubeclient/middleware.go

@@ -0,0 +1,147 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+type Middleware interface {
+	Handle(ctx context.Context, rt RoundTrip)
+}
+
+var _ Middleware = MiddlewareFunc(nil)
+
+type MiddlewareFunc func(ctx context.Context, rt RoundTrip)
+
+func (f MiddlewareFunc) Handle(ctx context.Context, rt RoundTrip) {
+	f(ctx, rt)
+}
+
+var _ Middleware = Middlewares{}
+
+type Middlewares []Middleware
+
+func (m Middlewares) Handle(ctx context.Context, rt RoundTrip) {
+	for _, middleware := range m {
+		middleware := middleware
+		middleware.Handle(ctx, rt)
+	}
+}
+
+type RoundTrip interface {
+	Verb() Verb
+	Namespace() string // this is the only valid way to check namespace, Object.GetNamespace() will almost always be empty
+	NamespaceScoped() bool
+	Resource() schema.GroupVersionResource
+	Subresource() string
+	MutateRequest(f func(obj Object))
+	MutateResponse(f func(obj Object))
+}
+
+type Object interface {
+	runtime.Object // generic access to TypeMeta
+	metav1.Object  // generic access to ObjectMeta
+}
+
+var _ RoundTrip = &request{}
+
+type request struct {
+	verb                Verb
+	namespace           string
+	resource            schema.GroupVersionResource
+	reqFuncs, respFuncs []func(obj Object)
+	subresource         string
+}
+
+func (r *request) Verb() Verb {
+	return r.verb
+}
+
+func (r *request) Namespace() string {
+	return r.namespace
+}
+
+//nolint: gochecknoglobals
+var namespaceGVR = corev1.SchemeGroupVersion.WithResource("namespaces")
+
+func (r *request) NamespaceScoped() bool {
+	if r.Resource() == namespaceGVR {
+		return false // always consider namespaces to be cluster scoped
+	}
+
+	return len(r.Namespace()) != 0
+}
+
+func (r *request) Resource() schema.GroupVersionResource {
+	return r.resource
+}
+
+func (r *request) Subresource() string {
+	return r.subresource
+}
+
+func (r *request) MutateRequest(f func(obj Object)) {
+	r.reqFuncs = append(r.reqFuncs, f)
+}
+
+func (r *request) MutateResponse(f func(obj Object)) {
+	r.respFuncs = append(r.respFuncs, f)
+}
+
+type mutationResult struct {
+	origGVK, newGVK     schema.GroupVersionKind
+	gvkChanged, mutated bool
+}
+
+func (r *request) mutateRequest(obj Object) (*mutationResult, error) {
+	origGVK := obj.GetObjectKind().GroupVersionKind()
+	if origGVK.Empty() {
+		return nil, fmt.Errorf("invalid empty orig GVK for %T: %#v", obj, r)
+	}
+
+	origObj, ok := obj.DeepCopyObject().(Object)
+	if !ok {
+		return nil, fmt.Errorf("invalid deep copy semantics for %T: %#v", obj, r)
+	}
+
+	for _, reqFunc := range r.reqFuncs {
+		reqFunc := reqFunc
+		reqFunc(obj)
+	}
+
+	newGVK := obj.GetObjectKind().GroupVersionKind()
+	if newGVK.Empty() {
+		return nil, fmt.Errorf("invalid empty new GVK for %T: %#v", obj, r)
+	}
+
+	return &mutationResult{
+		origGVK:    origGVK,
+		newGVK:     newGVK,
+		gvkChanged: origGVK != newGVK,
+		mutated:    len(r.respFuncs) != 0 || !apiequality.Semantic.DeepEqual(origObj, obj),
+	}, nil
+}
+
+func (r *request) mutateResponse(obj Object) (bool, error) {
+	origObj, ok := obj.DeepCopyObject().(Object)
+	if !ok {
+		return false, fmt.Errorf("invalid deep copy semantics for %T: %#v", obj, r)
+	}
+
+	for _, respFunc := range r.respFuncs {
+		respFunc := respFunc
+		respFunc(obj)
+	}
+
+	mutated := !apiequality.Semantic.DeepEqual(origObj, obj)
+	return mutated, nil
+}

internal/kubeclient/middleware_test.go

@@ -0,0 +1,74 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func Test_request_mutate(t *testing.T) {
+	tests := []struct {
+		name     string
+		reqFuncs []func(Object)
+		obj      Object
+		want     *mutationResult
+		wantObj  Object
+		wantErr  string
+	}{
+		{
+			name: "mutate config map data",
+			reqFuncs: []func(Object){
+				func(obj Object) {
+					cm := obj.(*corev1.ConfigMap)
+					cm.Data = map[string]string{"new": "stuff"}
+				},
+			},
+			obj: &corev1.ConfigMap{
+				TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"},
+				Data:       map[string]string{"old": "things"},
+				BinaryData: map[string][]byte{"weee": nil},
+			},
+			want: &mutationResult{
+				origGVK:    schema.GroupVersionKind{Group: "", Version: "v1", Kind: "ConfigMap"},
+				newGVK:     schema.GroupVersionKind{Group: "", Version: "v1", Kind: "ConfigMap"},
+				gvkChanged: false,
+				mutated:    true,
+			},
+			wantObj: &corev1.ConfigMap{
+				TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"},
+				Data:       map[string]string{"new": "stuff"},
+				BinaryData: map[string][]byte{"weee": nil},
+			},
+			wantErr: "",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			r := &request{reqFuncs: tt.reqFuncs}
+			orig := tt.obj.DeepCopyObject()
+
+			got, err := r.mutateRequest(tt.obj)
+
+			if len(tt.wantErr) > 0 {
+				require.EqualError(t, err, tt.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+
+			require.Equal(t, tt.want, got)
+
+			if tt.wantObj != nil {
+				require.Equal(t, tt.wantObj, tt.obj)
+			} else {
+				require.Equal(t, orig, tt.obj)
+			}
+		})
+	}
+}

internal/kubeclient/option.go

@@ -20,6 +20,10 @@ func WithConfig(config *restclient.Config) Option {
 
 func WithMiddleware(middleware Middleware) Option {
 	return func(c *clientConfig) {
+		if middleware == nil {
+			return // support passing in a nil middleware as a no-op
+		}
+
 		c.middlewares = append(c.middlewares, middleware)
 	}
 }

internal/kubeclient/path.go

@@ -0,0 +1,75 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"path"
+	"strings"
+
+	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	restclient "k8s.io/client-go/rest"
+)
+
+func updatePathNewGVK(reqURL *url.URL, result *mutationResult, apiPathPrefix string, reqInfo *genericapirequest.RequestInfo) (*url.URL, error) {
+	if !result.gvkChanged {
+		return reqURL, nil
+	}
+
+	if len(result.origGVK.Group) == 0 {
+		return nil, fmt.Errorf("invalid attempt to change core group")
+	}
+
+	newURL := &url.URL{}
+	*newURL = *reqURL
+
+	// replace old GVK with new GVK
+	apiRoot := path.Join(apiPathPrefix, reqInfo.APIPrefix)
+	oldPrefix := restclient.DefaultVersionedAPIPath(apiRoot, result.origGVK.GroupVersion())
+	newPrefix := restclient.DefaultVersionedAPIPath(apiRoot, result.newGVK.GroupVersion())
+
+	newURL.Path = path.Join(newPrefix, strings.TrimPrefix(newURL.Path, oldPrefix))
+
+	return newURL, nil
+}
+
+func getHostAndAPIPathPrefix(config *restclient.Config) (string, string, error) {
+	hostURL, _, err := defaultServerUrlFor(config)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to parse host URL from rest config: %w", err)
+	}
+
+	return hostURL.String(), hostURL.Path, nil
+}
+
+func reqWithoutPrefix(req *http.Request, hostURL, apiPathPrefix string) *http.Request {
+	if len(apiPathPrefix) == 0 {
+		return req
+	}
+
+	if !strings.HasSuffix(hostURL, "/") {
+		hostURL += "/"
+	}
+
+	if !strings.HasPrefix(req.URL.String(), hostURL) {
+		return req
+	}
+
+	if !strings.HasPrefix(apiPathPrefix, "/") {
+		apiPathPrefix = "/" + apiPathPrefix
+	}
+	if !strings.HasSuffix(apiPathPrefix, "/") {
+		apiPathPrefix += "/"
+	}
+
+	reqCopy := req.WithContext(req.Context())
+	urlCopy := &url.URL{}
+	*urlCopy = *reqCopy.URL
+	urlCopy.Path = "/" + strings.TrimPrefix(urlCopy.Path, apiPathPrefix)
+	reqCopy.URL = urlCopy
+
+	return reqCopy
+}

internal/kubeclient/path_test.go

@@ -0,0 +1,231 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+
+	loginv1alpha1 "go.pinniped.dev/generated/1.20/apis/concierge/login/v1alpha1"
+	configv1alpha1 "go.pinniped.dev/generated/1.20/apis/supervisor/config/v1alpha1"
+)
+
+func Test_updatePathNewGVK(t *testing.T) {
+	type args struct {
+		reqURL        *url.URL
+		result        *mutationResult
+		apiPathPrefix string
+		reqInfo       *genericapirequest.RequestInfo
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *url.URL
+		wantErr bool
+	}{
+		{
+			name: "no gvk change",
+			args: args{
+				reqURL: mustParse(t, "https://walrus.tld/api/v1/pods"),
+				result: &mutationResult{},
+			},
+			want: mustParse(t, "https://walrus.tld/api/v1/pods"),
+		},
+		{
+			name: "no original gvk group",
+			args: args{
+				result: &mutationResult{
+					origGVK: schema.GroupVersionKind{
+						Group: "",
+					},
+					gvkChanged: true,
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "cluster-scoped list path",
+			args: args{
+				reqURL: mustParse(t, "https://walrus.tld/apis/"+loginv1alpha1.SchemeGroupVersion.String()+"/tokencredentialrequests"),
+				result: &mutationResult{
+					origGVK: loginv1alpha1.SchemeGroupVersion.WithKind("TokenCredentialRequest"),
+					newGVK: schema.GroupVersionKind{
+						Group:   "login.concierge.tuna.io",
+						Version: loginv1alpha1.SchemeGroupVersion.Version,
+						Kind:    "TokenCredentialRequest",
+					},
+					gvkChanged: true,
+				},
+				apiPathPrefix: "/apis",
+				reqInfo:       &genericapirequest.RequestInfo{},
+			},
+			want: mustParse(t, "https://walrus.tld/apis/login.concierge.tuna.io/v1alpha1/tokencredentialrequests"),
+		},
+		{
+			name: "cluster-scoped get path",
+			args: args{
+				reqURL: mustParse(t, "https://walrus.tld/apis/"+loginv1alpha1.SchemeGroupVersion.String()+"/tokencredentialrequests/some-name"),
+				result: &mutationResult{
+					origGVK: loginv1alpha1.SchemeGroupVersion.WithKind("TokenCredentialRequest"),
+					newGVK: schema.GroupVersionKind{
+						Group:   "login.concierge.tuna.io",
+						Version: loginv1alpha1.SchemeGroupVersion.Version,
+						Kind:    "TokenCredentialRequest",
+					},
+					gvkChanged: true,
+				},
+				apiPathPrefix: "/apis",
+				reqInfo:       &genericapirequest.RequestInfo{},
+			},
+			want: mustParse(t, "https://walrus.tld/apis/login.concierge.tuna.io/v1alpha1/tokencredentialrequests/some-name"),
+		},
+		{
+			name: "namespace-scoped list path",
+			args: args{
+				reqURL: mustParse(t, "https://walrus.tld/apis/"+configv1alpha1.SchemeGroupVersion.String()+"/namespaces/default/federationdomains"),
+				result: &mutationResult{
+					origGVK: configv1alpha1.SchemeGroupVersion.WithKind("FederationDomain"),
+					newGVK: schema.GroupVersionKind{
+						Group:   "config.supervisor.tuna.io",
+						Version: configv1alpha1.SchemeGroupVersion.Version,
+						Kind:    "FederationDomain",
+					},
+					gvkChanged: true,
+				},
+				apiPathPrefix: "/apis",
+				reqInfo:       &genericapirequest.RequestInfo{},
+			},
+			want: mustParse(t, "https://walrus.tld/apis/config.supervisor.tuna.io/v1alpha1/namespaces/default/federationdomains"),
+		},
+		{
+			name: "namespace-scoped get path",
+			args: args{
+				reqURL: mustParse(t, "https://walrus.tld/apis/"+configv1alpha1.SchemeGroupVersion.String()+"/namespaces/default/federationdomains/some-name"),
+				result: &mutationResult{
+					origGVK: configv1alpha1.SchemeGroupVersion.WithKind("FederationDomain"),
+					newGVK: schema.GroupVersionKind{
+						Group:   "config.supervisor.tuna.io",
+						Version: configv1alpha1.SchemeGroupVersion.Version,
+						Kind:    "FederationDomain",
+					},
+					gvkChanged: true,
+				},
+				apiPathPrefix: "/apis",
+				reqInfo:       &genericapirequest.RequestInfo{},
+			},
+			want: mustParse(t, "https://walrus.tld/apis/config.supervisor.tuna.io/v1alpha1/namespaces/default/federationdomains/some-name"),
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := updatePathNewGVK(tt.args.reqURL, tt.args.result, tt.args.apiPathPrefix, tt.args.reqInfo)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("updatePathNewGVK() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("updatePathNewGVK() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_reqWithoutPrefix(t *testing.T) {
+	body := ioutil.NopCloser(bytes.NewBuffer([]byte("some body")))
+	newReq := func(rawurl string) *http.Request {
+		req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, rawurl, body)
+		require.NoError(t, err)
+		return req
+	}
+
+	type args struct {
+		req           *http.Request
+		hostURL       string
+		apiPathPrefix string
+	}
+	tests := []struct {
+		name string
+		args args
+		want *http.Request
+	}{
+		{
+			name: "happy path",
+			args: args{
+				req:           newReq("https://walrus.tld/apis/some/path"),
+				hostURL:       "https://walrus.tld",
+				apiPathPrefix: "/apis",
+			},
+			want: newReq("https://walrus.tld/some/path"),
+		},
+		{
+			name: "host url already has slash suffix",
+			args: args{
+				req:           newReq("https://walrus.tld/apis/some/path"),
+				hostURL:       "https://walrus.tld/",
+				apiPathPrefix: "/apis",
+			},
+			want: newReq("https://walrus.tld/some/path"),
+		},
+		{
+			name: "api prefix already has slash prefix",
+			args: args{
+				req:           newReq("https://walrus.tld/apis/some/path"),
+				hostURL:       "https://walrus.tld",
+				apiPathPrefix: "apis",
+			},
+			want: newReq("https://walrus.tld/some/path"),
+		},
+		{
+			name: "api prefix already has slash suffix",
+			args: args{
+				req:           newReq("https://walrus.tld/apis/some/path"),
+				hostURL:       "https://walrus.tld",
+				apiPathPrefix: "/apis/",
+			},
+			want: newReq("https://walrus.tld/some/path"),
+		},
+		{
+			name: "no api path prefix",
+			args: args{
+				req: newReq("https://walrus.tld"),
+			},
+			want: newReq("https://walrus.tld"),
+		},
+		{
+			name: "hostURL and req URL mismatch",
+			args: args{
+				req:           newReq("https://walrus.tld.some-other-url/some/path"),
+				hostURL:       "https://walrus.tld",
+				apiPathPrefix: "/apis",
+			},
+			want: newReq("https://walrus.tld.some-other-url/some/path"),
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			req := *tt.args.req
+			if got := reqWithoutPrefix(&req, tt.args.hostURL, tt.args.apiPathPrefix); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("reqWithoutPrefix() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func mustParse(t *testing.T, rawurl string) *url.URL {
+	t.Helper()
+	url, err := url.Parse(rawurl)
+	require.NoError(t, err)
+	return url
+}

internal/kubeclient/roundtrip.go

@@ -9,14 +9,27 @@ import (
 	"io/ioutil"
 	"net/http"
 
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/server"
 	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/transport"
+
+	"go.pinniped.dev/internal/plog"
 )
 
-// TODO unit test
+func configWithWrapper(config *restclient.Config, scheme *runtime.Scheme, negotiatedSerializer runtime.NegotiatedSerializer, middlewares []Middleware) *restclient.Config {
+	hostURL, apiPathPrefix, err := getHostAndAPIPathPrefix(config)
+	if err != nil {
+		plog.DebugErr("invalid rest config", err)
+		return config // invalid input config, will fail existing client-go validation
+	}
 
-func configWithWrapper(config *restclient.Config, negotiatedSerializer runtime.NegotiatedSerializer, middlewares []Middleware) *restclient.Config {
 	// no need for any wrapping when we have no middleware to inject
 	if len(middlewares) == 0 {
 		return config
@@ -26,97 +39,346 @@ func configWithWrapper(config *restclient.Config, negotiatedSerializer runtime.N
 	if !ok {
 		panic(fmt.Errorf("unknown content type: %s ", config.ContentType)) // static input, programmer error
 	}
-	serializer := info.Serializer // should perform no conversion
+	regSerializer := info.Serializer // should perform no conversion
 
-	f := func(rt http.RoundTripper) http.RoundTripper {
-		return roundTripperFunc(func(req *http.Request) (*http.Response, error) {
-			// ignore everything that has an unreadable body
-			if req.GetBody == nil {
-				return rt.RoundTrip(req)
-			}
+	resolver := server.NewRequestInfoResolver(server.NewConfig(serializer.CodecFactory{}))
 
-			var reqMiddlewares []Middleware
-			for _, middleware := range middlewares {
-				middleware := middleware
-				if middleware.Handles(req.Method) {
-					reqMiddlewares = append(reqMiddlewares, middleware)
-				}
-			}
+	schemeRestMapperFunc := schemeRestMapper(scheme)
 
-			// no middleware to handle this request
-			if len(reqMiddlewares) == 0 {
-				return rt.RoundTrip(req)
-			}
-
-			body, err := req.GetBody()
-			if err != nil {
-				return nil, fmt.Errorf("get body failed: %w", err)
-			}
-			defer body.Close()
-			data, err := ioutil.ReadAll(body)
-			if err != nil {
-				return nil, fmt.Errorf("read body failed: %w", err)
-			}
-
-			// attempt to decode with no defaults or into specified, i.e. defer to the decoder
-			// this should result in the a straight decode with no conversion
-			obj, _, err := serializer.Decode(data, nil, nil)
-			if err != nil {
-				return nil, fmt.Errorf("body decode failed: %w", err)
-			}
-
-			accessor, err := meta.Accessor(obj)
-			if err != nil {
-				return rt.RoundTrip(req) // ignore everything that has no object meta for now
-			}
-
-			// run all the mutating operations
-			var reqMutated bool
-			for _, reqMiddleware := range reqMiddlewares {
-				mutated := reqMiddleware.Mutate(accessor)
-				reqMutated = mutated || reqMutated
-			}
-
-			// no mutation occurred, keep the original request
-			if !reqMutated {
-				return rt.RoundTrip(req)
-			}
-
-			// we plan on making a new request so make sure to close the original request's body
-			_ = req.Body.Close()
-
-			newData, err := runtime.Encode(serializer, obj)
-			if err != nil {
-				return nil, fmt.Errorf("new body encode failed: %w", err)
-			}
-
-			// TODO log newData at high loglevel similar to REST client
-
-			// simplest way to reuse the body creation logic
-			newReqForBody, err := http.NewRequest(req.Method, req.URL.String(), bytes.NewReader(newData))
-			if err != nil {
-				return nil, fmt.Errorf("failed to create new req for body: %w", err) // this should never happen
-			}
-
-			// shallow copy because we want to preserve all the headers and such but not mutate the original request
-			newReq := req.WithContext(req.Context())
-
-			// replace the body with the new data
-			newReq.ContentLength = newReqForBody.ContentLength
-			newReq.Body = newReqForBody.Body
-			newReq.GetBody = newReqForBody.GetBody
-
-			return rt.RoundTrip(newReq)
-		})
-	}
+	f := newWrapper(hostURL, apiPathPrefix, config, resolver, regSerializer, negotiatedSerializer, schemeRestMapperFunc, middlewares)
 
 	cc := restclient.CopyConfig(config)
 	cc.Wrap(f)
 	return cc
 }
 
-type roundTripperFunc func(req *http.Request) (*http.Response, error)
+type roundTripperFunc func(req *http.Request) (bool, *http.Response, error)
 
 func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
-	return f(req)
+	// always attempt to close the body, as long as we are the ones that handled the request
+	// see http.RoundTripper doc:
+	//   "RoundTrip must always close the body, including on errors, ..."
+	handled, resp, err := f(req)
+	if handled && req.Body != nil {
+		_ = req.Body.Close()
+	}
+	return resp, err
+}
+
+func newWrapper(
+	hostURL, apiPathPrefix string,
+	config *restclient.Config,
+	resolver genericapirequest.RequestInfoResolver,
+	regSerializer runtime.Serializer,
+	negotiatedSerializer runtime.NegotiatedSerializer,
+	schemeRestMapperFunc func(schema.GroupVersionResource, Verb) (schema.GroupVersionKind, bool),
+	middlewares []Middleware,
+) transport.WrapperFunc {
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return roundTripperFunc(func(req *http.Request) (bool, *http.Response, error) {
+			reqInfo, err := resolver.NewRequestInfo(reqWithoutPrefix(req, hostURL, apiPathPrefix))
+			if err != nil || !reqInfo.IsResourceRequest {
+				resp, err := rt.RoundTrip(req) // we only handle kube resource requests
+				return false, resp, err
+			}
+
+			middlewareReq := &request{
+				verb:      verb(reqInfo.Verb),
+				namespace: reqInfo.Namespace,
+				resource: schema.GroupVersionResource{
+					Group:    reqInfo.APIGroup,
+					Version:  reqInfo.APIVersion,
+					Resource: reqInfo.Resource,
+				},
+				subresource: reqInfo.Subresource,
+			}
+
+			for _, middleware := range middlewares {
+				middleware := middleware
+				middleware.Handle(req.Context(), middlewareReq)
+			}
+
+			if len(middlewareReq.reqFuncs) == 0 && len(middlewareReq.respFuncs) == 0 {
+				resp, err := rt.RoundTrip(req) // no middleware wanted to mutate this request
+				return false, resp, err
+			}
+
+			switch v := middlewareReq.Verb(); v {
+			case VerbCreate, VerbUpdate:
+				return handleCreateOrUpdate(req, middlewareReq, regSerializer, rt, apiPathPrefix, reqInfo, config, negotiatedSerializer)
+
+			case VerbGet, VerbList, VerbDelete, VerbDeleteCollection, VerbPatch, VerbWatch:
+				return handleOtherVerbs(v, req, middlewareReq, schemeRestMapperFunc, rt, apiPathPrefix, reqInfo, config, negotiatedSerializer)
+
+			case VerbProxy: // for now we do not support proxy interception
+				fallthrough
+
+			default:
+				resp, err := rt.RoundTrip(req) // we only handle certain verbs
+				return false, resp, err
+			}
+		})
+	}
+}
+
+func handleOtherVerbs(
+	v Verb,
+	req *http.Request,
+	middlewareReq *request,
+	schemeRestMapperFunc func(schema.GroupVersionResource, Verb) (schema.GroupVersionKind, bool),
+	rt http.RoundTripper,
+	apiPathPrefix string,
+	reqInfo *genericapirequest.RequestInfo,
+	config *restclient.Config,
+	negotiatedSerializer runtime.NegotiatedSerializer,
+) (bool, *http.Response, error) {
+	mapperGVK, ok := schemeRestMapperFunc(middlewareReq.Resource(), v)
+	if !ok {
+		return true, nil, fmt.Errorf("unable to determine GVK for middleware request %#v", middlewareReq)
+	}
+
+	// no need to do anything with object meta since we only support GVK changes
+	obj := &metav1.PartialObjectMetadata{}
+	obj.APIVersion, obj.Kind = mapperGVK.ToAPIVersionAndKind()
+
+	result, err := middlewareReq.mutateRequest(obj)
+	if err != nil {
+		return true, nil, err
+	}
+
+	if !result.mutated {
+		resp, err := rt.RoundTrip(req) // no middleware mutated the request
+		return false, resp, err
+	}
+
+	// sanity check to make sure mutation is to type meta and/or the response
+	unexpectedMutation := len(middlewareReq.respFuncs) == 0 && !result.gvkChanged
+	metaIsZero := apiequality.Semantic.DeepEqual(obj.ObjectMeta, metav1.ObjectMeta{})
+	if unexpectedMutation || !metaIsZero {
+		return true, nil, fmt.Errorf("invalid object meta mutation: %#v", middlewareReq)
+	}
+
+	reqURL, err := updatePathNewGVK(req.URL, result, apiPathPrefix, reqInfo)
+	if err != nil {
+		return true, nil, err
+	}
+
+	// shallow copy because we want to preserve all the headers and such but not mutate the original request
+	newReq := req.WithContext(req.Context())
+
+	// replace the body and path with the new data
+	newReq.URL = reqURL
+
+	glogBody("mutated request url", []byte(reqURL.String()))
+
+	resp, err := rt.RoundTrip(newReq)
+	if err != nil {
+		return true, nil, fmt.Errorf("middleware request for %#v failed: %w", middlewareReq, err)
+	}
+
+	switch v {
+	case VerbDelete, VerbDeleteCollection:
+		return true, resp, nil // we do not need to fix the response on delete
+
+	case VerbWatch:
+		resp, err := handleWatchResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
+		return true, resp, err
+
+	default: // VerbGet, VerbList, VerbPatch
+		resp, err := handleResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
+		return true, resp, err
+	}
+}
+
+func handleCreateOrUpdate(
+	req *http.Request,
+	middlewareReq *request,
+	regSerializer runtime.Serializer,
+	rt http.RoundTripper,
+	apiPathPrefix string,
+	reqInfo *genericapirequest.RequestInfo,
+	config *restclient.Config,
+	negotiatedSerializer runtime.NegotiatedSerializer,
+) (bool, *http.Response, error) {
+	if req.GetBody == nil {
+		return true, nil, fmt.Errorf("unreadible body for request: %#v", middlewareReq) // this should never happen
+	}
+
+	body, err := req.GetBody()
+	if err != nil {
+		return true, nil, fmt.Errorf("get body failed: %w", err)
+	}
+	defer body.Close()
+	data, err := ioutil.ReadAll(body)
+	if err != nil {
+		return true, nil, fmt.Errorf("read body failed: %w", err)
+	}
+
+	// attempt to decode with no defaults or into specified, i.e. defer to the decoder
+	// this should result in the a straight decode with no conversion
+	decodedObj, err := runtime.Decode(regSerializer, data)
+	if err != nil {
+		return true, nil, fmt.Errorf("body decode failed: %w", err)
+	}
+
+	obj, ok := decodedObj.(Object)
+	if !ok {
+		return true, nil, fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
+	}
+
+	result, err := middlewareReq.mutateRequest(obj)
+	if err != nil {
+		return true, nil, err
+	}
+
+	if !result.mutated {
+		resp, err := rt.RoundTrip(req) // no middleware mutated the request
+		return false, resp, err
+	}
+
+	reqURL, err := updatePathNewGVK(req.URL, result, apiPathPrefix, reqInfo)
+	if err != nil {
+		return true, nil, err
+	}
+
+	newData, err := runtime.Encode(regSerializer, obj)
+	if err != nil {
+		return true, nil, fmt.Errorf("new body encode failed: %w", err)
+	}
+
+	// simplest way to reuse the body creation logic
+	newReqForBody, err := http.NewRequest(req.Method, reqURL.String(), bytes.NewReader(newData))
+	if err != nil {
+		return true, nil, fmt.Errorf("failed to create new req for body: %w", err) // this should never happen
+	}
+
+	// shallow copy because we want to preserve all the headers and such but not mutate the original request
+	newReq := req.WithContext(req.Context())
+
+	// replace the body and path with the new data
+	newReq.URL = reqURL
+	newReq.ContentLength = newReqForBody.ContentLength
+	newReq.Body = newReqForBody.Body
+	newReq.GetBody = newReqForBody.GetBody
+
+	glogBody("mutated request", newData)
+
+	resp, err := rt.RoundTrip(newReq)
+	if err != nil {
+		return true, nil, fmt.Errorf("middleware request for %#v failed: %w", middlewareReq, err)
+	}
+
+	if !result.gvkChanged && len(middlewareReq.respFuncs) == 0 {
+		return true, resp, nil // we did not change the GVK, so we do not need to mess with the incoming data
+	}
+
+	resp, err = handleResponseNewGVK(config, negotiatedSerializer, resp, middlewareReq, result)
+	return true, resp, err
+}
+
+func handleResponseNewGVK(
+	config *restclient.Config,
+	negotiatedSerializer runtime.NegotiatedSerializer,
+	resp *http.Response,
+	middlewareReq *request,
+	result *mutationResult,
+) (*http.Response, error) {
+	// defer these status codes to client-go
+	switch {
+	case resp.StatusCode == http.StatusSwitchingProtocols,
+		resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
+		return resp, nil
+	}
+
+	// always make sure we close the body, even if reading from it fails
+	defer resp.Body.Close()
+	respData, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	serializerInfo, err := getSerializerInfo(config, negotiatedSerializer, resp, middlewareReq)
+	if err != nil {
+		return nil, err
+	}
+
+	fixedRespData, err := maybeRestoreGVK(serializerInfo.Serializer, respData, result)
+	if err != nil {
+		return nil, fmt.Errorf("unable to restore GVK for %#v: %w", middlewareReq, err)
+	}
+
+	fixedRespData, err = maybeMutateResponse(serializerInfo.Serializer, fixedRespData, middlewareReq, result)
+	if err != nil {
+		return nil, fmt.Errorf("unable to mutate response for %#v: %w", middlewareReq, err)
+	}
+
+	newResp := &http.Response{}
+	*newResp = *resp
+
+	newResp.Body = ioutil.NopCloser(bytes.NewBuffer(fixedRespData))
+	return newResp, nil
+}
+
+func maybeMutateResponse(serializer runtime.Serializer, fixedRespData []byte, middlewareReq *request, result *mutationResult) ([]byte, error) {
+	if len(middlewareReq.respFuncs) == 0 {
+		return fixedRespData, nil
+	}
+
+	decodedObj, err := runtime.Decode(serializer, fixedRespData)
+	if err != nil {
+		return fixedRespData, nil // if we cannot decode it, it is not for us - let client-go figure out what to do
+	}
+
+	if decodedObj.GetObjectKind().GroupVersionKind() != result.origGVK {
+		return fixedRespData, nil
+	}
+
+	var mutated bool
+
+	switch middlewareReq.Verb() {
+	case VerbList:
+		if err := meta.EachListItem(decodedObj, func(listObj runtime.Object) error {
+			obj, ok := listObj.(Object)
+			if !ok {
+				return fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
+			}
+
+			singleMutated, err := middlewareReq.mutateResponse(obj)
+			if err != nil {
+				return fmt.Errorf("response mutation failed for %#v: %w", middlewareReq, err)
+			}
+
+			mutated = mutated || singleMutated
+
+			return nil
+		}); err != nil {
+			return nil, fmt.Errorf("failed to iterate over list for %#v: %T", middlewareReq, decodedObj)
+		}
+
+	default:
+		obj, ok := decodedObj.(Object)
+		if !ok {
+			return nil, fmt.Errorf("middleware request for %#v has invalid object semantics: %T", middlewareReq, decodedObj)
+		}
+
+		mutated, err = middlewareReq.mutateResponse(obj)
+		if err != nil {
+			return nil, fmt.Errorf("response mutation failed for %#v: %w", middlewareReq, err)
+		}
+	}
+
+	if !mutated {
+		return fixedRespData, nil
+	}
+
+	newData, err := runtime.Encode(serializer, decodedObj)
+	if err != nil {
+		return nil, fmt.Errorf("new body encode failed: %w", err)
+	}
+
+	// only log if we mutated the response; we only need to log the unmutated response since client-go
+	// will log the mutated response for us
+	glogBody("unmutated response", fixedRespData)
+
+	return newData, nil
 }

internal/kubeclient/scheme.go

@@ -0,0 +1,84 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"fmt"
+
+	"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/gengo/namer"
+	"k8s.io/gengo/types"
+)
+
+type objectList interface {
+	runtime.Object       // generic access to TypeMeta
+	metav1.ListInterface // generic access to ListMeta
+}
+
+func schemeRestMapper(scheme *runtime.Scheme) func(schema.GroupVersionResource, Verb) (schema.GroupVersionKind, bool) {
+	// we are assuming that no code uses the `// +resourceName=CUSTOM_RESOURCE_NAME` directive
+	// and that no Kube code generator is passed a --plural-exceptions argument
+	pluralExceptions := map[string]string{"Endpoints": "Endpoints"} // default copied from client-gen
+	lowercaseNamer := namer.NewAllLowercasePluralNamer(pluralExceptions)
+
+	listVerbMapping := map[schema.GroupVersionResource]schema.GroupVersionKind{}
+	nonListVerbMapping := map[schema.GroupVersionResource]schema.GroupVersionKind{}
+
+	for gvk := range scheme.AllKnownTypes() {
+		obj, err := scheme.New(gvk)
+		if err != nil {
+			panic(err) // programmer error (internal scheme code is broken)
+		}
+
+		switch t := obj.(type) {
+		case interface {
+			Object
+			objectList
+		}:
+			panic(fmt.Errorf("type is both list and non-list: %T", t))
+
+		case Object:
+			resource := lowercaseNamer.Name(types.Ref("ignored", gvk.Kind))
+			gvr := gvk.GroupVersion().WithResource(resource)
+			nonListVerbMapping[gvr] = gvk
+
+		case objectList:
+			if _, ok := t.(*metav1.Status); ok {
+				continue // ignore status since it does not have an Items field
+			}
+
+			itemsPtr, err := meta.GetItemsPtr(obj)
+			if err != nil {
+				panic(err) // programmer error (internal scheme code is broken)
+			}
+			items, err := conversion.EnforcePtr(itemsPtr)
+			if err != nil {
+				panic(err) // programmer error (internal scheme code is broken)
+			}
+			nonListKind := items.Type().Elem().Name()
+			resource := lowercaseNamer.Name(types.Ref("ignored", nonListKind))
+			gvr := gvk.GroupVersion().WithResource(resource)
+			listVerbMapping[gvr] = gvk
+
+		default:
+			// ignore stuff like ListOptions
+		}
+	}
+
+	return func(resource schema.GroupVersionResource, v Verb) (schema.GroupVersionKind, bool) {
+		switch v {
+		case VerbList:
+			gvk, ok := listVerbMapping[resource]
+			return gvk, ok
+
+		default:
+			gvk, ok := nonListVerbMapping[resource]
+			return gvk, ok
+		}
+	}
+}

internal/kubeclient/scheme_test.go

@@ -0,0 +1,156 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	kubescheme "k8s.io/client-go/kubernetes/scheme"
+	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
+	aggregatorclientscheme "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/scheme"
+
+	loginv1alpha1 "go.pinniped.dev/generated/1.20/apis/concierge/login/v1alpha1"
+	idpv1alpha1 "go.pinniped.dev/generated/1.20/apis/supervisor/idp/v1alpha1"
+	pinnipedconciergeclientsetscheme "go.pinniped.dev/generated/1.20/client/concierge/clientset/versioned/scheme"
+	pinnipedsupervisorclientsetscheme "go.pinniped.dev/generated/1.20/client/supervisor/clientset/versioned/scheme"
+)
+
+func Test_schemeRestMapper(t *testing.T) {
+	type args struct {
+		scheme *runtime.Scheme
+		gvr    schema.GroupVersionResource
+		v      Verb
+	}
+	tests := []struct {
+		name string
+		args args
+		want schema.GroupVersionKind
+	}{
+		{
+			name: "config map get",
+			args: args{
+				scheme: kubescheme.Scheme,
+				gvr:    corev1.SchemeGroupVersion.WithResource("configmaps"),
+				v:      VerbGet,
+			},
+			want: corev1.SchemeGroupVersion.WithKind("ConfigMap"),
+		},
+		{
+			name: "config map list",
+			args: args{
+				scheme: kubescheme.Scheme,
+				gvr:    corev1.SchemeGroupVersion.WithResource("configmaps"),
+				v:      VerbList,
+			},
+			want: corev1.SchemeGroupVersion.WithKind("ConfigMapList"),
+		},
+		{
+			name: "endpoints patch",
+			args: args{
+				scheme: kubescheme.Scheme,
+				gvr:    corev1.SchemeGroupVersion.WithResource("endpoints"),
+				v:      VerbPatch,
+			},
+			want: corev1.SchemeGroupVersion.WithKind("Endpoints"),
+		},
+		{
+			name: "endpoints list",
+			args: args{
+				scheme: kubescheme.Scheme,
+				gvr:    corev1.SchemeGroupVersion.WithResource("endpoints"),
+				v:      VerbList,
+			},
+			want: corev1.SchemeGroupVersion.WithKind("EndpointsList"),
+		},
+		{
+			name: "api service create",
+			args: args{
+				scheme: aggregatorclientscheme.Scheme,
+				gvr:    apiregistrationv1.SchemeGroupVersion.WithResource("apiservices"),
+				v:      VerbCreate,
+			},
+			want: apiregistrationv1.SchemeGroupVersion.WithKind("APIService"),
+		},
+		{
+			name: "api service create - wrong scheme",
+			args: args{
+				scheme: kubescheme.Scheme,
+				gvr:    apiregistrationv1.SchemeGroupVersion.WithResource("apiservices"),
+				v:      VerbCreate,
+			},
+		},
+		{
+			name: "api service list",
+			args: args{
+				scheme: aggregatorclientscheme.Scheme,
+				gvr:    apiregistrationv1.SchemeGroupVersion.WithResource("apiservices"),
+				v:      VerbList,
+			},
+			want: apiregistrationv1.SchemeGroupVersion.WithKind("APIServiceList"),
+		},
+		{
+			name: "token credential delete",
+			args: args{
+				scheme: pinnipedconciergeclientsetscheme.Scheme,
+				gvr:    loginv1alpha1.SchemeGroupVersion.WithResource("tokencredentialrequests"),
+				v:      VerbDelete,
+			},
+			want: loginv1alpha1.SchemeGroupVersion.WithKind("TokenCredentialRequest"),
+		},
+		{
+			name: "token credential list",
+			args: args{
+				scheme: pinnipedconciergeclientsetscheme.Scheme,
+				gvr:    loginv1alpha1.SchemeGroupVersion.WithResource("tokencredentialrequests"),
+				v:      VerbList,
+			},
+			want: loginv1alpha1.SchemeGroupVersion.WithKind("TokenCredentialRequestList"),
+		},
+		{
+			name: "oidc idp update",
+			args: args{
+				scheme: pinnipedsupervisorclientsetscheme.Scheme,
+				gvr:    idpv1alpha1.SchemeGroupVersion.WithResource("oidcidentityproviders"),
+				v:      VerbUpdate,
+			},
+			want: idpv1alpha1.SchemeGroupVersion.WithKind("OIDCIdentityProvider"),
+		},
+		{
+			name: "oidc idp list",
+			args: args{
+				scheme: pinnipedsupervisorclientsetscheme.Scheme,
+				gvr:    idpv1alpha1.SchemeGroupVersion.WithResource("oidcidentityproviders"),
+				v:      VerbList,
+			},
+			want: idpv1alpha1.SchemeGroupVersion.WithKind("OIDCIdentityProviderList"),
+		},
+		{
+			name: "oidc idp list - wrong scheme",
+			args: args{
+				scheme: pinnipedconciergeclientsetscheme.Scheme,
+				gvr:    idpv1alpha1.SchemeGroupVersion.WithResource("oidcidentityproviders"),
+				v:      VerbList,
+			},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			schemeRestMapperFunc := schemeRestMapper(tt.args.scheme)
+			gvk, ok := schemeRestMapperFunc(tt.args.gvr, tt.args.v)
+
+			if tt.want.Empty() {
+				require.True(t, gvk.Empty())
+				require.False(t, ok)
+			} else {
+				require.Equal(t, tt.want, gvk)
+				require.True(t, ok)
+			}
+		})
+	}
+}

internal/kubeclient/serializer.go

@@ -0,0 +1,39 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"fmt"
+	"mime"
+	"net/http"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	restclient "k8s.io/client-go/rest"
+)
+
+type passthroughDecoder struct{}
+
+func (d passthroughDecoder) Decode(data []byte, _ *schema.GroupVersionKind, _ runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
+	return &runtime.Unknown{Raw: data}, &schema.GroupVersionKind{}, nil
+}
+
+func getSerializerInfo(config *restclient.Config, negotiatedSerializer runtime.NegotiatedSerializer, resp *http.Response, middlewareReq *request) (runtime.SerializerInfo, error) {
+	contentType := resp.Header.Get("Content-Type")
+	if len(contentType) == 0 {
+		contentType = config.ContentType
+	}
+
+	mediaType, _, err := mime.ParseMediaType(contentType)
+	if err != nil {
+		return runtime.SerializerInfo{}, fmt.Errorf("failed to parse content type for %#v: %w", middlewareReq, err)
+	}
+
+	respInfo, ok := runtime.SerializerInfoForMediaType(negotiatedSerializer.SupportedMediaTypes(), mediaType)
+	if !ok || respInfo.Serializer == nil || respInfo.StreamSerializer == nil || respInfo.StreamSerializer.Serializer == nil || respInfo.StreamSerializer.Framer == nil {
+		return runtime.SerializerInfo{}, fmt.Errorf("unable to find resp serialier for %#v with content-type %s", middlewareReq, mediaType)
+	}
+
+	return respInfo, nil
+}

internal/kubeclient/verb.go

@@ -0,0 +1,27 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+type Verb interface {
+	verb() // private method to prevent creation of verbs outside this package
+}
+
+const (
+	VerbCreate           verb = "create"
+	VerbUpdate           verb = "update"
+	VerbDelete           verb = "delete"
+	VerbDeleteCollection verb = "deletecollection"
+	VerbGet              verb = "get"
+	VerbList             verb = "list"
+	VerbWatch            verb = "watch"
+	VerbPatch            verb = "patch"
+
+	VerbProxy verb = "proxy" // proxy unsupported for now
+)
+
+var _, _ Verb = VerbGet, verb("")
+
+type verb string
+
+func (verb) verb() {}

internal/kubeclient/verb_test.go

@@ -0,0 +1,54 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_verb(t *testing.T) {
+	tests := []struct {
+		name string
+		f    func() string
+		want string
+	}{
+		{
+			name: "error: string format",
+			f: func() string {
+				return fmt.Errorf("%s", VerbGet).Error()
+			},
+			want: "get",
+		},
+		{
+			name: "error: value format",
+			f: func() string {
+				return fmt.Errorf("%v", VerbUpdate).Error()
+			},
+			want: "update",
+		},
+		{
+			name: "error: go value format",
+			f: func() string {
+				return fmt.Errorf("%#v", VerbDelete).Error()
+			},
+			want: `"delete"`,
+		},
+		{
+			name: "error: go value format in middelware request",
+			f: func() string {
+				return fmt.Errorf("%#v", request{verb: VerbPatch}).Error()
+			},
+			want: `kubeclient.request{verb:"patch", namespace:"", resource:schema.GroupVersionResource{Group:"", Version:"", Resource:""}, reqFuncs:[]func(kubeclient.Object)(nil), respFuncs:[]func(kubeclient.Object)(nil), subresource:""}`,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, tt.f())
+		})
+	}
+}

internal/kubeclient/watch.go

@@ -0,0 +1,163 @@
+// Copyright 2021 the Pinniped contributors. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeclient
+
+import (
+	stderrors "errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer/streaming"
+	"k8s.io/apimachinery/pkg/util/net"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	restclient "k8s.io/client-go/rest"
+	restclientwatch "k8s.io/client-go/rest/watch"
+
+	"go.pinniped.dev/internal/plog"
+)
+
+func handleWatchResponseNewGVK(
+	config *restclient.Config,
+	negotiatedSerializer runtime.NegotiatedSerializer,
+	resp *http.Response,
+	middlewareReq *request,
+	result *mutationResult,
+) (*http.Response, error) {
+	// defer non-success cases to client-go
+	if resp.StatusCode != http.StatusOK {
+		return resp, nil
+	}
+
+	var goRoutineStarted bool
+	defer func() {
+		if goRoutineStarted {
+			return
+		}
+		// always drain and close the body if we do not get to the point of starting our go routine
+		drainAndMaybeCloseBody(resp, true)
+	}()
+
+	serializerInfo, err := getSerializerInfo(config, negotiatedSerializer, resp, middlewareReq)
+	if err != nil {
+		return nil, err
+	}
+
+	newResp := &http.Response{}
+	*newResp = *resp
+
+	newBodyReader, newBodyWriter := io.Pipe()
+
+	newResp.Body = newBodyReader // client-go is responsible for closing this reader
+
+	goRoutineStarted = true
+	go func() {
+		var sourceDecoder watch.Decoder
+		defer utilruntime.HandleCrash()
+		defer func() {
+			// the sourceDecoder will close the resp body. we want to make sure the drain the body before
+			// we do that
+			drainAndMaybeCloseBody(resp, false)
+			if sourceDecoder != nil {
+				sourceDecoder.Close()
+			}
+		}()
+		defer newBodyWriter.Close()
+
+		frameReader := serializerInfo.StreamSerializer.Framer.NewFrameReader(resp.Body)
+		watchEventDecoder := streaming.NewDecoder(frameReader, serializerInfo.StreamSerializer.Serializer)
+		sourceDecoder = restclientwatch.NewDecoder(watchEventDecoder, &passthroughDecoder{})
+		defer sourceDecoder.Close()
+
+		frameWriter := serializerInfo.StreamSerializer.Framer.NewFrameWriter(newBodyWriter)
+		watchEventEncoder := streaming.NewEncoder(frameWriter, serializerInfo.StreamSerializer.Serializer)
+
+		for {
+			ok, err := sendWatchEvent(sourceDecoder, serializerInfo.Serializer, middlewareReq, result, watchEventEncoder)
+			if err != nil {
+				if stderrors.Is(err, io.ErrClosedPipe) {
+					return // calling newBodyReader.Close() will send this to all newBodyWriter.Write()
+				}
+
+				// CloseWithError always returns nil
+				// all newBodyReader.Read() will get this error
+				_ = newBodyWriter.CloseWithError(err)
+
+				return
+			}
+
+			if !ok {
+				return
+			}
+		}
+	}()
+
+	return newResp, nil
+}
+
+func sendWatchEvent(sourceDecoder watch.Decoder, s runtime.Serializer, middlewareReq *request, result *mutationResult, watchEventEncoder streaming.Encoder) (bool, error) {
+	// partially copied from watch.NewStreamWatcher.receive
+	eventType, obj, err := sourceDecoder.Decode()
+	if err != nil {
+		switch {
+		case stderrors.Is(err, io.EOF):
+			// watch closed normally
+		case stderrors.Is(err, io.ErrUnexpectedEOF):
+			plog.InfoErr("Unexpected EOF during watch stream event decoding", err)
+		case net.IsProbableEOF(err), net.IsTimeout(err):
+			plog.TraceErr("Unable to decode an event from the watch stream", err)
+		default:
+			return false, fmt.Errorf("unexpected watch decode error for %#v: %w", middlewareReq, err)
+		}
+		return false, nil // all errors end watch
+	}
+
+	unknown, ok := obj.(*runtime.Unknown)
+	if !ok || len(unknown.Raw) == 0 {
+		return false, fmt.Errorf("unexpected decode type: %T", obj)
+	}
+
+	respData := unknown.Raw
+	fixedRespData, err := maybeRestoreGVK(s, respData, result)
+	if err != nil {
+		return false, fmt.Errorf("unable to restore GVK for %#v: %w", middlewareReq, err)
+	}
+
+	fixedRespData, err = maybeMutateResponse(s, fixedRespData, middlewareReq, result)
+	if err != nil {
+		return false, fmt.Errorf("unable to mutate response for %#v: %w", middlewareReq, err)
+	}
+
+	event := &metav1.WatchEvent{
+		Type:   string(eventType),
+		Object: runtime.RawExtension{Raw: fixedRespData},
+	}
+
+	if err := watchEventEncoder.Encode(event); err != nil {
+		return false, fmt.Errorf("failed to encode watch event for %#v: %w", middlewareReq, err)
+	}
+
+	return true, nil
+}
+
+// drainAndMaybeCloseBody attempts to drain and optionallt close the provided body.
+//
+// We want to drain used HTTP response bodies so that the underlying TCP connection can be
+// reused. However, if the underlying response body is extremely large or a never-ending stream,
+// then we don't want to wait for the read to finish. In these cases, we give up on the TCP
+// connection and just close the body.
+func drainAndMaybeCloseBody(resp *http.Response, close bool) {
+	// from k8s.io/client-go/rest/request.go...
+	const maxBodySlurpSize = 2 << 10
+	if resp.ContentLength <= maxBodySlurpSize {
+		_, _ = io.Copy(ioutil.Discard, &io.LimitedReader{R: resp.Body, N: maxBodySlurpSize})
+	}
+	if close {
+		resp.Body.Close()
+	}
+}