plugin/secondary: serve catalog member zones (#8230)

Signed-off-by: houyuwushang <liuluoqianqiu@outlook.com>

plugin/file/file.go

@@ -24,11 +24,15 @@ type (
 		Next plugin.Handler
 		Zones
 		Xfer *transfer.Transfer
+		ZoneLookupFunc
 		TransferInFunc
 
 		Fall fall.F
 	}
 
+	// ZoneLookupFunc looks up the authoritative zone for qname.
+	ZoneLookupFunc func(qname string) (zone string, z *Zone, ok bool)
+
 	// Zones maps zone names to a *Zone.
 	Zones struct {
 		Z     map[string]*Zone // A map mapping zone (origin) to the Zone's data
@@ -41,9 +45,8 @@ func (f File) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (i
 	state := request.Request{W: w, Req: r}
 
 	qname := state.Name()
-	// TODO(miek): match the qname better in the map
+	zone, z, ok := f.lookupZone(qname)
-	zone := plugin.Zones(f.Zones.Names).Matches(qname)
+	if !ok {
-	if zone == "" {
 		// If no next plugin is configured, it's more correct to return REFUSED as file acts as an authoritative server
 		if f.Next == nil {
 			return dns.RcodeRefused, nil
@@ -51,8 +54,7 @@ func (f File) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (i
 		return plugin.NextOrFailure(f.Name(), f.Next, ctx, w, r)
 	}
 
-	z, ok := f.Z[zone]
+	if z == nil {
-	if !ok || z == nil {
 		return dns.RcodeServerFailure, nil
 	}
 
@@ -134,6 +136,22 @@ func (f File) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (i
 // Name implements the Handler interface.
 func (f File) Name() string { return "file" }
 
+func (f File) lookupZone(qname string) (string, *Zone, bool) {
+	if f.ZoneLookupFunc != nil {
+		return f.ZoneLookupFunc(qname)
+	}
+	// TODO(miek): match the qname better in the map
+	zone := plugin.Zones(f.Zones.Names).Matches(qname)
+	if zone == "" {
+		return "", nil, false
+	}
+	z, ok := f.Z[zone]
+	if !ok {
+		return zone, nil, true
+	}
+	return zone, z, true
+}
+
 func (f File) transferIn(z *Zone, t *transfer.Transfer) error {
 	if f.TransferInFunc != nil {
 		return f.TransferInFunc(z, t)

plugin/file/xfr.go

@@ -9,8 +9,8 @@ import (
 
 // Transfer implements the transfer.Transfer interface.
 func (f File) Transfer(zone string, serial uint32) (<-chan []dns.RR, error) {
-	z, ok := f.Z[zone]
+	matched, z, ok := f.lookupZone(zone)
-	if !ok || z == nil {
+	if !ok || matched != zone || z == nil {
 		return nil, transfer.ErrNotAuthoritative
 	}
 	return z.Transfer(serial)

plugin/file/xfr_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/coredns/coredns/plugin/pkg/dnstest"
 	"github.com/coredns/coredns/plugin/test"
+	"github.com/coredns/coredns/plugin/transfer"
 
 	"github.com/miekg/dns"
 )
@@ -70,3 +71,15 @@ func TestAXFRWithOutTransferPlugin(t *testing.T) {
 		t.Errorf("Expecting REFUSED, got %d", code)
 	}
 }
+
+func TestTransferRequiresExactZone(t *testing.T) {
+	zone, err := Parse(strings.NewReader(dbMiekNL), testzone, "stdin", 0)
+	if err != nil {
+		t.Fatalf("Expected no error when reading zone, got %q", err)
+	}
+
+	fm := File{Zones: Zones{Z: map[string]*Zone{testzone: zone}, Names: []string{testzone}}}
+	if _, err := fm.Transfer("www."+testzone, 0); err != transfer.ErrNotAuthoritative {
+		t.Fatalf("expected ErrNotAuthoritative for subdomain transfer, got %v", err)
+	}
+}

plugin/secondary/README.md

@@ -36,8 +36,9 @@ secondary [zones...] {
    times; if one does not work, another will be tried. Transferring this zone outwards again can be
    done by enabling the *transfer* plugin.
 
-*  `catalog` treats the transferred zone as an RFC 9432 catalog zone and parses the catalog member
+*  `catalog` treats the transferred zone as an RFC 9432 catalog zone. After each successful catalog
-   list after each successful transfer.
+   transfer, CoreDNS adds and removes the catalog member zones and transfers those member zones from
+   the same primary servers.
 
 *  `fallthrough` If a query for a record in the zone results in NXDOMAIN, the query will be passed
    to the next plugin in the chain. If **[ZONES...]** are listed, then only queries for those zones

plugin/secondary/catalog.go

@@ -3,6 +3,7 @@ package secondary
 import (
 	"github.com/coredns/coredns/plugin/file"
 	"github.com/coredns/coredns/plugin/pkg/catalog"
+	"github.com/coredns/coredns/plugin/pkg/upstream"
 	"github.com/coredns/coredns/plugin/transfer"
 
 	"github.com/miekg/dns"
@@ -34,6 +35,92 @@ func (s *Secondary) transferIn(origin string, z *file.Zone, t *transfer.Transfer
 	}
 	s.catalogs[origin] = parsed
 	s.catalogMu.Unlock()
+
+	s.applyCatalog(origin, parsed, z, t)
 	log.Infof("Parsed catalog zone %s with %d member zones", origin, len(parsed.Members))
 	return nil
 }
+
+type dynamicZoneStart struct {
+	origin   string
+	zone     *file.Zone
+	shutdown chan bool
+}
+
+func (s *Secondary) applyCatalog(origin string, cat *catalog.Catalog, catalogZone *file.Zone, t *transfer.Transfer) {
+	memberZones := make(map[string]struct{}, len(cat.Members))
+	var starts []dynamicZoneStart
+
+	s.zoneMu.Lock()
+	s.ensureZoneStateLocked()
+
+	for _, member := range cat.Members {
+		memberZones[member.Zone] = struct{}{}
+
+		if existing, ok := s.Z[member.Zone]; ok {
+			if dyn, ok := s.dynamicZones[member.Zone]; ok && dyn.catalog == origin && existing != nil {
+				continue
+			}
+			log.Warningf("Skipping catalog member zone %s from %s: zone already exists", member.Zone, origin)
+			continue
+		}
+
+		z := file.NewZone(member.Zone, "stdin")
+		if catalogZone != nil {
+			z.TransferFrom = append([]string(nil), catalogZone.TransferFrom...)
+		}
+		z.Upstream = upstream.New()
+
+		shutdown := make(chan bool)
+		s.Z[member.Zone] = z
+		s.Names = append(s.Names, member.Zone)
+		s.zoneNames[z] = member.Zone
+		s.dynamicZones[member.Zone] = &dynamicZone{catalog: origin, shutdown: shutdown}
+		starts = append(starts, dynamicZoneStart{origin: member.Zone, zone: z, shutdown: shutdown})
+		log.Infof("Added catalog member zone %s from catalog %s", member.Zone, origin)
+	}
+
+	for member := range s.catalogMemberZones[origin] {
+		if _, ok := memberZones[member]; ok {
+			continue
+		}
+		dyn, ok := s.dynamicZones[member]
+		if !ok || dyn.catalog != origin {
+			continue
+		}
+		dyn.stopOnce.Do(func() { close(dyn.shutdown) })
+		delete(s.dynamicZones, member)
+		if z := s.Z[member]; z != nil {
+			delete(s.zoneNames, z)
+		}
+		delete(s.Z, member)
+		s.Names = removeZoneName(s.Names, member)
+		log.Infof("Removed catalog member zone %s from catalog %s", member, origin)
+	}
+	s.catalogMemberZones[origin] = memberZones
+	s.zoneMu.Unlock()
+
+	for _, start := range starts {
+		go s.transferAndUpdate(start.origin, start.zone, t, start.shutdown)
+	}
+}
+
+func (s *Secondary) ensureZoneStateLocked() {
+	if s.Z == nil {
+		s.Z = make(map[string]*file.Zone)
+	}
+	if s.zoneNames == nil {
+		s.zoneNames = make(map[*file.Zone]string, len(s.Z))
+		for name, zone := range s.Z {
+			if zone != nil {
+				s.zoneNames[zone] = name
+			}
+		}
+	}
+	if s.dynamicZones == nil {
+		s.dynamicZones = make(map[string]*dynamicZone)
+	}
+	if s.catalogMemberZones == nil {
+		s.catalogMemberZones = make(map[string]map[string]struct{})
+	}
+}

plugin/secondary/catalog_test.go

@@ -1,29 +1,34 @@
 package secondary
 
 import (
+	"context"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/coredns/coredns/plugin/file"
-	"github.com/coredns/coredns/plugin/pkg/catalog"
 	"github.com/coredns/coredns/plugin/pkg/dnstest"
+	"github.com/coredns/coredns/plugin/pkg/fall"
+	plugintest "github.com/coredns/coredns/plugin/test"
 
 	"github.com/miekg/dns"
 )
 
 func TestTransferInCatalog(t *testing.T) {
 	const origin = "catalog.example."
-	rrs := catalogZoneRecords(t, true)
+	zones := newTestTransferZones(map[string][]dns.RR{
+		origin:         catalogZoneRecords(t, true),
+		"example.org.": memberZoneRecords(t),
+	})
 
-	server := dnstest.NewServer(transferHandler(rrs))
+	server := dnstest.NewServer(zones.handler())
 	defer server.Close()
 
 	z := file.NewZone(origin, "stdin")
 	z.TransferFrom = []string{server.Addr}
-	s := &Secondary{
+	s := newTestSecondary(origin, z, true)
-		catalogs:     make(map[string]*catalog.Catalog),
+	t.Cleanup(s.stopDynamicZones)
-		catalogZones: map[string]struct{}{origin: {}},
-	}
 
 	if err := s.transferIn(origin, z, nil); err != nil {
 		t.Fatalf("transferIn returned error: %v", err)
@@ -47,21 +52,59 @@ func TestTransferInCatalog(t *testing.T) {
 	if z.SOA == nil {
 		t.Fatal("expected zone data to be live after valid catalog transfer")
 	}
+
+	msg := waitForAnswer(t, s, "www.example.org.", dns.TypeA)
+	if len(msg.Answer) != 1 {
+		t.Fatalf("expected 1 member zone answer, got %d", len(msg.Answer))
+	}
+	a, ok := msg.Answer[0].(*dns.A)
+	if !ok {
+		t.Fatalf("expected A answer, got %T", msg.Answer[0])
+	}
+	if a.A.String() != "192.0.2.1" {
+		t.Fatalf("expected 192.0.2.1, got %s", a.A.String())
+	}
+}
+
+func TestTransferInCatalogRemovesMemberZone(t *testing.T) {
+	const origin = "catalog.example."
+	zones := newTestTransferZones(map[string][]dns.RR{
+		origin:         catalogZoneRecords(t, true),
+		"example.org.": memberZoneRecords(t),
+	})
+
+	server := dnstest.NewServer(zones.handler())
+	defer server.Close()
+
+	z := file.NewZone(origin, "stdin")
+	z.TransferFrom = []string{server.Addr}
+	s := newTestSecondary(origin, z, true)
+	t.Cleanup(s.stopDynamicZones)
+
+	if err := s.transferIn(origin, z, nil); err != nil {
+		t.Fatalf("transferIn returned error: %v", err)
+	}
+	waitForAnswer(t, s, "www.example.org.", dns.TypeA)
+
+	zones.set(origin, catalogZoneRecordsWithoutMembers(t))
+	if err := s.transferIn(origin, z, nil); err != nil {
+		t.Fatalf("transferIn returned error: %v", err)
+	}
+	if _, _, ok := s.lookupZone("www.example.org."); ok {
+		t.Fatal("expected member zone to be removed after catalog update")
+	}
 }
 
 func TestTransferInCatalogRejectsInvalidCatalog(t *testing.T) {
 	const origin = "catalog.example."
 	rrs := catalogZoneRecords(t, false)
 
-	server := dnstest.NewServer(transferHandler(rrs))
+	server := dnstest.NewServer(newTestTransferZones(map[string][]dns.RR{origin: rrs}).handler())
 	defer server.Close()
 
 	z := file.NewZone(origin, "stdin")
 	z.TransferFrom = []string{server.Addr}
-	s := &Secondary{
+	s := newTestSecondary(origin, z, true)
-		catalogs:     make(map[string]*catalog.Catalog),
-		catalogZones: map[string]struct{}{origin: {}},
-	}
 
 	err := s.transferIn(origin, z, nil)
 	if err == nil {
@@ -82,14 +125,12 @@ func TestTransferInSkipsCatalogParseForRegularZone(t *testing.T) {
 	const origin = "catalog.example."
 	rrs := catalogZoneRecords(t, false)
 
-	server := dnstest.NewServer(transferHandler(rrs))
+	server := dnstest.NewServer(newTestTransferZones(map[string][]dns.RR{origin: rrs}).handler())
 	defer server.Close()
 
 	z := file.NewZone(origin, "stdin")
 	z.TransferFrom = []string{server.Addr}
-	s := &Secondary{
+	s := newTestSecondary(origin, z, false)
-		catalogs: make(map[string]*catalog.Catalog),
-	}
 
 	if err := s.transferIn(origin, z, nil); err != nil {
 		t.Fatalf("transferIn returned error: %v", err)
@@ -102,11 +143,31 @@ func TestTransferInSkipsCatalogParseForRegularZone(t *testing.T) {
 	}
 }
 
-func transferHandler(rrs []dns.RR) dns.HandlerFunc {
+type testTransferZones struct {
+	mu      sync.RWMutex
+	records map[string][]dns.RR
+}
+
+func newTestTransferZones(records map[string][]dns.RR) *testTransferZones {
+	return &testTransferZones{records: records}
+}
+
+func (z *testTransferZones) set(zone string, rrs []dns.RR) {
+	z.mu.Lock()
+	z.records[zone] = rrs
+	z.mu.Unlock()
+}
+
+func (z *testTransferZones) handler() dns.HandlerFunc {
 	return func(w dns.ResponseWriter, req *dns.Msg) {
 		m := new(dns.Msg)
 		m.SetReply(req)
 		if len(req.Question) > 0 {
+			qname := strings.ToLower(dns.Fqdn(req.Question[0].Name))
+			z.mu.RLock()
+			rrs := append([]dns.RR(nil), z.records[qname]...)
+			z.mu.RUnlock()
+
 			switch req.Question[0].Qtype {
 			case dns.TypeSOA:
 				for _, rr := range rrs {
@@ -123,6 +184,32 @@ func transferHandler(rrs []dns.RR) dns.HandlerFunc {
 	}
 }
 
+func newTestSecondary(origin string, z *file.Zone, catalog bool) *Secondary {
+	catalogZones := map[string]struct{}{}
+	if catalog {
+		catalogZones[origin] = struct{}{}
+	}
+	return newSecondary(file.Zones{Z: map[string]*file.Zone{origin: z}, Names: []string{origin}}, fall.F{}, catalogZones)
+}
+
+func waitForAnswer(t *testing.T, s *Secondary, name string, qtype uint16) *dns.Msg {
+	t.Helper()
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		req := new(dns.Msg)
+		req.SetQuestion(name, qtype)
+		rec := dnstest.NewRecorder(&plugintest.ResponseWriter{})
+		code, err := s.ServeDNS(context.Background(), rec, req)
+		if err == nil && code == dns.RcodeSuccess && rec.Msg != nil && rec.Msg.Rcode == dns.RcodeSuccess && len(rec.Msg.Answer) > 0 {
+			return rec.Msg
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for answer to %s", name)
+	return nil
+}
+
 func catalogZoneRecords(t *testing.T, includeVersion bool) []dns.RR {
 	t.Helper()
 
@@ -142,6 +229,30 @@ func catalogZoneRecords(t *testing.T, includeVersion bool) []dns.RR {
 	return rrs
 }
 
+func catalogZoneRecordsWithoutMembers(t *testing.T) []dns.RR {
+	t.Helper()
+
+	soa := mustRR(t, "catalog.example. 0 IN SOA invalid. hostmaster.invalid. 2 3600 600 604800 0")
+	return []dns.RR{
+		soa,
+		mustRR(t, "catalog.example. 0 IN NS invalid."),
+		mustRR(t, `version.catalog.example. 0 IN TXT "2"`),
+		soa,
+	}
+}
+
+func memberZoneRecords(t *testing.T) []dns.RR {
+	t.Helper()
+
+	soa := mustRR(t, "example.org. 0 IN SOA ns.example.org. hostmaster.example.org. 1 3600 600 604800 0")
+	return []dns.RR{
+		soa,
+		mustRR(t, "example.org. 0 IN NS ns.example.org."),
+		mustRR(t, "www.example.org. 0 IN A 192.0.2.1"),
+		soa,
+	}
+}
+
 func mustRR(t *testing.T, s string) dns.RR {
 	t.Helper()
 

plugin/secondary/secondary.go

@@ -13,9 +13,20 @@ import (
 type Secondary struct {
 	file.File
 
-	catalogMu    sync.RWMutex
+	zoneMu       sync.RWMutex
-	catalogs     map[string]*catalog.Catalog
+	zoneNames    map[*file.Zone]string
-	catalogZones map[string]struct{}
+	dynamicZones map[string]*dynamicZone
+
+	catalogMu          sync.RWMutex
+	catalogs           map[string]*catalog.Catalog
+	catalogZones       map[string]struct{}
+	catalogMemberZones map[string]map[string]struct{}
+}
+
+type dynamicZone struct {
+	catalog  string
+	shutdown chan bool
+	stopOnce sync.Once
 }
 
 // Name implements the Handler interface.

plugin/secondary/setup.go

@@ -26,18 +26,7 @@ func setup(c *caddy.Controller) error {
 		return plugin.Error("secondary", err)
 	}
 
-	s := &Secondary{
+	s := newSecondary(zones, fall, catalogZones)
-		File:         file.File{Zones: zones, Fall: fall},
-		catalogs:     make(map[string]*catalog.Catalog),
-		catalogZones: catalogZones,
-	}
-	zoneNames := make(map[*file.Zone]string, len(zones.Z))
-	for name, zone := range zones.Z {
-		zoneNames[zone] = name
-	}
-	s.TransferInFunc = func(z *file.Zone, t *transfer.Transfer) error {
-		return s.transferIn(zoneNames[z], z, t)
-	}
 	var x *transfer.Transfer
 	c.OnStartup(func() error {
 		t := dnsserver.GetConfig(c).Handler("transfer")
@@ -59,32 +48,7 @@ func setup(c *caddy.Controller) error {
 
 			c.OnStartup(func() error {
 				z.StartupOnce.Do(func() {
-					go func() {
+					go s.transferAndUpdate(n, z, x, updateShutdown)
-						dur := time.Millisecond * 250
-						max := time.Second * 10
-						for {
-							err := s.transferIn(n, z, x)
-							if err == nil {
-								break
-							}
-							log.Warningf("All '%s' masters failed to transfer, retrying in %s: %s", n, dur.String(), err)
-							if waitForTransferRetry(updateShutdown, dur) {
-								return
-							}
-							dur <<= 1 // double the duration
-							if dur > max {
-								dur = max
-							}
-						}
-						select {
-						case <-updateShutdown:
-							return
-						default:
-						}
-						z.UpdateWithTransfer(updateShutdown, x, func(z *file.Zone, t *transfer.Transfer) error {
-							return s.transferIn(n, z, t)
-						})
-					}()
 				})
 				return nil
 			})
@@ -94,6 +58,10 @@ func setup(c *caddy.Controller) error {
 			})
 		}
 	}
+	c.OnShutdown(func() error {
+		s.stopDynamicZones()
+		return nil
+	})
 
 	dnsserver.GetConfig(c).AddPlugin(func(next plugin.Handler) plugin.Handler {
 		s.Next = next
@@ -103,6 +71,52 @@ func setup(c *caddy.Controller) error {
 	return nil
 }
 
+func newSecondary(zones file.Zones, fall fall.F, catalogZones map[string]struct{}) *Secondary {
+	s := &Secondary{
+		File:               file.File{Zones: zones, Fall: fall},
+		zoneNames:          make(map[*file.Zone]string, len(zones.Z)),
+		dynamicZones:       make(map[string]*dynamicZone),
+		catalogs:           make(map[string]*catalog.Catalog),
+		catalogZones:       catalogZones,
+		catalogMemberZones: make(map[string]map[string]struct{}),
+	}
+	for name, zone := range zones.Z {
+		s.zoneNames[zone] = name
+	}
+	s.ZoneLookupFunc = s.lookupZone
+	s.TransferInFunc = func(z *file.Zone, t *transfer.Transfer) error {
+		return s.transferIn(s.zoneName(z), z, t)
+	}
+	return s
+}
+
+func (s *Secondary) transferAndUpdate(origin string, z *file.Zone, x *transfer.Transfer, updateShutdown chan bool) {
+	dur := time.Millisecond * 250
+	max := time.Second * 10
+	for {
+		err := s.transferIn(origin, z, x)
+		if err == nil {
+			break
+		}
+		log.Warningf("All '%s' masters failed to transfer, retrying in %s: %s", origin, dur.String(), err)
+		if waitForTransferRetry(updateShutdown, dur) {
+			return
+		}
+		dur <<= 1 // double the duration
+		if dur > max {
+			dur = max
+		}
+	}
+	select {
+	case <-updateShutdown:
+		return
+	default:
+	}
+	z.UpdateWithTransfer(updateShutdown, x, func(z *file.Zone, t *transfer.Transfer) error {
+		return s.transferIn(origin, z, t)
+	})
+}
+
 func waitForTransferRetry(updateShutdown <-chan bool, dur time.Duration) bool {
 	timer := time.NewTimer(dur)
 	defer timer.Stop()

plugin/secondary/zones.go

@@ -0,0 +1,46 @@
+package secondary
+
+import (
+	"github.com/coredns/coredns/plugin"
+	"github.com/coredns/coredns/plugin/file"
+)
+
+func (s *Secondary) lookupZone(qname string) (string, *file.Zone, bool) {
+	s.zoneMu.RLock()
+	defer s.zoneMu.RUnlock()
+
+	zone := plugin.Zones(s.Names).Matches(qname)
+	if zone == "" {
+		return "", nil, false
+	}
+	z, ok := s.Z[zone]
+	if !ok {
+		return zone, nil, true
+	}
+	return zone, z, true
+}
+
+func (s *Secondary) zoneName(z *file.Zone) string {
+	s.zoneMu.RLock()
+	defer s.zoneMu.RUnlock()
+
+	return s.zoneNames[z]
+}
+
+func (s *Secondary) stopDynamicZones() {
+	s.zoneMu.Lock()
+	defer s.zoneMu.Unlock()
+
+	for _, dyn := range s.dynamicZones {
+		dyn.stopOnce.Do(func() { close(dyn.shutdown) })
+	}
+}
+
+func removeZoneName(names []string, name string) []string {
+	for i, n := range names {
+		if n == name {
+			return append(names[:i], names[i+1:]...)
+		}
+	}
+	return names
+}