diff --git a/plugin/federation/kubernetes_api_test.go b/plugin/federation/kubernetes_api_test.go index 3900ae0b8..0d8df64d3 100644 --- a/plugin/federation/kubernetes_api_test.go +++ b/plugin/federation/kubernetes_api_test.go @@ -9,8 +9,10 @@ import ( type APIConnFederationTest struct{} -func (APIConnFederationTest) Run() { return } -func (APIConnFederationTest) Stop() error { return nil } +func (APIConnFederationTest) Run() { return } +func (APIConnFederationTest) Stop() error { return nil } +func (APIConnFederationTest) SvcIndexReverse(string) []*api.Service { return nil } +func (APIConnFederationTest) EpIndexReverse(string) []*api.Endpoints { return nil } func (APIConnFederationTest) PodIndex(string) []*api.Pod { a := []*api.Pod{{ @@ -24,6 +26,49 @@ func (APIConnFederationTest) PodIndex(string) []*api.Pod { return a } +func (APIConnFederationTest) SvcIndex(string) []*api.Service { + svcs := []*api.Service{ + { + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: "10.0.0.1", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: api.ClusterIPNone, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "external", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ExternalName: "ext.interwebs.test", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + } + return svcs +} + func (APIConnFederationTest) ServiceList() []*api.Service { svcs := []*api.Service{ { @@ -67,6 +112,35 @@ func (APIConnFederationTest) ServiceList() []*api.Service { return svcs } +func (APIConnFederationTest) EpIndex(string) []*api.Endpoints { + eps := []*api.Endpoints{ + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.1", + Hostname: "ep1a", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + }, + } + return eps +} + func (APIConnFederationTest) EndpointsList() []*api.Endpoints { eps := []*api.Endpoints{ { diff --git a/plugin/kubernetes/controller.go b/plugin/kubernetes/controller.go index a5b697c88..a3145bf3d 100644 --- a/plugin/kubernetes/controller.go +++ b/plugin/kubernetes/controller.go @@ -21,10 +21,18 @@ var ( ) const podIPIndex = "PodIP" +const svcNameNamespaceIndex = "NameNamespace" +const svcIPIndex = "ServiceIP" +const epNameNamespaceIndex = "EndpointNameNamespace" +const epIPIndex = "EndpointsIP" type dnsController interface { ServiceList() []*api.Service + SvcIndex(string) []*api.Service + SvcIndexReverse(string) []*api.Service PodIndex(string) []*api.Pod + EpIndex(string) []*api.Endpoints + EpIndexReverse(string) []*api.Endpoints EndpointsList() []*api.Endpoints GetNodeByName(string) (*api.Node, error) @@ -44,7 +52,7 @@ type dnsControl struct { svcLister cache.Indexer podLister cache.Indexer - epLister cache.Store + epLister cache.Indexer // stopLock is used to enforce only a single call to Stop is active. // Needed because we allow stopping through an http endpoint and @@ -77,7 +85,7 @@ func newdnsController(kubeClient *kubernetes.Clientset, opts dnsControlOpts) *dn &api.Service{}, opts.resyncPeriod, cache.ResourceEventHandlerFuncs{}, - cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + cache.Indexers{svcNameNamespaceIndex: svcNameNamespaceIndexFunc, svcIPIndex: svcIPIndexFunc}) if opts.initPodCache { dns.podLister, dns.podController = cache.NewIndexerInformer( @@ -90,14 +98,15 @@ func newdnsController(kubeClient *kubernetes.Clientset, opts dnsControlOpts) *dn cache.ResourceEventHandlerFuncs{}, cache.Indexers{podIPIndex: podIPIndexFunc}) } - dns.epLister, dns.epController = cache.NewInformer( + dns.epLister, dns.epController = cache.NewIndexerInformer( &cache.ListWatch{ ListFunc: endpointsListFunc(dns.client, namespace, dns.selector), WatchFunc: endpointsWatchFunc(dns.client, namespace, dns.selector), }, &api.Endpoints{}, opts.resyncPeriod, - cache.ResourceEventHandlerFuncs{}) + cache.ResourceEventHandlerFuncs{}, + cache.Indexers{epNameNamespaceIndex: epNameNamespaceIndexFunc, epIPIndex: epIPIndexFunc}) return &dns } @@ -110,6 +119,38 @@ func podIPIndexFunc(obj interface{}) ([]string, error) { return []string{p.Status.PodIP}, nil } +func svcIPIndexFunc(obj interface{}) ([]string, error) { + svc, ok := obj.(*api.Service) + if !ok { + return nil, errors.New("obj was not an *api.Service") + } + return []string{svc.Spec.ClusterIP}, nil +} + +func svcNameNamespaceIndexFunc(obj interface{}) ([]string, error) { + s, ok := obj.(*api.Service) + if !ok { + return nil, errors.New("obj was not an *api.Service") + } + return []string{s.ObjectMeta.Name + "." + s.ObjectMeta.Namespace}, nil +} + +func epNameNamespaceIndexFunc(obj interface{}) ([]string, error) { + s, ok := obj.(*api.Endpoints) + if !ok { + return nil, errors.New("obj was not an *api.Endpoints") + } + return []string{s.ObjectMeta.Name + "." + s.ObjectMeta.Namespace}, nil +} + +func epIPIndexFunc(obj interface{}) ([]string, error) { + ep, ok := obj.(*api.EndpointAddress) + if !ok { + return nil, errors.New("obj was not an *api.EndpointAddress") + } + return []string{ep.IP}, nil +} + func serviceListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) func(meta.ListOptions) (runtime.Object, error) { return func(opts meta.ListOptions) (runtime.Object, error) { if s != nil { @@ -255,6 +296,79 @@ func (dns *dnsControl) PodIndex(ip string) (pods []*api.Pod) { return pods } +func (dns *dnsControl) SvcIndex(idx string) (svcs []*api.Service) { + if dns.svcLister == nil { + return nil + } + os, err := dns.svcLister.ByIndex(svcNameNamespaceIndex, idx) + if err != nil { + return nil + } + for _, o := range os { + s, ok := o.(*api.Service) + if !ok { + continue + } + svcs = append(svcs, s) + } + return svcs +} + +func (dns *dnsControl) SvcIndexReverse(ip string) (svcs []*api.Service) { + if dns.svcLister == nil { + return nil + } + os, err := dns.svcLister.ByIndex(svcIPIndex, ip) + if err != nil { + return nil + } + + for _, o := range os { + s, ok := o.(*api.Service) + if !ok { + continue + } + svcs = append(svcs, s) + } + return svcs +} + +func (dns *dnsControl) EpIndex(idx string) (ep []*api.Endpoints) { + if dns.epLister == nil { + return nil + } + os, err := dns.epLister.ByIndex(epNameNamespaceIndex, idx) + if err != nil { + return nil + } + for _, o := range os { + e, ok := o.(*api.Endpoints) + if !ok { + continue + } + ep = append(ep, e) + } + return ep +} + +func (dns *dnsControl) EpIndexReverse(ip string) (ep []*api.Endpoints) { + if dns.svcLister == nil { + return nil + } + os, err := dns.epLister.ByIndex(epIPIndex, ip) + if err != nil { + return nil + } + for _, o := range os { + e, ok := o.(*api.Endpoints) + if !ok { + continue + } + ep = append(ep, e) + } + return ep +} + func (dns *dnsControl) EndpointsList() (eps []*api.Endpoints) { os := dns.epLister.List() for _, o := range os { diff --git a/plugin/kubernetes/handler_test.go b/plugin/kubernetes/handler_test.go index 388428e3a..77edae236 100644 --- a/plugin/kubernetes/handler_test.go +++ b/plugin/kubernetes/handler_test.go @@ -191,8 +191,10 @@ func TestServeDNS(t *testing.T) { type APIConnServeTest struct{} -func (APIConnServeTest) Run() { return } -func (APIConnServeTest) Stop() error { return nil } +func (APIConnServeTest) Run() { return } +func (APIConnServeTest) Stop() error { return nil } +func (APIConnServeTest) EpIndexReverse(string) []*api.Endpoints { return nil } +func (APIConnServeTest) SvcIndexReverse(string) []*api.Service { return nil } func (APIConnServeTest) PodIndex(string) []*api.Pod { a := []*api.Pod{{ @@ -206,6 +208,49 @@ func (APIConnServeTest) PodIndex(string) []*api.Pod { return a } +func (APIConnServeTest) SvcIndex(string) []*api.Service { + svcs := []*api.Service{ + { + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: "10.0.0.1", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: api.ClusterIPNone, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "external", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ExternalName: "ext.interwebs.test", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + } + return svcs +} + func (APIConnServeTest) ServiceList() []*api.Service { svcs := []*api.Service{ { @@ -249,6 +294,93 @@ func (APIConnServeTest) ServiceList() []*api.Service { return svcs } +func (APIConnServeTest) EpIndex(string) []*api.Endpoints { + n := "test.node.foo.bar" + + eps := []*api.Endpoints{ + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.1", + Hostname: "ep1a", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.2", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.3", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "10.9.8.7", + NodeName: &n, + }, + }, + }, + }, + }, + } + return eps +} + func (APIConnServeTest) EndpointsList() []*api.Endpoints { n := "test.node.foo.bar" diff --git a/plugin/kubernetes/kubernetes.go b/plugin/kubernetes/kubernetes.go index d8173769c..f51ee8e3b 100644 --- a/plugin/kubernetes/kubernetes.go +++ b/plugin/kubernetes/kubernetes.go @@ -331,8 +331,18 @@ func (k *Kubernetes) findPods(r recordRequest, zone string) (pods []msg.Service, func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.Service, err error) { zonePath := msg.Path(zone, "coredns") err = errNoItems // Set to errNoItems to signal really nothing found, gets reset when name is matched. - - for _, svc := range k.APIConn.ServiceList() { + var ( + endpointsList []*api.Endpoints + serviceList []*api.Service + idx string + ) + if wildcard(r.service) || wildcard(r.namespace) { + serviceList = k.APIConn.ServiceList() + } else { + idx = r.service + "." + r.namespace + serviceList = k.APIConn.SvcIndex(idx) + } + for _, svc := range serviceList { if !(match(r.namespace, svc.Namespace) && match(r.service, svc.Name)) { continue @@ -346,8 +356,13 @@ func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg. // Endpoint query or headless service if svc.Spec.ClusterIP == api.ClusterIPNone || r.endpoint != "" { - - for _, ep := range k.APIConn.EndpointsList() { + if wildcard(r.service) || wildcard(r.namespace) { + endpointsList = k.APIConn.EndpointsList() + } else { + idx = r.service + "." + r.namespace + endpointsList = k.APIConn.EpIndex(idx) + } + for _, ep := range endpointsList { if ep.ObjectMeta.Name != svc.Name || ep.ObjectMeta.Namespace != svc.Namespace { continue } diff --git a/plugin/kubernetes/kubernetes_test.go b/plugin/kubernetes/kubernetes_test.go index 4a3bb3d07..fecc7c396 100644 --- a/plugin/kubernetes/kubernetes_test.go +++ b/plugin/kubernetes/kubernetes_test.go @@ -51,9 +51,54 @@ func TestEndpointHostname(t *testing.T) { type APIConnServiceTest struct{} -func (APIConnServiceTest) Run() { return } -func (APIConnServiceTest) Stop() error { return nil } -func (APIConnServiceTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnServiceTest) Run() { return } +func (APIConnServiceTest) Stop() error { return nil } +func (APIConnServiceTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnServiceTest) SvcIndexReverse(string) []*api.Service { return nil } +func (APIConnServiceTest) EpIndexReverse(string) []*api.Endpoints { return nil } + +func (APIConnServiceTest) SvcIndex(string) []*api.Service { + svcs := []*api.Service{ + { + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: "10.0.0.1", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ClusterIP: api.ClusterIPNone, + }, + }, + { + ObjectMeta: meta.ObjectMeta{ + Name: "external", + Namespace: "testns", + }, + Spec: api.ServiceSpec{ + ExternalName: "coredns.io", + Ports: []api.ServicePort{{ + Name: "http", + Protocol: "tcp", + Port: 80, + }}, + }, + }, + } + return svcs +} func (APIConnServiceTest) ServiceList() []*api.Service { svcs := []*api.Service{ @@ -98,6 +143,93 @@ func (APIConnServiceTest) ServiceList() []*api.Service { return svcs } +func (APIConnServiceTest) EpIndex(string) []*api.Endpoints { + n := "test.node.foo.bar" + + eps := []*api.Endpoints{ + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.1", + Hostname: "ep1a", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "svc1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.2", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "172.0.0.3", + }, + }, + Ports: []api.EndpointPort{ + { + Port: 80, + Protocol: "tcp", + Name: "http", + }, + }, + }, + }, + ObjectMeta: meta.ObjectMeta{ + Name: "hdls1", + Namespace: "testns", + }, + }, + { + Subsets: []api.EndpointSubset{ + { + Addresses: []api.EndpointAddress{ + { + IP: "10.9.8.7", + NodeName: &n, + }, + }, + }, + }, + }, + } + return eps +} + func (APIConnServiceTest) EndpointsList() []*api.Endpoints { n := "test.node.foo.bar" diff --git a/plugin/kubernetes/local.go b/plugin/kubernetes/local.go index ef3651a91..961eb9410 100644 --- a/plugin/kubernetes/local.go +++ b/plugin/kubernetes/local.go @@ -28,7 +28,7 @@ func (k *Kubernetes) localNodeName() string { } // Find endpoint matching localIP - for _, ep := range k.APIConn.EndpointsList() { + for _, ep := range k.APIConn.EpIndexReverse(localIP.String()) { for _, eps := range ep.Subsets { for _, addr := range eps.Addresses { if localIP.Equal(net.ParseIP(addr.IP)) { diff --git a/plugin/kubernetes/ns.go b/plugin/kubernetes/ns.go index c4b05d5b5..16fa36994 100644 --- a/plugin/kubernetes/ns.go +++ b/plugin/kubernetes/ns.go @@ -23,7 +23,7 @@ func (k *Kubernetes) nsAddr() *dns.A { rr.A = localIP FindEndpoint: - for _, ep := range k.APIConn.EndpointsList() { + for _, ep := range k.APIConn.EpIndexReverse(localIP.String()) { for _, eps := range ep.Subsets { for _, addr := range eps.Addresses { if localIP.Equal(net.ParseIP(addr.IP)) { diff --git a/plugin/kubernetes/ns_test.go b/plugin/kubernetes/ns_test.go index 20dfef1ec..f179231a9 100644 --- a/plugin/kubernetes/ns_test.go +++ b/plugin/kubernetes/ns_test.go @@ -9,9 +9,13 @@ import ( type APIConnTest struct{} -func (APIConnTest) Run() { return } -func (APIConnTest) Stop() error { return nil } -func (APIConnTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnTest) Run() { return } +func (APIConnTest) Stop() error { return nil } +func (APIConnTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnTest) SvcIndex(string) []*api.Service { return nil } +func (APIConnTest) SvcIndexReverse(string) []*api.Service { return nil } +func (APIConnTest) EpIndex(string) []*api.Endpoints { return nil } +func (APIConnTest) EndpointsList() []*api.Endpoints { return nil } func (APIConnTest) ServiceList() []*api.Service { svcs := []*api.Service{ @@ -28,7 +32,7 @@ func (APIConnTest) ServiceList() []*api.Service { return svcs } -func (APIConnTest) EndpointsList() []*api.Endpoints { +func (APIConnTest) EpIndexReverse(string) []*api.Endpoints { eps := []*api.Endpoints{ { Subsets: []api.EndpointSubset{ diff --git a/plugin/kubernetes/reverse.go b/plugin/kubernetes/reverse.go index 84baae964..a67f59a5f 100644 --- a/plugin/kubernetes/reverse.go +++ b/plugin/kubernetes/reverse.go @@ -35,7 +35,7 @@ func (k *Kubernetes) serviceRecordForIP(ip, name string) []msg.Service { } } // If no cluster ips match, search endpoints - for _, ep := range k.APIConn.EndpointsList() { + for _, ep := range k.APIConn.EpIndexReverse(ip) { if (len(k.Namespaces) > 0) && !k.namespaceExposed(ep.ObjectMeta.Namespace) { continue } diff --git a/plugin/kubernetes/reverse_test.go b/plugin/kubernetes/reverse_test.go index ca3d5a8b9..f9102f244 100644 --- a/plugin/kubernetes/reverse_test.go +++ b/plugin/kubernetes/reverse_test.go @@ -14,9 +14,13 @@ import ( type APIConnReverseTest struct{} -func (APIConnReverseTest) Run() { return } -func (APIConnReverseTest) Stop() error { return nil } -func (APIConnReverseTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnReverseTest) Run() { return } +func (APIConnReverseTest) Stop() error { return nil } +func (APIConnReverseTest) PodIndex(string) []*api.Pod { return nil } +func (APIConnReverseTest) SvcIndex(string) []*api.Service { return nil } +func (APIConnReverseTest) SvcIndexReverse(string) []*api.Service { return nil } +func (APIConnReverseTest) EpIndex(string) []*api.Endpoints { return nil } +func (APIConnReverseTest) EndpointsList() []*api.Endpoints { return nil } func (APIConnReverseTest) ServiceList() []*api.Service { svcs := []*api.Service{ @@ -38,7 +42,7 @@ func (APIConnReverseTest) ServiceList() []*api.Service { return svcs } -func (APIConnReverseTest) EndpointsList() []*api.Endpoints { +func (APIConnReverseTest) EpIndexReverse(string) []*api.Endpoints { eps := []*api.Endpoints{ { Subsets: []api.EndpointSubset{