123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302 |
- // Copyright (C) 2020 The Syncthing Authors.
- //
- // This Source Code Form is subject to the terms of the Mozilla Public
- // License, v. 2.0. If a copy of the MPL was not distributed with this file,
- // You can obtain one at https://mozilla.org/MPL/2.0/.
- //go:generate -command counterfeiter go run github.com/maxbrunsfeld/counterfeiter/v6
- //go:generate counterfeiter -o mocks/manager.go --fake-name Manager . Manager
- package discover
- import (
- "context"
- "crypto/tls"
- "fmt"
- "sort"
- "time"
- "github.com/thejerf/suture/v4"
- "github.com/syncthing/syncthing/lib/config"
- "github.com/syncthing/syncthing/lib/connections/registry"
- "github.com/syncthing/syncthing/lib/events"
- "github.com/syncthing/syncthing/lib/protocol"
- "github.com/syncthing/syncthing/lib/stringutil"
- "github.com/syncthing/syncthing/lib/svcutil"
- "github.com/syncthing/syncthing/lib/sync"
- )
- // The Manager aggregates results from multiple Finders. Each Finder has
- // an associated cache time and negative cache time. The cache time sets how
- // long we cache and return successful lookup results, the negative cache
- // time sets how long we refrain from asking about the same device ID after
- // receiving a negative answer. The value of zero disables caching (positive
- // or negative).
- type Manager interface {
- FinderService
- ChildErrors() map[string]error
- }
- type manager struct {
- *suture.Supervisor
- myID protocol.DeviceID
- cfg config.Wrapper
- cert tls.Certificate
- evLogger events.Logger
- addressLister AddressLister
- registry *registry.Registry
- finders map[string]cachedFinder
- mut sync.RWMutex
- }
- func NewManager(myID protocol.DeviceID, cfg config.Wrapper, cert tls.Certificate, evLogger events.Logger, lister AddressLister, registry *registry.Registry) Manager {
- m := &manager{
- Supervisor: suture.New("discover.Manager", svcutil.SpecWithDebugLogger(l)),
- myID: myID,
- cfg: cfg,
- cert: cert,
- evLogger: evLogger,
- addressLister: lister,
- registry: registry,
- finders: make(map[string]cachedFinder),
- mut: sync.NewRWMutex(),
- }
- m.Add(svcutil.AsService(m.serve, m.String()))
- return m
- }
- func (m *manager) serve(ctx context.Context) error {
- m.cfg.Subscribe(m)
- m.CommitConfiguration(config.Configuration{}, m.cfg.RawCopy())
- <-ctx.Done()
- m.cfg.Unsubscribe(m)
- return nil
- }
- func (m *manager) addLocked(identity string, finder Finder, cacheTime, negCacheTime time.Duration) {
- entry := cachedFinder{
- Finder: finder,
- cacheTime: cacheTime,
- negCacheTime: negCacheTime,
- cache: newCache(),
- token: nil,
- }
- if service, ok := finder.(suture.Service); ok {
- token := m.Supervisor.Add(service)
- entry.token = &token
- }
- m.finders[identity] = entry
- l.Infoln("Using discovery mechanism:", identity)
- }
- func (m *manager) removeLocked(identity string) {
- entry, ok := m.finders[identity]
- if !ok {
- return
- }
- if entry.token != nil {
- err := m.Supervisor.Remove(*entry.token)
- if err != nil {
- l.Warnf("removing discovery %s: %s", identity, err)
- }
- }
- delete(m.finders, identity)
- l.Infoln("Stopped using discovery mechanism: ", identity)
- }
- // Lookup attempts to resolve the device ID using any of the added Finders,
- // while obeying the cache settings.
- func (m *manager) Lookup(ctx context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
- m.mut.RLock()
- for _, finder := range m.finders {
- if cacheEntry, ok := finder.cache.Get(deviceID); ok {
- // We have a cache entry. Lets see what it says.
- if cacheEntry.found && time.Since(cacheEntry.when) < finder.cacheTime {
- // It's a positive, valid entry. Use it.
- l.Debugln("cached discovery entry for", deviceID, "at", finder)
- l.Debugln(" cache:", cacheEntry)
- addresses = append(addresses, cacheEntry.Addresses...)
- continue
- }
- valid := time.Now().Before(cacheEntry.validUntil) || time.Since(cacheEntry.when) < finder.negCacheTime
- if !cacheEntry.found && valid {
- // It's a negative, valid entry. We should not make another
- // attempt right now.
- l.Debugln("negative cache entry for", deviceID, "at", finder, "valid until", cacheEntry.when.Add(finder.negCacheTime), "or", cacheEntry.validUntil)
- continue
- }
- // It's expired. Ignore and continue.
- }
- // Perform the actual lookup and cache the result.
- if addrs, err := finder.Lookup(ctx, deviceID); err == nil {
- l.Debugln("lookup for", deviceID, "at", finder)
- l.Debugln(" addresses:", addrs)
- addresses = append(addresses, addrs...)
- finder.cache.Set(deviceID, CacheEntry{
- Addresses: addrs,
- when: time.Now(),
- found: len(addrs) > 0,
- })
- } else {
- // Lookup returned error, add a negative cache entry.
- entry := CacheEntry{
- when: time.Now(),
- found: false,
- }
- if err, ok := err.(cachedError); ok {
- entry.validUntil = time.Now().Add(err.CacheFor())
- }
- finder.cache.Set(deviceID, entry)
- }
- }
- m.mut.RUnlock()
- addresses = stringutil.UniqueTrimmedStrings(addresses)
- sort.Strings(addresses)
- l.Debugln("lookup results for", deviceID)
- l.Debugln(" addresses: ", addresses)
- return addresses, nil
- }
- func (*manager) String() string {
- return "discovery cache"
- }
- func (*manager) Error() error {
- return nil
- }
- func (m *manager) ChildErrors() map[string]error {
- children := make(map[string]error, len(m.finders))
- m.mut.RLock()
- for _, f := range m.finders {
- children[f.String()] = f.Error()
- }
- m.mut.RUnlock()
- return children
- }
- func (m *manager) Cache() map[protocol.DeviceID]CacheEntry {
- // Res will be the "total" cache, i.e. the union of our cache and all our
- // children's caches.
- res := make(map[protocol.DeviceID]CacheEntry)
- m.mut.RLock()
- for _, finder := range m.finders {
- // Each finder[i] has a corresponding cache. Go through
- // it and populate the total, appending any addresses and keeping
- // the newest "when" time. We skip any negative cache finders.
- for k, v := range finder.cache.Cache() {
- if v.found {
- cur := res[k]
- if v.when.After(cur.when) {
- cur.when = v.when
- }
- cur.Addresses = append(cur.Addresses, v.Addresses...)
- res[k] = cur
- }
- }
- // Then ask the finder itself for its cache and do the same. If this
- // finder is a global discovery client, it will have no cache. If it's
- // a local discovery client, this will be its current state.
- for k, v := range finder.Cache() {
- if v.found {
- cur := res[k]
- if v.when.After(cur.when) {
- cur.when = v.when
- }
- cur.Addresses = append(cur.Addresses, v.Addresses...)
- res[k] = cur
- }
- }
- }
- m.mut.RUnlock()
- for k, v := range res {
- v.Addresses = stringutil.UniqueTrimmedStrings(v.Addresses)
- res[k] = v
- }
- return res
- }
- func (m *manager) CommitConfiguration(_, to config.Configuration) (handled bool) {
- m.mut.Lock()
- defer m.mut.Unlock()
- toIdentities := make(map[string]struct{})
- if to.Options.GlobalAnnEnabled {
- for _, srv := range to.Options.GlobalDiscoveryServers() {
- toIdentities[globalDiscoveryIdentity(srv)] = struct{}{}
- }
- }
- if to.Options.LocalAnnEnabled {
- toIdentities[ipv4Identity(to.Options.LocalAnnPort)] = struct{}{}
- toIdentities[ipv6Identity(to.Options.LocalAnnMCAddr)] = struct{}{}
- }
- // Remove things that we're not expected to have.
- for identity := range m.finders {
- if _, ok := toIdentities[identity]; !ok {
- m.removeLocked(identity)
- }
- }
- // Add things we don't have.
- if to.Options.GlobalAnnEnabled {
- for _, srv := range to.Options.GlobalDiscoveryServers() {
- identity := globalDiscoveryIdentity(srv)
- // Skip, if it's already running.
- if _, ok := m.finders[identity]; ok {
- continue
- }
- gd, err := NewGlobal(srv, m.cert, m.addressLister, m.evLogger, m.registry)
- if err != nil {
- l.Warnln("Global discovery:", err)
- continue
- }
- // Each global discovery server gets its results cached for five
- // minutes, and is not asked again for a minute when it's returned
- // unsuccessfully.
- m.addLocked(identity, gd, 5*time.Minute, time.Minute)
- }
- }
- if to.Options.LocalAnnEnabled {
- // v4 broadcasts
- v4Identity := ipv4Identity(to.Options.LocalAnnPort)
- if _, ok := m.finders[v4Identity]; !ok {
- bcd, err := NewLocal(m.myID, fmt.Sprintf(":%d", to.Options.LocalAnnPort), m.addressLister, m.evLogger)
- if err != nil {
- l.Warnln("IPv4 local discovery:", err)
- } else {
- m.addLocked(v4Identity, bcd, 0, 0)
- }
- }
- // v6 multicasts
- v6Identity := ipv6Identity(to.Options.LocalAnnMCAddr)
- if _, ok := m.finders[v6Identity]; !ok {
- mcd, err := NewLocal(m.myID, to.Options.LocalAnnMCAddr, m.addressLister, m.evLogger)
- if err != nil {
- l.Warnln("IPv6 local discovery:", err)
- } else {
- m.addLocked(v6Identity, mcd, 0, 0)
- }
- }
- }
- return true
- }
|