Background updates for names

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...

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
+		}
+	}
+}

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 {