Plugin/Kubernetes: Service and Endpoint Indexing (#1149)

* indexing

* corrections

plugin/federation/kubernetes_api_test.go

@@ -11,6 +11,8 @@ type APIConnFederationTest struct{}
 
 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{
 		{

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 {

plugin/kubernetes/handler_test.go

@@ -193,6 +193,8 @@ type APIConnServeTest struct{}
 
 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"
 

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
 				}

plugin/kubernetes/kubernetes_test.go

@@ -54,6 +54,51 @@ type APIConnServiceTest struct{}
 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"
 

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)) {

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)) {

plugin/kubernetes/ns_test.go

@@ -12,6 +12,10 @@ type APIConnTest struct{}
 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{

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
 		}

plugin/kubernetes/reverse_test.go

@@ -17,6 +17,10 @@ type APIConnReverseTest struct{}
 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{