Codebase list golang-github-go-kit-kit / 4b48129
Add TTL for etcd sd miao 7 years ago
3 changed file(s) with 48 addition(s) and 1 deletion(s). Raw diff Collapse all Expand all
+4
-0
sd/etcd/client.go less more
155155 if s.Value == "" {
156156 return ErrNoValue
157157 }
158 if s.TTL != nil {
159 _, err := c.keysAPI.Set(c.ctx, s.Key, s.Value, &etcd.SetOptions{PrevExist: etcd.PrevIgnore, TTL: s.TTL.TTL})
160 return err
161 }
158162 _, err := c.keysAPI.Create(c.ctx, s.Key, s.Value)
159163 return err
160164 }
+43
-0
sd/etcd/registrar.go less more
33 etcd "github.com/coreos/etcd/client"
44
55 "github.com/go-kit/kit/log"
6 "time"
7 )
8
9 const (
10 MinHeartBeatTime = time.Millisecond * 500
611 )
712
813 // Registrar registers service instance liveness information to etcd.
1015 client Client
1116 service Service
1217 logger log.Logger
18 quit chan struct{}
1319 }
1420
1521 // Service holds the instance identifying data you want to publish to etcd. Key
1824 type Service struct {
1925 Key string // unique key, e.g. "/service/foobar/1.2.3.4:8080"
2026 Value string // returned to subscribers, e.g. "http://1.2.3.4:8080"
27 TTL *TTLOption
2128 DeleteOptions *etcd.DeleteOptions
29 }
30
31 // TTLOption allow setting a key with a TTL, and regularly refreshes the lease with a goroutine
32 type TTLOption struct {
33 Heartbeat time.Duration
34 TTL time.Duration
35 }
36
37 // NewTTLOption returns a TTLOption
38 func NewTTLOption(heartbeat, ttl time.Duration) *TTLOption {
39 if heartbeat <= MinHeartBeatTime {
40 heartbeat = MinHeartBeatTime
41 }
42 if ttl <= heartbeat {
43 ttl = heartbeat * 3
44 }
45 return &TTLOption{heartbeat, ttl}
2246 }
2347
2448 // NewRegistrar returns a etcd Registrar acting on the provided catalog
4266 } else {
4367 r.logger.Log("action", "register")
4468 }
69 if r.service.TTL == nil {
70 return
71 }
72 r.quit = make(chan struct{})
73 go func() {
74 for {
75 select {
76 case <-r.quit:
77 return
78 case <-time.After(r.service.TTL.Heartbeat):
79 if err := r.client.Register(r.service); err != nil {
80 r.logger.Log("err", err)
81 }
82 }
83 }
84 }()
4585 }
4686
4787 // Deregister implements the sd.Registrar interface. Call it when you want your
5292 } else {
5393 r.logger.Log("action", "deregister")
5494 }
95 if r.quit != nil {
96 close(r.quit)
97 }
5598 }
+1
-1
sd/etcd/registrar_test.go less more
2929 }
3030
3131 // default service used to build registrar in our tests
32 var testService = Service{"testKey", "testValue", nil}
32 var testService = Service{"testKey", "testValue", nil, nil}
3333
3434 // NewRegistar should return a registar with a logger using the service key and value
3535 func TestNewRegistar(t *testing.T) {