vendor/github.com/go-kit/kit/sd/dnssrv/instancer.go

   1 package dnssrv
   2
   3 import (
   4         "fmt"
   5         "net"
   6         "time"
   7
   8         "github.com/go-kit/kit/log"
   9         "github.com/go-kit/kit/sd"
  10         "github.com/go-kit/kit/sd/internal/instance"
  11 )
  12
  13 // Instancer yields instances from the named DNS SRV record. The name is
  14 // resolved on a fixed schedule. Priorities and weights are ignored.
  15 type Instancer struct {
  16         cache  *instance.Cache
  17         name   string
  18         logger log.Logger
  19         quit   chan struct{}
  20 }
  21
  22 // NewInstancer returns a DNS SRV instancer.
  23 func NewInstancer(
  24         name string,
  25         ttl time.Duration,
  26         logger log.Logger,
  27 ) *Instancer {
  28         return NewInstancerDetailed(name, time.NewTicker(ttl), net.LookupSRV, logger)
  29 }
  30
  31 // NewInstancerDetailed is the same as NewInstancer, but allows users to
  32 // provide an explicit lookup refresh ticker instead of a TTL, and specify the
  33 // lookup function instead of using net.LookupSRV.
  34 func NewInstancerDetailed(
  35         name string,
  36         refresh *time.Ticker,
  37         lookup Lookup,
  38         logger log.Logger,
  39 ) *Instancer {
  40         p := &Instancer{
  41                 cache:  instance.NewCache(),
  42                 name:   name,
  43                 logger: logger,
  44                 quit:   make(chan struct{}),
  45         }
  46
  47         instances, err := p.resolve(lookup)
  48         if err == nil {
  49                 logger.Log("name", name, "instances", len(instances))
  50         } else {
  51                 logger.Log("name", name, "err", err)
  52         }
  53         p.cache.Update(sd.Event{Instances: instances, Err: err})
  54
  55         go p.loop(refresh, lookup)
  56         return p
  57 }
  58
  59 // Stop terminates the Instancer.
  60 func (p *Instancer) Stop() {
  61         close(p.quit)
  62 }
  63
  64 func (p *Instancer) loop(t *time.Ticker, lookup Lookup) {
  65         defer t.Stop()
  66         for {
  67                 select {
  68                 case <-t.C:
  69                         instances, err := p.resolve(lookup)
  70                         if err != nil {
  71                                 p.logger.Log("name", p.name, "err", err)
  72                                 p.cache.Update(sd.Event{Err: err})
  73                                 continue // don't replace potentially-good with bad
  74                         }
  75                         p.cache.Update(sd.Event{Instances: instances})
  76
  77                 case <-p.quit:
  78                         return
  79                 }
  80         }
  81 }
  82
  83 func (p *Instancer) resolve(lookup Lookup) ([]string, error) {
  84         _, addrs, err := lookup("", "", p.name)
  85         if err != nil {
  86                 return nil, err
  87         }
  88         instances := make([]string, len(addrs))
  89         for i, addr := range addrs {
  90                 instances[i] = net.JoinHostPort(addr.Target, fmt.Sprint(addr.Port))
  91         }
  92         return instances, nil
  93 }
  94
  95 // Register implements Instancer.
  96 func (s *Instancer) Register(ch chan<- sd.Event) {
  97         s.cache.Register(ch)
  98 }
  99
 100 // Deregister implements Instancer.
 101 func (s *Instancer) Deregister(ch chan<- sd.Event) {
 102         s.cache.Deregister(ch)
 103 }