From 4bdc9a3e128b01b40ce61173a0199d72bb10c976 Mon Sep 17 00:00:00 2001 From: bryanqiu Date: Fri, 28 Oct 2022 17:24:46 +0800 Subject: [PATCH] initial version --- redis.go | 45 ++++++ sentinel/sentinel.go | 371 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 416 insertions(+) create mode 100644 redis.go create mode 100644 sentinel/sentinel.go diff --git a/redis.go b/redis.go new file mode 100644 index 0000000..725dc8d --- /dev/null +++ b/redis.go @@ -0,0 +1,45 @@ +package redis + +import ( + "fmt" + "time" + + "github.com/gomodule/redigo/redis" + "qoobing.com/gomod/redis/sentinel" +) + +func newSentinelPool() *redis.Pool { + sntnl := &sentinel.Sentinel{ + Addrs: []string{":26379", ":26380", ":26381"}, + MasterName: "mymaster", + Dial: func(addr string) (redis.Conn, error) { + timeout := 500 * time.Millisecond + c, err := redis.DialTimeout("tcp", addr, timeout, timeout, timeout) + if err != nil { + return nil, err + } + return c, nil + }, + } + return &redis.Pool{ + MaxIdle: 3, + MaxActive: 64, + Wait: true, + IdleTimeout: 240 * time.Second, + Dial: func() (redis.Conn, error) { + masterAddr, err := sntnl.MasterAddr() + if err != nil { + return nil, err + } + c, err := redis.Dial("tcp", masterAddr) + if err != nil { + return nil, err + } + if !sentinel.TestRole(c, "master") { + c.Close() + return nil, fmt.Errorf("%s is not redis master", masterAddr) + } + return c, nil + }, + } +} diff --git a/sentinel/sentinel.go b/sentinel/sentinel.go new file mode 100644 index 0000000..22e844a --- /dev/null +++ b/sentinel/sentinel.go @@ -0,0 +1,371 @@ +package sentinel + +import ( + "errors" + "fmt" + "net" + "strings" + "sync" + "time" + + "github.com/gomodule/redigo/redis" +) + +// Sentinel provides a way to add high availability (HA) to Redis Pool using +// preconfigured addresses of Sentinel servers and name of master which Sentinels +// monitor. It works with Redis >= 2.8.12 (mostly because of ROLE command that +// was introduced in that version, it's possible though to support old versions +// using INFO command). +// +// Example of the simplest usage to contact master "mymaster": +// +// func newSentinelPool() *redis.Pool { +// sntnl := &sentinel.Sentinel{ +// Addrs: []string{":26379", ":26380", ":26381"}, +// MasterName: "mymaster", +// Dial: func(addr string) (redis.Conn, error) { +// timeout := 500 * time.Millisecond +// c, err := redis.DialTimeout("tcp", addr, timeout, timeout, timeout) +// if err != nil { +// return nil, err +// } +// return c, nil +// }, +// } +// return &redis.Pool{ +// MaxIdle: 3, +// MaxActive: 64, +// Wait: true, +// IdleTimeout: 240 * time.Second, +// Dial: func() (redis.Conn, error) { +// masterAddr, err := sntnl.MasterAddr() +// if err != nil { +// return nil, err +// } +// c, err := redis.Dial("tcp", masterAddr) +// if err != nil { +// return nil, err +// } +// if !sentinel.TestRole(c, "master") { +// c.Close() +// return nil, fmt.Errorf("%s is not redis master", masterAddr) +// } +// return c, nil +// }, +// } +// } +type Sentinel struct { + // Addrs is a slice with known Sentinel addresses. + Addrs []string + + // MasterName is a name of Redis master Sentinel servers monitor. + MasterName string + + // Dial is a user supplied function to connect to Sentinel on given address. This + // address will be chosen from Addrs slice. + // Note that as per the redis-sentinel client guidelines, a timeout is mandatory + // while connecting to Sentinels, and should not be set to 0. + Dial func(addr string) (redis.Conn, error) + + // Pool is a user supplied function returning custom connection pool to Sentinel. + // This can be useful to tune options if you are not satisfied with what default + // Sentinel pool offers. See defaultPool() method for default pool implementation. + // In most cases you only need to provide Dial function and let this be nil. + Pool func(addr string) *redis.Pool + + mu sync.RWMutex + pools map[string]*redis.Pool + addr string +} + +// NoSentinelsAvailable is returned when all sentinels in the list are exhausted +// (or none configured), and contains the last error returned by Dial (which +// may be nil) +type NoSentinelsAvailable struct { + lastError error +} + +func (ns NoSentinelsAvailable) Error() string { + if ns.lastError != nil { + return fmt.Sprintf("redigo: no sentinels available; last error: %s", ns.lastError.Error()) + } + return fmt.Sprintf("redigo: no sentinels available") +} + +// putToTop puts Sentinel address to the top of address list - this means +// that all next requests will use Sentinel on this address first. +// +// From Sentinel guidelines: +// +// The first Sentinel replying to the client request should be put at the +// start of the list, so that at the next reconnection, we'll try first +// the Sentinel that was reachable in the previous connection attempt, +// minimizing latency. +func (s *Sentinel) putToTop(addr string) { + s.mu.Lock() + defer s.mu.Unlock() + for i, a := range s.Addrs { + if a == addr { + s.Addrs[0], s.Addrs[i] = s.Addrs[i], s.Addrs[0] + break + } + } +} + +// putToBottom puts Sentinel address to the bottom of address list. +// We call this method internally when see that some Sentinel failed to answer +// on application request so next time we start with another one. +func (s *Sentinel) putToBottom(addr string) { + s.mu.Lock() + defer s.mu.Unlock() + for i, a := range s.Addrs { + if a == addr { + copy(s.Addrs[i:], s.Addrs[i+1:]) + s.Addrs[len(s.Addrs)-1] = a + break + } + } +} + +// defaultPool returns a connection pool to one Sentinel. This allows +// us to call concurrent requests to Sentinel using connection Do method. +func (s *Sentinel) defaultPool(addr string) *redis.Pool { + return &redis.Pool{ + MaxIdle: 3, + MaxActive: 10, + Wait: true, + IdleTimeout: 240 * time.Second, + Dial: func() (redis.Conn, error) { + return s.Dial(addr) + }, + TestOnBorrow: func(c redis.Conn, t time.Time) error { + _, err := c.Do("PING") + return err + }, + } +} + +func (s *Sentinel) poolForAddr(addr string) *redis.Pool { + s.mu.Lock() + defer s.mu.Unlock() + if pool, ok := s.pools[addr]; ok { + return pool + } + s.mu.Unlock() + newPool := s.newPool(addr) + s.mu.Lock() + if pool, ok := s.pools[addr]; ok { + return pool + } + if s.pools == nil { + s.pools = make(map[string]*redis.Pool) + } + s.pools[addr] = newPool + return newPool +} + +func (s *Sentinel) newPool(addr string) *redis.Pool { + if s.Pool != nil { + return s.Pool(addr) + } + return s.defaultPool(addr) +} + +// close connection pool to Sentinel. +// Lock must be hold by caller. +func (s *Sentinel) close() { + for _, pool := range s.pools { + pool.Close() + } + s.pools = nil +} + +func (s *Sentinel) doUntilSuccess(f func(redis.Conn) (interface{}, error)) (interface{}, error) { + s.mu.RLock() + addrs := s.Addrs + s.mu.RUnlock() + + var lastErr error + + for _, addr := range addrs { + conn := s.poolForAddr(addr).Get() + reply, err := f(conn) + conn.Close() + if err != nil { + lastErr = err + s.putToBottom(addr) + continue + } + s.putToTop(addr) + return reply, nil + } + + return nil, NoSentinelsAvailable{lastError: lastErr} +} + +// MasterAddr returns an address of current Redis master instance. +func (s *Sentinel) MasterAddr() (string, error) { + return redis.String(s.doUntilSuccess(func(c redis.Conn) (interface{}, error) { + return queryForMaster(c, s.MasterName) + })) +} + +// SlaveAddrs returns a slice with known slave addresses of current master instance. +func (s *Sentinel) SlaveAddrs() ([]string, error) { + return redis.Strings(s.doUntilSuccess(func(c redis.Conn) (interface{}, error) { + return queryForSlaveAddrs(c, s.MasterName) + })) +} + +// Slave represents a Redis slave instance which is known by Sentinel. +type Slave struct { + ip string + port string + flags string +} + +// Addr returns an address of slave. +func (s *Slave) Addr() string { + return net.JoinHostPort(s.ip, s.port) +} + +// Available returns if slave is in working state at moment based on information in slave flags. +func (s *Slave) Available() bool { + return !strings.Contains(s.flags, "disconnected") && !strings.Contains(s.flags, "s_down") +} + +// Slaves returns a slice with known slaves of master instance. +func (s *Sentinel) Slaves() ([]*Slave, error) { + res, err := s.doUntilSuccess(func(c redis.Conn) (interface{}, error) { + return queryForSlaves(c, s.MasterName) + }) + if err != nil { + return nil, err + } + return res.([]*Slave), nil +} + +// SentinelAddrs returns a slice of known Sentinel addresses Sentinel server aware of. +func (s *Sentinel) SentinelAddrs() ([]string, error) { + return redis.Strings(s.doUntilSuccess(func(c redis.Conn) (interface{}, error) { + return queryForSentinels(c, s.MasterName) + })) +} + +// Discover allows to update list of known Sentinel addresses. From docs: +// +// A client may update its internal list of Sentinel nodes following this procedure: +// 1) Obtain a list of other Sentinels for this master using the command SENTINEL sentinels . +// 2) Add every ip:port pair not already existing in our list at the end of the list. +func (s *Sentinel) Discover() error { + addrs, err := s.SentinelAddrs() + if err != nil { + return err + } + s.mu.Lock() + for _, addr := range addrs { + if !stringInSlice(addr, s.Addrs) { + s.Addrs = append(s.Addrs, addr) + } + } + s.mu.Unlock() + return nil +} + +// Close closes current connection to Sentinel. +func (s *Sentinel) Close() error { + s.mu.Lock() + s.close() + s.mu.Unlock() + return nil +} + +// TestRole wraps GetRole in a test to verify if the role matches an expected +// role string. If there was any error in querying the supplied connection, +// the function returns false. Works with Redis >= 2.8.12. +// It's not goroutine safe, but if you call this method on pooled connections +// then you are OK. +func TestRole(c redis.Conn, expectedRole string) bool { + role, err := getRole(c) + return err == nil && role == expectedRole +} + +// getRole is a convenience function supplied to query an instance (master or +// slave) for its role. It attempts to use the ROLE command introduced in +// redis 2.8.12. +func getRole(c redis.Conn) (string, error) { + res, err := c.Do("ROLE") + if err != nil { + return "", err + } + rres, ok := res.([]interface{}) + if ok { + return redis.String(rres[0], nil) + } + return "", errors.New("redigo: can not transform ROLE reply to string") +} + +func queryForMaster(conn redis.Conn, masterName string) (string, error) { + res, err := redis.Strings(conn.Do("SENTINEL", "get-master-addr-by-name", masterName)) + if err != nil { + return "", err + } + if len(res) < 2 { + return "", errors.New("redigo: malformed get-master-addr-by-name reply") + } + masterAddr := net.JoinHostPort(res[0], res[1]) + return masterAddr, nil +} + +func queryForSlaveAddrs(conn redis.Conn, masterName string) ([]string, error) { + slaves, err := queryForSlaves(conn, masterName) + if err != nil { + return nil, err + } + slaveAddrs := make([]string, len(slaves)) + for i, slave := range slaves { + slaveAddrs[i] = slave.Addr() + } + return slaveAddrs, nil +} + +func queryForSlaves(conn redis.Conn, masterName string) ([]*Slave, error) { + res, err := redis.Values(conn.Do("SENTINEL", "slaves", masterName)) + if err != nil { + return nil, err + } + slaves := make([]*Slave, len(res)) + for i, a := range res { + sm, err := redis.StringMap(a, err) + if err != nil { + return nil, err + } + slaves[i] = &Slave{ip: sm["ip"], port: sm["port"], flags: sm["flags"]} + } + return slaves, nil +} + +func queryForSentinels(conn redis.Conn, masterName string) ([]string, error) { + res, err := redis.Values(conn.Do("SENTINEL", "sentinels", masterName)) + if err != nil { + return nil, err + } + sentinels := make([]string, len(res)) + for i, a := range res { + sm, err := redis.StringMap(a, err) + if err != nil { + return nil, err + } + sentinels[i] = fmt.Sprintf("%s:%s", sm["ip"], sm["port"]) + } + return sentinels, nil +} + +func stringInSlice(str string, slice []string) bool { + for _, s := range slice { + if s == str { + return true + } + } + return false +}