mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-24 04:06:03 +00:00
Allow garbage collection to work against different API prefixes
The GC needs to build clients based only on Resource or Kind. Hoist the restmapper out of the controller and the clientpool, support a new ClientForGroupVersionKind and ClientForGroupVersionResource, and use the appropriate one in both places.
This commit is contained in:
parent
4ab5a76338
commit
97c35fcc67
@ -36,6 +36,7 @@ import (
|
|||||||
"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
|
"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
"k8s.io/kubernetes/pkg/apis/batch"
|
"k8s.io/kubernetes/pkg/apis/batch"
|
||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||||
unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
|
unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
|
||||||
@ -311,9 +312,12 @@ func StartControllers(s *options.CMServer, kubeconfig *restclient.Config, stop <
|
|||||||
glog.Fatalf("Failed to get supported resources from server: %v", err)
|
glog.Fatalf("Failed to get supported resources from server: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TODO: should use a dynamic RESTMapper built from the discovery results.
|
||||||
|
restMapper := registered.RESTMapper()
|
||||||
|
|
||||||
// Find the list of namespaced resources via discovery that the namespace controller must manage
|
// Find the list of namespaced resources via discovery that the namespace controller must manage
|
||||||
namespaceKubeClient := client("namespace-controller")
|
namespaceKubeClient := client("namespace-controller")
|
||||||
namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), dynamic.LegacyAPIPathResolverFunc)
|
namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), restMapper, dynamic.LegacyAPIPathResolverFunc)
|
||||||
groupVersionResources, err := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources()
|
groupVersionResources, err := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources()
|
||||||
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)
|
||||||
@ -519,12 +523,13 @@ func StartControllers(s *options.CMServer, kubeconfig *restclient.Config, stop <
|
|||||||
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)
|
||||||
}
|
}
|
||||||
|
|
||||||
config := restclient.AddUserAgent(kubeconfig, "generic-garbage-collector")
|
config := restclient.AddUserAgent(kubeconfig, "generic-garbage-collector")
|
||||||
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
||||||
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
metaOnlyClientPool := dynamic.NewClientPool(config, restMapper, dynamic.LegacyAPIPathResolverFunc)
|
||||||
config.ContentConfig.NegotiatedSerializer = nil
|
config.ContentConfig.NegotiatedSerializer = nil
|
||||||
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(config, restMapper, dynamic.LegacyAPIPathResolverFunc)
|
||||||
garbageCollector, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, groupVersionResources)
|
garbageCollector, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, restMapper, 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 {
|
||||||
|
@ -30,6 +30,7 @@ import (
|
|||||||
"k8s.io/kubernetes/contrib/mesos/pkg/node"
|
"k8s.io/kubernetes/contrib/mesos/pkg/node"
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||||
"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"
|
||||||
@ -234,7 +235,7 @@ func (s *CMServer) Run(_ []string) error {
|
|||||||
|
|
||||||
// Find the list of namespaced resources via discovery that the namespace controller must manage
|
// Find the list of namespaced resources via discovery that the namespace controller must manage
|
||||||
namespaceKubeClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller"))
|
namespaceKubeClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller"))
|
||||||
namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), dynamic.LegacyAPIPathResolverFunc)
|
namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
groupVersionResources, err := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources()
|
groupVersionResources, err := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources()
|
||||||
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)
|
||||||
|
@ -160,8 +160,7 @@ type RESTMapping struct {
|
|||||||
// to API groups. In other words, kinds and resources should not be assumed to be
|
// to API groups. In other words, kinds and resources should not be assumed to be
|
||||||
// unique across groups.
|
// unique across groups.
|
||||||
//
|
//
|
||||||
// TODO(caesarxuchao): Add proper multi-group support so that kinds & resources are
|
// TODO: split into sub-interfaces
|
||||||
// scoped to groups. See http://issues.k8s.io/12413 and http://issues.k8s.io/10009.
|
|
||||||
type RESTMapper interface {
|
type RESTMapper interface {
|
||||||
// KindFor takes a partial resource and returns the single match. Returns an error if there are multiple matches
|
// KindFor takes a partial resource and returns the single match. Returns an error if there are multiple matches
|
||||||
KindFor(resource unversioned.GroupVersionResource) (unversioned.GroupVersionKind, error)
|
KindFor(resource unversioned.GroupVersionResource) (unversioned.GroupVersionKind, error)
|
||||||
|
@ -20,6 +20,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
|
"k8s.io/kubernetes/pkg/api/meta"
|
||||||
"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"
|
||||||
@ -28,46 +29,72 @@ import (
|
|||||||
|
|
||||||
// ClientPool manages a pool of dynamic clients.
|
// ClientPool manages a pool of dynamic clients.
|
||||||
type ClientPool interface {
|
type ClientPool interface {
|
||||||
// ClientForGroupVersion returns a client configured for the specified groupVersion.
|
// ClientForGroupVersionKind returns a client configured for the specified groupVersionResource.
|
||||||
ClientForGroupVersion(groupVersion unversioned.GroupVersion) (*Client, error)
|
// Resource may be empty.
|
||||||
|
ClientForGroupVersionResource(resource unversioned.GroupVersionResource) (*Client, error)
|
||||||
|
// ClientForGroupVersionKind returns a client configured for the specified groupVersionKind.
|
||||||
|
// Kind may be empty.
|
||||||
|
ClientForGroupVersionKind(kind unversioned.GroupVersionKind) (*Client, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
// APIPathResolverFunc knows how to convert a groupVersion to its API path.
|
// APIPathResolverFunc knows how to convert a groupVersion to its API path. The Kind field is
|
||||||
type APIPathResolverFunc func(groupVersion unversioned.GroupVersion) string
|
// optional.
|
||||||
|
type APIPathResolverFunc func(kind unversioned.GroupVersionKind) string
|
||||||
|
|
||||||
// LegacyAPIPathResolverFunc can resolve paths properly with the legacy API.
|
// LegacyAPIPathResolverFunc can resolve paths properly with the legacy API.
|
||||||
func LegacyAPIPathResolverFunc(groupVersion unversioned.GroupVersion) string {
|
func LegacyAPIPathResolverFunc(kind unversioned.GroupVersionKind) string {
|
||||||
if len(groupVersion.Group) == 0 {
|
if len(kind.Group) == 0 {
|
||||||
return "/api"
|
return "/api"
|
||||||
}
|
}
|
||||||
return "/apis"
|
return "/apis"
|
||||||
}
|
}
|
||||||
|
|
||||||
// clientPoolImpl implements Factory
|
// clientPoolImpl implements ClientPool and caches clients for the resource group versions
|
||||||
|
// is asked to retrieve. This type is thread safe.
|
||||||
type clientPoolImpl struct {
|
type clientPoolImpl struct {
|
||||||
lock sync.RWMutex
|
lock sync.RWMutex
|
||||||
config *restclient.Config
|
config *restclient.Config
|
||||||
clients map[unversioned.GroupVersion]*Client
|
clients map[unversioned.GroupVersion]*Client
|
||||||
apiPathResolverFunc APIPathResolverFunc
|
apiPathResolverFunc APIPathResolverFunc
|
||||||
|
mapper meta.RESTMapper
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewClientPool returns a ClientPool from the specified config
|
// NewClientPool returns a ClientPool from the specified config. It reuses clients for the the same
|
||||||
func NewClientPool(config *restclient.Config, apiPathResolverFunc APIPathResolverFunc) ClientPool {
|
// group version. It is expected this type may be wrapped by specific logic that special cases certain
|
||||||
|
// resources or groups.
|
||||||
|
func NewClientPool(config *restclient.Config, mapper meta.RESTMapper, apiPathResolverFunc APIPathResolverFunc) ClientPool {
|
||||||
confCopy := *config
|
confCopy := *config
|
||||||
return &clientPoolImpl{
|
return &clientPoolImpl{
|
||||||
config: &confCopy,
|
config: &confCopy,
|
||||||
clients: map[unversioned.GroupVersion]*Client{},
|
clients: map[unversioned.GroupVersion]*Client{},
|
||||||
apiPathResolverFunc: apiPathResolverFunc,
|
apiPathResolverFunc: apiPathResolverFunc,
|
||||||
|
mapper: mapper,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ClientForGroupVersion returns a client for the specified groupVersion, creates one if none exists
|
// ClientForGroupVersionResource uses the provided RESTMapper to identify the appropriate resource. Resource may
|
||||||
func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVersion) (*Client, error) {
|
// be empty. If no matching kind is found the underlying client for that group is still returned.
|
||||||
|
func (c *clientPoolImpl) ClientForGroupVersionResource(resource unversioned.GroupVersionResource) (*Client, error) {
|
||||||
|
kinds, err := c.mapper.KindsFor(resource)
|
||||||
|
if err != nil {
|
||||||
|
if meta.IsNoMatchError(err) {
|
||||||
|
return c.ClientForGroupVersionKind(unversioned.GroupVersionKind{Group: resource.Group, Version: resource.Version})
|
||||||
|
}
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return c.ClientForGroupVersionKind(kinds[0])
|
||||||
|
}
|
||||||
|
|
||||||
|
// ClientForGroupVersion returns a client for the specified groupVersion, creates one if none exists. Kind
|
||||||
|
// in the GroupVersionKind may be empty.
|
||||||
|
func (c *clientPoolImpl) ClientForGroupVersionKind(kind unversioned.GroupVersionKind) (*Client, error) {
|
||||||
c.lock.Lock()
|
c.lock.Lock()
|
||||||
defer c.lock.Unlock()
|
defer c.lock.Unlock()
|
||||||
|
|
||||||
|
gv := kind.GroupVersion()
|
||||||
|
|
||||||
// do we have a client already configured?
|
// do we have a client already configured?
|
||||||
if existingClient, found := c.clients[groupVersion]; found {
|
if existingClient, found := c.clients[gv]; found {
|
||||||
return existingClient, nil
|
return existingClient, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -76,10 +103,10 @@ func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVer
|
|||||||
conf := &confCopy
|
conf := &confCopy
|
||||||
|
|
||||||
// we need to set the api path based on group version, if no group, default to legacy path
|
// we need to set the api path based on group version, if no group, default to legacy path
|
||||||
conf.APIPath = c.apiPathResolverFunc(groupVersion)
|
conf.APIPath = c.apiPathResolverFunc(kind)
|
||||||
|
|
||||||
// we need to make a client
|
// we need to make a client
|
||||||
conf.GroupVersion = &groupVersion
|
conf.GroupVersion = &gv
|
||||||
|
|
||||||
if conf.NegotiatedSerializer == nil {
|
if conf.NegotiatedSerializer == nil {
|
||||||
streamingInfo, _ := api.Codecs.StreamingSerializerForMediaType("application/json;stream=watch", nil)
|
streamingInfo, _ := api.Codecs.StreamingSerializerForMediaType("application/json;stream=watch", nil)
|
||||||
@ -90,6 +117,6 @@ func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVer
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
c.clients[groupVersion] = dynamicClient
|
c.clients[gv] = dynamicClient
|
||||||
return dynamicClient, nil
|
return dynamicClient, nil
|
||||||
}
|
}
|
||||||
|
@ -29,7 +29,6 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/meta/metatypes"
|
"k8s.io/kubernetes/pkg/api/meta/metatypes"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
"k8s.io/kubernetes/pkg/api/v1"
|
"k8s.io/kubernetes/pkg/api/v1"
|
||||||
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
|
||||||
"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/garbagecollector/metaonly"
|
"k8s.io/kubernetes/pkg/controller/garbagecollector/metaonly"
|
||||||
@ -478,7 +477,7 @@ func (gc *GarbageCollector) monitorFor(resource unversioned.GroupVersionResource
|
|||||||
// 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 := gc.metaOnlyClientPool.ClientForGroupVersion(resource.GroupVersion())
|
client, err := gc.metaOnlyClientPool.ClientForGroupVersionKind(kind)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return monitor, err
|
return monitor, err
|
||||||
}
|
}
|
||||||
@ -537,12 +536,11 @@ var ignoredResources = map[unversioned.GroupVersionResource]struct{}{
|
|||||||
unversioned.GroupVersionResource{Group: "authorization.k8s.io", Version: "v1beta1", Resource: "localsubjectaccessreviews"}: {},
|
unversioned.GroupVersionResource{Group: "authorization.k8s.io", Version: "v1beta1", Resource: "localsubjectaccessreviews"}: {},
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewGarbageCollector(metaOnlyClientPool dynamic.ClientPool, clientPool dynamic.ClientPool, resources []unversioned.GroupVersionResource) (*GarbageCollector, error) {
|
func NewGarbageCollector(metaOnlyClientPool dynamic.ClientPool, clientPool dynamic.ClientPool, mapper meta.RESTMapper, resources []unversioned.GroupVersionResource) (*GarbageCollector, error) {
|
||||||
gc := &GarbageCollector{
|
gc := &GarbageCollector{
|
||||||
metaOnlyClientPool: metaOnlyClientPool,
|
metaOnlyClientPool: metaOnlyClientPool,
|
||||||
clientPool: clientPool,
|
clientPool: clientPool,
|
||||||
// TODO: should use a dynamic RESTMapper built from the discovery results.
|
restMapper: mapper,
|
||||||
restMapper: registered.RESTMapper(),
|
|
||||||
clock: clock.RealClock{},
|
clock: clock.RealClock{},
|
||||||
dirtyQueue: workqueue.NewTimedWorkQueue(),
|
dirtyQueue: workqueue.NewTimedWorkQueue(),
|
||||||
orphanQueue: workqueue.NewTimedWorkQueue(),
|
orphanQueue: workqueue.NewTimedWorkQueue(),
|
||||||
@ -611,7 +609,7 @@ func (gc *GarbageCollector) apiResource(apiVersion, kind string, namespaced bool
|
|||||||
|
|
||||||
func (gc *GarbageCollector) deleteObject(item objectReference) error {
|
func (gc *GarbageCollector) deleteObject(item objectReference) error {
|
||||||
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
||||||
client, err := gc.clientPool.ClientForGroupVersion(fqKind.GroupVersion())
|
client, err := gc.clientPool.ClientForGroupVersionKind(fqKind)
|
||||||
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
||||||
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -625,7 +623,7 @@ func (gc *GarbageCollector) deleteObject(item objectReference) error {
|
|||||||
|
|
||||||
func (gc *GarbageCollector) getObject(item objectReference) (*runtime.Unstructured, error) {
|
func (gc *GarbageCollector) getObject(item objectReference) (*runtime.Unstructured, error) {
|
||||||
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
||||||
client, err := gc.clientPool.ClientForGroupVersion(fqKind.GroupVersion())
|
client, err := gc.clientPool.ClientForGroupVersionKind(fqKind)
|
||||||
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
||||||
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -636,7 +634,7 @@ func (gc *GarbageCollector) getObject(item objectReference) (*runtime.Unstructur
|
|||||||
|
|
||||||
func (gc *GarbageCollector) updateObject(item objectReference, obj *runtime.Unstructured) (*runtime.Unstructured, error) {
|
func (gc *GarbageCollector) updateObject(item objectReference, obj *runtime.Unstructured) (*runtime.Unstructured, error) {
|
||||||
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
||||||
client, err := gc.clientPool.ClientForGroupVersion(fqKind.GroupVersion())
|
client, err := gc.clientPool.ClientForGroupVersionKind(fqKind)
|
||||||
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
||||||
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -647,7 +645,7 @@ func (gc *GarbageCollector) updateObject(item objectReference, obj *runtime.Unst
|
|||||||
|
|
||||||
func (gc *GarbageCollector) patchObject(item objectReference, patch []byte) (*runtime.Unstructured, error) {
|
func (gc *GarbageCollector) patchObject(item objectReference, patch []byte) (*runtime.Unstructured, error) {
|
||||||
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
fqKind := unversioned.FromAPIVersionAndKind(item.APIVersion, item.Kind)
|
||||||
client, err := gc.clientPool.ClientForGroupVersion(fqKind.GroupVersion())
|
client, err := gc.clientPool.ClientForGroupVersionKind(fqKind)
|
||||||
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
gc.registeredRateLimiter.registerIfNotPresent(fqKind.GroupVersion(), client, "garbage_collector_operation")
|
||||||
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
resource, err := gc.apiResource(item.APIVersion, item.Kind, len(item.Namespace) != 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -728,7 +726,7 @@ func (gc *GarbageCollector) processItem(item *node) error {
|
|||||||
// prevent objects having references to an old resource from being
|
// prevent objects having references to an old resource from being
|
||||||
// deleted during a cluster upgrade.
|
// deleted during a cluster upgrade.
|
||||||
fqKind := unversioned.FromAPIVersionAndKind(reference.APIVersion, reference.Kind)
|
fqKind := unversioned.FromAPIVersionAndKind(reference.APIVersion, reference.Kind)
|
||||||
client, err := gc.clientPool.ClientForGroupVersion(fqKind.GroupVersion())
|
client, err := gc.clientPool.ClientForGroupVersionKind(fqKind)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -23,17 +23,19 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
_ "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/install"
|
||||||
|
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/meta/metatypes"
|
"k8s.io/kubernetes/pkg/api/meta/metatypes"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
"k8s.io/kubernetes/pkg/api/v1"
|
"k8s.io/kubernetes/pkg/api/v1"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
"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/metaonly"
|
||||||
|
"k8s.io/kubernetes/pkg/runtime/serializer"
|
||||||
"k8s.io/kubernetes/pkg/types"
|
"k8s.io/kubernetes/pkg/types"
|
||||||
"k8s.io/kubernetes/pkg/util/clock"
|
"k8s.io/kubernetes/pkg/util/clock"
|
||||||
"k8s.io/kubernetes/pkg/util/json"
|
"k8s.io/kubernetes/pkg/util/json"
|
||||||
@ -44,11 +46,11 @@ import (
|
|||||||
func TestNewGarbageCollector(t *testing.T) {
|
func TestNewGarbageCollector(t *testing.T) {
|
||||||
config := &restclient.Config{}
|
config := &restclient.Config{}
|
||||||
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
||||||
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
metaOnlyClientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
config.ContentConfig.NegotiatedSerializer = nil
|
config.ContentConfig.NegotiatedSerializer = nil
|
||||||
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
|
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
|
||||||
gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, podResource)
|
gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, registered.RESTMapper(), podResource)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -107,11 +109,11 @@ func testServerAndClientConfig(handler func(http.ResponseWriter, *http.Request))
|
|||||||
|
|
||||||
func setupGC(t *testing.T, config *restclient.Config) *GarbageCollector {
|
func setupGC(t *testing.T, config *restclient.Config) *GarbageCollector {
|
||||||
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
||||||
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
metaOnlyClientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
config.ContentConfig.NegotiatedSerializer = nil
|
config.ContentConfig.NegotiatedSerializer = nil
|
||||||
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
|
podResource := []unversioned.GroupVersionResource{{Version: "v1", Resource: "pods"}}
|
||||||
gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, podResource)
|
gc, err := NewGarbageCollector(metaOnlyClientPool, clientPool, registered.RESTMapper(), podResource)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -340,9 +342,9 @@ func TestGCListWatcher(t *testing.T) {
|
|||||||
testHandler := &fakeActionHandler{}
|
testHandler := &fakeActionHandler{}
|
||||||
srv, clientConfig := testServerAndClientConfig(testHandler.ServeHTTP)
|
srv, clientConfig := testServerAndClientConfig(testHandler.ServeHTTP)
|
||||||
defer srv.Close()
|
defer srv.Close()
|
||||||
clientPool := dynamic.NewClientPool(clientConfig, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(clientConfig, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
podResource := unversioned.GroupVersionResource{Version: "v1", Resource: "pods"}
|
podResource := unversioned.GroupVersionResource{Version: "v1", Resource: "pods"}
|
||||||
client, err := clientPool.ClientForGroupVersion(podResource.GroupVersion())
|
client, err := clientPool.ClientForGroupVersionResource(podResource)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
@ -28,6 +28,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
"k8s.io/kubernetes/pkg/api/errors"
|
"k8s.io/kubernetes/pkg/api/errors"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||||
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
|
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
|
||||||
"k8s.io/kubernetes/pkg/client/restclient"
|
"k8s.io/kubernetes/pkg/client/restclient"
|
||||||
@ -113,7 +114,7 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *unversioned.APIV
|
|||||||
groupVersionResources := testGroupVersionResources()
|
groupVersionResources := testGroupVersionResources()
|
||||||
for _, groupVersionResource := range groupVersionResources {
|
for _, groupVersionResource := range groupVersionResources {
|
||||||
urlPath := path.Join([]string{
|
urlPath := path.Join([]string{
|
||||||
dynamic.LegacyAPIPathResolverFunc(groupVersionResource.GroupVersion()),
|
dynamic.LegacyAPIPathResolverFunc(unversioned.GroupVersionKind{Group: groupVersionResource.Group, Version: groupVersionResource.Version}),
|
||||||
groupVersionResource.Group,
|
groupVersionResource.Group,
|
||||||
groupVersionResource.Version,
|
groupVersionResource.Version,
|
||||||
"namespaces",
|
"namespaces",
|
||||||
@ -155,7 +156,7 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *unversioned.APIV
|
|||||||
defer srv.Close()
|
defer srv.Close()
|
||||||
|
|
||||||
mockClient := fake.NewSimpleClientset(testInput.testNamespace)
|
mockClient := fake.NewSimpleClientset(testInput.testNamespace)
|
||||||
clientPool := dynamic.NewClientPool(clientConfig, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(clientConfig, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
|
|
||||||
err := syncNamespace(mockClient, clientPool, operationNotSupportedCache{}, groupVersionResources, testInput.testNamespace, api.FinalizerKubernetes)
|
err := syncNamespace(mockClient, clientPool, operationNotSupportedCache{}, groupVersionResources, testInput.testNamespace, api.FinalizerKubernetes)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -274,7 +274,7 @@ func deleteAllContentForGroupVersionResource(
|
|||||||
glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - estimate - namespace: %s, gvr: %v, estimate: %v", namespace, gvr, estimate)
|
glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - estimate - namespace: %s, gvr: %v, estimate: %v", namespace, gvr, estimate)
|
||||||
|
|
||||||
// get a client for this group version...
|
// get a client for this group version...
|
||||||
dynamicClient, err := clientPool.ClientForGroupVersion(gvr.GroupVersion())
|
dynamicClient, err := clientPool.ClientForGroupVersionResource(gvr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - unable to get client - namespace: %s, gvr: %v, err: %v", namespace, gvr, err)
|
glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - unable to get client - namespace: %s, gvr: %v, err: %v", namespace, gvr, err)
|
||||||
return estimate, err
|
return estimate, err
|
||||||
|
@ -34,6 +34,7 @@ import (
|
|||||||
apierrs "k8s.io/kubernetes/pkg/api/errors"
|
apierrs "k8s.io/kubernetes/pkg/api/errors"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
"k8s.io/kubernetes/pkg/api/v1"
|
"k8s.io/kubernetes/pkg/api/v1"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
"k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
|
"k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
|
||||||
"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"
|
||||||
@ -197,7 +198,7 @@ func (f *Framework) BeforeEach() {
|
|||||||
clientRepoConfig := getClientRepoConfig(config)
|
clientRepoConfig := getClientRepoConfig(config)
|
||||||
f.StagingClient, err = staging.NewForConfig(clientRepoConfig)
|
f.StagingClient, err = staging.NewForConfig(clientRepoConfig)
|
||||||
Expect(err).NotTo(HaveOccurred())
|
Expect(err).NotTo(HaveOccurred())
|
||||||
f.ClientPool = dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
f.ClientPool = dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
}
|
}
|
||||||
|
|
||||||
if f.federated {
|
if f.federated {
|
||||||
|
@ -437,7 +437,7 @@ func SkipUnlessFederated(c *client.Client) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func SkipIfMissingResource(clientPool dynamic.ClientPool, gvr unversioned.GroupVersionResource, namespace string) {
|
func SkipIfMissingResource(clientPool dynamic.ClientPool, gvr unversioned.GroupVersionResource, namespace string) {
|
||||||
dynamicClient, err := clientPool.ClientForGroupVersion(gvr.GroupVersion())
|
dynamicClient, err := clientPool.ClientForGroupVersionResource(gvr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
Failf("Unexpected error getting dynamic client for %v: %v", gvr.GroupVersion(), err)
|
Failf("Unexpected error getting dynamic client for %v: %v", gvr.GroupVersion(), err)
|
||||||
}
|
}
|
||||||
@ -1230,7 +1230,7 @@ func hasRemainingContent(c *client.Client, clientPool dynamic.ClientPool, namesp
|
|||||||
// dump how many of resource type is on the server in a log.
|
// dump how many of resource type is on the server in a log.
|
||||||
for _, gvr := range groupVersionResources {
|
for _, gvr := range groupVersionResources {
|
||||||
// get a client for this group version...
|
// get a client for this group version...
|
||||||
dynamicClient, err := clientPool.ClientForGroupVersion(gvr.GroupVersion())
|
dynamicClient, err := clientPool.ClientForGroupVersionResource(gvr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// not all resource types support list, so some errors here are normal depending on the resource type.
|
// not all resource types support list, so some errors here are normal depending on the resource type.
|
||||||
Logf("namespace: %s, unable to get client - gvr: %v, error: %v", namespace, gvr, err)
|
Logf("namespace: %s, unable to get client - gvr: %v, error: %v", namespace, gvr, err)
|
||||||
|
@ -20,6 +20,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"k8s.io/kubernetes/pkg/api"
|
"k8s.io/kubernetes/pkg/api"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||||
"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"
|
||||||
@ -54,7 +55,7 @@ func (n *NamespaceController) Start() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
resources, err := client.Discovery().ServerPreferredNamespacedResources()
|
resources, err := client.Discovery().ServerPreferredNamespacedResources()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
@ -33,6 +33,7 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/api/errors"
|
"k8s.io/kubernetes/pkg/api/errors"
|
||||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||||
"k8s.io/kubernetes/pkg/api/v1"
|
"k8s.io/kubernetes/pkg/api/v1"
|
||||||
|
"k8s.io/kubernetes/pkg/apimachinery/registered"
|
||||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_3"
|
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_3"
|
||||||
"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"
|
||||||
@ -133,10 +134,10 @@ func setup(t *testing.T) (*httptest.Server, *garbagecollector.GarbageCollector,
|
|||||||
}
|
}
|
||||||
config := &restclient.Config{Host: s.URL}
|
config := &restclient.Config{Host: s.URL}
|
||||||
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
config.ContentConfig.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: metaonly.NewMetadataCodecFactory()}
|
||||||
metaOnlyClientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
metaOnlyClientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
config.ContentConfig.NegotiatedSerializer = nil
|
config.ContentConfig.NegotiatedSerializer = nil
|
||||||
clientPool := dynamic.NewClientPool(config, dynamic.LegacyAPIPathResolverFunc)
|
clientPool := dynamic.NewClientPool(config, registered.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
|
||||||
gc, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, groupVersionResources)
|
gc, err := garbagecollector.NewGarbageCollector(metaOnlyClientPool, clientPool, registered.RESTMapper(), groupVersionResources)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to create garbage collector")
|
t.Fatalf("Failed to create garbage collector")
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user