Better handling of moving sentinel servers

This commit is contained in:
dave 2019-03-05 23:19:41 -08:00
parent d51e77efdd
commit d7b8e82b5c
1 changed files with 149 additions and 87 deletions

236
main.go
View File

@ -1,119 +1,181 @@
package main package main
import ( import (
"errors" "errors"
"flag" "flag"
"fmt" "fmt"
"io" "io"
"log" "log"
"net" "net"
"strings" "strings"
"time" "sync"
"time"
) )
var ( var (
masterAddr *net.TCPAddr saddr *net.TCPAddr // Address of the sentinel service
raddr *net.TCPAddr slock *sync.Mutex // Guard for above var
saddr *net.TCPAddr
localAddr = flag.String("listen", ":9999", "local address") masterAddr *net.TCPAddr // Address of the redis master
sentinelAddr = flag.String("sentinel", ":26379", "remote address") mlock *sync.Mutex // Guard for above var
masterName = flag.String("master", "", "name of the master redis node")
localAddr = flag.String("listen", ":9999", "local address")
sentinelAddr = flag.String("sentinel", ":26379", "remote address")
masterName = flag.String("master", "", "name of the master redis node")
) )
func main() { func main() {
flag.Parse() flag.Parse()
laddr, err := net.ResolveTCPAddr("tcp", *localAddr) slock = &sync.Mutex{}
if err != nil { mlock = &sync.Mutex{}
log.Fatal("Failed to resolve local address: %s", err)
}
saddr, err = net.ResolveTCPAddr("tcp", *sentinelAddr)
if err != nil {
log.Fatal("Failed to resolve sentinel address: %s", err)
}
go master() laddr, err := net.ResolveTCPAddr("tcp", *localAddr)
if err != nil {
log.Fatalf("Failed to resolve local address: %s", err.Error())
}
resolveSentinel(*sentinelAddr)
listener, err := net.ListenTCP("tcp", laddr) // If sentinel's address is set to nil, this goroutine will resolve it and set the var again
if err != nil { go sentinelUpdater(*sentinelAddr)
log.Fatal(err)
}
for { // Continuously query sentinel for the master address, updating masterAddr when needed
conn, err := listener.AcceptTCP() go master()
if err != nil {
log.Println(err)
continue
}
go proxy(conn, masterAddr) listener, err := net.ListenTCP("tcp", laddr)
} if err != nil {
log.Fatal(err)
}
for {
conn, err := listener.AcceptTCP()
if err != nil {
log.Println(err)
continue
}
go proxy(conn, masterAddr)
}
}
func sentinelUpdater(sentinelAddr string) {
// Resolve the address of sentinel when needed
for {
if saddr == nil {
log.Print("Resolving sentinel address")
resolveSentinel(sentinelAddr)
}
time.Sleep(1 * time.Second)
}
}
func resolveSentinel(sentinelAddr string) {
// var err error
addr, err := net.ResolveTCPAddr("tcp", sentinelAddr)
if err != nil {
log.Printf("Failed to resolve sentinel address: %s", err.Error())
return
}
slock.Lock()
saddr = addr
slock.Unlock()
// TODO other cases when saddr isn't valid
} }
func master() { func master() {
var err error var err error
for { var tempAddr *net.TCPAddr
masterAddr, err = getMasterAddr(saddr, *masterName) for {
if err != nil { tempAddr, err = getMasterAddr()
log.Println(err) if err != nil {
} log.Printf("Failed to get master addres: %s", err.Error())
time.Sleep(1 * time.Second) } else {
} mlock.Lock()
masterAddr = tempAddr
mlock.Unlock()
}
time.Sleep(1 * time.Second)
}
} }
func pipe(r io.Reader, w io.WriteCloser) { func pipe(r io.Reader, w io.WriteCloser) {
io.Copy(w, r) io.Copy(w, r)
w.Close() w.Close()
} }
func proxy(local io.ReadWriteCloser, remoteAddr *net.TCPAddr) { func proxy(local io.ReadWriteCloser, remoteAddr *net.TCPAddr) {
remote, err := net.DialTCP("tcp", nil, remoteAddr) remote, err := net.DialTCP("tcp", nil, remoteAddr)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
local.Close() local.Close()
return return
} }
go pipe(local, remote) go pipe(local, remote)
go pipe(remote, local) go pipe(remote, local)
} }
func getMasterAddr(sentinelAddress *net.TCPAddr, masterName string) (*net.TCPAddr, error) { // Connect to Sentinel and query it to find the redis master
conn, err := net.DialTCP("tcp", nil, sentinelAddress) func getMasterAddr() (*net.TCPAddr, error) {
if err != nil { // Connect to sentinel
return nil, err // If the connection times out, that master is probably gone.
} // Mark saddr as nil so that the resolver thread will update it later.
// Create a local copy of the sentinel address, it can change under our feet
slock.Lock()
if saddr == nil {
defer slock.Unlock()
return nil, errors.New("Sentinel address not available")
}
local_saddr := *saddr
slock.Unlock()
defer conn.Close() sentConn, err := dialTimeout(&local_saddr, 5*time.Second)
if err != nil {
log.Printf("Connecting to sentinel master timed out/failed: %s\n", err.Error())
slock.Lock()
saddr = nil
slock.Unlock()
return nil, err
}
defer sentConn.Close()
conn.Write([]byte(fmt.Sprintf("sentinel get-master-addr-by-name %s\n", masterName))) // We connected to the master, ask for the redis master
sentConn.Write([]byte(fmt.Sprintf("sentinel get-master-addr-by-name %s\n", *masterName)))
b := make([]byte, 256) b := make([]byte, 256)
_, err = conn.Read(b) _, err = sentConn.Read(b)
if err != nil { if err != nil {
log.Fatal(err) return nil, err
} }
parts := strings.Split(string(b), "\r\n")
if len(parts) < 5 {
err = errors.New("Couldn't get master address from sentinel")
return nil, err
}
parts := strings.Split(string(b), "\r\n") // Parse the address for the master node
stringaddr := fmt.Sprintf("%s:%s", parts[2], parts[4])
addr, err := net.ResolveTCPAddr("tcp", stringaddr)
if err != nil {
return nil, err
}
if len(parts) < 5 { // Verify the returned address is actually listening
err = errors.New("Couldn't get master address from sentinel") // TODO is this really needed?
return nil, err conn2, err := dialTimeout(addr, 5*time.Second)
} if err != nil {
return nil, err
//getting the string address for the master node }
stringaddr := fmt.Sprintf("%s:%s", parts[2], parts[4]) defer conn2.Close()
addr, err := net.ResolveTCPAddr("tcp", stringaddr) return addr, err
}
if err != nil {
return nil, err // Connect to a TCPAddr, failing if a timeout is exceeded or other error encountered
} func dialTimeout(destAddr *net.TCPAddr, timeout time.Duration) (*net.TCPConn, error) {
d := net.Dialer{Timeout: timeout}
//check that there's actually someone listening on that address netcon, err := d.Dial("tcp", fmt.Sprintf("%s:%d", destAddr.IP, destAddr.Port))
conn2, err := net.DialTCP("tcp", nil, addr) if err != nil {
if err == nil { return nil, err
defer conn2.Close() }
} conn, _ := netcon.(*net.TCPConn)
return conn, nil
return addr, err
} }