diff options
Diffstat (limited to 'internal/pkg/factory')
-rw-r--r-- | internal/pkg/factory/factory.go | 318 | ||||
-rw-r--r-- | internal/pkg/factory/factory_test.go | 686 |
2 files changed, 0 insertions, 1004 deletions
diff --git a/internal/pkg/factory/factory.go b/internal/pkg/factory/factory.go deleted file mode 100644 index a635e3f..0000000 --- a/internal/pkg/factory/factory.go +++ /dev/null @@ -1,318 +0,0 @@ -package factory - -import ( - "fmt" - "reflect" - "sync" - "sync/atomic" - "time" - - "github.com/sirupsen/logrus" - - kapi "k8s.io/api/core/v1" - knet "k8s.io/api/networking/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" - informerfactory "k8s.io/client-go/informers" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/tools/cache" -) - -type informer struct { - sync.Mutex - oType reflect.Type - inf cache.SharedIndexInformer - handlers map[uint64]cache.ResourceEventHandler -} - -func (i *informer) forEachHandler(obj interface{}, f func(id uint64, handler cache.ResourceEventHandler)) { - i.Lock() - defer i.Unlock() - - objType := reflect.TypeOf(obj) - if objType != i.oType { - logrus.Errorf("object type %v did not match expected %v", objType, i.oType) - return - } - - for id, handler := range i.handlers { - f(id, handler) - } -} - -// WatchFactory initializes and manages common kube watches -type WatchFactory struct { - iFactory informerfactory.SharedInformerFactory - informers map[reflect.Type]*informer - handlerCounter uint64 -} - -const ( - resyncInterval = 12 * time.Hour -) - -func newInformer(oType reflect.Type, inf cache.SharedIndexInformer) *informer { - return &informer{ - oType: oType, - inf: inf, - handlers: make(map[uint64]cache.ResourceEventHandler), - } -} - -var ( - podType reflect.Type = reflect.TypeOf(&kapi.Pod{}) - serviceType reflect.Type = reflect.TypeOf(&kapi.Service{}) - endpointsType reflect.Type = reflect.TypeOf(&kapi.Endpoints{}) - policyType reflect.Type = reflect.TypeOf(&knet.NetworkPolicy{}) - namespaceType reflect.Type = reflect.TypeOf(&kapi.Namespace{}) - nodeType reflect.Type = reflect.TypeOf(&kapi.Node{}) -) - -// NewWatchFactory initializes a new watch factory -func NewWatchFactory(c kubernetes.Interface, stopChan <-chan struct{}) (*WatchFactory, error) { - // resync time is 12 hours, none of the resources being watched in ovn-kubernetes have - // any race condition where a resync may be required e.g. cni executable on node watching for - // events on pods and assuming that an 'ADD' event will contain the annotations put in by - // ovnkube master (currently, it is just a 'get' loop) - // the downside of making it tight (like 10 minutes) is needless spinning on all resources - wf := &WatchFactory{ - iFactory: informerfactory.NewSharedInformerFactory(c, resyncInterval), - informers: make(map[reflect.Type]*informer), - } - - // Create shared informers we know we'll use - wf.informers[podType] = newInformer(podType, wf.iFactory.Core().V1().Pods().Informer()) - wf.informers[serviceType] = newInformer(serviceType, wf.iFactory.Core().V1().Services().Informer()) - wf.informers[endpointsType] = newInformer(endpointsType, wf.iFactory.Core().V1().Endpoints().Informer()) - wf.informers[policyType] = newInformer(policyType, wf.iFactory.Networking().V1().NetworkPolicies().Informer()) - wf.informers[namespaceType] = newInformer(namespaceType, wf.iFactory.Core().V1().Namespaces().Informer()) - wf.informers[nodeType] = newInformer(nodeType, wf.iFactory.Core().V1().Nodes().Informer()) - - wf.iFactory.Start(stopChan) - res := wf.iFactory.WaitForCacheSync(stopChan) - for oType, synced := range res { - if !synced { - return nil, fmt.Errorf("error in syncing cache for %v informer", oType) - } - informer := wf.informers[oType] - informer.inf.AddEventHandler(wf.newFederatedHandler(informer)) - } - - return wf, nil -} - -func (wf *WatchFactory) newFederatedHandler(inf *informer) cache.ResourceEventHandlerFuncs { - return cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - inf.forEachHandler(obj, func(id uint64, handler cache.ResourceEventHandler) { - logrus.Debugf("running %v ADD event for handler %d", inf.oType, id) - handler.OnAdd(obj) - }) - }, - UpdateFunc: func(oldObj, newObj interface{}) { - inf.forEachHandler(newObj, func(id uint64, handler cache.ResourceEventHandler) { - logrus.Debugf("running %v UPDATE event for handler %d", inf.oType, id) - handler.OnUpdate(oldObj, newObj) - }) - }, - DeleteFunc: func(obj interface{}) { - if inf.oType != reflect.TypeOf(obj) { - tombstone, ok := obj.(cache.DeletedFinalStateUnknown) - if !ok { - logrus.Errorf("couldn't get object from tombstone: %+v", obj) - return - } - obj = tombstone.Obj - objType := reflect.TypeOf(obj) - if inf.oType != objType { - logrus.Errorf("expected tombstone object resource type %v but got %v", inf.oType, objType) - return - } - } - inf.forEachHandler(obj, func(id uint64, handler cache.ResourceEventHandler) { - logrus.Debugf("running %v DELETE event for handler %d", inf.oType, id) - handler.OnDelete(obj) - }) - }, - } -} - -func getObjectMeta(objType reflect.Type, obj interface{}) (*metav1.ObjectMeta, error) { - switch objType { - case podType: - if pod, ok := obj.(*kapi.Pod); ok { - return &pod.ObjectMeta, nil - } - case serviceType: - if service, ok := obj.(*kapi.Service); ok { - return &service.ObjectMeta, nil - } - case endpointsType: - if endpoints, ok := obj.(*kapi.Endpoints); ok { - return &endpoints.ObjectMeta, nil - } - case policyType: - if policy, ok := obj.(*knet.NetworkPolicy); ok { - return &policy.ObjectMeta, nil - } - case namespaceType: - if namespace, ok := obj.(*kapi.Namespace); ok { - return &namespace.ObjectMeta, nil - } - case nodeType: - if node, ok := obj.(*kapi.Node); ok { - return &node.ObjectMeta, nil - } - } - return nil, fmt.Errorf("cannot get ObjectMeta from type %v", objType) -} - -func (wf *WatchFactory) addHandler(objType reflect.Type, namespace string, lsel *metav1.LabelSelector, funcs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - inf, ok := wf.informers[objType] - if !ok { - return 0, fmt.Errorf("unknown object type %v", objType) - } - - sel, err := metav1.LabelSelectorAsSelector(lsel) - if err != nil { - return 0, fmt.Errorf("error creating label selector: %v", err) - } - - filterFunc := func(obj interface{}) bool { - if namespace == "" && lsel == nil { - // Unfiltered handler - return true - } - meta, err := getObjectMeta(objType, obj) - if err != nil { - logrus.Errorf("watch handler filter error: %v", err) - return false - } - if namespace != "" && meta.Namespace != namespace { - return false - } - if lsel != nil && !sel.Matches(labels.Set(meta.Labels)) { - return false - } - return true - } - - // Process existing items as a set so the caller can clean up - // after a restart or whatever - existingItems := inf.inf.GetStore().List() - if processExisting != nil { - items := make([]interface{}, 0) - for _, obj := range existingItems { - if filterFunc(obj) { - items = append(items, obj) - } - } - processExisting(items) - } - - handlerID := atomic.AddUint64(&wf.handlerCounter, 1) - - inf.Lock() - defer inf.Unlock() - - inf.handlers[handlerID] = cache.FilteringResourceEventHandler{ - FilterFunc: filterFunc, - Handler: funcs, - } - logrus.Debugf("added %v event handler %d", objType, handlerID) - - // Send existing items to the handler's add function; informers usually - // do this but since we share informers, it's long-since happened so - // we must emulate that here - for _, obj := range existingItems { - inf.handlers[handlerID].OnAdd(obj) - } - - return handlerID, nil -} - -func (wf *WatchFactory) removeHandler(objType reflect.Type, handlerID uint64) error { - inf, ok := wf.informers[objType] - if !ok { - return fmt.Errorf("tried to remove unknown object type %v event handler", objType) - } - - inf.Lock() - defer inf.Unlock() - if _, ok := inf.handlers[handlerID]; !ok { - return fmt.Errorf("tried to remove unknown object type %v event handler %d", objType, handlerID) - } - delete(inf.handlers, handlerID) - logrus.Debugf("removed %v event handler %d", objType, handlerID) - return nil -} - -// AddPodHandler adds a handler function that will be executed on Pod object changes -func (wf *WatchFactory) AddPodHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(podType, "", nil, handlerFuncs, processExisting) -} - -// AddFilteredPodHandler adds a handler function that will be executed when Pod objects that match the given filters change -func (wf *WatchFactory) AddFilteredPodHandler(namespace string, lsel *metav1.LabelSelector, handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(podType, namespace, lsel, handlerFuncs, processExisting) -} - -// RemovePodHandler removes a Pod object event handler function -func (wf *WatchFactory) RemovePodHandler(handlerID uint64) error { - return wf.removeHandler(podType, handlerID) -} - -// AddServiceHandler adds a handler function that will be executed on Service object changes -func (wf *WatchFactory) AddServiceHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(serviceType, "", nil, handlerFuncs, processExisting) -} - -// RemoveServiceHandler removes a Service object event handler function -func (wf *WatchFactory) RemoveServiceHandler(handlerID uint64) error { - return wf.removeHandler(serviceType, handlerID) -} - -// AddEndpointsHandler adds a handler function that will be executed on Endpoints object changes -func (wf *WatchFactory) AddEndpointsHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(endpointsType, "", nil, handlerFuncs, processExisting) -} - -// RemoveEndpointsHandler removes a Endpoints object event handler function -func (wf *WatchFactory) RemoveEndpointsHandler(handlerID uint64) error { - return wf.removeHandler(endpointsType, handlerID) -} - -// AddPolicyHandler adds a handler function that will be executed on NetworkPolicy object changes -func (wf *WatchFactory) AddPolicyHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(policyType, "", nil, handlerFuncs, processExisting) -} - -// RemovePolicyHandler removes a NetworkPolicy object event handler function -func (wf *WatchFactory) RemovePolicyHandler(handlerID uint64) error { - return wf.removeHandler(policyType, handlerID) -} - -// AddNamespaceHandler adds a handler function that will be executed on Namespace object changes -func (wf *WatchFactory) AddNamespaceHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(namespaceType, "", nil, handlerFuncs, processExisting) -} - -// AddFilteredNamespaceHandler adds a handler function that will be executed when Namespace objects that match the given filters change -func (wf *WatchFactory) AddFilteredNamespaceHandler(namespace string, lsel *metav1.LabelSelector, handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(namespaceType, namespace, lsel, handlerFuncs, processExisting) -} - -// RemoveNamespaceHandler removes a Namespace object event handler function -func (wf *WatchFactory) RemoveNamespaceHandler(handlerID uint64) error { - return wf.removeHandler(namespaceType, handlerID) -} - -// AddNodeHandler adds a handler function that will be executed on Node object changes -func (wf *WatchFactory) AddNodeHandler(handlerFuncs cache.ResourceEventHandler, processExisting func([]interface{})) (uint64, error) { - return wf.addHandler(nodeType, "", nil, handlerFuncs, processExisting) -} - -// RemoveNodeHandler removes a Node object event handler function -func (wf *WatchFactory) RemoveNodeHandler(handlerID uint64) error { - return wf.removeHandler(nodeType, handlerID) -} diff --git a/internal/pkg/factory/factory_test.go b/internal/pkg/factory/factory_test.go deleted file mode 100644 index 38fb77e..0000000 --- a/internal/pkg/factory/factory_test.go +++ /dev/null @@ -1,686 +0,0 @@ -package factory - -import ( - "reflect" - "testing" - - "k8s.io/api/core/v1" - knet "k8s.io/api/networking/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/types" - "k8s.io/apimachinery/pkg/watch" - "k8s.io/client-go/kubernetes/fake" - core "k8s.io/client-go/testing" - "k8s.io/client-go/tools/cache" - - . "github.com/onsi/ginkgo" - . "github.com/onsi/gomega" -) - -func TestFactory(t *testing.T) { - RegisterFailHandler(Fail) - RunSpecs(t, "Watch Factory Suite") -} - -func newObjectMeta(name, namespace string) metav1.ObjectMeta { - return metav1.ObjectMeta{ - Name: name, - UID: types.UID(name), - Namespace: namespace, - Labels: map[string]string{ - "name": name, - }, - } -} - -func newPod(name, namespace string) *v1.Pod { - return &v1.Pod{ - Status: v1.PodStatus{ - Phase: v1.PodRunning, - }, - ObjectMeta: newObjectMeta(name, namespace), - Spec: v1.PodSpec{ - Containers: []v1.Container{ - { - Name: "containerName", - Image: "containerImage", - }, - }, - NodeName: "mynode", - }, - } -} - -func newNamespace(name string) *v1.Namespace { - return &v1.Namespace{ - Status: v1.NamespaceStatus{ - Phase: v1.NamespaceActive, - }, - ObjectMeta: newObjectMeta(name, name), - } -} - -func newNode(name string) *v1.Node { - return &v1.Node{ - Status: v1.NodeStatus{ - Phase: v1.NodeRunning, - }, - ObjectMeta: newObjectMeta(name, ""), - } -} - -func newPolicy(name, namespace string) *knet.NetworkPolicy { - return &knet.NetworkPolicy{ - ObjectMeta: newObjectMeta(name, namespace), - } -} - -func newEndpoints(name, namespace string) *v1.Endpoints { - return &v1.Endpoints{ - ObjectMeta: newObjectMeta(name, namespace), - } -} - -func newService(name, namespace string) *v1.Service { - return &v1.Service{ - ObjectMeta: metav1.ObjectMeta{ - Name: name, - UID: types.UID(name), - Namespace: namespace, - Labels: map[string]string{ - "name": name, - }, - }, - } -} - -func objSetup(c *fake.Clientset, objType string, listFn func(core.Action) (bool, runtime.Object, error)) *watch.FakeWatcher { - w := watch.NewFake() - c.AddWatchReactor(objType, core.DefaultWatchReactor(w, nil)) - c.AddReactor("list", objType, listFn) - return w -} - -var _ = Describe("Watch Factory Operations", func() { - var ( - fakeClient *fake.Clientset - podWatch, namespaceWatch, nodeWatch *watch.FakeWatcher - policyWatch, endpointsWatch, serviceWatch *watch.FakeWatcher - pods []*v1.Pod - namespaces []*v1.Namespace - nodes []*v1.Node - policies []*knet.NetworkPolicy - endpoints []*v1.Endpoints - services []*v1.Service - stop chan struct{} - numAdded, numUpdated, numDeleted int - ) - - BeforeEach(func() { - fakeClient = &fake.Clientset{} - stop = make(chan struct{}) - - pods = make([]*v1.Pod, 0) - podWatch = objSetup(fakeClient, "pods", func(core.Action) (bool, runtime.Object, error) { - obj := &v1.PodList{} - for _, p := range pods { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - namespaces = make([]*v1.Namespace, 0) - namespaceWatch = objSetup(fakeClient, "namespaces", func(core.Action) (bool, runtime.Object, error) { - obj := &v1.NamespaceList{} - for _, p := range namespaces { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - nodes = make([]*v1.Node, 0) - nodeWatch = objSetup(fakeClient, "nodes", func(core.Action) (bool, runtime.Object, error) { - obj := &v1.NodeList{} - for _, p := range nodes { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - policies = make([]*knet.NetworkPolicy, 0) - policyWatch = objSetup(fakeClient, "networkpolicies", func(core.Action) (bool, runtime.Object, error) { - obj := &knet.NetworkPolicyList{} - for _, p := range policies { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - endpoints = make([]*v1.Endpoints, 0) - endpointsWatch = objSetup(fakeClient, "endpoints", func(core.Action) (bool, runtime.Object, error) { - obj := &v1.EndpointsList{} - for _, p := range endpoints { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - services = make([]*v1.Service, 0) - serviceWatch = objSetup(fakeClient, "services", func(core.Action) (bool, runtime.Object, error) { - obj := &v1.ServiceList{} - for _, p := range services { - obj.Items = append(obj.Items, *p) - } - return true, obj, nil - }) - - numAdded = 0 - numUpdated = 0 - numDeleted = 0 - }) - - Context("when a processExisting is given", func() { - testExisting := func(objType reflect.Type, namespace string, lsel *metav1.LabelSelector) { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - id, err := wf.addHandler(objType, namespace, lsel, - cache.ResourceEventHandlerFuncs{}, - func(objs []interface{}) { - Expect(len(objs)).To(Equal(1)) - }) - Expect(err).NotTo(HaveOccurred()) - Expect(id).To(BeNumerically(">", uint64(0))) - wf.removeHandler(objType, id) - close(stop) - } - - It("is called for each existing pod", func() { - pods = append(pods, newPod("pod1", "default")) - testExisting(podType, "", nil) - }) - - It("is called for each existing namespace", func() { - namespaces = append(namespaces, newNamespace("default")) - testExisting(namespaceType, "", nil) - }) - - It("is called for each existing node", func() { - nodes = append(nodes, newNode("default")) - testExisting(nodeType, "", nil) - }) - - It("is called for each existing policy", func() { - policies = append(policies, newPolicy("denyall", "default")) - testExisting(policyType, "", nil) - }) - - It("is called for each existing endpoints", func() { - endpoints = append(endpoints, newEndpoints("myendpoint", "default")) - testExisting(endpointsType, "", nil) - }) - - It("is called for each existing service", func() { - services = append(services, newService("myservice", "default")) - testExisting(serviceType, "", nil) - }) - - It("is called for each existing pod that matches a given namespace and label", func() { - pod := newPod("pod1", "default") - pod.ObjectMeta.Labels["blah"] = "foobar" - pods = append(pods, pod) - testExisting(podType, "default", &metav1.LabelSelector{ - MatchLabels: map[string]string{"blah": "foobar"}, - }) - }) - }) - - Context("when existing items are known to the informer", func() { - testExisting := func(objType reflect.Type) { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - id, err := wf.addHandler(objType, "", nil, - cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - numAdded++ - }, - UpdateFunc: func(old, new interface{}) {}, - DeleteFunc: func(obj interface{}) {}, - }, nil) - Expect(err).NotTo(HaveOccurred()) - Expect(numAdded).To(Equal(2)) - wf.removeHandler(objType, id) - close(stop) - } - - It("calls ADD for each existing pod", func() { - pods = append(pods, newPod("pod1", "default")) - pods = append(pods, newPod("pod2", "default")) - testExisting(podType) - }) - - It("calls ADD for each existing namespace", func() { - namespaces = append(namespaces, newNamespace("default")) - namespaces = append(namespaces, newNamespace("default2")) - testExisting(namespaceType) - }) - - It("calls ADD for each existing node", func() { - nodes = append(nodes, newNode("default")) - nodes = append(nodes, newNode("default2")) - testExisting(nodeType) - }) - - It("calls ADD for each existing policy", func() { - policies = append(policies, newPolicy("denyall", "default")) - policies = append(policies, newPolicy("denyall2", "default")) - testExisting(policyType) - }) - - It("calls ADD for each existing endpoints", func() { - endpoints = append(endpoints, newEndpoints("myendpoint", "default")) - endpoints = append(endpoints, newEndpoints("myendpoint2", "default")) - testExisting(endpointsType) - }) - - It("calls ADD for each existing service", func() { - services = append(services, newService("myservice", "default")) - services = append(services, newService("myservice2", "default")) - testExisting(serviceType) - }) - }) - - addFilteredHandler := func(wf *WatchFactory, objType reflect.Type, namespace string, lsel *metav1.LabelSelector, funcs cache.ResourceEventHandlerFuncs) uint64 { - id, err := wf.addHandler(objType, namespace, lsel, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - defer GinkgoRecover() - numAdded++ - funcs.AddFunc(obj) - }, - UpdateFunc: func(old, new interface{}) { - defer GinkgoRecover() - numUpdated++ - funcs.UpdateFunc(old, new) - }, - DeleteFunc: func(obj interface{}) { - defer GinkgoRecover() - numDeleted++ - funcs.DeleteFunc(obj) - }, - }, nil) - Expect(err).NotTo(HaveOccurred()) - Expect(id).To(BeNumerically(">", uint64(0))) - return id - } - - addHandler := func(wf *WatchFactory, objType reflect.Type, funcs cache.ResourceEventHandlerFuncs) uint64 { - return addFilteredHandler(wf, objType, "", nil, funcs) - } - - It("responds to pod add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newPod("pod1", "default") - id := addHandler(wf, podType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - pod := obj.(*v1.Pod) - Expect(reflect.DeepEqual(pod, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newPod := new.(*v1.Pod) - Expect(reflect.DeepEqual(newPod, added)).To(BeTrue()) - Expect(newPod.Spec.NodeName).To(Equal("foobar")) - }, - DeleteFunc: func(obj interface{}) { - pod := obj.(*v1.Pod) - Expect(reflect.DeepEqual(pod, added)).To(BeTrue()) - }, - }) - - pods = append(pods, added) - podWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Spec.NodeName = "foobar" - podWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - pods = pods[:0] - podWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.RemovePodHandler(id) - close(stop) - }) - - It("responds to namespace add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newNamespace("default") - id := addHandler(wf, namespaceType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - ns := obj.(*v1.Namespace) - Expect(reflect.DeepEqual(ns, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newNS := new.(*v1.Namespace) - Expect(reflect.DeepEqual(newNS, added)).To(BeTrue()) - Expect(newNS.Status.Phase).To(Equal(v1.NamespaceTerminating)) - }, - DeleteFunc: func(obj interface{}) { - ns := obj.(*v1.Namespace) - Expect(reflect.DeepEqual(ns, added)).To(BeTrue()) - }, - }) - - namespaces = append(namespaces, added) - namespaceWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Status.Phase = v1.NamespaceTerminating - namespaceWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - namespaces = namespaces[:0] - namespaceWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.RemoveNamespaceHandler(id) - close(stop) - }) - - It("responds to node add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newNode("mynode") - id := addHandler(wf, nodeType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - node := obj.(*v1.Node) - Expect(reflect.DeepEqual(node, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newNode := new.(*v1.Node) - Expect(reflect.DeepEqual(newNode, added)).To(BeTrue()) - Expect(newNode.Status.Phase).To(Equal(v1.NodeTerminated)) - }, - DeleteFunc: func(obj interface{}) { - node := obj.(*v1.Node) - Expect(reflect.DeepEqual(node, added)).To(BeTrue()) - }, - }) - - nodes = append(nodes, added) - nodeWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Status.Phase = v1.NodeTerminated - nodeWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - nodes = nodes[:0] - nodeWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.removeHandler(nodeType, id) - close(stop) - }) - - It("responds to policy add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newPolicy("mypolicy", "default") - id := addHandler(wf, policyType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - np := obj.(*knet.NetworkPolicy) - Expect(reflect.DeepEqual(np, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newNP := new.(*knet.NetworkPolicy) - Expect(reflect.DeepEqual(newNP, added)).To(BeTrue()) - Expect(newNP.Spec.PolicyTypes).To(Equal([]knet.PolicyType{knet.PolicyTypeIngress})) - }, - DeleteFunc: func(obj interface{}) { - np := obj.(*knet.NetworkPolicy) - Expect(reflect.DeepEqual(np, added)).To(BeTrue()) - }, - }) - - policies = append(policies, added) - policyWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Spec.PolicyTypes = []knet.PolicyType{knet.PolicyTypeIngress} - policyWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - policies = policies[:0] - policyWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.removeHandler(policyType, id) - close(stop) - }) - - It("responds to endpoints add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newEndpoints("myendpoints", "default") - id := addHandler(wf, endpointsType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - eps := obj.(*v1.Endpoints) - Expect(reflect.DeepEqual(eps, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newEPs := new.(*v1.Endpoints) - Expect(reflect.DeepEqual(newEPs, added)).To(BeTrue()) - Expect(len(newEPs.Subsets)).To(Equal(1)) - }, - DeleteFunc: func(obj interface{}) { - eps := obj.(*v1.Endpoints) - Expect(reflect.DeepEqual(eps, added)).To(BeTrue()) - }, - }) - - endpoints = append(endpoints, added) - endpointsWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Subsets = append(added.Subsets, v1.EndpointSubset{ - Ports: []v1.EndpointPort{ - { - Name: "foobar", - Port: 1234, - }, - }, - }) - endpointsWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - endpoints = endpoints[:0] - endpointsWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.removeHandler(endpointsType, id) - close(stop) - }) - - It("responds to service add/update/delete events", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newService("myservice", "default") - id := addHandler(wf, serviceType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - service := obj.(*v1.Service) - Expect(reflect.DeepEqual(service, added)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newService := new.(*v1.Service) - Expect(reflect.DeepEqual(newService, added)).To(BeTrue()) - Expect(newService.Spec.ClusterIP).To(Equal("1.1.1.1")) - }, - DeleteFunc: func(obj interface{}) { - service := obj.(*v1.Service) - Expect(reflect.DeepEqual(service, added)).To(BeTrue()) - }, - }) - - services = append(services, added) - serviceWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - added.Spec.ClusterIP = "1.1.1.1" - serviceWatch.Modify(added) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - services = services[:0] - serviceWatch.Delete(added) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - wf.removeHandler(serviceType, id) - close(stop) - }) - - It("stops processing events after the handler is removed", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - added := newNamespace("default") - id := addHandler(wf, namespaceType, cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) {}, - UpdateFunc: func(old, new interface{}) {}, - DeleteFunc: func(obj interface{}) {}, - }) - - namespaces = append(namespaces, added) - namespaceWatch.Add(added) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - wf.RemoveNamespaceHandler(id) - - added2 := newNamespace("other") - namespaces = append(namespaces, added2) - namespaceWatch.Add(added2) - Consistently(func() int { return numAdded }, 2).Should(Equal(1)) - - added2.Status.Phase = v1.NamespaceTerminating - namespaceWatch.Modify(added2) - Consistently(func() int { return numUpdated }, 2).Should(Equal(0)) - namespaces = []*v1.Namespace{added} - namespaceWatch.Delete(added2) - Consistently(func() int { return numDeleted }, 2).Should(Equal(0)) - - close(stop) - }) - - It("filters correctly by label and namespace", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - passesFilter := newPod("pod1", "default") - passesFilter.ObjectMeta.Labels["blah"] = "foobar" - failsFilter := newPod("pod2", "default") - failsFilter.ObjectMeta.Labels["blah"] = "baz" - failsFilter2 := newPod("pod3", "otherns") - failsFilter2.ObjectMeta.Labels["blah"] = "foobar" - - addFilteredHandler(wf, - podType, - "default", - &metav1.LabelSelector{ - MatchLabels: map[string]string{"blah": "foobar"}, - }, - cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - pod := obj.(*v1.Pod) - Expect(reflect.DeepEqual(pod, passesFilter)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) { - newPod := new.(*v1.Pod) - Expect(reflect.DeepEqual(newPod, passesFilter)).To(BeTrue()) - }, - DeleteFunc: func(obj interface{}) { - pod := obj.(*v1.Pod) - Expect(reflect.DeepEqual(pod, passesFilter)).To(BeTrue()) - }, - }) - - pods = append(pods, passesFilter) - podWatch.Add(passesFilter) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - - // numAdded should remain 1 - pods = append(pods, failsFilter) - podWatch.Add(failsFilter) - Consistently(func() int { return numAdded }, 2).Should(Equal(1)) - - // numAdded should remain 1 - pods = append(pods, failsFilter2) - podWatch.Add(failsFilter2) - Consistently(func() int { return numAdded }, 2).Should(Equal(1)) - - passesFilter.Status.Phase = v1.PodFailed - podWatch.Modify(passesFilter) - Eventually(func() int { return numUpdated }, 2).Should(Equal(1)) - - // numAdded should remain 1 - failsFilter.Status.Phase = v1.PodFailed - podWatch.Modify(failsFilter) - Consistently(func() int { return numUpdated }, 2).Should(Equal(1)) - - failsFilter2.Status.Phase = v1.PodFailed - podWatch.Modify(failsFilter2) - Consistently(func() int { return numUpdated }, 2).Should(Equal(1)) - - pods = []*v1.Pod{failsFilter, failsFilter2} - podWatch.Delete(passesFilter) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - - close(stop) - }) - - It("correctly handles object updates that cause filter changes", func() { - wf, err := NewWatchFactory(fakeClient, stop) - Expect(err).NotTo(HaveOccurred()) - - pod := newPod("pod1", "default") - pod.ObjectMeta.Labels["blah"] = "baz" - - equalPod := pod - id := addFilteredHandler(wf, - podType, - "default", - &metav1.LabelSelector{ - MatchLabels: map[string]string{"blah": "foobar"}, - }, - cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - p := obj.(*v1.Pod) - Expect(reflect.DeepEqual(p, equalPod)).To(BeTrue()) - }, - UpdateFunc: func(old, new interface{}) {}, - DeleteFunc: func(obj interface{}) { - p := obj.(*v1.Pod) - Expect(reflect.DeepEqual(p, equalPod)).To(BeTrue()) - }, - }) - - pods = append(pods, pod) - - // Pod doesn't pass filter; shouldn't be added - podWatch.Add(pod) - Consistently(func() int { return numAdded }, 2).Should(Equal(0)) - - // Update pod to pass filter; should be treated as add. Need - // to deep-copy pod when modifying because it's a pointer all - // the way through when using FakeClient - podCopy := pod.DeepCopy() - podCopy.ObjectMeta.Labels["blah"] = "foobar" - pods = []*v1.Pod{podCopy} - equalPod = podCopy - podWatch.Modify(podCopy) - Eventually(func() int { return numAdded }, 2).Should(Equal(1)) - - // Update pod to fail filter; should be treated as delete - pod.ObjectMeta.Labels["blah"] = "baz" - podWatch.Modify(pod) - Eventually(func() int { return numDeleted }, 2).Should(Equal(1)) - Consistently(func() int { return numAdded }, 2).Should(Equal(1)) - Consistently(func() int { return numUpdated }, 2).Should(Equal(0)) - - wf.RemovePodHandler(id) - close(stop) - }) -}) |