forked from mediocregopher/radix.v2
/
sentinel.go
292 lines (260 loc) · 7.5 KB
/
sentinel.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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
// Package sentinel provides a convenient interface with a redis sentinel which
// will automatically handle pooling connections and automatic failover.
//
// Here's an example of creating a sentinel client and then using it to perform
// some commands
//
// func example() error {
// // If there exists sentinel masters "bucket0" and "bucket1", and we want
// // out client to create pools for both:
// client, err := sentinel.NewClient("tcp", "localhost:6379", 100, "bucket0", "bucket1")
// if err != nil {
// return err
// }
//
// if err := exampleCmd(client); err != nil {
// return err
// }
//
// return nil
// }
//
// func exampleCmd(client *sentinel.Client) error {
// conn, err := client.GetMaster("bucket0")
// if err != nil {
// return redisErr
// }
// defer client.PutMaster("bucket0", conn)
//
// i, err := conn.Cmd("GET", "foo").Int()
// if err != nil {
// return err
// }
//
// if err := conn.Cmd("SET", "foo", i+1); err != nil {
// return err
// }
//
// return nil
// }
//
// This package only guarantees that when GetMaster is called the returned
// connection will be a connection to the master as of the moment that method is
// called. It is still possible that there is a failover as that connection is
// being used by the application.
//
// As a final note, a Client can be interacted with from multiple routines at
// once safely, except for the Close method. To safely Close, ensure that only
// one routine ever makes the call and that once the call is made no other
// methods are ever called by any routines.
package sentinel
import (
"errors"
"strings"
"github.com/mediocregopher/radix.v2/pool"
"github.com/mediocregopher/radix.v2/pubsub"
"github.com/mediocregopher/radix.v2/redis"
)
// ClientError is an error wrapper returned by operations in this package. It
// implements the error interface and can therefore be passed around as a normal
// error.
type ClientError struct {
err error
// If this is true the error is due to a problem with the sentinel
// connection, either it being closed or otherwise unavailable. If false the
// error is due to some other circumstances. This is useful if you want to
// implement some kind of reconnecting to sentinel on an error.
SentinelErr bool
}
// Error implements the error protocol
func (ce *ClientError) Error() string {
return ce.err.Error()
}
type getReqRet struct {
conn *redis.Client
err *ClientError
}
type getReq struct {
name string
retCh chan *getReqRet
}
type putReq struct {
name string
conn *redis.Client
}
type switchMaster struct {
name string
addr string
}
// Client communicates with a sentinel instance and manages connection pools of
// active masters
type Client struct {
poolSize int
masterPools map[string]*pool.Pool
subClient *pubsub.SubClient
// This is pool.DialFunc instead of the package's DialFunc
// as it's only used when calling pool.NewCustom. Otherwise it
// will have to be cast on each invocation.
dialFunc pool.DialFunc
getCh chan *getReq
putCh chan *putReq
closeCh chan struct{}
alwaysErr *ClientError
alwaysErrCh chan *ClientError
switchMasterCh chan *switchMaster
}
// DialFunc is a function which can be passed into NewClientCustom
type DialFunc func(network, addr string) (*redis.Client, error)
// NewClient creates a sentinel client. Connects to the given sentinel instance,
// pulls the information for the masters of the given names, and creates an
// initial pool of connections for each master. The client will automatically
// replace the pool for any master should sentinel decide to fail the master
// over. The returned error is a *ClientError.
func NewClient(
network, address string, poolSize int, names ...string,
) (
*Client, error,
) {
return NewClientCustom(network, address, poolSize, redis.Dial, names...)
}
// NewClientCustom is the same as NewClient, except it takes in a DialFunc which
// will be used to create all new connections to the master instances. This can
// be used to implement authentication, custom timeouts, etc...
func NewClientCustom(
network, address string, poolSize int, df DialFunc, names ...string,
) (
*Client, error,
) {
// We use this to fetch initial details about masters before we upgrade it
// to a pubsub client
client, err := redis.Dial(network, address)
if err != nil {
return nil, &ClientError{err: err}
}
masterPools := map[string]*pool.Pool{}
for _, name := range names {
r := client.Cmd("SENTINEL", "MASTER", name)
l, err := r.List()
if err != nil {
return nil, &ClientError{err: err, SentinelErr: true}
}
addr := l[3] + ":" + l[5]
pool, err := pool.NewCustom("tcp", addr, poolSize, (pool.DialFunc)(df))
if err != nil {
return nil, &ClientError{err: err}
}
masterPools[name] = pool
}
subClient := pubsub.NewSubClient(client)
r := subClient.Subscribe("+switch-master")
if r.Err != nil {
return nil, &ClientError{err: r.Err, SentinelErr: true}
}
c := &Client{
poolSize: poolSize,
masterPools: masterPools,
subClient: subClient,
dialFunc: (pool.DialFunc)(df),
getCh: make(chan *getReq),
putCh: make(chan *putReq),
closeCh: make(chan struct{}),
alwaysErrCh: make(chan *ClientError),
switchMasterCh: make(chan *switchMaster),
}
go c.subSpin()
go c.spin()
return c, nil
}
func (c *Client) subSpin() {
alwaysErr := func(err error) {
select {
case c.alwaysErrCh <- &ClientError{err: err, SentinelErr: true}:
case <-c.closeCh:
}
}
for {
if err := c.subClient.Ping().Err; err != nil {
alwaysErr(err)
return
}
r := c.subClient.Receive()
if r.Timeout() {
continue
}
if r.Err != nil {
alwaysErr(r.Err)
return
}
sMsg := strings.Split(r.Message, " ")
name := sMsg[0]
newAddr := sMsg[3] + ":" + sMsg[4]
select {
case c.switchMasterCh <- &switchMaster{name, newAddr}:
case <-c.closeCh:
return
}
}
}
func (c *Client) spin() {
for {
select {
case req := <-c.getCh:
if c.alwaysErr != nil {
req.retCh <- &getReqRet{nil, c.alwaysErr}
continue
}
pool, ok := c.masterPools[req.name]
if !ok {
err := errors.New("unknown name: " + req.name)
req.retCh <- &getReqRet{nil, &ClientError{err: err}}
continue
}
conn, err := pool.Get()
if err != nil {
req.retCh <- &getReqRet{nil, &ClientError{err: err}}
continue
}
req.retCh <- &getReqRet{conn, nil}
case req := <-c.putCh:
if pool, ok := c.masterPools[req.name]; ok {
pool.Put(req.conn)
}
case err := <-c.alwaysErrCh:
c.alwaysErr = err
case sm := <-c.switchMasterCh:
if p, ok := c.masterPools[sm.name]; ok {
p.Empty()
p, _ = pool.NewCustom("tcp", sm.addr, c.poolSize, c.dialFunc)
c.masterPools[sm.name] = p
}
case <-c.closeCh:
for name := range c.masterPools {
c.masterPools[name].Empty()
}
c.subClient.Client.Close()
close(c.getCh)
close(c.putCh)
return
}
}
}
// GetMaster retrieves a connection for the master of the given name. If
// sentinel has become unreachable this will always return an error. Close
// should be called in that case. The returned error is a *ClientError.
func (c *Client) GetMaster(name string) (*redis.Client, error) {
req := getReq{name, make(chan *getReqRet)}
c.getCh <- &req
ret := <-req.retCh
if ret.err != nil {
return nil, ret.err
}
return ret.conn, nil
}
// PutMaster return a connection for a master of a given name
func (c *Client) PutMaster(name string, client *redis.Client) {
c.putCh <- &putReq{name, client}
}
// Close closes all connection pools as well as the connection to sentinel.
func (c *Client) Close() {
close(c.closeCh)
}