From 1e01c0ad7c4502ec1a60a010d7b97206483a0ecb Mon Sep 17 00:00:00 2001 From: houyuwushang Date: Fri, 10 Jul 2026 08:42:30 +0800 Subject: [PATCH] plugin/secondary: serve catalog member zones (#8230) Signed-off-by: houyuwushang --- plugin/file/file.go | 28 ++++-- plugin/file/xfr.go | 4 +- plugin/file/xfr_test.go | 13 +++ plugin/secondary/README.md | 5 +- plugin/secondary/catalog.go | 87 +++++++++++++++++++ plugin/secondary/catalog_test.go | 145 +++++++++++++++++++++++++++---- plugin/secondary/secondary.go | 17 +++- plugin/secondary/setup.go | 90 +++++++++++-------- plugin/secondary/zones.go | 46 ++++++++++ 9 files changed, 368 insertions(+), 67 deletions(-) create mode 100644 plugin/secondary/zones.go diff --git a/plugin/file/file.go b/plugin/file/file.go index a7c3eded6..41e615131 100644 --- a/plugin/file/file.go +++ b/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 := plugin.Zones(f.Zones.Names).Matches(qname) - if zone == "" { + zone, z, ok := f.lookupZone(qname) + if !ok { // 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 !ok || z == nil { + if 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) diff --git a/plugin/file/xfr.go b/plugin/file/xfr.go index d0a2bdc16..1e182cc54 100644 --- a/plugin/file/xfr.go +++ b/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] - if !ok || z == nil { + matched, z, ok := f.lookupZone(zone) + if !ok || matched != zone || z == nil { return nil, transfer.ErrNotAuthoritative } return z.Transfer(serial) diff --git a/plugin/file/xfr_test.go b/plugin/file/xfr_test.go index f8d4cafcd..e58301622 100644 --- a/plugin/file/xfr_test.go +++ b/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) + } +} diff --git a/plugin/secondary/README.md b/plugin/secondary/README.md index 4a904a2c9..d7ef1bdb9 100644 --- a/plugin/secondary/README.md +++ b/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 - list after each successful transfer. +* `catalog` treats the transferred zone as an RFC 9432 catalog zone. After each successful catalog + 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 diff --git a/plugin/secondary/catalog.go b/plugin/secondary/catalog.go index 66d4946de..348fb6580 100644 --- a/plugin/secondary/catalog.go +++ b/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{}) + } +} diff --git a/plugin/secondary/catalog_test.go b/plugin/secondary/catalog_test.go index 0c4c79c15..999326a96 100644 --- a/plugin/secondary/catalog_test.go +++ b/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{ - catalogs: make(map[string]*catalog.Catalog), - catalogZones: map[string]struct{}{origin: {}}, - } + 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) @@ -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{ - catalogs: make(map[string]*catalog.Catalog), - catalogZones: map[string]struct{}{origin: {}}, - } + s := newTestSecondary(origin, z, true) 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{ - catalogs: make(map[string]*catalog.Catalog), - } + s := newTestSecondary(origin, z, false) 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() diff --git a/plugin/secondary/secondary.go b/plugin/secondary/secondary.go index 5500d6e10..1241f1f43 100644 --- a/plugin/secondary/secondary.go +++ b/plugin/secondary/secondary.go @@ -13,9 +13,20 @@ import ( type Secondary struct { file.File - catalogMu sync.RWMutex - catalogs map[string]*catalog.Catalog - catalogZones map[string]struct{} + zoneMu sync.RWMutex + zoneNames map[*file.Zone]string + 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. diff --git a/plugin/secondary/setup.go b/plugin/secondary/setup.go index bbcccca27..e2bc96f30 100644 --- a/plugin/secondary/setup.go +++ b/plugin/secondary/setup.go @@ -26,18 +26,7 @@ func setup(c *caddy.Controller) error { return plugin.Error("secondary", err) } - s := &Secondary{ - 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) - } + s := newSecondary(zones, fall, catalogZones) 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() { - 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) - }) - }() + go s.transferAndUpdate(n, z, x, updateShutdown) }) 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() diff --git a/plugin/secondary/zones.go b/plugin/secondary/zones.go new file mode 100644 index 000000000..180f44763 --- /dev/null +++ b/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 +}