Additional optimizations to the encode/decode paths

pkg/api/install/install.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/conversion"
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/util/sets"
+	"k8s.io/kubernetes/pkg/watch/versioned"
 )
 
 const importPrefix = "k8s.io/kubernetes/pkg/api"
@@ -233,6 +234,17 @@ func addVersionsToScheme(externalVersions ...unversioned.GroupVersion) {
 			case *v1.Endpoints:
 				return true, v1.Convert_api_Endpoints_To_v1_Endpoints(a, b, s)
 			}
+
+		case *versioned.Event:
+			switch b := objB.(type) {
+			case *versioned.InternalEvent:
+				return true, versioned.Convert_versioned_Event_to_versioned_InternalEvent(a, b, s)
+			}
+		case *versioned.InternalEvent:
+			switch b := objB.(type) {
+			case *versioned.Event:
+				return true, versioned.Convert_versioned_InternalEvent_to_versioned_Event(a, b, s)
+			}
 		}
 		return false, nil
 	})

pkg/api/ref.go

@@ -57,11 +57,11 @@ func GetReference(obj runtime.Object) (*ObjectReference, error) {
 	kind := gvk.Kind
 	if len(kind) == 0 {
 		// TODO: this is wrong
-		gvk, err := Scheme.ObjectKind(obj)
+		gvks, _, err := Scheme.ObjectKinds(obj)
 		if err != nil {
 			return nil, err
 		}
-		kind = gvk.Kind
+		kind = gvks[0].Kind
 	}
 
 	// if the object referenced is actually persisted, we can also get version from meta

pkg/api/ref_test.go

@@ -40,7 +40,7 @@ func TestGetReference(t *testing.T) {
 	// when vendoring kube, if you don't force the set of registered versions (like this hack/test-go.sh does)
 	// then you run into trouble because the types aren't registered in the scheme by anything.  This does the
 	// register manually to allow unit test execution
-	if _, err := Scheme.ObjectKind(&Pod{}); err != nil {
+	if _, _, err := Scheme.ObjectKinds(&Pod{}); err != nil {
 		AddToScheme(Scheme)
 	}
 

pkg/api/rest/create.go

@@ -112,11 +112,11 @@ func objectMetaAndKind(typer runtime.ObjectTyper, obj runtime.Object) (*api.Obje
 	if err != nil {
 		return nil, unversioned.GroupVersionKind{}, errors.NewInternalError(err)
 	}
-	kind, err := typer.ObjectKind(obj)
+	kinds, _, err := typer.ObjectKinds(obj)
 	if err != nil {
 		return nil, unversioned.GroupVersionKind{}, errors.NewInternalError(err)
 	}
-	return objectMeta, kind, nil
+	return objectMeta, kinds[0], nil
 }
 
 // NamespaceScopedStrategy has a method to tell if the object must be in a namespace.

pkg/api/serialization_proto_test.go

@@ -37,7 +37,7 @@ import (
 
 func init() {
 	codecsToTest = append(codecsToTest, func(version unversioned.GroupVersion, item runtime.Object) (runtime.Codec, error) {
-		s := protobuf.NewSerializer(api.Scheme, runtime.ObjectTyperToTyper(api.Scheme), "application/arbitrary.content.type")
+		s := protobuf.NewSerializer(api.Scheme, api.Scheme, "application/arbitrary.content.type")
 		return api.Codecs.CodecForVersions(s, s, testapi.ExternalGroupVersions(), nil), nil
 	})
 }
@@ -138,7 +138,7 @@ func BenchmarkEncodeProtobufGeneratedMarshal(b *testing.B) {
 func BenchmarkDecodeCodecToInternalProtobuf(b *testing.B) {
 	items := benchmarkItems()
 	width := len(items)
-	s := protobuf.NewSerializer(api.Scheme, runtime.ObjectTyperToTyper(api.Scheme), "application/arbitrary.content.type")
+	s := protobuf.NewSerializer(api.Scheme, api.Scheme, "application/arbitrary.content.type")
 	encoder := api.Codecs.EncoderForVersion(s, v1.SchemeGroupVersion)
 	var encoded [][]byte
 	for i := range items {

pkg/api/testapi/testapi.go

@@ -354,10 +354,11 @@ func ExternalGroupVersions() []unversioned.GroupVersion {
 
 // Get codec based on runtime.Object
 func GetCodecForObject(obj runtime.Object) (runtime.Codec, error) {
-	kind, err := api.Scheme.ObjectKind(obj)
+	kinds, _, err := api.Scheme.ObjectKinds(obj)
 	if err != nil {
 		return nil, fmt.Errorf("unexpected encoding error: %v", err)
 	}
+	kind := kinds[0]
 
 	for _, group := range Groups {
 		if group.GroupVersion().Group != kind.Group {

pkg/api/testing/fuzzer.go

@@ -195,6 +195,8 @@ func FuzzerFor(t *testing.T, version unversioned.GroupVersion, src rand.Source)
 			randomQuantity := func() resource.Quantity {
 				var q resource.Quantity
 				c.Fuzz(&q)
+				// precalc the string for benchmarking purposes
+				_ = q.String()
 				return q
 			}
 			q.Limits = make(api.ResourceList)

pkg/apiserver/api_installer.go

@@ -118,7 +118,7 @@ func (a *APIInstaller) getResourceKind(path string, storage rest.Storage) (unver
 	}
 
 	object := storage.New()
-	fqKinds, err := a.group.Typer.ObjectKinds(object)
+	fqKinds, _, err := a.group.Typer.ObjectKinds(object)
 	if err != nil {
 		return unversioned.GroupVersionKind{}, err
 	}
@@ -233,8 +233,11 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 	var versionedList interface{}
 	if isLister {
 		list := lister.NewList()
-		listGVK, err := a.group.Typer.ObjectKind(list)
-		versionedListPtr, err := a.group.Creater.New(a.group.GroupVersion.WithKind(listGVK.Kind))
+		listGVKs, _, err := a.group.Typer.ObjectKinds(list)
+		if err != nil {
+			return nil, err
+		}
+		versionedListPtr, err := a.group.Creater.New(a.group.GroupVersion.WithKind(listGVKs[0].Kind))
 		if err != nil {
 			return nil, err
 		}
@@ -272,10 +275,11 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 	)
 	if isGetterWithOptions {
 		getOptions, getSubpath, _ = getterWithOptions.NewGetOptions()
-		getOptionsInternalKind, err = a.group.Typer.ObjectKind(getOptions)
+		getOptionsInternalKinds, _, err := a.group.Typer.ObjectKinds(getOptions)
 		if err != nil {
 			return nil, err
 		}
+		getOptionsInternalKind = getOptionsInternalKinds[0]
 		versionedGetOptions, err = a.group.Creater.New(optionsExternalVersion.WithKind(getOptionsInternalKind.Kind))
 		if err != nil {
 			return nil, err
@@ -300,12 +304,16 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 	if isConnecter {
 		connectOptions, connectSubpath, _ = connecter.NewConnectOptions()
 		if connectOptions != nil {
-			connectOptionsInternalKind, err = a.group.Typer.ObjectKind(connectOptions)
+			connectOptionsInternalKinds, _, err := a.group.Typer.ObjectKinds(connectOptions)
 			if err != nil {
 				return nil, err
 			}
 
+			connectOptionsInternalKind = connectOptionsInternalKinds[0]
 			versionedConnectOptions, err = a.group.Creater.New(optionsExternalVersion.WithKind(connectOptionsInternalKind.Kind))
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 
@@ -390,18 +398,26 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 
 		resourcePath := namespacedPath
 		resourceParams := namespaceParams
+		itemPathPrefix := gpath.Join(a.prefix, scope.ParamName()) + "/"
 		itemPath := namespacedPath + "/{name}"
+		itemPathMiddle := "/" + resource + "/"
 		nameParams := append(namespaceParams, nameParam)
 		proxyParams := append(nameParams, pathParam)
+		itemPathSuffix := ""
 		if hasSubresource {
-			itemPath = itemPath + "/" + subresource
+			itemPathSuffix = "/" + subresource
+			itemPath = itemPath + itemPathSuffix
 			resourcePath = itemPath
 			resourceParams = nameParams
 		}
 		apiResource.Name = path
 		apiResource.Namespaced = true
 		apiResource.Kind = resourceKind
-		namer := scopeNaming{scope, a.group.Linker, gpath.Join(a.prefix, itemPath), false}
+
+		itemPathFn := func(name, namespace string) string {
+			return itemPathPrefix + namespace + itemPathMiddle + name + itemPathSuffix
+		}
+		namer := scopeNaming{scope, a.group.Linker, itemPathFn, false}
 
 		actions = appendIf(actions, action{"LIST", resourcePath, resourceParams, namer}, isLister)
 		actions = appendIf(actions, action{"POST", resourcePath, resourceParams, namer}, isCreater)
@@ -430,7 +446,7 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 		// For ex: LIST all pods in all namespaces by sending a LIST request at /api/apiVersion/pods.
 		// TODO: more strongly type whether a resource allows these actions on "all namespaces" (bulk delete)
 		if !hasSubresource {
-			namer = scopeNaming{scope, a.group.Linker, gpath.Join(a.prefix, itemPath), true}
+			namer = scopeNaming{scope, a.group.Linker, itemPathFn, true}
 			actions = appendIf(actions, action{"LIST", resource, params, namer}, isLister)
 			actions = appendIf(actions, action{"WATCHLIST", "watch/" + resource, params, namer}, allowWatchList)
 		}
@@ -775,7 +791,7 @@ func (n rootScopeNaming) ObjectName(obj runtime.Object) (namespace, name string,
 type scopeNaming struct {
 	scope meta.RESTScope
 	runtime.SelfLinker
-	itemPath      string
+	itemPathFn    func(name, namespace string) string
 	allNamespaces bool
 }
 
@@ -822,9 +838,8 @@ func (n scopeNaming) GenerateLink(req *restful.Request, obj runtime.Object) (pat
 	if len(name) == 0 {
 		return "", "", errEmptyName
 	}
-	path = strings.Replace(n.itemPath, "{name}", name, 1)
-	path = strings.Replace(path, "{"+n.scope.ArgumentName()+"}", namespace, 1)
-	return path, "", nil
+
+	return n.itemPathFn(name, namespace), "", nil
 }
 
 // GenerateListLink returns the appropriate path and query to locate a list by its canonical path.

pkg/apiserver/api_installer_test.go

@@ -36,7 +36,9 @@ func TestScopeNamingGenerateLink(t *testing.T) {
 	s := scopeNaming{
 		meta.RESTScopeNamespace,
 		selfLinker,
-		"/api/v1/namespaces/{namespace}/services/{name}",
+		func(name, namespace string) string {
+			return "/api/v1/namespaces/" + namespace + "/services/" + name
+		},
 		true,
 	}
 	service := &api.Service{

pkg/apiserver/resthandler.go

@@ -23,7 +23,6 @@ import (
 	"math/rand"
 	"net/http"
 	"net/url"
-	gpath "path"
 	"strings"
 	"time"
 
@@ -940,10 +939,11 @@ func finishRequest(timeout time.Duration, fn resultFunc) (result runtime.Object,
 
 // transformDecodeError adds additional information when a decode fails.
 func transformDecodeError(typer runtime.ObjectTyper, baseErr error, into runtime.Object, gvk *unversioned.GroupVersionKind, body []byte) error {
-	objGVK, err := typer.ObjectKind(into)
+	objGVKs, _, err := typer.ObjectKinds(into)
 	if err != nil {
 		return err
 	}
+	objGVK := objGVKs[0]
 	if gvk != nil && len(gvk.Kind) > 0 {
 		return errors.NewBadRequest(fmt.Sprintf("%s in version %q cannot be handled as a %s: %v", gvk.Kind, gvk.Version, objGVK.Kind, baseErr))
 	}
@@ -962,7 +962,7 @@ func setSelfLink(obj runtime.Object, req *restful.Request, namer ScopeNamer) err
 
 	newURL := *req.Request.URL
 	// use only canonical paths
-	newURL.Path = gpath.Clean(path)
+	newURL.Path = path
 	newURL.RawQuery = query
 	newURL.Fragment = ""
 

pkg/apiserver/watch.go

@@ -167,31 +167,38 @@ func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	w.WriteHeader(http.StatusOK)
 	flusher.Flush()
 
+	var unknown runtime.Unknown
+	internalEvent := &versioned.InternalEvent{}
 	buf := &bytes.Buffer{}
+	ch := s.watching.ResultChan()
 	for {
 		select {
 		case <-cn.CloseNotify():
 			return
 		case <-timeoutCh:
 			return
-		case event, ok := <-s.watching.ResultChan():
+		case event, ok := <-ch:
 			if !ok {
 				// End of results.
 				return
 			}
+
 			obj := event.Object
 			s.fixup(obj)
 			if err := s.embeddedEncoder.EncodeToStream(obj, buf); err != nil {
 				// unexpected error
 				utilruntime.HandleError(fmt.Errorf("unable to encode watch object: %v", err))
 				return
 			}
-			event.Object = &runtime.Unknown{
-				Raw: buf.Bytes(),
-				// ContentType is not required here because we are defaulting to the serializer
-				// type
-			}
-			if err := e.Encode((*versioned.InternalEvent)(&event)); err != nil {
+
+			// ContentType is not required here because we are defaulting to the serializer
+			// type
+			unknown.Raw = buf.Bytes()
+			event.Object = &unknown
+
+			// the internal event will be versioned by the encoder
+			*internalEvent = versioned.InternalEvent(event)
+			if err := e.Encode(internalEvent); err != nil {
 				utilruntime.HandleError(fmt.Errorf("unable to encode watch object: %v (%#v)", err, e))
 				// client disconnect.
 				return
@@ -208,14 +215,18 @@ func (s *WatchServer) HandleWS(ws *websocket.Conn) {
 	defer ws.Close()
 	done := make(chan struct{})
 	go wsstream.IgnoreReceives(ws, 0)
+
+	var unknown runtime.Unknown
+	internalEvent := &versioned.InternalEvent{}
 	buf := &bytes.Buffer{}
 	streamBuf := &bytes.Buffer{}
+	ch := s.watching.ResultChan()
 	for {
 		select {
 		case <-done:
 			s.watching.Stop()
 			return
-		case event, ok := <-s.watching.ResultChan():
+		case event, ok := <-ch:
 			if !ok {
 				// End of results.
 				return
@@ -227,14 +238,15 @@ func (s *WatchServer) HandleWS(ws *websocket.Conn) {
 				utilruntime.HandleError(fmt.Errorf("unable to encode watch object: %v", err))
 				return
 			}
-			event.Object = &runtime.Unknown{
-				Raw: buf.Bytes(),
-				// ContentType is not required here because we are defaulting to the serializer
-				// type
-			}
+
+			// ContentType is not required here because we are defaulting to the serializer
+			// type
+			unknown.Raw = buf.Bytes()
+			event.Object = &unknown
+
 			// the internal event will be versioned by the encoder
-			internalEvent := versioned.InternalEvent(event)
-			if err := s.encoder.EncodeToStream(&internalEvent, streamBuf); err != nil {
+			*internalEvent = versioned.InternalEvent(event)
+			if err := s.encoder.EncodeToStream(internalEvent, streamBuf); err != nil {
 				// encoding error
 				utilruntime.HandleError(fmt.Errorf("unable to encode event: %v", err))
 				s.watching.Stop()

pkg/apiserver/watch_test.go

@@ -561,6 +561,7 @@ func benchmarkItems() []api.Pod {
 	items := make([]api.Pod, 3)
 	for i := range items {
 		apiObjectFuzzer.Fuzz(&items[i])
+		items[i].Spec.InitContainers, items[i].Status.InitContainerStatuses = nil, nil
 	}
 	return items
 }

pkg/client/testing/core/fixture.go

@@ -210,11 +210,11 @@ func (o objects) Kind(kind unversioned.GroupVersionKind, name string) (runtime.O
 }
 
 func (o objects) Add(obj runtime.Object) error {
-	gvk, err := o.scheme.ObjectKind(obj)
+	gvks, _, err := o.scheme.ObjectKinds(obj)
 	if err != nil {
 		return err
 	}
-	kind := gvk.Kind
+	kind := gvks[0].Kind
 
 	switch {
 	case meta.IsListType(obj):