From dbce56ed30d71dbe921cf7f0b96972a4e2321295 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Thu, 30 Apr 2015 12:56:58 +0200 Subject: [PATCH] Background updates for names --- nameserver/zone.go | 120 +++++++++++++++++++++++++++++- nameserver/zone_lookup.go | 149 ++++++++++++++++++++++++++++++++++++++ weavedns/main.go | 35 +++++---- 3 files changed, 287 insertions(+), 17 deletions(-) diff --git a/nameserver/zone.go b/nameserver/zone.go index 95ab4809..c863ddf0 100644 --- a/nameserver/zone.go +++ b/nameserver/zone.go @@ -17,7 +17,11 @@ const ( const ( DefaultLocalDomain = "weave.local." // The default name used for the local domain + DefaultRefreshInterval = int(localTTL) // Period for background updates with mDNS + DefaultRelevantTime = 60 // When to forget info about remote info if nobody asks... + defaultNumUpdaters = 1 // Number of background updaters + defaultRefreshMailbox = 1000 // Number of on-the-fly background queries defaultRemoteIdent = "weave:remote" // Ident used for info obtained from mDNS ) @@ -58,7 +62,7 @@ type Zone interface { /////////////////////////////////////////////////////////////////// // -// Zone database: +// Zone database overview: // // idents names entry // *-------* *------* *------* @@ -148,6 +152,7 @@ type name struct { name string ipv4 map[IPv4]*recordEntry // all the IPv4 records for this name observers []ZoneRecordObserver // the observers for this name + lastRefreshTime time.Time // last time this name was updated by broadcasting a query lastAccessTime time.Time // last time a Lookup* used info from this name } @@ -224,6 +229,7 @@ func (n *name) updateIPs(records []ZoneRecord, now time.Time) (added int, remove // notify observers and update times if added > 0 || removed > 0 { + n.lastRefreshTime = now n.notifyNameObservers() } return @@ -404,11 +410,23 @@ func (irs identRecordsSet) String() string { /////////////////////////////////////////////////////////////////// +// Invariant on updateRequests: they are in ascending order of time. +type refreshRequest struct { + name string + time time.Time +} + type ZoneConfig struct { // The local domain Domain string // The interface where mDNS operates Iface *net.Interface + // Refresh interval for local names in the zone database (disabled by default) (in seconds) + RefreshInterval int + // Forget about remote info if nobody asks for it in this time (in seconds) + RelevantTime int + // Max pending refresh requests + MaxRefreshRequests int // Force a specific mDNS client MDNSClient ZoneMDNSClient // Force a specific mDNS server @@ -429,11 +447,23 @@ type zoneDb struct { mdnsSrv ZoneMDNSServer iface *net.Interface domain string // the local domain + relevantLimit time.Duration // if no one asks for something in this time, it is not relevant a anymore... clock clock.Clock + + refreshChan chan refreshRequest // channel for enqueing requests for updating names + refreshSchedChan chan refreshRequest // channel for enqueing requests for updating names + refreshCloseChan chan bool + refreshWg sync.WaitGroup + refreshInterval time.Duration } // Create a new zone database func NewZoneDb(config ZoneConfig) (zone *zoneDb, err error) { + maxRefreshRequests := defaultRefreshMailbox + if config.MaxRefreshRequests > 0 { + maxRefreshRequests = config.MaxRefreshRequests + } + zone = &zoneDb{ domain: config.Domain, idents: make(identRecordsSet), @@ -441,6 +471,10 @@ func NewZoneDb(config ZoneConfig) (zone *zoneDb, err error) { mdnsSrv: config.MDNSServer, iface: config.Iface, clock: config.Clock, + relevantLimit: time.Duration(DefaultRelevantTime) * time.Second, + refreshChan: make(chan refreshRequest, maxRefreshRequests), + refreshSchedChan: make(chan refreshRequest, maxRefreshRequests), + refreshCloseChan: make(chan bool), } // fix the default configuration parameters @@ -450,6 +484,12 @@ func NewZoneDb(config ZoneConfig) (zone *zoneDb, err error) { if len(zone.domain) == 0 { zone.domain = DefaultLocalDomain } + if config.RefreshInterval > 0 { + zone.refreshInterval = time.Duration(config.RefreshInterval) * time.Second + } + if config.RelevantTime > 0 { + zone.relevantLimit = time.Duration(config.RelevantTime) * time.Second + } // Create the mDNS client and server if zone.mdnsCli == nil { @@ -469,7 +509,7 @@ func NewZoneDb(config ZoneConfig) (zone *zoneDb, err error) { // Start the zone database func (zone *zoneDb) Start() (err error) { if zone.iface != nil { - Info.Printf("[zonedb] Using mDNS on %s", zone.iface) + Info.Printf("[zonedb] Using mDNS on %+v", zone.iface) } else { Info.Printf("[zonedb] Using mDNS on all interfaces") } @@ -480,11 +520,34 @@ func (zone *zoneDb) Start() (err error) { if err = zone.mdnsSrv.Start(zone.iface); err != nil { return } + + // Start some name refreshers... + if zone.refreshInterval > 0 { + Info.Printf("[zonedb] Starting %d background name updaters", defaultNumUpdaters) + for i := 0; i < defaultNumUpdaters; i++ { + zone.refreshWg.Add(1) + go zone.updater(i) + } + + zone.refreshWg.Add(1) + go zone.periodicUpdater() + } else { + // TODO: the refresher also garbage collects irrelevant remote info... we should have an alternative mechanism + // when we don't use refreshers. + Warning.Printf("[zonedb] Remote info will not be garbage collected") + } + return } // Perform a graceful shutdown of the zone database func (zone *zoneDb) Stop() error { + if zone.refreshInterval > 0 { + Debug.Printf("[zonedb] Closing background updaters...") + close(zone.refreshCloseChan) + zone.refreshWg.Wait() + } + Debug.Printf("[zonedb] Exiting mDNS client and server...") zone.mdnsCli.Stop() zone.mdnsSrv.Stop() @@ -598,7 +661,7 @@ func (zone *zoneDb) ObserveInaddr(inaddr string, observer ZoneRecordObserver) (e } // Get the string representation of a zone -func (zone zoneDb) String() string { +func (zone *zoneDb) String() string { zone.mx.RLock() defer zone.mx.RUnlock() return zone.idents.String() @@ -610,6 +673,57 @@ func (zone *zoneDb) ContainerDied(ident string) error { return zone.DeleteRecordsFor(ident) } +// Return true if a remote name is still relevant +func (zone *zoneDb) IsNameRelevant(n string) bool { + zone.mx.Lock() + defer zone.mx.Unlock() + + lastAccess := time.Time{} + for _, nameset := range zone.idents { + if t := nameset.getNameLastAccess(n); !t.IsZero() { + if lastAccess.IsZero() || t.After(lastAccess) { + lastAccess = t + } + } + } + + return !lastAccess.IsZero() && zone.clock.Now().Sub(lastAccess) < zone.relevantLimit +} + +// Returns true if the info we have for a remote name has expired. Local names are never expired +// Returns false if we do not have info for this name from remote peers. +func (zone *zoneDb) IsNameExpired(n string) bool { + zone.mx.Lock() + defer zone.mx.Unlock() + + now := zone.clock.Now() + remoteNamesSet := zone.getNamesSet(defaultRemoteIdent) + entries := remoteNamesSet.getEntriesForName(n) + if entries != nil { + for _, entry := range entries { + if entry.hasExpired(now) { + return true + } + } + } + return false +} + +func (zone *zoneDb) HasNameLocalInfo(n string) bool { + zone.mx.Lock() + defer zone.mx.Unlock() + + remoteNamesSet := zone.getNamesSet(defaultRemoteIdent) + for _, nameset := range zone.idents { + if nameset != remoteNamesSet { + if nameset.hasName(n) { + return true + } + } + } + return false +} + // Return true if the we have obtained some information for a name from remote peers // We can have both local (eg, introduced through the HTTP API) and remote (eg, mDNS) // information for a name. This method only checks if there is remote information... diff --git a/nameserver/zone_lookup.go b/nameserver/zone_lookup.go index f64c7fb3..247df6bd 100644 --- a/nameserver/zone_lookup.go +++ b/nameserver/zone_lookup.go @@ -3,6 +3,7 @@ package nameserver import ( "github.com/miekg/dns" . "github.com/weaveworks/weave/common" + "time" ) // Perform a lookup for a name in the zone @@ -83,6 +84,10 @@ func (zone *zoneDb) DomainLookupName(name string) (res []ZoneRecord, err error) } if len(res) > 0 { + // start (if we need to) a background refresh for the name + // note: we do not spend time trying to update names that did not return an initial response... + zone.startUpdatingName(name) + return res, nil } @@ -174,6 +179,13 @@ func (zone *zoneDb) DomainLookupInaddr(inaddr string) (res []ZoneRecord, err err } if len(res) > 0 { + // note: even for reverse addresses, we perform the background updates in the name, not in the IP + // this simplifies the process and produces basically the same results... + // note: we do not spend time trying to update names that did not return an initial response... + for _, name := range res { + zone.startUpdatingName(name.Name()) + } + return res, nil } @@ -181,3 +193,140 @@ func (zone *zoneDb) DomainLookupInaddr(inaddr string) (res []ZoneRecord, err err } +////////////////////////////////////////////////////////////////////////////// + +// Names updates try to find all the IPs for a given name with a mDNS query +// +// There are two types of names updates: +// +// - immediate updates. +// After a `DomainLookup*()` for a name not in the database we will return the +// first IP we can get with mDNS from other peers. Waiting for more responses would +// mean more latency in the response to the client, so we send only one answer BUT +// we also trigger an immediate update request for that name in order to get all +// the other IPs we didn't wait for... +// +// - periodic updates +// once we have obtained the first group of IPs for a name, we schedule a periodic +// refresh for that name, so we keep the list of IPs for that name up to date. +// +// These names updates are repeated until either +// +// a) there is no interest in the name, determined by a global 'relevant time' +// and the last time some local client asked about the name, +// or +// b) no peers answer one of our refresh requests (because the name has probably +// disappeared from the network) +// + +// TODO: for the sake of simplicity, we implement this mechanism with two channels: one for immediate +// updates and another one for scheduled requests. We could use a heap of "next-time"s and a timer, +// but that would require timers cancellations/updates/etc on insertions/removals/etc, and +// it is probably not worth the trouble as we always use the same refresh period. +// It could be useful if we move to a solution where we set update times from the responses TTLs, +// but we currently use a fixed TTL (30secs), the same as the refresh period... +// Anyway, maybe we will move to a gossip-based solution instead of doing this polling... + +// Check if we must start updating a name and, in that case, trigger a immediate update +func (zone *zoneDb) startUpdatingName(name string) { + if zone.refreshInterval > 0 { + zone.mx.Lock() + defer zone.mx.Unlock() + + // check if we should enqueue a refresh request for this name + n := zone.getNamesSet(defaultRemoteIdent).getName(name, true) + if n.lastRefreshTime.IsZero() { + now := zone.clock.Now() + n.lastRefreshTime = now + + Debug.Printf("[zonedb] Creating new immediate refresh request for '%s'", name) + zone.refreshChan <- refreshRequest{name: name, time: now} + } + } +} + +// A worker for updating the list of IPs we have for a name +func (zone *zoneDb) updater(num int) { + defer zone.refreshWg.Done() + + Debug.Printf("[zonedb] Starting background updater #%d...", num) + for { + select { + case <-zone.refreshCloseChan: + Debug.Printf("[zonedb] Background updater #%d: interrupted while waiting for requests: exiting", num) + return + + case request := <-zone.refreshChan: + // if nobody has asked for this name for long time, just forget about it... + // this will eventually garbage collect the `refreshChan` and all remote info in absence of activity + if !zone.IsNameRelevant(request.name) || zone.IsNameExpired(request.name) { + Debug.Printf("[zonedb] '%s' seem to be irrelevant now: removing any remote information", request.name) + zone.mx.Lock() + zone.getNamesSet(defaultRemoteIdent).deleteName(request.name) + zone.mx.Unlock() + continue + } + + // perform the refresh for this name + name := dns.Fqdn(request.name) + Debug.Printf("[zonedb] Refreshing name '%s' with mDNS...", name) + res, _ := zone.mdnsCli.InsistentLookupName(request.name) + if res != nil && len(res) > 0 { + numIps := len(res) + zone.mx.Lock() + now := zone.clock.Now() + added, removed := zone.getNamesSet(defaultRemoteIdent).getName(name, true).updateIPs(res, now) + zone.mx.Unlock() + Debug.Printf("[zonedb] Obtained %d IPs for name '%s' with mDNS: %d added, %d removed", + numIps, name, added, removed) + + // once the name has been updated, we insert the request (back) in the periodic requests channel + now = zone.clock.Now() + request.time = now.Add(zone.refreshInterval) + Debug.Printf("[zonedb] Rescheduling update for '%s' in %.2f secs", + request.name, zone.refreshInterval.Seconds()) + zone.refreshSchedChan <- request + } else { + Debug.Printf("[zonedb] nobody knows about '%s'... removing", name) + zone.mx.Lock() + zone.getNamesSet(defaultRemoteIdent).deleteName(request.name) + zone.mx.Unlock() + } + } + } +} + +// The periodic updater +// Consume requests from the `refreshSchedChan`, where requests with increasing scheduling time are enqueued +// for refreshing names... +func (zone *zoneDb) periodicUpdater() { + defer zone.refreshWg.Done() + for { + select { + case <-zone.refreshCloseChan: + Debug.Printf("[zonedb] Periodic updater: interrupted while waiting for requests: exiting") + return + + case request := <-zone.refreshSchedChan: + now := zone.clock.Now() + // we can sleep until the update time has arrived, as requests are sorted by scheduled time (new request + // in this channel will always be scheduled later than the last item in the channel), so we can + // safely suspend this goroutine until then... + if request.time.After(now) { + ddiff := time.Duration(request.time.Sub(now).Nanoseconds()) + timer := zone.clock.Timer(ddiff) + Debug.Printf("[zonedb] Periodic updater: new request for %s: sleeping for %.2f secs...", + request.name, ddiff.Seconds()) + select { + case <-zone.refreshCloseChan: + Debug.Printf("[zonedb] Periodic updater: interrupted while sleeping: exiting") + return + case <-timer.C: + } + } + + // once the time has arrived, we insert the request in the immediate refresh channel + zone.refreshChan <- request + } + } +} diff --git a/weavedns/main.go b/weavedns/main.go index a944f7c3..f5b7e1f2 100644 --- a/weavedns/main.go +++ b/weavedns/main.go @@ -15,19 +15,21 @@ var version = "(unreleased version)" func main() { var ( - justVersion bool - ifaceName string - apiPath string - domain string - dnsPort int - httpPort int - wait int - timeout int - udpbuf int - cacheLen int - watch bool - debug bool - err error + justVersion bool + ifaceName string + apiPath string + domain string + dnsPort int + httpPort int + wait int + timeout int + udpbuf int + refreshInterval int + relevantTime int + cacheLen int + watch bool + debug bool + err error ) flag.BoolVar(&justVersion, "version", false, "print version and exit") @@ -42,6 +44,9 @@ func main() { flag.IntVar(&cacheLen, "cache", weavedns.DefaultCacheLen, "cache length") flag.BoolVar(&watch, "watch", true, "watch the docker socket for container events") flag.BoolVar(&debug, "debug", false, "output debugging info to stderr") + // advanced options + flag.IntVar(&refreshInterval, "refresh", weavedns.DefaultRefreshInterval, "refresh interval (in secs) for local names (0=disable)") + flag.IntVar(&relevantTime, "relevant", weavedns.DefaultRelevantTime, "life time for info in the absence of queries (in secs)") flag.Parse() if justVersion { @@ -50,7 +55,7 @@ func main() { } InitDefaultLogging(debug) - Info.Printf("WeaveDNS version %s\n", version) // first thing in log: the version + Info.Printf("[main] WeaveDNS version %s\n", version) // first thing in log: the version var iface *net.Interface if ifaceName != "" { @@ -67,6 +72,8 @@ func main() { zoneConfig := weavedns.ZoneConfig{ Domain: domain, Iface: iface, + RefreshInterval: refreshInterval, + RelevantTime: relevantTime, } zone, err := weavedns.NewZoneDb(zoneConfig) if err != nil {