mirror of
https://github.com/redis/go-redis.git
synced 2026-10-24 07:59:20 +03:00
fix(sentinel): bound failover dials by DialTimeout (#4067)
* fix(sentinel): bound failover dials by DialTimeout * test(sentinel): cover Subscribe dial timeout
This commit is contained in:
committed by
GitHub
parent
5a9916dd5a
commit
cfbdbccf35
@@ -589,7 +589,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
|
||||
}
|
||||
|
||||
opt := failoverOpt.clientOptions()
|
||||
opt.Dialer = masterReplicaDialer(failover)
|
||||
opt.Dialer = masterReplicaDialer(failover, opt)
|
||||
opt.init()
|
||||
|
||||
rdb := &Client{
|
||||
@@ -702,6 +702,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
|
||||
|
||||
func masterReplicaDialer(
|
||||
failover *sentinelFailover,
|
||||
opt *Options,
|
||||
) func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
return func(ctx context.Context, network, _ string) (net.Conn, error) {
|
||||
var addr string
|
||||
@@ -723,7 +724,7 @@ func masterReplicaDialer(
|
||||
}
|
||||
|
||||
netDialer := &net.Dialer{
|
||||
Timeout: failover.opt.DialTimeout,
|
||||
Timeout: opt.DialTimeout,
|
||||
KeepAliveConfig: defaultKeepAliveConfig,
|
||||
}
|
||||
if failover.opt.TLSConfig == nil {
|
||||
|
||||
143
sentinel_dial_timeout_test.go
Normal file
143
sentinel_dial_timeout_test.go
Normal file
@@ -0,0 +1,143 @@
|
||||
package redis_test
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/ecdsa"
|
||||
"crypto/elliptic"
|
||||
"crypto/rand"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"math/big"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// TestFailoverClientSubscribeBoundedByDialTimeout checks that the master dial
|
||||
// uses the default DialTimeout when FailoverOptions.DialTimeout is unset. The
|
||||
// pub/sub pool passes the caller ctx to the dialer as is, so an unbounded dial
|
||||
// made Subscribe hang. The fake master accepts TCP but never completes the TLS
|
||||
// handshake, which stands in for an unreachable host without network tricks.
|
||||
func TestFailoverClientSubscribeBoundedByDialTimeout(t *testing.T) {
|
||||
cert := selfSignedCert(t)
|
||||
|
||||
master, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
stalled := make(chan net.Conn, 16)
|
||||
defer func() {
|
||||
master.Close()
|
||||
for {
|
||||
select {
|
||||
case conn := <-stalled:
|
||||
conn.Close()
|
||||
default:
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
for {
|
||||
conn, err := master.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
stalled <- conn
|
||||
}
|
||||
}()
|
||||
masterHost, masterPort, _ := net.SplitHostPort(master.Addr().String())
|
||||
|
||||
sentinel, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{cert}})
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
defer sentinel.Close()
|
||||
go func() {
|
||||
for {
|
||||
conn, err := sentinel.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
go serveFakeSentinel(conn, masterHost, masterPort)
|
||||
}
|
||||
}()
|
||||
|
||||
client := redis.NewFailoverClient(&redis.FailoverOptions{
|
||||
MasterName: "mymaster",
|
||||
SentinelAddrs: []string{sentinel.Addr().String()},
|
||||
TLSConfig: &tls.Config{InsecureSkipVerify: true},
|
||||
})
|
||||
defer client.Close()
|
||||
|
||||
dialTimeout := client.Options().DialTimeout
|
||||
errc := make(chan error, 1)
|
||||
start := time.Now()
|
||||
go func() {
|
||||
pubsub := client.Subscribe(context.Background())
|
||||
defer pubsub.Close()
|
||||
errc <- pubsub.Subscribe(context.Background(), "ch")
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-errc:
|
||||
if err == nil {
|
||||
t.Fatal("expected a dial error, got nil")
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed > dialTimeout+time.Second {
|
||||
t.Fatalf("Subscribe took %s, want about %s", elapsed, dialTimeout)
|
||||
}
|
||||
case <-time.After(2 * dialTimeout):
|
||||
t.Fatalf("Subscribe did not return within %s", 2*dialTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func serveFakeSentinel(conn net.Conn, masterHost, masterPort string) {
|
||||
defer conn.Close()
|
||||
rd := bufio.NewReader(conn)
|
||||
for {
|
||||
args, err := readCommand(rd)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var reply string
|
||||
switch {
|
||||
case len(args) == 3 && strings.EqualFold(args[0], "sentinel") &&
|
||||
strings.EqualFold(args[1], "get-master-addr-by-name"):
|
||||
reply = "*2\r\n$" + strconv.Itoa(len(masterHost)) + "\r\n" + masterHost + "\r\n$" +
|
||||
strconv.Itoa(len(masterPort)) + "\r\n" + masterPort + "\r\n"
|
||||
case strings.EqualFold(args[0], "psubscribe"):
|
||||
// Leave the sentinel pub/sub connection idle.
|
||||
continue
|
||||
default:
|
||||
reply = "-ERR unknown command\r\n"
|
||||
}
|
||||
if _, err := conn.Write([]byte(reply)); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func selfSignedCert(t *testing.T) tls.Certificate {
|
||||
t.Helper()
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatalf("generate key: %v", err)
|
||||
}
|
||||
tmpl := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(1),
|
||||
NotBefore: time.Now().Add(-time.Hour),
|
||||
NotAfter: time.Now().Add(time.Hour),
|
||||
IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1)},
|
||||
}
|
||||
der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
|
||||
if err != nil {
|
||||
t.Fatalf("create certificate: %v", err)
|
||||
}
|
||||
return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
|
||||
}
|
||||
Reference in New Issue
Block a user