File: instancer.go

package info (click to toggle)
golang-github-go-kit-kit 0.6.0-2
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, experimental, sid
  • size: 4,848 kB
  • sloc: sh: 65; makefile: 14
file content (95 lines) | stat: -rw-r--r-- 2,383 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
package zk

import (
	"github.com/samuel/go-zookeeper/zk"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/sd"
	"github.com/go-kit/kit/sd/internal/instance"
)

// Instancer yield instances stored in a certain ZooKeeper path. Any kind of
// change in that path is watched and will update the subscribers.
type Instancer struct {
	cache  *instance.Cache
	client Client
	path   string
	logger log.Logger
	quitc  chan struct{}
}

// NewInstancer returns a ZooKeeper Instancer. ZooKeeper will start watching
// the given path for changes and update the Instancer endpoints.
func NewInstancer(c Client, path string, logger log.Logger) (*Instancer, error) {
	s := &Instancer{
		cache:  instance.NewCache(),
		client: c,
		path:   path,
		logger: logger,
		quitc:  make(chan struct{}),
	}

	err := s.client.CreateParentNodes(s.path)
	if err != nil {
		return nil, err
	}

	instances, eventc, err := s.client.GetEntries(s.path)
	if err != nil {
		logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
		// other implementations continue here, but we exit because we don't know if eventc is valid
		return nil, err
	}
	logger.Log("path", s.path, "instances", len(instances))
	s.cache.Update(sd.Event{Instances: instances})

	go s.loop(eventc)

	return s, nil
}

func (s *Instancer) loop(eventc <-chan zk.Event) {
	var (
		instances []string
		err       error
	)
	for {
		select {
		case <-eventc:
			// We received a path update notification. Call GetEntries to
			// retrieve child node data, and set a new watch, as ZK watches are
			// one-time triggers.
			instances, eventc, err = s.client.GetEntries(s.path)
			if err != nil {
				s.logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
				s.cache.Update(sd.Event{Err: err})
				continue
			}
			s.logger.Log("path", s.path, "instances", len(instances))
			s.cache.Update(sd.Event{Instances: instances})

		case <-s.quitc:
			return
		}
	}
}

// Stop terminates the Instancer.
func (s *Instancer) Stop() {
	close(s.quitc)
}

// Register implements Instancer.
func (s *Instancer) Register(ch chan<- sd.Event) {
	s.cache.Register(ch)
}

// Deregister implements Instancer.
func (s *Instancer) Deregister(ch chan<- sd.Event) {
	s.cache.Deregister(ch)
}

// state returns the current state of instance.Cache, only for testing
func (s *Instancer) state() sd.Event {
	return s.cache.State()
}