Update kubernetai plugin to support transfers via the transfers plugin.

plugin/kubernetai/axfr_test.go

@@ -0,0 +1,145 @@
+package kubernetai
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/coredns/coredns/plugin/transfer"
+
+	"github.com/miekg/dns"
+)
+
+func TestKubernetesTransferNonAuthZone(t *testing.T) {
+	type fields struct {
+		name          string
+		kubernetes    []*mockK8sPlugin
+		zone          string
+		serial        uint32
+		expectedZone  string
+		expectedError error
+	}
+	tests := []fields{
+		{
+			name: "TestSingleKubernetesTransferNonAuthZone",
+			kubernetes: []*mockK8sPlugin{
+				{
+					zones:       []string{"cluster.local"},
+					transferErr: transfer.ErrNotAuthoritative,
+				},
+			},
+			zone:          "example.com",
+			expectedError: transfer.ErrNotAuthoritative,
+		},
+		{
+			name: "TestSingleKubernetesTransferAuthZone",
+			kubernetes: []*mockK8sPlugin{
+				{
+					zones: []string{"cluster.local"},
+					transfer: `
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+cluster.local.	5	IN	NS	ns.dns.cluster.local.
+ns.dns.cluster.local.	5	IN	A	10.0.0.10
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+`,
+					transferErr: nil,
+				},
+			},
+			zone: "cluster.local",
+			expectedZone: `
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+cluster.local.	5	IN	NS	ns.dns.cluster.local.
+ns.dns.cluster.local.	5	IN	A	10.0.0.10
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+`,
+			expectedError: nil,
+		},
+		{
+			name: "TestMultipleNonAuthorititativeSingleAuthoritative",
+			kubernetes: []*mockK8sPlugin{
+				{
+					zones: []string{"fluster.local"},
+					transfer: `
+fluster.local.	5	IN	SOA	ns.dns.fluster.local. hostmaster.fluster.local. 3 7200 1800 86400 5
+fluster.local.	5	IN	NS	ns.dns.fluster.local.
+ns.dns.fluster.local.	5	IN	A	10.0.0.10
+fluster.local.	5	IN	SOA	ns.dns.fluster.local. hostmaster.fluster.local. 3 7200 1800 86400 5
+`,
+					transferErr: transfer.ErrNotAuthoritative,
+				},
+				{
+					zones:       []string{"bluster.local"},
+					transferErr: transfer.ErrNotAuthoritative,
+				},
+				{
+					zones: []string{"cluster.local"},
+					transfer: `
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+cluster.local.	5	IN	NS	ns.dns.cluster.local.
+ns.dns.cluster.local.	5	IN	A	10.0.0.10
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+`,
+					transferErr: nil,
+				},
+				{
+					zones:       []string{"muster.local"},
+					transferErr: transfer.ErrNotAuthoritative,
+				},
+			},
+			zone: "cluster.local",
+			expectedZone: `
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+cluster.local.	5	IN	NS	ns.dns.cluster.local.
+ns.dns.cluster.local.	5	IN	A	10.0.0.10
+cluster.local.	5	IN	SOA	ns.dns.cluster.local. hostmaster.cluster.local. 3 7200 1800 86400 5
+`,
+			expectedError: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// create kubernetai with mock kubernetes plugins
+			kai := Kubernetai{}
+			for _, plug := range tt.kubernetes {
+				kai.Kubernetes = append(kai.Kubernetes, plug)
+			}
+
+			// create a axfr test message with test zone
+			dnsmsg := &dns.Msg{}
+			dnsmsg.SetAxfr(tt.zone)
+
+			// perform AXFR
+			ch, err := kai.Transfer(tt.zone, tt.serial)
+			if err != nil {
+				if err != tt.expectedError {
+					t.Errorf("expected error %+v but received %+v", tt.expectedError, err)
+				}
+				return
+			}
+			validateAXFR(t, ch, tt.expectedZone)
+		})
+	}
+}
+
+func validateAXFR(t *testing.T, ch <-chan []dns.RR, expectedZone string) {
+	xfr := []dns.RR{}
+	for rrs := range ch {
+		xfr = append(xfr, rrs...)
+	}
+	if xfr[0].Header().Rrtype != dns.TypeSOA {
+		t.Error("Invalid transfer response, does not start with SOA record")
+	}
+
+	zp := dns.NewZoneParser(strings.NewReader(expectedZone), "", "")
+	i := 0
+	for rr, ok := zp.Next(); ok; rr, ok = zp.Next() {
+		if !dns.IsDuplicate(rr, xfr[i]) {
+			t.Fatalf("Record %d, expected\n%v\n, got\n%v", i, rr, xfr[i])
+		}
+		i++
+	}
+
+	if err := zp.Err(); err != nil {
+		t.Fatal(err)
+	}
+}

plugin/kubernetai/kubernetai.go

@@ -1,33 +1,76 @@
+// Package kubernetai implements a plugin which can embed a number of kubernetes plugins in the same dns server.
 package kubernetai
 
 import (
 	"context"
 
 	"github.com/coredns/coredns/plugin"
 	"github.com/coredns/coredns/plugin/kubernetes"
+	"github.com/coredns/coredns/plugin/kubernetes/object"
 	clog "github.com/coredns/coredns/plugin/pkg/log"
+	"github.com/coredns/coredns/plugin/transfer"
 	"github.com/coredns/coredns/request"
 	"github.com/miekg/dns"
 )
 
 var log = clog.NewWithPlugin("kubernetai")
 
+// embeddedKubernetesPluginInterface describes the kubernetes plugin interface that kubernetai requires/uses.
+type embeddedKubernetesPluginInterface interface {
+	plugin.Handler
+	transfer.Transferer
+	PodWithIP(ip string) (pod *object.Pod)
+	Zones() (zones plugin.Zones)
+}
+
+// embeddedKubernetes wraps a real kubernetes plugin
+type embeddedKubernetes struct {
+	*kubernetes.Kubernetes
+}
+
+var _ embeddedKubernetesPluginInterface = &embeddedKubernetes{}
+
+func newEmbeddedKubernetes(k *kubernetes.Kubernetes) *embeddedKubernetes {
+	return &embeddedKubernetes{
+		Kubernetes: k,
+	}
+}
+
+// PodWithIP satisfies the embeddedKubernetesPluginInterface by adding this additional method not exported from the kubernetes plugin.
+func (ek embeddedKubernetes) PodWithIP(ip string) *object.Pod {
+	if ek.Kubernetes == nil {
+		return nil
+	}
+	ps := ek.Kubernetes.APIConn.PodIndex(ip)
+	if len(ps) == 0 {
+		return nil
+	}
+	return ps[0]
+}
+
+// Zones satisfies the embeddedKubernetesPluginInterface by providing access to the kubernetes plugin Zones.
+func (ek embeddedKubernetes) Zones() plugin.Zones {
+	if ek.Kubernetes == nil {
+		return nil
+	}
+	return plugin.Zones(ek.Kubernetes.Zones)
+}
+
 // Kubernetai handles multiple Kubernetes
 type Kubernetai struct {
 	Zones          []string
-	Kubernetes     []*kubernetes.Kubernetes
+	Kubernetes     []embeddedKubernetesPluginInterface
 	autoPathSearch []string // Local search path from /etc/resolv.conf. Needed for autopath.
-	p              podHandlerItf
 }
 
 // New creates a Kubernetai containing one Kubernetes with zones
 func New(zones []string) (Kubernetai, *kubernetes.Kubernetes) {
 	h := Kubernetai{
 		autoPathSearch: searchFromResolvConf(),
-		p:              &podHandler{},
 	}
 	k := kubernetes.New(zones)
-	h.Kubernetes = append(h.Kubernetes, k)
+	ek := newEmbeddedKubernetes(k)
+	h.Kubernetes = append(h.Kubernetes, ek)
 	return h, k
 }
 
@@ -43,7 +86,7 @@ func (k8i Kubernetai) AutoPath(state request.Request) []string {
 	// Abort if zone is not in kubernetai stanza.
 	var zMatch bool
 	for _, k8s := range k8i.Kubernetes {
-		zone := plugin.Zones(k8s.Zones).Matches(state.Name())
+		zone := k8s.Zones().Matches(state.Name())
 		if zone != "" {
 			zMatch = true
 			break
@@ -55,13 +98,13 @@ func (k8i Kubernetai) AutoPath(state request.Request) []string {
 
 	// Add autopath result for the handled zones
 	for _, k := range k8i.Kubernetes {
-		pod := k8i.p.PodWithIP(*k, state.IP())
+		pod := k.PodWithIP(state.IP())
 		if pod == nil {
 			return nil
 		}
 
 		search := make([]string, 3)
-		for _, z := range k.Zones {
+		for _, z := range k.Zones() {
 			if z == "." {
 				search[0] = pod.Namespace + ".svc."
 				search[1] = "svc."
@@ -80,6 +123,20 @@ func (k8i Kubernetai) AutoPath(state request.Request) []string {
 	return searchPath
 }
 
+// Transfer supports the transfer plugin, implementing the Transferer interface, by calling Transfer on each of the embedded plugins.
+// It will return a channel to the FIRST kubernetai stanza that reports that it is authoritative for the requested zone.
+func (k8i Kubernetai) Transfer(zone string, serial uint32) (retCh <-chan []dns.RR, err error) {
+	for _, k := range k8i.Kubernetes {
+		retCh, err = k.Transfer(zone, serial)
+		if err == transfer.ErrNotAuthoritative {
+			continue
+		}
+		return
+	}
+	// none of the embedded plugins were authoritative
+	return nil, transfer.ErrNotAuthoritative
+}
+
 func searchFromResolvConf() []string {
 	rc, err := dns.ClientConfigFromFile("/etc/resolv.conf")
 	if err != nil {
@@ -93,7 +150,7 @@ func searchFromResolvConf() []string {
 func (k8i Kubernetai) Health() bool {
 	healthy := true
 	for _, k := range k8i.Kubernetes {
-		healthy = healthy && k.APIConn.HasSynced()
+		healthy = healthy && k.(*embeddedKubernetes).APIConn.HasSynced()
 		if !healthy {
 			break
 		}