Commit 4d235063 authored by Chao Xu's avatar Chao Xu

only store typeMeta and objectMeta in the gc store

parent d52204a1
...@@ -51,6 +51,7 @@ import ( ...@@ -51,6 +51,7 @@ import (
endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint" endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint"
"k8s.io/kubernetes/pkg/controller/framework/informers" "k8s.io/kubernetes/pkg/controller/framework/informers"
"k8s.io/kubernetes/pkg/controller/garbagecollector" "k8s.io/kubernetes/pkg/controller/garbagecollector"
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
"k8s.io/kubernetes/pkg/controller/job" "k8s.io/kubernetes/pkg/controller/job"
namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace" namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
nodecontroller "k8s.io/kubernetes/pkg/controller/node" nodecontroller "k8s.io/kubernetes/pkg/controller/node"
...@@ -69,6 +70,7 @@ import ( ...@@ -69,6 +70,7 @@ import (
persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume" persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
"k8s.io/kubernetes/pkg/healthz" "k8s.io/kubernetes/pkg/healthz"
quotainstall "k8s.io/kubernetes/pkg/quota/install" quotainstall "k8s.io/kubernetes/pkg/quota/install"
"k8s.io/kubernetes/pkg/runtime/serializer"
"k8s.io/kubernetes/pkg/serviceaccount" "k8s.io/kubernetes/pkg/serviceaccount"
"k8s.io/kubernetes/pkg/util/configz" "k8s.io/kubernetes/pkg/util/configz"
"k8s.io/kubernetes/pkg/util/crypto" "k8s.io/kubernetes/pkg/util/crypto"
...@@ -495,8 +497,12 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig ...@@ -495,8 +497,12 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
if err != nil { if err != nil {
glog.Fatalf("Failed to get supported resources from server: %v", err) glog.Fatalf("Failed to get supported resources from server: %v", err)
} }
clientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "generic-garbage-collector"), dynamic.LegacyAPIPathResolverFunc) config := restclient.AddUserAgent(kubeconfig, "generic-garbage-collector")
garbageCollector, err := garbagecollector.NewGarbageCollector(clientPool, groupVersionResources) config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
config.ContentConfig.NegotiatedSerializer = nil
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
garbageCollector, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, groupVersionResources)
if err != nil { if err != nil {
glog.Errorf("Failed to start the generic garbage collector: %v", err) glog.Errorf("Failed to start the generic garbage collector: %v", err)
} else { } else {
......
...@@ -1911,6 +1911,7 @@ func TestDeleteWithOptions(t *testing.T) { ...@@ -1911,6 +1911,7 @@ func TestDeleteWithOptions(t *testing.T) {
if simpleStorage.deleted != ID { if simpleStorage.deleted != ID {
t.Errorf("Unexpected delete: %s, expected %s", simpleStorage.deleted, ID) t.Errorf("Unexpected delete: %s, expected %s", simpleStorage.deleted, ID)
} }
simpleStorage.deleteOptions.GetObjectKind().SetGroupVersionKind(unversioned.GroupVersionKind{})
if !api.Semantic.DeepEqual(simpleStorage.deleteOptions, item) { if !api.Semantic.DeepEqual(simpleStorage.deleteOptions, item) {
t.Errorf("unexpected delete options: %s", diff.ObjectDiff(simpleStorage.deleteOptions, item)) t.Errorf("unexpected delete options: %s", diff.ObjectDiff(simpleStorage.deleteOptions, item))
} }
...@@ -2700,6 +2701,7 @@ func TestCreate(t *testing.T) { ...@@ -2700,6 +2701,7 @@ func TestCreate(t *testing.T) {
t.Errorf("unexpected error: %v %#v", err, response) t.Errorf("unexpected error: %v %#v", err, response)
} }
itemOut.GetObjectKind().SetGroupVersionKind(unversioned.GroupVersionKind{})
if !reflect.DeepEqual(&itemOut, simple) { if !reflect.DeepEqual(&itemOut, simple) {
t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body)) t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body))
} }
...@@ -2769,6 +2771,7 @@ func TestCreateYAML(t *testing.T) { ...@@ -2769,6 +2771,7 @@ func TestCreateYAML(t *testing.T) {
t.Fatalf("unexpected error: %v %#v", err, response) t.Fatalf("unexpected error: %v %#v", err, response)
} }
itemOut.GetObjectKind().SetGroupVersionKind(unversioned.GroupVersionKind{})
if !reflect.DeepEqual(&itemOut, simple) { if !reflect.DeepEqual(&itemOut, simple) {
t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body)) t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body))
} }
...@@ -2827,6 +2830,7 @@ func TestCreateInNamespace(t *testing.T) { ...@@ -2827,6 +2830,7 @@ func TestCreateInNamespace(t *testing.T) {
t.Fatalf("unexpected error: %v\n%s", err, data) t.Fatalf("unexpected error: %v\n%s", err, data)
} }
itemOut.GetObjectKind().SetGroupVersionKind(unversioned.GroupVersionKind{})
if !reflect.DeepEqual(&itemOut, simple) { if !reflect.DeepEqual(&itemOut, simple) {
t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body)) t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simple, string(body))
} }
......
...@@ -54,13 +54,9 @@ func NewClient(conf *restclient.Config) (*Client, error) { ...@@ -54,13 +54,9 @@ func NewClient(conf *restclient.Config) (*Client, error) {
confCopy := *conf confCopy := *conf
conf = &confCopy conf = &confCopy
codec := dynamicCodec{}
// TODO: it's questionable that this should be using anything other than unstructured schema and JSON // TODO: it's questionable that this should be using anything other than unstructured schema and JSON
conf.ContentType = runtime.ContentTypeJSON conf.ContentType = runtime.ContentTypeJSON
conf.AcceptContentTypes = runtime.ContentTypeJSON conf.AcceptContentTypes = runtime.ContentTypeJSON
streamingInfo, _ := api.Codecs.StreamingSerializerForMediaType("application/json;stream=watch", nil)
conf.NegotiatedSerializer = serializer.NegotiatedSerializerWrapper(runtime.SerializerInfo{Serializer: codec}, streamingInfo)
if conf.APIPath == "" { if conf.APIPath == "" {
conf.APIPath = "/api" conf.APIPath = "/api"
...@@ -69,6 +65,10 @@ func NewClient(conf *restclient.Config) (*Client, error) { ...@@ -69,6 +65,10 @@ func NewClient(conf *restclient.Config) (*Client, error) {
if len(conf.UserAgent) == 0 { if len(conf.UserAgent) == 0 {
conf.UserAgent = restclient.DefaultKubernetesUserAgent() conf.UserAgent = restclient.DefaultKubernetesUserAgent()
} }
if conf.NegotiatedSerializer == nil {
streamingInfo, _ := api.Codecs.StreamingSerializerForMediaType("application/json;stream=watch", nil)
conf.NegotiatedSerializer = serializer.NegotiatedSerializerWrapper(runtime.SerializerInfo{Serializer: dynamicCodec{}}, streamingInfo)
}
cl, err := restclient.RESTClientFor(conf) cl, err := restclient.RESTClientFor(conf)
if err != nil { if err != nil {
...@@ -119,19 +119,17 @@ type ResourceClient struct { ...@@ -119,19 +119,17 @@ type ResourceClient struct {
} }
// List returns a list of objects for this resource. // List returns a list of objects for this resource.
func (rc *ResourceClient) List(opts runtime.Object) (*runtime.UnstructuredList, error) { func (rc *ResourceClient) List(opts runtime.Object) (runtime.Object, error) {
result := new(runtime.UnstructuredList)
parameterEncoder := rc.parameterCodec parameterEncoder := rc.parameterCodec
if parameterEncoder == nil { if parameterEncoder == nil {
parameterEncoder = defaultParameterEncoder parameterEncoder = defaultParameterEncoder
} }
err := rc.cl.Get(). return rc.cl.Get().
NamespaceIfScoped(rc.ns, rc.resource.Namespaced). NamespaceIfScoped(rc.ns, rc.resource.Namespaced).
Resource(rc.resource.Name). Resource(rc.resource.Name).
VersionedParams(opts, parameterEncoder). VersionedParams(opts, parameterEncoder).
Do(). Do().
Into(result) Get()
return result, err
} }
// Get gets the resource with the specified name. // Get gets the resource with the specified name.
......
...@@ -19,8 +19,11 @@ package dynamic ...@@ -19,8 +19,11 @@ package dynamic
import ( import (
"sync" "sync"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/client/restclient" "k8s.io/kubernetes/pkg/client/restclient"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/runtime/serializer"
) )
// ClientPool manages a pool of dynamic clients. // ClientPool manages a pool of dynamic clients.
...@@ -77,6 +80,12 @@ func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVer ...@@ -77,6 +80,12 @@ func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVer
// we need to make a client // we need to make a client
conf.GroupVersion = &groupVersion conf.GroupVersion = &groupVersion
if conf.NegotiatedSerializer == nil {
streamingInfo, _ := api.Codecs.StreamingSerializerForMediaType("application/json;stream=watch", nil)
conf.NegotiatedSerializer = serializer.NegotiatedSerializerWrapper(runtime.SerializerInfo{Serializer: dynamicCodec{}}, streamingInfo)
}
dynamicClient, err := NewClient(conf) dynamicClient, err := NewClient(conf)
if err != nil { if err != nil {
return nil, err return nil, err
......
...@@ -33,6 +33,7 @@ import ( ...@@ -33,6 +33,7 @@ import (
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
"k8s.io/kubernetes/pkg/client/typed/dynamic" "k8s.io/kubernetes/pkg/client/typed/dynamic"
"k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/controller/framework"
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
utilerrors "k8s.io/kubernetes/pkg/util/errors" utilerrors "k8s.io/kubernetes/pkg/util/errors"
...@@ -154,6 +155,7 @@ func (p *Propagator) addDependentToOwners(n *node, owners []metatypes.OwnerRefer ...@@ -154,6 +155,7 @@ func (p *Propagator) addDependentToOwners(n *node, owners []metatypes.OwnerRefer
dependentsLock: &sync.RWMutex{}, dependentsLock: &sync.RWMutex{},
dependents: make(map[*node]struct{}), dependents: make(map[*node]struct{}),
} }
glog.V(6).Infof("add virtual node.identity: %s\n\n", ownerNode.identity)
p.uidToNode.Write(ownerNode) p.uidToNode.Write(ownerNode)
p.gc.dirtyQueue.Add(ownerNode) p.gc.dirtyQueue.Add(ownerNode)
} }
...@@ -426,7 +428,12 @@ func (p *Propagator) processEvent() { ...@@ -426,7 +428,12 @@ func (p *Propagator) processEvent() {
// removing ownerReferences from the dependents if the owner is deleted with // removing ownerReferences from the dependents if the owner is deleted with
// DeleteOptions.OrphanDependents=true. // DeleteOptions.OrphanDependents=true.
type GarbageCollector struct { type GarbageCollector struct {
restMapper meta.RESTMapper restMapper meta.RESTMapper
// metaOnlyClientPool uses a special codec, which removes fields except for
// apiVersion, kind, and metadata during decoding.
metaOnlyClientPool dynamic.ClientPool
// clientPool uses the regular dynamicCodec. We need it to update
// finalizers. It can be removed if we support patching finalizers.
clientPool dynamic.ClientPool clientPool dynamic.ClientPool
dirtyQueue *workqueue.Type dirtyQueue *workqueue.Type
orphanQueue *workqueue.Type orphanQueue *workqueue.Type
...@@ -434,8 +441,6 @@ type GarbageCollector struct { ...@@ -434,8 +441,6 @@ type GarbageCollector struct {
propagator *Propagator propagator *Propagator
} }
// TODO: make special List and Watch function that removes fields other than
// ObjectMeta.
func gcListWatcher(client *dynamic.Client, resource unversioned.GroupVersionResource) *cache.ListWatch { func gcListWatcher(client *dynamic.Client, resource unversioned.GroupVersionResource) *cache.ListWatch {
return &cache.ListWatch{ return &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) { ListFunc: func(options api.ListOptions) (runtime.Object, error) {
...@@ -461,14 +466,21 @@ func gcListWatcher(client *dynamic.Client, resource unversioned.GroupVersionReso ...@@ -461,14 +466,21 @@ func gcListWatcher(client *dynamic.Client, resource unversioned.GroupVersionReso
} }
} }
func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversioned.GroupVersionResource) (monitor, error) { func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversioned.GroupVersionResource, kind unversioned.GroupVersionKind) (monitor, error) {
// TODO: consider store in one storage. // TODO: consider store in one storage.
glog.V(6).Infof("create storage for resource %s", resource) glog.V(6).Infof("create storage for resource %s", resource)
var monitor monitor var monitor monitor
client, err := clientPool.ClientForGroupVersion(resource.GroupVersion()) client, err := p.gc.metaOnlyClientPool.ClientForGroupVersion(resource.GroupVersion())
if err != nil { if err != nil {
return monitor, err return monitor, err
} }
setObjectTypeMeta := func(obj interface{}) {
runtimeObject, ok := obj.(runtime.Object)
if !ok {
utilruntime.HandleError(fmt.Errorf("expected runtime.Object, got %#v", obj))
}
runtimeObject.GetObjectKind().SetGroupVersionKind(kind)
}
monitor.store, monitor.controller = framework.NewInformer( monitor.store, monitor.controller = framework.NewInformer(
gcListWatcher(client, resource), gcListWatcher(client, resource),
nil, nil,
...@@ -476,6 +488,7 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion ...@@ -476,6 +488,7 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion
framework.ResourceEventHandlerFuncs{ framework.ResourceEventHandlerFuncs{
// add the event to the propagator's eventQueue. // add the event to the propagator's eventQueue.
AddFunc: func(obj interface{}) { AddFunc: func(obj interface{}) {
setObjectTypeMeta(obj)
event := event{ event := event{
eventType: addEvent, eventType: addEvent,
obj: obj, obj: obj,
...@@ -483,6 +496,8 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion ...@@ -483,6 +496,8 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion
p.eventQueue.Add(event) p.eventQueue.Add(event)
}, },
UpdateFunc: func(oldObj, newObj interface{}) { UpdateFunc: func(oldObj, newObj interface{}) {
setObjectTypeMeta(newObj)
setObjectTypeMeta(oldObj)
event := event{updateEvent, newObj, oldObj} event := event{updateEvent, newObj, oldObj}
p.eventQueue.Add(event) p.eventQueue.Add(event)
}, },
...@@ -491,6 +506,7 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion ...@@ -491,6 +506,7 @@ func monitorFor(p *Propagator, clientPool dynamic.ClientPool, resource unversion
if deletedFinalStateUnknown, ok := obj.(cache.DeletedFinalStateUnknown); ok { if deletedFinalStateUnknown, ok := obj.(cache.DeletedFinalStateUnknown); ok {
obj = deletedFinalStateUnknown.Obj obj = deletedFinalStateUnknown.Obj
} }
setObjectTypeMeta(obj)
event := event{ event := event{
eventType: deleteEvent, eventType: deleteEvent,
obj: obj, obj: obj,
...@@ -511,11 +527,12 @@ var ignoredResources = map[unversioned.GroupVersionResource]struct{}{ ...@@ -511,11 +527,12 @@ var ignoredResources = map[unversioned.GroupVersionResource]struct{}{
unversioned.GroupVersionResource{Group: "authorization.k8s.io", Version: "v1beta1", Resource: "subjectaccessreviews"}: {}, unversioned.GroupVersionResource{Group: "authorization.k8s.io", Version: "v1beta1", Resource: "subjectaccessreviews"}: {},
} }
func NewGarbageCollector(clientPool dynamic.ClientPool, resources []unversioned.GroupVersionResource) (*GarbageCollector, error) { func NewGarbageCollector(metaOnlyClientPool dynamic.ClientPool, clientPool dynamic.ClientPool, resources []unversioned.GroupVersionResource) (*GarbageCollector, error) {
gc := &GarbageCollector{ gc := &GarbageCollector{
clientPool: clientPool, metaOnlyClientPool: metaOnlyClientPool,
dirtyQueue: workqueue.New(), clientPool: clientPool,
orphanQueue: workqueue.New(), dirtyQueue: workqueue.New(),
orphanQueue: workqueue.New(),
// TODO: should use a dynamic RESTMapper built from the discovery results. // TODO: should use a dynamic RESTMapper built from the discovery results.
restMapper: registered.RESTMapper(), restMapper: registered.RESTMapper(),
} }
...@@ -532,7 +549,11 @@ func NewGarbageCollector(clientPool dynamic.ClientPool, resources []unversioned. ...@@ -532,7 +549,11 @@ func NewGarbageCollector(clientPool dynamic.ClientPool, resources []unversioned.
glog.V(6).Infof("ignore resource %#v", resource) glog.V(6).Infof("ignore resource %#v", resource)
continue continue
} }
monitor, err := monitorFor(gc.propagator, gc.clientPool, resource) kind, err := gc.restMapper.KindFor(resource)
if err != nil {
return nil, err
}
monitor, err := monitorFor(gc.propagator, gc.clientPool, resource, kind)
if err != nil { if err != nil {
return nil, err return nil, err
} }
...@@ -549,7 +570,7 @@ func (gc *GarbageCollector) worker() { ...@@ -549,7 +570,7 @@ func (gc *GarbageCollector) worker() {
defer gc.dirtyQueue.Done(key) defer gc.dirtyQueue.Done(key)
err := gc.processItem(key.(*node)) err := gc.processItem(key.(*node))
if err != nil { if err != nil {
utilruntime.HandleError(fmt.Errorf("Error syncing item %v: %v", key, err)) utilruntime.HandleError(fmt.Errorf("Error syncing item %#v: %v", key, err))
} }
} }
...@@ -623,6 +644,20 @@ func objectReferenceToUnstructured(ref objectReference) *runtime.Unstructured { ...@@ -623,6 +644,20 @@ func objectReferenceToUnstructured(ref objectReference) *runtime.Unstructured {
return ret return ret
} }
func objectReferenceToMetadataOnlyObject(ref objectReference) *metaonly.MetadataOnlyObject {
return &metaonly.MetadataOnlyObject{
TypeMeta: unversioned.TypeMeta{
APIVersion: ref.APIVersion,
Kind: ref.Kind,
},
ObjectMeta: v1.ObjectMeta{
Namespace: ref.Namespace,
UID: ref.UID,
Name: ref.Name,
},
}
}
func (gc *GarbageCollector) processItem(item *node) error { func (gc *GarbageCollector) processItem(item *node) error {
// Get the latest item from the API server // Get the latest item from the API server
latest, err := gc.getObject(item.identity) latest, err := gc.getObject(item.identity)
...@@ -634,15 +669,22 @@ func (gc *GarbageCollector) processItem(item *node) error { ...@@ -634,15 +669,22 @@ func (gc *GarbageCollector) processItem(item *node) error {
glog.V(6).Infof("item %v not found, generating a virtual delete event", item.identity) glog.V(6).Infof("item %v not found, generating a virtual delete event", item.identity)
event := event{ event := event{
eventType: deleteEvent, eventType: deleteEvent,
obj: objectReferenceToUnstructured(item.identity), obj: objectReferenceToMetadataOnlyObject(item.identity),
} }
glog.V(6).Infof("generating virtual delete event for %s\n\n", event.obj)
gc.propagator.eventQueue.Add(event) gc.propagator.eventQueue.Add(event)
return nil return nil
} }
return err return err
} }
if latest.GetUID() != item.identity.UID { if latest.GetUID() != item.identity.UID {
glog.V(6).Infof("UID doesn't match, item %v not found, ignore it", item.identity) glog.V(6).Infof("UID doesn't match, item %v not found, generating a virtual delete event", item.identity)
event := event{
eventType: deleteEvent,
obj: objectReferenceToMetadataOnlyObject(item.identity),
}
glog.V(6).Infof("generating virtual delete event for %s\n\n", event.obj)
gc.propagator.eventQueue.Add(event)
return nil return nil
} }
ownerReferences := latest.GetOwnerReferences() ownerReferences := latest.GetOwnerReferences()
......
...@@ -24,6 +24,8 @@ import ( ...@@ -24,6 +24,8 @@ import (
"testing" "testing"
_ "k8s.io/kubernetes/pkg/api/install" _ "k8s.io/kubernetes/pkg/api/install"
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
"k8s.io/kubernetes/pkg/runtime/serializer"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
...@@ -39,9 +41,13 @@ import ( ...@@ -39,9 +41,13 @@ import (
) )
func TestNewGarbageCollector(t *testing.T) { func TestNewGarbageCollector(t *testing.T) {
clientPool := dynamic.NewClientPool(&restclient.Config{}, dynamic.LegacyAPIPathResolverFunc) config := &restclient.Config{}
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
config.ContentConfig.NegotiatedSerializer = nil
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}} podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
gc, err := NewGarbageCollector(clientPool, podResource) gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, podResource)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
...@@ -141,8 +147,11 @@ func TestProcessItem(t *testing.T) { ...@@ -141,8 +147,11 @@ func TestProcessItem(t *testing.T) {
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}} podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
srv, clientConfig := testServerAndClientConfig(testHandler.ServeHTTP) srv, clientConfig := testServerAndClientConfig(testHandler.ServeHTTP)
defer srv.Close() defer srv.Close()
clientConfig.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
metaOnlyClientPool := dynamic.NewClientPool(clientConfig, dynamic.LegacyAPIPathResolverFunc)
clientConfig.ContentConfig.NegotiatedSerializer = nil
clientPool := dynamic.NewClientPool(clientConfig, dynamic.LegacyAPIPathResolverFunc) clientPool := dynamic.NewClientPool(clientConfig, dynamic.LegacyAPIPathResolverFunc)
gc, err := NewGarbageCollector(clientPool, podResource) gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, podResource)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
...@@ -293,9 +302,13 @@ func TestProcessEvent(t *testing.T) { ...@@ -293,9 +302,13 @@ func TestProcessEvent(t *testing.T) {
// TestDependentsRace relies on golang's data race detector to check if there is // TestDependentsRace relies on golang's data race detector to check if there is
// data race among in the dependents field. // data race among in the dependents field.
func TestDependentsRace(t *testing.T) { func TestDependentsRace(t *testing.T) {
clientPool := dynamic.NewClientPool(&restclient.Config{}, dynamic.LegacyAPIPathResolverFunc) config := &restclient.Config{}
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
config.ContentConfig.NegotiatedSerializer = nil
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}} podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
gc, err := NewGarbageCollector(clientPool, podResource) gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, podResource)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
......
/*
Copyright 2016 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package metaonly
import (
"fmt"
"strings"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/runtime/serializer"
)
func (obj *MetadataOnlyObject) GetObjectKind() unversioned.ObjectKind { return obj }
func (obj *MetadataOnlyObjectList) GetObjectKind() unversioned.ObjectKind { return obj }
type metaOnlyJSONScheme struct{}
// This function can be extended to mapping different gvk to different MetadataOnlyObject,
// which embedded with different version of ObjectMeta. Currently the system
// only supports v1.ObjectMeta.
func gvkToMetadataOnlyObject(gvk unversioned.GroupVersionKind) runtime.Object {
if strings.HasSuffix(gvk.Kind, "List") {
return &MetadataOnlyObjectList{}
} else {
return &MetadataOnlyObject{}
}
}
func NewMetadataCodecFactory() serializer.CodecFactory {
// populating another scheme from api.Scheme, registering every kind with
// MetadataOnlyObject (or MetadataOnlyObjectList).
scheme := runtime.NewScheme()
allTypes := api.Scheme.AllKnownTypes()
for kind := range allTypes {
if kind.Version == runtime.APIVersionInternal {
continue
}
metaOnlyObject := gvkToMetadataOnlyObject(kind)
scheme.AddKnownTypeWithName(kind, metaOnlyObject)
}
scheme.AddUnversionedTypes(api.Unversioned, &unversioned.Status{})
return serializer.NewCodecFactory(scheme)
}
// String converts a MetadataOnlyObject to a human-readable string.
func (metaOnly MetadataOnlyObject) String() string {
return fmt.Sprintf("%s/%s, name: %s, DeletionTimestamp:%v", metaOnly.TypeMeta.APIVersion, metaOnly.TypeMeta.Kind, metaOnly.ObjectMeta.Name, metaOnly.ObjectMeta.DeletionTimestamp)
}
/*
Copyright 2016 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package metaonly
import (
"encoding/json"
"reflect"
"testing"
_ "k8s.io/kubernetes/pkg/api/install"
"k8s.io/kubernetes/pkg/api/meta"
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/runtime/serializer"
)
func getPod() *v1.Pod {
return &v1.Pod{
TypeMeta: unversioned.TypeMeta{
Kind: "Pod",
APIVersion: "v1",
},
ObjectMeta: v1.ObjectMeta{
Name: "pod",
OwnerReferences: []v1.OwnerReference{
{UID: "1234"},
},
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{
Name: "fake-name",
Image: "fakeimage",
},
},
},
}
}
func getPodJson(t *testing.T) []byte {
data, err := json.Marshal(getPod())
if err != nil {
t.Fatal(err)
}
return data
}
func getPodListJson(t *testing.T) []byte {
data, err := json.Marshal(&v1.PodList{
TypeMeta: unversioned.TypeMeta{
Kind: "PodList",
APIVersion: "v1",
},
Items: []v1.Pod{
*getPod(),
*getPod(),
},
})
if err != nil {
t.Fatal(err)
}
return data
}
func verfiyMetadata(description string, t *testing.T, in *MetadataOnlyObject) {
pod := getPod()
if e, a := pod.ObjectMeta, in.ObjectMeta; !reflect.DeepEqual(e, a) {
t.Errorf("%s: expected %#v, got %#v", description, e, a)
}
}
func TestDecodeToMetadataOnlyObject(t *testing.T) {
data := getPodJson(t)
cf := serializer.DirectCodecFactory{CodecFactory: NewMetadataCodecFactory()}
serializer, ok := cf.SerializerForMediaType(runtime.ContentTypeJSON, nil)
if !ok {
t.Fatalf("expected to get a JSON serializer")
}
codec := cf.DecoderToVersion(serializer, unversioned.GroupVersion{Group: "SOMEGROUP", Version: "SOMEVERSION"})
// decode with into
into := &MetadataOnlyObject{}
ret, _, err := codec.Decode(data, nil, into)
if err != nil {
t.Fatal(err)
}
metaOnly, ok := ret.(*MetadataOnlyObject)
if !ok {
t.Fatalf("expected ret to be *runtime.MetadataOnlyObject")
}
verfiyMetadata("check returned metaonly with into", t, metaOnly)
verfiyMetadata("check into", t, into)
// decode without into
ret, _, err = codec.Decode(data, nil, nil)
if err != nil {
t.Fatal(err)
}
metaOnly, ok = ret.(*MetadataOnlyObject)
if !ok {
t.Fatalf("expected ret to be *runtime.MetadataOnlyObject")
}
verfiyMetadata("check returned metaonly without into", t, metaOnly)
}
func verifyListMetadata(t *testing.T, metaOnlyList *MetadataOnlyObjectList) {
items, err := meta.ExtractList(metaOnlyList)
if err != nil {
t.Fatal(err)
}
for _, item := range items {
metaOnly, ok := item.(*MetadataOnlyObject)
if !ok {
t.Fatalf("expected item to be *MetadataOnlyObject")
}
verfiyMetadata("check list", t, metaOnly)
}
}
func TestDecodeToMetadataOnlyObjectList(t *testing.T) {
data := getPodListJson(t)
cf := serializer.DirectCodecFactory{CodecFactory: NewMetadataCodecFactory()}
serializer, ok := cf.SerializerForMediaType(runtime.ContentTypeJSON, nil)
if !ok {
t.Fatalf("expected to get a JSON serializer")
}
codec := cf.DecoderToVersion(serializer, unversioned.GroupVersion{Group: "SOMEGROUP", Version: "SOMEVERSION"})
// decode with into
into := &MetadataOnlyObjectList{}
ret, _, err := codec.Decode(data, nil, into)
if err != nil {
t.Fatal(err)
}
metaOnlyList, ok := ret.(*MetadataOnlyObjectList)
if !ok {
t.Fatalf("expected ret to be *runtime.UnstructuredList")
}
verifyListMetadata(t, metaOnlyList)
verifyListMetadata(t, into)
// decode without into
ret, _, err = codec.Decode(data, nil, nil)
if err != nil {
t.Fatal(err)
}
metaOnlyList, ok = ret.(*MetadataOnlyObjectList)
if !ok {
t.Fatalf("expected ret to be *runtime.UnstructuredList")
}
verifyListMetadata(t, metaOnlyList)
}
/*
Copyright 2016 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package metaonly
import (
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/api/v1"
)
// MetadataOnlyObject allows decoding only the apiVersion, kind, and metadata fields of
// JSON data.
// TODO: enable meta-only decoding for protobuf.
type MetadataOnlyObject struct {
unversioned.TypeMeta `json:",inline"`
v1.ObjectMeta `json:"metadata,omitempty"`
}
// MetadataOnlyObjectList allows decoding from JSON data only the typemeta and metadata of
// a list, and those of the enclosing objects.
// TODO: enable meta-only decoding for protobuf.
type MetadataOnlyObjectList struct {
unversioned.TypeMeta `json:",inline"`
unversioned.ListMeta `json:"metadata,omitempty"`
Items []MetadataOnlyObject `json:"items"`
}
...@@ -272,7 +272,7 @@ func (f *fakeActionHandler) ServeHTTP(response http.ResponseWriter, request *htt ...@@ -272,7 +272,7 @@ func (f *fakeActionHandler) ServeHTTP(response http.ResponseWriter, request *htt
f.actions = append(f.actions, fakeAction{method: request.Method, path: request.URL.Path}) f.actions = append(f.actions, fakeAction{method: request.Method, path: request.URL.Path})
response.Header().Set("Content-Type", runtime.ContentTypeJSON) response.Header().Set("Content-Type", runtime.ContentTypeJSON)
response.WriteHeader(f.statusCode) response.WriteHeader(f.statusCode)
response.Write([]byte("{\"kind\": \"List\"}")) response.Write([]byte("{\"kind\": \"List\",\"items\":null}"))
} }
// testGroupVersionResources returns a mocked up set of resources across different api groups for testing namespace controller. // testGroupVersionResources returns a mocked up set of resources across different api groups for testing namespace controller.
......
...@@ -195,8 +195,12 @@ func listCollection( ...@@ -195,8 +195,12 @@ func listCollection(
} }
apiResource := unversioned.APIResource{Name: gvr.Resource, Namespaced: true} apiResource := unversioned.APIResource{Name: gvr.Resource, Namespaced: true}
unstructuredList, err := dynamicClient.Resource(&apiResource, namespace).List(&v1.ListOptions{}) obj, err := dynamicClient.Resource(&apiResource, namespace).List(&v1.ListOptions{})
if err == nil { if err == nil {
unstructuredList, ok := obj.(*runtime.UnstructuredList)
if !ok {
return nil, false, fmt.Errorf("resource: %s, expected *runtime.UnstructuredList, got %#v", apiResource.Name, obj)
}
return unstructuredList, true, nil return unstructuredList, true, nil
} }
......
...@@ -80,12 +80,14 @@ func EncodeOrDie(e Encoder, obj Object) string { ...@@ -80,12 +80,14 @@ func EncodeOrDie(e Encoder, obj Object) string {
// invokes the ObjectCreator to instantiate a new gvk. Returns an error if the typer cannot find the object. // invokes the ObjectCreator to instantiate a new gvk. Returns an error if the typer cannot find the object.
func UseOrCreateObject(t ObjectTyper, c ObjectCreater, gvk unversioned.GroupVersionKind, obj Object) (Object, error) { func UseOrCreateObject(t ObjectTyper, c ObjectCreater, gvk unversioned.GroupVersionKind, obj Object) (Object, error) {
if obj != nil { if obj != nil {
into, _, err := t.ObjectKinds(obj) kinds, _, err := t.ObjectKinds(obj)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if gvk == into[0] { for _, kind := range kinds {
return obj, nil if gvk == kind {
return obj, nil
}
} }
} }
return c.New(gvk) return c.New(gvk)
......
...@@ -211,6 +211,11 @@ func (s *Scheme) KnownTypes(gv unversioned.GroupVersion) map[string]reflect.Type ...@@ -211,6 +211,11 @@ func (s *Scheme) KnownTypes(gv unversioned.GroupVersion) map[string]reflect.Type
return types return types
} }
// AllKnownTypes returns the all known types.
func (s *Scheme) AllKnownTypes() map[unversioned.GroupVersionKind]reflect.Type {
return s.gvkToType
}
// ObjectKind returns the group,version,kind of the go object and true if this object // ObjectKind returns the group,version,kind of the go object and true if this object
// is considered unversioned, or an error if it's not a pointer or is unregistered. // is considered unversioned, or an error if it's not a pointer or is unregistered.
func (s *Scheme) ObjectKind(obj Object) (unversioned.GroupVersionKind, bool, error) { func (s *Scheme) ObjectKind(obj Object) (unversioned.GroupVersionKind, bool, error) {
......
...@@ -95,6 +95,7 @@ func (s unstructuredJSONScheme) decode(data []byte) (Object, error) { ...@@ -95,6 +95,7 @@ func (s unstructuredJSONScheme) decode(data []byte) (Object, error) {
err := s.decodeToUnstructured(data, unstruct) err := s.decodeToUnstructured(data, unstruct)
return unstruct, err return unstruct, err
} }
func (s unstructuredJSONScheme) decodeInto(data []byte, obj Object) error { func (s unstructuredJSONScheme) decodeInto(data []byte, obj Object) error {
switch x := obj.(type) { switch x := obj.(type) {
case *Unstructured: case *Unstructured:
......
...@@ -92,7 +92,11 @@ func TestDynamicClient(t *testing.T) { ...@@ -92,7 +92,11 @@ func TestDynamicClient(t *testing.T) {
} }
// check dynamic list // check dynamic list
unstructuredList, err := dynamicClient.Resource(&resource, ns.Name).List(&v1.ListOptions{}) obj, err := dynamicClient.Resource(&resource, ns.Name).List(&v1.ListOptions{})
unstructuredList, ok := obj.(*runtime.UnstructuredList)
if !ok {
t.Fatalf("expected *runtime.UnstructuredList, got %#v", obj)
}
if err != nil { if err != nil {
t.Fatalf("unexpected error when listing pods: %v", err) t.Fatalf("unexpected error when listing pods: %v", err)
} }
......
...@@ -35,7 +35,9 @@ import ( ...@@ -35,7 +35,9 @@ import (
"k8s.io/kubernetes/pkg/client/restclient" "k8s.io/kubernetes/pkg/client/restclient"
"k8s.io/kubernetes/pkg/client/typed/dynamic" "k8s.io/kubernetes/pkg/client/typed/dynamic"
"k8s.io/kubernetes/pkg/controller/garbagecollector" "k8s.io/kubernetes/pkg/controller/garbagecollector"
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
"k8s.io/kubernetes/pkg/registry/generic/registry" "k8s.io/kubernetes/pkg/registry/generic/registry"
"k8s.io/kubernetes/pkg/runtime/serializer"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util/wait" "k8s.io/kubernetes/pkg/util/wait"
"k8s.io/kubernetes/test/integration/framework" "k8s.io/kubernetes/test/integration/framework"
...@@ -127,8 +129,12 @@ func setup(t *testing.T) (*httptest.Server, *garbagecollector.GarbageCollector, ...@@ -127,8 +129,12 @@ func setup(t *testing.T) (*httptest.Server, *garbagecollector.GarbageCollector,
if err != nil { if err != nil {
t.Fatalf("Failed to get supported resources from server: %v", err) t.Fatalf("Failed to get supported resources from server: %v", err)
} }
clientPool := dynamic.NewClientPool(&restclient.Config{Host: s.URL}, dynamic.LegacyAPIPathResolverFunc) config := &restclient.Config{Host: s.URL}
gc, err := garbagecollector.NewGarbageCollector(clientPool, groupVersionResources) config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
config.ContentConfig.NegotiatedSerializer = nil
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
gc, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, groupVersionResources)
if err != nil { if err != nil {
t.Fatalf("Failed to create garbage collector") t.Fatalf("Failed to create garbage collector")
} }
...@@ -215,7 +221,7 @@ func TestCascadingDeletion(t *testing.T) { ...@@ -215,7 +221,7 @@ func TestCascadingDeletion(t *testing.T) {
// sometimes the deletion of the RC takes long time to be observed by // sometimes the deletion of the RC takes long time to be observed by
// the gc, so wait for the garbage collector to observe the deletion of // the gc, so wait for the garbage collector to observe the deletion of
// the toBeDeletedRC // the toBeDeletedRC
if err := wait.Poll(10*time.Second, 120*time.Second, func() (bool, error) { if err := wait.Poll(10*time.Second, 60*time.Second, func() (bool, error) {
return !gc.GraphHasUID([]types.UID{toBeDeletedRC.ObjectMeta.UID}), nil return !gc.GraphHasUID([]types.UID{toBeDeletedRC.ObjectMeta.UID}), nil
}); err != nil { }); err != nil {
t.Fatal(err) t.Fatal(err)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment