TrueCloudLab/coredns
16
0
Fork 2
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

middleware/kubernetes: Implement current federation beta (#723)

Browse source 
* federation initial commit

* UTs/bugfixes

* federation bits

* polish, cover UT gaps

* add TODO

* go fmt & todo note

* remove unrelated change

* pr changes

* start node watcher

* get real node name

* remove unused case
This commit is contained in:
Chris O'Haver 2017-06-14 09:38:00 -04:00 • committed by John Belamaric
parent 8e86fa6f23
commit 930c54ef62
9 changed files with 420 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

6
middleware/kubernetes/README.md
View file

@ -115,6 +115,12 @@ kubernetes coredns.local {
# a path to a file structured like resolv.conf.
upstream 12.34.56.78:53
# federation <federation-name> <federation-domain>
#
# Defines federation membership. One line for each federation membership.
# Each line consists of the name of the federation, and the domain.
federation myfed foo.example.com
# fallthrough
#
# If a query for a record in the cluster zone results in NXDOMAIN,

25
middleware/kubernetes/controller.go
View file

@ -39,6 +39,9 @@ type dnsController interface {
ServiceList() []*api.Service
PodIndex(string) []interface{}
EndpointsList() api.EndpointsList
GetNodeByName(string) (api.Node, error)
Run()
Stop() error
}
@ -52,6 +55,7 @@ type dnsControl struct {
podController *cache.Controller
nsController *cache.Controller
epController *cache.Controller
nodeController *cache.Controller
svcLister cache.StoreToServiceLister
podLister cache.StoreToPodLister
@ -66,8 +70,12 @@ type dnsControl struct {
stopCh chan struct{}
}
type dnsControlOpts struct {
initPodCache bool
}
// newDNSController creates a controller for CoreDNS.
func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Duration, lselector *labels.Selector, initPodCache bool) *dnsControl {
func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Duration, lselector *labels.Selector, opts dnsControlOpts) *dnsControl {
dns := dnsControl{
client: kubeClient,
selector: lselector,
@ -84,7 +92,7 @@ func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Durati
cache.ResourceEventHandlerFuncs{},
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
if initPodCache {
if opts.initPodCache {
dns.podLister.Indexer, dns.podController = cache.NewIndexerInformer(
&cache.ListWatch{
ListFunc: podListFunc(dns.client, namespace, dns.selector),
@ -368,3 +376,16 @@ func (dns *dnsControl) EndpointsList() api.EndpointsList {
return epl
}
func (dns *dnsControl) GetNodeByName(name string) (api.Node, error) {
v1node, err := dns.client.Core().Nodes().Get(name)
if err != nil {
return api.Node{}, err
}
var apinode api.Node
err = v1.Convert_v1_Node_To_api_Node(v1node, &apinode, nil)
if err != nil {
return api.Node{}, err
}
return apinode, nil
}

102
middleware/kubernetes/federation.go Normal file
View file

@ -0,0 +1,102 @@
package kubernetes
import (
"net"
"strings"
"github.com/coredns/coredns/middleware/etcd/msg"
)
type Federation struct {
name string
zone string
}
var localNodeName string
var federationZone string
var federationRegion string
const (
// TODO: Do not hardcode these labels. Pull them out of the API instead.
//
// We can get them via ....
// import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
// metav1.LabelZoneFailureDomain
// metav1.LabelZoneRegion
//
// But importing above breaks coredns with flag collision of 'log_dir'
LabelAvailabilityZone = "failure-domain.beta.kubernetes.io/zone"
LabelRegion = "failure-domain.beta.kubernetes.io/region"
)
// stripFederation removes the federation segment from the segment list, if it
// matches a configured federation name.
func (k *Kubernetes) stripFederation(segs []string) (string, []string) {
if len(segs) < 3 {
return "", segs
}
for _, f := range k.Federations {
if f.name == segs[len(segs)-2] {
fed := segs[len(segs)-2]
segs[len(segs)-2] = segs[len(segs)-1]
segs = segs[:len(segs)-1]
return fed, segs
}
}
return "", segs
}
// federationCNAMERecord returns a service record for the requested federated service
// with the target host in the federated CNAME format which the external DNS provider
// should be able to resolve
func (k *Kubernetes) federationCNAMERecord(r recordRequest) msg.Service {
myNodeName := k.localNodeName()
node, err := k.APIConn.GetNodeByName(myNodeName)
if err != nil {
return msg.Service{}
}
for _, f := range k.Federations {
if f.name != r.federation {
continue
}
if r.endpoint == "" {
return msg.Service{
Key: strings.Join([]string{msg.Path(r.zone, "coredns"), r.typeName, r.federation, r.namespace, r.service}, "/"),
Host: strings.Join([]string{r.service, r.namespace, r.federation, r.typeName, node.Labels[LabelAvailabilityZone], node.Labels[LabelRegion], f.zone}, "."),
}
}
return msg.Service{
Key: strings.Join([]string{msg.Path(r.zone, "coredns"), r.typeName, r.federation, r.namespace, r.service, r.endpoint}, "/"),
Host: strings.Join([]string{r.endpoint, r.service, r.namespace, r.federation, r.typeName, node.Labels[LabelAvailabilityZone], node.Labels[LabelRegion], f.zone}, "."),
}
}
return msg.Service{}
}
func (k *Kubernetes) localNodeName() string {
if localNodeName != "" {
return localNodeName
}
localIP := k.localPodIP()
if localIP == nil {
return ""
}
// Find endpoint matching localIP
endpointsList := k.APIConn.EndpointsList()
for _, ep := range endpointsList.Items {
for _, eps := range ep.Subsets {
for _, addr := range eps.Addresses {
if localIP.Equal(net.ParseIP(addr.IP)) {
localNodeName = *addr.NodeName
return localNodeName
}
}
}
}
return ""
}

104
middleware/kubernetes/federation_test.go Normal file
View file

@ -0,0 +1,104 @@
package kubernetes
import (
"net"
"strings"
"testing"
"github.com/coredns/coredns/middleware/etcd/msg"
"github.com/miekg/dns"
"k8s.io/client-go/1.5/pkg/api"
)
func testStripFederation(t *testing.T, k Kubernetes, input []string, expectedFed string, expectedSegs string) {
fed, segs := k.stripFederation(input)
if expectedSegs != strings.Join(segs, ".") {
t.Errorf("For '%v', expected segs result '%v'. Instead got result '%v'.", strings.Join(input, "."), expectedSegs, strings.Join(segs, "."))
}
if expectedFed != fed {
t.Errorf("For '%v', expected fed result '%v'. Instead got result '%v'.", strings.Join(input, "."), expectedFed, fed)
}
}
func TestStripFederation(t *testing.T) {
k := Kubernetes{Zones: []string{"inter.webs.test"}}
k.Federations = []Federation{{name: "fed", zone: "era.tion.com"}}
testStripFederation(t, k, []string{"service", "ns", "fed", "svc"}, "fed", "service.ns.svc")
testStripFederation(t, k, []string{"service", "ns", "foo", "svc"}, "", "service.ns.foo.svc")
testStripFederation(t, k, []string{"foo", "bar"}, "", "foo.bar")
}
type apiConnFedTest struct{}
func (apiConnFedTest) Run() { return }
func (apiConnFedTest) Stop() error { return nil }
func (apiConnFedTest) ServiceList() []*api.Service { return []*api.Service{} }
func (apiConnFedTest) PodIndex(string) []interface{} { return nil }
func (apiConnFedTest) EndpointsList() api.EndpointsList {
n := "test.node.foo.bar"
return api.EndpointsList{
Items: []api.Endpoints{
{
Subsets: []api.EndpointSubset{
{
Addresses: []api.EndpointAddress{
{
IP: "10.9.8.7",
NodeName: &n,
},
},
},
},
},
},
}
}
func (apiConnFedTest) GetNodeByName(name string) (api.Node, error) {
if name != "test.node.foo.bar" {
return api.Node{}, nil
}
return api.Node{
ObjectMeta: api.ObjectMeta{
Name: "test.node.foo.bar",
Labels: map[string]string{
LabelRegion: "fd-r",
LabelAvailabilityZone: "fd-az",
},
},
}, nil
}
func testFederationCNAMERecord(t *testing.T, k Kubernetes, input recordRequest, expected msg.Service) {
svc := k.federationCNAMERecord(input)
if expected.Host != svc.Host {
t.Errorf("For '%v', expected Host result '%v'. Instead got result '%v'.", input, expected.Host, svc.Host)
}
if expected.Key != svc.Key {
t.Errorf("For '%v', expected Key result '%v'. Instead got result '%v'.", input, expected.Key, svc.Key)
}
}
func TestFederationCNAMERecord(t *testing.T) {
k := Kubernetes{Zones: []string{"inter.webs"}}
k.Federations = []Federation{{name: "fed", zone: "era.tion.com"}}
k.APIConn = apiConnFedTest{}
var r recordRequest
r, _ = k.parseRequest("s1.ns.fed.svc.inter.webs", dns.TypeA)
localPodIP = net.ParseIP("10.9.8.7")
testFederationCNAMERecord(t, k, r, msg.Service{Key: "/coredns/webs/inter/svc/fed/ns/s1", Host: "s1.ns.fed.svc.fd-az.fd-r.era.tion.com"})
r, _ = k.parseRequest("ep1.s1.ns.fed.svc.inter.webs", dns.TypeA)
testFederationCNAMERecord(t, k, r, msg.Service{Key: "/coredns/webs/inter/svc/fed/ns/s1/ep1", Host: "ep1.s1.ns.fed.svc.fd-az.fd-r.era.tion.com"})
r, _ = k.parseRequest("ep1.s1.ns.foo.svc.inter.webs", dns.TypeA)
testFederationCNAMERecord(t, k, r, msg.Service{Key: "", Host: ""})
}

77
middleware/kubernetes/kubernetes.go
View file

@ -39,6 +39,7 @@ type Kubernetes struct {
APIConn dnsController
ResyncPeriod time.Duration
Namespaces []string
Federations []Federation
LabelSelector *unversionedapi.LabelSelector
Selector *labels.Selector
PodMode string
@ -78,9 +79,18 @@ type pod struct {
}
type recordRequest struct {
port, protocol, endpoint, service, namespace, typeName, zone string
port string
protocol string
endpoint string
service string
namespace string
typeName string
zone string
federation string
}
var localPodIP net.IP
var errNoItems = errors.New("no items found")
var errNsNotExposed = errors.New("namespace is not exposed")
var errInvalidRequest = errors.New("invalid query name")
@ -236,16 +246,20 @@ func (k *Kubernetes) InitKubeCache() (err error) {
log.Printf("[INFO] Kubernetes middleware configured with the label selector '%s'. Only kubernetes objects matching this label selector will be exposed.", unversionedapi.FormatLabelSelector(k.LabelSelector))
}
k.APIConn = newdnsController(kubeClient, k.ResyncPeriod, k.Selector, k.PodMode == PodModeVerified)
opts := dnsControlOpts{
initPodCache: k.PodMode == PodModeVerified,
}
k.APIConn = newdnsController(kubeClient, k.ResyncPeriod, k.Selector, opts)
return err
}
func (k *Kubernetes) parseRequest(lowerCasedName string, qtype uint16) (r recordRequest, err error) {
// 3 Possible cases
// SRV Request: _port._protocol.service.namespace.type.zone
// A Request (endpoint): endpoint.service.namespace.type.zone
// A Request (service): service.namespace.type.zone
// SRV Request: _port._protocol.service.namespace.[federation.]type.zone
// A Request (endpoint): endpoint.service.namespace.[federation.]type.zone
// A Request (service): service.namespace.[federation.]type.zone
// separate zone from rest of lowerCasedName
var segs []string
for _, z := range k.Zones {
@ -261,6 +275,8 @@ func (k *Kubernetes) parseRequest(lowerCasedName string, qtype uint16) (r record
return r, errZoneNotFound
}
r.federation, segs = k.stripFederation(segs)
if qtype == dns.TypeNS {
return r, nil
}
@ -339,11 +355,17 @@ func (k *Kubernetes) Records(r recordRequest) ([]msg.Service, error) {
return nil, err
}
if len(services) == 0 && len(pods) == 0 {
// Did not find item in k8s
// Did not find item in k8s, try federated
if r.federation != "" {
fedCNAME := k.federationCNAMERecord(r)
if fedCNAME.Key != "" {
return []msg.Service{fedCNAME}, nil
}
}
return nil, errNoItems
}
records := k.getRecordsForK8sItems(services, pods, r.zone)
records := k.getRecordsForK8sItems(services, pods, r)
return records, nil
}
@ -360,27 +382,37 @@ func endpointHostname(addr api.EndpointAddress) string {
return ""
}
func (k *Kubernetes) getRecordsForK8sItems(services []service, pods []pod, zone string) (records []msg.Service) {
zonePath := msg.Path(zone, "coredns")
func (k *Kubernetes) getRecordsForK8sItems(services []service, pods []pod, r recordRequest) (records []msg.Service) {
zonePath := msg.Path(r.zone, "coredns")
for _, svc := range services {
if svc.addr == api.ClusterIPNone {
// This is a headless service, create records for each endpoint
for _, ep := range svc.endpoints {
s := msg.Service{
Key: strings.Join([]string{zonePath, "svc", svc.namespace, svc.name, endpointHostname(ep.addr)}, "/"),
Host: ep.addr.IP,
Port: int(ep.port.Port),
}
if r.federation != "" {
s.Key = strings.Join([]string{zonePath, "svc", r.federation, svc.namespace, svc.name, endpointHostname(ep.addr)}, "/")
} else {
s.Key = strings.Join([]string{zonePath, "svc", svc.namespace, svc.name, endpointHostname(ep.addr)}, "/")
}
records = append(records, s)
}
} else {
// Create records for each exposed port...
for _, p := range svc.ports {
s := msg.Service{
Key: strings.Join([]string{zonePath, "svc", svc.namespace, svc.name}, "/"),
Host: svc.addr,
Port: int(p.Port)}
if r.federation != "" {
s.Key = strings.Join([]string{zonePath, "svc", r.federation, svc.namespace, svc.name}, "/")
} else {
s.Key = strings.Join([]string{zonePath, "svc", svc.namespace, svc.name}, "/")
}
records = append(records, s)
}
// If the addr is not an IP (i.e. an external service), add the record ...
@ -388,6 +420,11 @@ func (k *Kubernetes) getRecordsForK8sItems(services []service, pods []pod, zone
Key: strings.Join([]string{zonePath, "svc", svc.namespace, svc.name}, "/"),
Host: svc.addr}
if t, _ := s.HostType(); t == dns.TypeCNAME {
if r.federation != "" {
s.Key = strings.Join([]string{zonePath, "svc", r.federation, svc.namespace, svc.name}, "/")
} else {
s.Key = strings.Join([]string{zonePath, "svc", svc.namespace, svc.name}, "/")
}
records = append(records, s)
}
@ -575,3 +612,21 @@ func (k *Kubernetes) getServiceRecordForIP(ip, name string) []msg.Service {
func symbolContainsWildcard(symbol string) bool {
return (symbol == "*" || symbol == "any")
}
func (k *Kubernetes) localPodIP() net.IP {
if localPodIP != nil {
return localPodIP
}
addrs, _ := k.interfaceAddrs.interfaceAddrs()
for _, addr := range addrs {
ip, _, _ := net.ParseCIDR(addr.String())
ip = ip.To4()
if ip == nil || ip.IsLoopback() {
continue
}
localPodIP = ip
return localPodIP
}
return nil
}

30
middleware/kubernetes/kubernetes_test.go
View file

@ -339,6 +339,8 @@ func (APIConnServiceTest) PodIndex(string) []interface{} {
}
func (APIConnServiceTest) EndpointsList() api.EndpointsList {
n := "test.node.foo.bar"
return api.EndpointsList{
Items: []api.Endpoints{
{
@ -407,13 +409,37 @@ func (APIConnServiceTest) EndpointsList() api.EndpointsList {
Namespace: "testns",
},
},
{
Subsets: []api.EndpointSubset{
{
Addresses: []api.EndpointAddress{
{
IP: "10.9.8.7",
NodeName: &n,
},
},
},
},
},
},
}
}
func (APIConnServiceTest) GetNodeByName(name string) (api.Node, error) {
return api.Node{
ObjectMeta: api.ObjectMeta{
Name: "test.node.foo.bar",
Labels: map[string]string{
LabelRegion: "fd-r",
LabelAvailabilityZone: "fd-az",
},
},
}, nil
}
func TestServices(t *testing.T) {
k := Kubernetes{Zones: []string{"interwebs.test"}}
k.Federations = []Federation{{name: "fed", zone: "era.tion.com"}}
k.APIConn = &APIConnServiceTest{}
type svcAns struct {
@ -432,6 +458,10 @@ func TestServices(t *testing.T) {
// External Services
{qname: "external.testns.svc.interwebs.test.", qtype: dns.TypeCNAME, answer: svcAns{host: "coredns.io", key: "/coredns/test/interwebs/svc/testns/external"}},
// Federated Services
{qname: "svc1.testns.fed.svc.interwebs.test.", qtype: dns.TypeA, answer: svcAns{host: "10.0.0.1", key: "/coredns/test/interwebs/svc/fed/testns/svc1"}},
{qname: "svc0.testns.fed.svc.interwebs.test.", qtype: dns.TypeA, answer: svcAns{host: "svc0.testns.fed.svc.fd-az.fd-r.era.tion.com", key: "/coredns/test/interwebs/svc/fed/testns/svc0"}},
}
for _, test := range tests {

2
middleware/kubernetes/ns_test.go
View file

@ -104,6 +104,8 @@ func (APIConnTest) EndpointsList() api.EndpointsList {
}
}
func (APIConnTest) GetNodeByName(name string) (api.Node, error) { return api.Node{}, nil }
type interfaceAddrsTest struct{}
func (i interfaceAddrsTest) interfaceAddrs() ([]net.Addr, error) {

13
middleware/kubernetes/setup.go
View file

@ -177,6 +177,19 @@ func kubernetesParse(c *caddy.Controller) (*Kubernetes, error) {
return nil, err
}
k8s.Proxy = proxy.NewLookup(ups)
case "federation": // name zone
args := c.RemainingArgs()
if len(args) == 2 {
k8s.Federations = append(k8s.Federations, Federation{
name: args[0],
zone: args[1],
})
continue
} else {
return nil, fmt.Errorf("Incorrect number of arguments for federation. Got %v, expect 2.", len(args))
}
return nil, c.ArgErr()
}
}
return k8s, nil

70
middleware/kubernetes/setup_test.go
View file

@ -29,6 +29,7 @@ func TestKubernetesParse(t *testing.T) {
expectedCidrs []net.IPNet
expectedFallthrough bool
expectedUpstreams []string
expectedFederations []Federation
}{
// positive
{
@ -44,6 +45,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"kubernetes keyword with multiple zones",
@ -58,6 +60,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"kubernetes keyword with zone and empty braces",
@ -73,6 +76,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"endpoint keyword with url",
@ -89,6 +93,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"namespaces keyword with one namespace",
@ -105,6 +110,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
nil,
},
{
"namespaces keyword with multiple namespaces",
@ -121,6 +127,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"resync period in seconds",
@ -137,6 +144,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"resync period in minutes",
@ -153,6 +161,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"basic label selector",
@ -169,6 +178,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"multi-label selector",
@ -185,6 +195,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"fully specified valid config",
@ -205,6 +216,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
true,
nil,
[]Federation{},
},
// negative
{
@ -220,6 +232,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"kubernetes keyword without a zone",
@ -234,6 +247,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"endpoint keyword without an endpoint value",
@ -250,6 +264,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"namespace keyword without a namespace value",
@ -266,6 +281,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"resyncperiod keyword without a duration value",
@ -282,6 +298,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"resync period no units",
@ -298,6 +315,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"resync period invalid",
@ -314,6 +332,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"labels with no selector value",
@ -330,6 +349,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
{
"labels with invalid selector value",
@ -346,6 +366,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// pods disabled
{
@ -363,6 +384,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// pods insecure
{
@ -380,6 +402,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// pods verified
{
@ -397,6 +420,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// pods invalid
{
@ -414,6 +438,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// cidrs ok
{
@ -431,6 +456,7 @@ func TestKubernetesParse(t *testing.T) {
[]net.IPNet{parseCidr("10.0.0.0/24"), parseCidr("10.0.1.0/24")},
false,
nil,
[]Federation{},
},
// cidrs ok
{
@ -448,6 +474,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// fallthrough invalid
{
@ -465,6 +492,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// Valid upstream
{
@ -482,6 +510,7 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
[]string{"13.14.15.16:53"},
[]Federation{},
},
// Invalid upstream
{
@ -499,6 +528,47 @@ func TestKubernetesParse(t *testing.T) {
nil,
false,
nil,
[]Federation{},
},
// Valid federations
{
"valid upstream",
`kubernetes coredns.local {
federation foo bar.crawl.com
federation fed era.tion.com
}`,
false,
"",
1,
0,
defaultResyncPeriod,
"",
defaultPodMode,
nil,
false,
nil,
[]Federation{
{name: "foo", zone: "bar.crawl.com"},
{name: "fed", zone: "era.tion.com"},
},
},
// Invalid federations
{
"valid upstream",
`kubernetes coredns.local {
federation starship
}`,
true,
`Incorrect number of arguments for federation. Got 1, expect 2.`,
-1,
0,
defaultResyncPeriod,
"",
defaultPodMode,
nil,
false,
nil,
[]Federation{},
},
}