-
Notifications
You must be signed in to change notification settings - Fork 58
/
consul.go
119 lines (106 loc) · 2.88 KB
/
consul.go
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
package consul
import (
"context"
"fmt"
"sort"
"time"
"github.com/hashicorp/consul/api"
"github.com/jpillora/backoff"
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/resolver"
)
// init function needs for auto-register in resolvers registry
func init() {
resolver.Register(&builder{})
}
// consulResolver watch for enpoints changes and push to the GRPC only diffs
// consulResolver implements resolver.Resolver from the GRPC package
type resolvr struct {
cancelFunc context.CancelFunc
}
// ResolveNow will be skipped due unnecessary in this case
func (r *resolvr) ResolveNow(resolver.ResolveNowOption) {}
// Close closes the resolver.
func (r *resolvr) Close() {
r.cancelFunc()
}
type servicer interface {
Service(string, string, bool, *api.QueryOptions) ([]*api.ServiceEntry, *api.QueryMeta, error)
}
func watchConsulService(ctx context.Context, s servicer, tgt target, out chan<- []string) {
res := make(chan []string)
bck := &backoff.Backoff{
Factor: 2,
Jitter: true,
Min: 10 * time.Millisecond,
Max: tgt.MaxBackoff,
}
go func() {
var lastIndex uint64
for {
ss, meta, err := s.Service(
tgt.Service,
tgt.Tag,
tgt.Healthy,
&api.QueryOptions{
WaitIndex: lastIndex,
Near: tgt.Near,
WaitTime: tgt.Wait,
},
)
if err != nil {
grpclog.Errorf("[Consul resolver] Couldn't fetch endpoints. target={%s}", tgt.String())
time.Sleep(bck.Duration())
continue
}
bck.Reset()
lastIndex = meta.LastIndex
grpclog.Infof("[Consul resolver] %d endpoints fetched in(+wait) %s for target={%s}",
len(ss),
meta.RequestTime,
tgt.String(),
)
ee := make([]string, 0, len(ss))
for _, s := range ss {
ee = append(ee, fmt.Sprintf("%s:%d", s.Service.Address, s.Service.Port))
}
if tgt.Limit != 0 && len(ee) > tgt.Limit {
ee = ee[:tgt.Limit]
}
res <- ee
}
}()
for {
select {
case ee := <-res:
out <- ee
case <-ctx.Done():
return
}
}
}
func populateEndpoints(ctx context.Context, clientConn resolver.ClientConn, input <-chan []string) {
for {
select {
case cc := <-input:
connsSet := make(map[string]struct{}, len(cc))
for _, c := range cc {
connsSet[c] = struct{}{}
}
conns := make([]resolver.Address, 0, len(connsSet))
for c := range connsSet {
conns = append(conns, resolver.Address{Addr: c})
}
sort.Sort(byAddressString(conns)) // Don't replace the same address list in the balancer
clientConn.NewAddress(conns)
case <-ctx.Done():
grpclog.Info("[Consul resolver] Watch has been finished")
return
}
}
}
// byAddressString sorts resolver.Address by Address Field sorting in increasing order.
type byAddressString []resolver.Address
func (p byAddressString) Len() int { return len(p) }
func (p byAddressString) Less(i, j int) bool { return p[i].Addr < p[j].Addr }
func (p byAddressString) Swap(i, j int) { p[i], p[j] = p[j], p[i] }