Mirrors/coredns
1
0
Fork 0
mirror of https://github.com/coredns/coredns.git synced 2026-08-20 23:08:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

plugin/secondary: support catalog migration and member scoping (#8288)

Browse Source 
* plugin/secondary: support catalog zone ownership migration

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

* plugin/secondary: scope catalog member zones

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

---------

Signed-off-by: houyuwushang <liuluoqianqiu@outlook.com>
This commit is contained in:
houyuwushang
2026-07-30 09:11:36 +08:00
committed by GitHub
parent a3135de7ec
commit 73cd1efe5b
6 changed files with 617 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
plugin/secondary/catalog.go
View File

@@ -50,39 +50,52 @@ type dynamicZoneStart struct {
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
rejected := 0
s.zoneMu.Lock()
s.ensureZoneStateLocked()
for _, member := range cat.Members {
if !s.catalogMemberAllowed(origin, member.Zone) {
rejected++
continue
}
memberZones[member.Zone] = struct{}{}
if existing, ok := s.Z[member.Zone]; ok {
dyn, dynamic := s.dynamicZones[member.Zone]
if !dynamic || dyn.catalog != origin || existing == nil {
if !dynamic || existing == nil {
log.Warningf("Skipping catalog member zone %s from %s: zone already exists", member.Zone, origin)
continue
}
if dyn.memberID == member.ID {
if dyn.catalog == origin {
if dyn.memberID == member.ID {
continue
}
previousID := dyn.memberID
start := s.replaceDynamicZoneLocked(member, origin, catalogZone, nil)
starts = append(starts, start)
log.Infof("Reset catalog member zone %s from %s after member ID changed from %s to %s", member.Zone, origin, previousID, member.ID)
continue
}
previousID := dyn.memberID
s.removeDynamicZoneLocked(member.Zone, origin)
log.Infof("Reset catalog member zone %s from %s after member ID changed from %s to %s", member.Zone, origin, previousID, member.ID)
s.catalogMu.RLock()
sourceMember, ok := catalogMember(s.catalogs[dyn.catalog], member.Zone)
if !ok || sourceMember.ChangeOfOwnership != origin {
s.catalogMu.RUnlock()
log.Warningf("Skipping catalog member zone %s from %s: zone already exists", member.Zone, origin)
continue
}
start, preserved := s.migrateCatalogMemberLocked(existing, dyn, sourceMember, member, origin, catalogZone)
s.catalogMu.RUnlock()
starts = append(starts, start)
s.logCatalogMigration(member.Zone, dyn.catalog, origin, sourceMember.ID, member.ID, preserved)
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, memberID: member.ID, shutdown: shutdown}
starts = append(starts, dynamicZoneStart{origin: member.Zone, zone: z, shutdown: shutdown})
start := s.replaceDynamicZoneLocked(member, origin, catalogZone, nil)
starts = append(starts, start)
log.Infof("Added catalog member zone %s from catalog %s", member.Zone, origin)
}
@@ -97,11 +110,91 @@ func (s *Secondary) applyCatalog(origin string, cat *catalog.Catalog, catalogZon
s.catalogMemberZones[origin] = memberZones
s.zoneMu.Unlock()
if rejected > 0 {
log.Warningf("Skipped %d member zones from catalog %s: outside configured member zones", rejected, origin)
}
for _, start := range starts {
go s.transferAndUpdate(start.origin, start.zone, t, start.shutdown)
}
}
func (s *Secondary) catalogMemberAllowed(origin, member string) bool {
zones, ok := s.catalogZones[origin]
return ok && (len(zones) == 0 || zones.Matches(member) != "")
}
func catalogMember(cat *catalog.Catalog, zone string) (catalog.Member, bool) {
if cat == nil {
return catalog.Member{}, false
}
for _, member := range cat.Members {
if member.Zone == zone {
return member, true
}
}
return catalog.Member{}, false
}
// migrateCatalogMemberLocked moves ownership to target. State is retained only
// when the active, source, and target member IDs all describe the same member.
func (s *Secondary) migrateCatalogMemberLocked(existing *file.Zone, dyn *dynamicZone, source, target catalog.Member, targetOrigin string, targetZone *file.Zone) (dynamicZoneStart, bool) {
preserved := dyn.memberID == source.ID && source.ID == target.ID
var preserveFrom *file.Zone
if preserved {
preserveFrom = existing
}
return s.replaceDynamicZoneLocked(target, targetOrigin, targetZone, preserveFrom), preserved
}
// replaceDynamicZoneLocked atomically installs a new transfer generation. A
// stopped generation may still finish an in-flight transfer, but it can only
// update the detached Zone pointer and cannot overwrite the new owner.
func (s *Secondary) replaceDynamicZoneLocked(member catalog.Member, origin string, catalogZone, preserveFrom *file.Zone) dynamicZoneStart {
if dyn, ok := s.dynamicZones[member.Zone]; ok {
dyn.stopOnce.Do(func() { close(dyn.shutdown) })
}
z := file.NewZone(member.Zone, "stdin")
if catalogZone != nil {
z.TransferFrom = append([]string(nil), catalogZone.TransferFrom...)
}
if preserveFrom != nil {
preserveDynamicZoneState(z, preserveFrom)
}
z.Upstream = upstream.New()
if previous, ok := s.Z[member.Zone]; ok {
if previous != nil {
delete(s.zoneNames, previous)
}
} else {
s.Names = append(s.Names, member.Zone)
}
shutdown := make(chan bool)
s.Z[member.Zone] = z
s.zoneNames[z] = member.Zone
s.dynamicZones[member.Zone] = &dynamicZone{catalog: origin, memberID: member.ID, shutdown: shutdown}
return dynamicZoneStart{origin: member.Zone, zone: z, shutdown: shutdown}
}
func preserveDynamicZoneState(target, source *file.Zone) {
// Dynamic transfers replace the live Apex and Tree instead of mutating them,
// so this snapshot remains isolated from any old in-flight transfer.
source.RLock()
target.Apex = source.Apex
target.Tree = source.Tree
target.Expired = source.Expired
source.RUnlock()
}
func (s *Secondary) logCatalogMigration(zone, source, target, sourceID, targetID string, preserved bool) {
if preserved {
log.Infof("Migrated catalog member zone %s from catalog %s to %s, preserving state for member ID %s", zone, source, target, sourceID)
return
}
log.Infof("Migrated catalog member zone %s from catalog %s to %s with state reset after member identity changed from %s to %s", zone, source, target, sourceID, targetID)
}
// removeDynamicZoneLocked removes a zone only when it belongs to catalog.
// The caller must hold s.zoneMu for writing.
func (s *Secondary) removeDynamicZoneLocked(zone, catalog string) bool {