instancer.go

Documentation: github.com/go-kit/kit/sd/zk

     1  package zk
     2  
     3  import (
     4  	"github.com/go-zookeeper/zk"
     5  
     6  	"github.com/go-kit/kit/sd"
     7  	"github.com/go-kit/kit/sd/internal/instance"
     8  	"github.com/go-kit/log"
     9  )
    10  
    11  // Instancer yield instances stored in a certain ZooKeeper path. Any kind of
    12  // change in that path is watched and will update the subscribers.
    13  type Instancer struct {
    14  	cache  *instance.Cache
    15  	client Client
    16  	path   string
    17  	logger log.Logger
    18  	quitc  chan struct{}
    19  }
    20  
    21  // NewInstancer returns a ZooKeeper Instancer. ZooKeeper will start watching
    22  // the given path for changes and update the Instancer endpoints.
    23  func NewInstancer(c Client, path string, logger log.Logger) (*Instancer, error) {
    24  	s := &Instancer{
    25  		cache:  instance.NewCache(),
    26  		client: c,
    27  		path:   path,
    28  		logger: logger,
    29  		quitc:  make(chan struct{}),
    30  	}
    31  
    32  	err := s.client.CreateParentNodes(s.path)
    33  	if err != nil {
    34  		return nil, err
    35  	}
    36  
    37  	instances, eventc, err := s.client.GetEntries(s.path)
    38  	if err != nil {
    39  		logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
    40  		// other implementations continue here, but we exit because we don't know if eventc is valid
    41  		return nil, err
    42  	}
    43  	logger.Log("path", s.path, "instances", len(instances))
    44  	s.cache.Update(sd.Event{Instances: instances})
    45  
    46  	go s.loop(eventc)
    47  
    48  	return s, nil
    49  }
    50  
    51  func (s *Instancer) loop(eventc <-chan zk.Event) {
    52  	var (
    53  		instances []string
    54  		err       error
    55  	)
    56  	for {
    57  		select {
    58  		case <-eventc:
    59  			// We received a path update notification. Call GetEntries to
    60  			// retrieve child node data, and set a new watch, as ZK watches are
    61  			// one-time triggers.
    62  			instances, eventc, err = s.client.GetEntries(s.path)
    63  			if err != nil {
    64  				s.logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
    65  				s.cache.Update(sd.Event{Err: err})
    66  				continue
    67  			}
    68  			s.logger.Log("path", s.path, "instances", len(instances))
    69  			s.cache.Update(sd.Event{Instances: instances})
    70  
    71  		case <-s.quitc:
    72  			return
    73  		}
    74  	}
    75  }
    76  
    77  // Stop terminates the Instancer.
    78  func (s *Instancer) Stop() {
    79  	close(s.quitc)
    80  }
    81  
    82  // Register implements Instancer.
    83  func (s *Instancer) Register(ch chan<- sd.Event) {
    84  	s.cache.Register(ch)
    85  }
    86  
    87  // Deregister implements Instancer.
    88  func (s *Instancer) Deregister(ch chan<- sd.Event) {
    89  	s.cache.Deregister(ch)
    90  }
    91  
    92  // state returns the current state of instance.Cache, only for testing
    93  func (s *Instancer) state() sd.Event {
    94  	return s.cache.State()
    95  }
    96
View as plain text