From 4d479aff37edfcdb17ac5a5f89b1e3f95d118f30 Mon Sep 17 00:00:00 2001 From: p0lyn0mial Date: Mon, 17 Sep 2018 19:07:00 +0200 Subject: [PATCH] adds dynamic informer factory --- staging/src/k8s.io/client-go/dynamic/BUILD | 1 + .../client-go/dynamic/dynamicinformer/BUILD | 53 ++++++ .../dynamic/dynamicinformer/informer.go | 155 +++++++++++++++++ .../dynamic/dynamicinformer/informer_test.go | 160 ++++++++++++++++++ .../dynamic/dynamicinformer/interface.go | 34 ++++ .../client-go/dynamic/dynamiclister/BUILD | 2 + .../client-go/dynamic/dynamiclister/shim.go | 87 ++++++++++ 7 files changed, 492 insertions(+) create mode 100644 staging/src/k8s.io/client-go/dynamic/dynamicinformer/BUILD create mode 100644 staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer.go create mode 100644 staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer_test.go create mode 100644 staging/src/k8s.io/client-go/dynamic/dynamicinformer/interface.go create mode 100644 staging/src/k8s.io/client-go/dynamic/dynamiclister/shim.go diff --git a/staging/src/k8s.io/client-go/dynamic/BUILD b/staging/src/k8s.io/client-go/dynamic/BUILD index 8c6297220c..5a923a9b4f 100644 --- a/staging/src/k8s.io/client-go/dynamic/BUILD +++ b/staging/src/k8s.io/client-go/dynamic/BUILD @@ -59,6 +59,7 @@ filegroup( name = "all-srcs", srcs = [ ":package-srcs", + "//staging/src/k8s.io/client-go/dynamic/dynamicinformer:all-srcs", "//staging/src/k8s.io/client-go/dynamic/dynamiclister:all-srcs", "//staging/src/k8s.io/client-go/dynamic/fake:all-srcs", ], diff --git a/staging/src/k8s.io/client-go/dynamic/dynamicinformer/BUILD b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/BUILD new file mode 100644 index 0000000000..0a708d17a9 --- /dev/null +++ b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/BUILD @@ -0,0 +1,53 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "informer.go", + "interface.go", + ], + importmap = "k8s.io/kubernetes/vendor/k8s.io/client-go/dynamic/dynamicinformer", + importpath = "k8s.io/client-go/dynamic/dynamicinformer", + visibility = ["//visibility:public"], + deps = [ + "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library", + "//staging/src/k8s.io/client-go/dynamic:go_default_library", + "//staging/src/k8s.io/client-go/dynamic/dynamiclister:go_default_library", + "//staging/src/k8s.io/client-go/informers:go_default_library", + "//staging/src/k8s.io/client-go/tools/cache:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["informer_test.go"], + embed = [":go_default_library"], + deps = [ + "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library", + "//staging/src/k8s.io/client-go/dynamic/fake:go_default_library", + "//staging/src/k8s.io/client-go/tools/cache:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [":package-srcs"], + tags = ["automanaged"], + visibility = ["//visibility:public"], +) diff --git a/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer.go b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer.go new file mode 100644 index 0000000000..88a2e500c0 --- /dev/null +++ b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer.go @@ -0,0 +1,155 @@ +/* +Copyright 2018 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 dynamicinformer + +import ( + "sync" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/dynamic/dynamiclister" + "k8s.io/client-go/informers" + "k8s.io/client-go/tools/cache" +) + +// NewDynamicSharedInformerFactory constructs a new instance of dynamicSharedInformerFactory for all namespaces. +func NewDynamicSharedInformerFactory(client dynamic.Interface, defaultResync time.Duration) DynamicSharedInformerFactory { + return NewFilteredDynamicSharedInformerFactory(client, defaultResync, metav1.NamespaceAll, nil) +} + +// NewFilteredDynamicSharedInformerFactory constructs a new instance of dynamicSharedInformerFactory. +// Listers obtained via this factory will be subject to the same filters as specified here. +func NewFilteredDynamicSharedInformerFactory(client dynamic.Interface, defaultResync time.Duration, namespace string, tweakListOptions TweakListOptionsFunc) DynamicSharedInformerFactory { + return &dynamicSharedInformerFactory{ + client: client, + defaultResync: defaultResync, + namespace: metav1.NamespaceAll, + informers: map[schema.GroupVersionResource]informers.GenericInformer{}, + startedInformers: make(map[schema.GroupVersionResource]bool), + } +} + +type dynamicSharedInformerFactory struct { + client dynamic.Interface + defaultResync time.Duration + namespace string + + lock sync.Mutex + informers map[schema.GroupVersionResource]informers.GenericInformer + // startedInformers is used for tracking which informers have been started. + // This allows Start() to be called multiple times safely. + startedInformers map[schema.GroupVersionResource]bool +} + +var _ DynamicSharedInformerFactory = &dynamicSharedInformerFactory{} + +func (f *dynamicSharedInformerFactory) ForResource(gvr schema.GroupVersionResource) informers.GenericInformer { + f.lock.Lock() + defer f.lock.Unlock() + + key := gvr + informer, exists := f.informers[key] + if exists { + return informer + } + + informer = NewFilteredDynamicInformer(f.client, gvr, f.namespace, f.defaultResync, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, nil) + f.informers[key] = informer + + return informer +} + +// Start initializes all requested informers. +func (f *dynamicSharedInformerFactory) Start(stopCh <-chan struct{}) { + f.lock.Lock() + defer f.lock.Unlock() + + for informerType, informer := range f.informers { + if !f.startedInformers[informerType] { + go informer.Informer().Run(stopCh) + f.startedInformers[informerType] = true + } + } +} + +// WaitForCacheSync waits for all started informers' cache were synced. +func (f *dynamicSharedInformerFactory) WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool { + informers := func() map[schema.GroupVersionResource]cache.SharedIndexInformer { + f.lock.Lock() + defer f.lock.Unlock() + + informers := map[schema.GroupVersionResource]cache.SharedIndexInformer{} + for informerType, informer := range f.informers { + if f.startedInformers[informerType] { + informers[informerType] = informer.Informer() + } + } + return informers + }() + + res := map[schema.GroupVersionResource]bool{} + for informType, informer := range informers { + res[informType] = cache.WaitForCacheSync(stopCh, informer.HasSynced) + } + return res +} + +// NewFilteredDynamicInformer constructs a new informer for a dynamic type. +func NewFilteredDynamicInformer(client dynamic.Interface, gvr schema.GroupVersionResource, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions TweakListOptionsFunc) informers.GenericInformer { + return &dynamicInformer{ + gvr: gvr, + informer: cache.NewSharedIndexInformer( + &cache.ListWatch{ + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { + if tweakListOptions != nil { + tweakListOptions(&options) + } + return client.Resource(gvr).Namespace(namespace).List(options) + }, + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { + if tweakListOptions != nil { + tweakListOptions(&options) + } + return client.Resource(gvr).Namespace(namespace).Watch(options) + }, + }, + &unstructured.Unstructured{}, + resyncPeriod, + indexers, + ), + } +} + +type dynamicInformer struct { + informer cache.SharedIndexInformer + gvr schema.GroupVersionResource +} + +var _ informers.GenericInformer = &dynamicInformer{} + +func (d *dynamicInformer) Informer() cache.SharedIndexInformer { + return d.informer +} + +func (d *dynamicInformer) Lister() cache.GenericLister { + return dynamiclister.NewRuntimeObjectShim(dynamiclister.New(d.informer.GetIndexer(), d.gvr)) +} diff --git a/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer_test.go b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer_test.go new file mode 100644 index 0000000000..67c8560206 --- /dev/null +++ b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/informer_test.go @@ -0,0 +1,160 @@ +/* +Copyright 2018 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 dynamicinformer_test + +import ( + "context" + "testing" + "time" + + "k8s.io/apimachinery/pkg/api/equality" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/diff" + "k8s.io/client-go/dynamic/dynamicinformer" + "k8s.io/client-go/dynamic/fake" + "k8s.io/client-go/tools/cache" +) + +func TestDynamicSharedInformerFactory(t *testing.T) { + scenarios := []struct { + name string + existingObj *unstructured.Unstructured + gvr schema.GroupVersionResource + ns string + trigger func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured + handler func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs + }{ + // scenario 1 + { + name: "scenario 1: test if adding an object triggers AddFunc", + ns: "ns-foo", + gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"}, + trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, _ *unstructured.Unstructured) *unstructured.Unstructured { + testObject := newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo") + createdObj, err := fakeClient.Resource(gvr).Namespace(ns).Create(testObject, metav1.CreateOptions{}) + if err != nil { + t.Error(err) + } + return createdObj + }, + handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs { + return &cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + rcvCh <- obj.(*unstructured.Unstructured) + }, + } + }, + }, + + // scenario 2 + { + name: "scenario 2: tests if updating an object triggers UpdateFunc", + ns: "ns-foo", + gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"}, + existingObj: newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"), + trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured { + testObject.Object["spec"] = "updatedName" + updatedObj, err := fakeClient.Resource(gvr).Namespace(ns).Update(testObject, metav1.UpdateOptions{}) + if err != nil { + t.Error(err) + } + return updatedObj + }, + handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs { + return &cache.ResourceEventHandlerFuncs{ + UpdateFunc: func(old, updated interface{}) { + rcvCh <- updated.(*unstructured.Unstructured) + }, + } + }, + }, + + // scenario 3 + { + name: "scenario 3: test if deleting an object triggers DeleteFunc", + ns: "ns-foo", + gvr: schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "deployments"}, + existingObj: newUnstructured("extensions/v1beta1", "Deployment", "ns-foo", "name-foo"), + trigger: func(gvr schema.GroupVersionResource, ns string, fakeClient *fake.FakeDynamicClient, testObject *unstructured.Unstructured) *unstructured.Unstructured { + err := fakeClient.Resource(gvr).Namespace(ns).Delete(testObject.GetName(), &metav1.DeleteOptions{}) + if err != nil { + t.Error(err) + } + return testObject + }, + handler: func(rcvCh chan<- *unstructured.Unstructured) *cache.ResourceEventHandlerFuncs { + return &cache.ResourceEventHandlerFuncs{ + DeleteFunc: func(obj interface{}) { + rcvCh <- obj.(*unstructured.Unstructured) + }, + } + }, + }, + } + + for _, ts := range scenarios { + t.Run(ts.name, func(t *testing.T) { + // test data + timeout := time.Duration(3 * time.Second) + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + scheme := runtime.NewScheme() + informerReciveObjectCh := make(chan *unstructured.Unstructured, 1) + objs := []runtime.Object{} + if ts.existingObj != nil { + objs = append(objs, ts.existingObj) + } + fakeClient := fake.NewSimpleDynamicClient(scheme, objs...) + target := dynamicinformer.NewDynamicSharedInformerFactory(fakeClient, 0) + + // act + informerListerForGvr := target.ForResource(ts.gvr) + informerListerForGvr.Informer().AddEventHandler(ts.handler(informerReciveObjectCh)) + target.Start(ctx.Done()) + if synced := target.WaitForCacheSync(ctx.Done()); !synced[ts.gvr] { + t.Errorf("informer for %s hasn't synced", ts.gvr) + } + + testObject := ts.trigger(ts.gvr, ts.ns, fakeClient, ts.existingObj) + select { + case objFromInformer := <-informerReciveObjectCh: + if !equality.Semantic.DeepEqual(testObject, objFromInformer) { + t.Fatalf("%v", diff.ObjectDiff(testObject, objFromInformer)) + } + case <-ctx.Done(): + t.Errorf("tested informer haven't received an object, waited %v", timeout) + } + }) + } +} + +func newUnstructured(apiVersion, kind, namespace, name string) *unstructured.Unstructured { + return &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": apiVersion, + "kind": kind, + "metadata": map[string]interface{}{ + "namespace": namespace, + "name": name, + }, + "spec": name, + }, + } +} diff --git a/staging/src/k8s.io/client-go/dynamic/dynamicinformer/interface.go b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/interface.go new file mode 100644 index 0000000000..083977c301 --- /dev/null +++ b/staging/src/k8s.io/client-go/dynamic/dynamicinformer/interface.go @@ -0,0 +1,34 @@ +/* +Copyright 2018 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 dynamicinformer + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/informers" +) + +// DynamicSharedInformerFactory provides access to a shared informer and lister for dynamic client +type DynamicSharedInformerFactory interface { + Start(stopCh <-chan struct{}) + ForResource(gvr schema.GroupVersionResource) informers.GenericInformer + WaitForCacheSync(stopCh <-chan struct{}) map[schema.GroupVersionResource]bool +} + +// TweakListOptionsFunc defines the signature of a helper function +// that wants to provide more listing options to API +type TweakListOptionsFunc func(*metav1.ListOptions) diff --git a/staging/src/k8s.io/client-go/dynamic/dynamiclister/BUILD b/staging/src/k8s.io/client-go/dynamic/dynamiclister/BUILD index 5c65e9ab2f..c1bb09e9b7 100644 --- a/staging/src/k8s.io/client-go/dynamic/dynamiclister/BUILD +++ b/staging/src/k8s.io/client-go/dynamic/dynamiclister/BUILD @@ -5,6 +5,7 @@ go_library( srcs = [ "interface.go", "lister.go", + "shim.go", ], importmap = "k8s.io/kubernetes/vendor/k8s.io/client-go/dynamic/dynamiclister", importpath = "k8s.io/client-go/dynamic/dynamiclister", @@ -13,6 +14,7 @@ go_library( "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library", "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library", "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library", + "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library", "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library", "//staging/src/k8s.io/client-go/tools/cache:go_default_library", ], diff --git a/staging/src/k8s.io/client-go/dynamic/dynamiclister/shim.go b/staging/src/k8s.io/client-go/dynamic/dynamiclister/shim.go new file mode 100644 index 0000000000..92a5f54af9 --- /dev/null +++ b/staging/src/k8s.io/client-go/dynamic/dynamiclister/shim.go @@ -0,0 +1,87 @@ +/* +Copyright 2018 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 dynamiclister + +import ( + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/cache" +) + +var _ cache.GenericLister = &dynamicListerShim{} +var _ cache.GenericNamespaceLister = &dynamicNamespaceListerShim{} + +// dynamicListerShim implements the cache.GenericLister interface. +type dynamicListerShim struct { + lister Lister +} + +// NewRuntimeObjectShim returns a new shim for Lister. +// It wraps Lister so that it implements cache.GenericLister interface +func NewRuntimeObjectShim(lister Lister) cache.GenericLister { + return &dynamicListerShim{lister: lister} +} + +// List will return all objects across namespaces +func (s *dynamicListerShim) List(selector labels.Selector) (ret []runtime.Object, err error) { + objs, err := s.lister.List(selector) + if err != nil { + return nil, err + } + + ret = make([]runtime.Object, len(objs)) + for index, obj := range objs { + ret[index] = obj + } + return ret, err +} + +// Get will attempt to retrieve assuming that name==key +func (s *dynamicListerShim) Get(name string) (runtime.Object, error) { + return s.lister.Get(name) +} + +func (s *dynamicListerShim) ByNamespace(namespace string) cache.GenericNamespaceLister { + return &dynamicNamespaceListerShim{ + namespaceLister: s.lister.Namespace(namespace), + } +} + +// dynamicNamespaceListerShim implements the NamespaceLister interface. +// It wraps NamespaceLister so that it implements cache.GenericNamespaceLister interface +type dynamicNamespaceListerShim struct { + namespaceLister NamespaceLister +} + +// List will return all objects in this namespace +func (ns *dynamicNamespaceListerShim) List(selector labels.Selector) (ret []runtime.Object, err error) { + objs, err := ns.namespaceLister.List(selector) + if err != nil { + return nil, err + } + + ret = make([]runtime.Object, len(objs)) + for index, obj := range objs { + ret[index] = obj + } + return ret, err +} + +// Get will attempt to retrieve by namespace and name +func (ns *dynamicNamespaceListerShim) Get(name string) (runtime.Object, error) { + return ns.namespaceLister.Get(name) +}