From 47ebccf41a91cc54f2fa117c13a644dac11c71fc Mon Sep 17 00:00:00 2001 From: Periyasamy Palanisamy Date: Tue, 29 Aug 2023 17:29:46 +0200 Subject: [PATCH 1/3] Use Echo method to send inactivity probe This commit uses client's Echo method to send inactivity probe as it can be made as a time bound call so that rpc read lock can not be held indefinitely. This would make disconnect happens immediately when current ovsdb leader is accidentally gone away (this happens in a very rare scenario in which case sendEcho method returns with unexpected EOF error after 12 mins only) and reconnects with new ovsdb leader. Signed-off-by: Periyasamy Palanisamy --- client/client.go | 63 +++++++++-------------------------------------- client/options.go | 4 +++ 2 files changed, 16 insertions(+), 51 deletions(-) diff --git a/client/client.go b/client/client.go index 5d7217bb..33ceb7da 100644 --- a/client/client.go +++ b/client/client.go @@ -1190,72 +1190,33 @@ func (o *ovsdbClient) handleClientErrors(stopCh <-chan struct{}) { } } -func (o *ovsdbClient) sendEcho(args []any, reply *[]any) *rpc2.Call { - o.rpcMutex.RLock() - defer o.rpcMutex.RUnlock() - if o.rpcClient == nil { - return nil - } - return o.rpcClient.Go("echo", args, reply, make(chan *rpc2.Call, 1)) -} - func (o *ovsdbClient) handleInactivityProbes() { defer o.handlerShutdown.Done() - echoReplied := make(chan string) - var lastEcho string stopCh := o.stopCh trafficSeen := o.trafficSeen + timer := time.NewTimer(o.options.inactivityTimeout) for { select { case <-stopCh: return case <-trafficSeen: // We got some traffic from the server, restart our timer - case ts := <-echoReplied: - // Got a response from the server, check it against lastEcho; if same clear lastEcho; if not same Disconnect() - if ts != lastEcho { - o.Disconnect() - return + if !timer.Stop() { + <-timer.C } - lastEcho = "" - case <-time.After(o.options.inactivityTimeout): - // If there's a lastEcho already, then we didn't get a server reply, disconnect - if lastEcho != "" { - o.Disconnect() - return - } - // Otherwise send an echo - thisEcho := fmt.Sprintf("%d", time.Now().UnixMicro()) - args := []any{"libovsdb echo", thisEcho} - var reply []any - // Can't use o.Echo() because it blocks; we need the Call object direct from o.rpcClient.Go() - call := o.sendEcho(args, &reply) - if call == nil { - o.Disconnect() - return - } - lastEcho = thisEcho + case <-timer.C: + // Otherwise send an echo in a goroutine so that transactions don't block go func() { - // Wait for the echo reply - select { - case <-stopCh: - return - case <-call.Done: - if call.Error != nil { - // RPC timeout; disconnect - o.logger.V(3).Error(call.Error, "server echo reply error") - o.Disconnect() - } else if !reflect.DeepEqual(args, reply) { - o.logger.V(3).Info("warning: incorrect server echo reply", - "expected", args, "reply", reply) - o.Disconnect() - } else { - // Otherwise stuff thisEcho into the echoReplied channel - echoReplied <- thisEcho - } + ctx, cancel := context.WithTimeout(context.Background(), o.options.timeout) + err := o.Echo(ctx) + if err != nil { + o.logger.V(3).Error(err, "server echo reply error") + o.Disconnect() } + cancel() }() } + timer.Reset(o.options.inactivityTimeout) } } diff --git a/client/options.go b/client/options.go index 81ccffe2..6f9b7b17 100644 --- a/client/options.go +++ b/client/options.go @@ -2,6 +2,7 @@ package client import ( "crypto/tls" + "errors" "net/url" "time" @@ -120,6 +121,9 @@ func WithReconnect(timeout time.Duration, backoff backoff.BackOff) Option { func WithInactivityCheck(inactivityTimeout, reconnectTimeout time.Duration, reconnectBackoff backoff.BackOff) Option { return func(o *options) error { + if reconnectTimeout >= inactivityTimeout { + return errors.New("inactivity timeout value should be greater than reconnect timeout value") + } o.reconnect = true o.timeout = reconnectTimeout o.backoff = reconnectBackoff From 598b053d78b841d88a3e952465b43de56f4994ba Mon Sep 17 00:00:00 2001 From: Periyasamy Palanisamy Date: Wed, 8 Nov 2023 11:25:33 +0530 Subject: [PATCH 2/3] Set deadline for tcp connection This sets up tcp connection deadline for write and read requests to fail after a deadline has been exceeded. Signed-off-by: Periyasamy Palanisamy --- client/client.go | 34 ++++++++++++++++++++++++++-------- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/client/client.go b/client/client.go index 33ceb7da..48de2026 100644 --- a/client/client.go +++ b/client/client.go @@ -84,6 +84,7 @@ type ovsdbClient struct { metrics metrics connected bool rpcClient *rpc2.Client + conn net.Conn rpcMutex sync.RWMutex // endpoints contains all possible endpoints; the first element is // the active endpoint if connected=true @@ -420,6 +421,7 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) { if o.options.inactivityTimeout > 0 { o.trafficSeen = make(chan struct{}) } + o.conn = conn o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn)) o.rpcClient.SetBlocking(true) o.rpcClient.Handle("echo", func(_ *rpc2.Client, args []any, reply *[]any) error { @@ -741,7 +743,7 @@ func (o *ovsdbClient) update3(params []json.RawMessage, reply *[]any) error { func (o *ovsdbClient) getSchema(ctx context.Context, dbName string) (ovsdb.DatabaseSchema, error) { args := ovsdb.NewGetSchemaArgs(dbName) var reply ovsdb.DatabaseSchema - err := o.rpcClient.CallWithContext(ctx, "get_schema", args, &reply) + err := o.CallWithContext(ctx, "get_schema", args, &reply) if err != nil { if err == rpc2.ErrShutdown { return ovsdb.DatabaseSchema{}, ErrNotConnected @@ -756,7 +758,7 @@ func (o *ovsdbClient) getSchema(ctx context.Context, dbName string) (ovsdb.Datab // Should only be called when mutex is held func (o *ovsdbClient) listDbs(ctx context.Context) ([]string, error) { var dbs []string - err := o.rpcClient.CallWithContext(ctx, "list_dbs", nil, &dbs) + err := o.CallWithContext(ctx, "list_dbs", nil, &dbs) if err != nil { if err == rpc2.ErrShutdown { return nil, ErrNotConnected @@ -829,7 +831,7 @@ func (o *ovsdbClient) transact(ctx context.Context, dbName string, skipChWrite b if dbgLogger.Enabled() { dbgLogger.Info("transacting operations", "operations", fmt.Sprintf("%+v", operation)) } - err := o.rpcClient.CallWithContext(ctx, "transact", args, &reply) + err := o.CallWithContext(ctx, "transact", args, &reply) if err != nil { if err == rpc2.ErrShutdown { return nil, ErrNotConnected @@ -862,7 +864,7 @@ func (o *ovsdbClient) MonitorCancel(ctx context.Context, cookie MonitorCookie) e if o.rpcClient == nil { return ErrNotConnected } - err := o.rpcClient.CallWithContext(ctx, "monitor_cancel", args, &reply) + err := o.CallWithContext(ctx, "monitor_cancel", args, &reply) if err != nil { if err == rpc2.ErrShutdown { return ErrNotConnected @@ -974,15 +976,15 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne switch monitor.Method { case ovsdb.MonitorRPC: var reply ovsdb.TableUpdates - err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply) + err = o.CallWithContext(ctx, monitor.Method, args, &reply) tableUpdates = reply case ovsdb.ConditionalMonitorRPC: var reply ovsdb.TableUpdates2 - err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply) + err = o.CallWithContext(ctx, monitor.Method, args, &reply) tableUpdates = reply case ovsdb.ConditionalMonitorSinceRPC: var reply ovsdb.MonitorCondSinceReply - err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply) + err = o.CallWithContext(ctx, monitor.Method, args, &reply) if err == nil && reply.Found { monitor.LastTransactionID = reply.LastTransactionID lastTransactionFound = true @@ -1073,7 +1075,7 @@ func (o *ovsdbClient) Echo(ctx context.Context) error { if o.rpcClient == nil { return ErrNotConnected } - err := o.rpcClient.CallWithContext(ctx, "echo", args, &reply) + err := o.CallWithContext(ctx, "echo", args, &reply) if err != nil { if err == rpc2.ErrShutdown { return ErrNotConnected @@ -1432,3 +1434,19 @@ func (o *ovsdbClient) WhereAll(m model.Model, conditions ...model.Condition) Con func (o *ovsdbClient) WhereCache(predicate any) ConditionalAPI { return o.primaryDB().api.WhereCache(predicate) } + +// CallWithContext invokes the named function, waits for it to complete, and +// returns its error status, or an error from Context timeout. +func (o *ovsdbClient) CallWithContext(ctx context.Context, method string, args interface{}, reply interface{}) error { + // Set up read/write deadline for tcp connection before making + // a rpc request to the server. + if tcpConn, ok := o.conn.(*net.TCPConn); ok { + if o.options.timeout > 0 { + err := tcpConn.SetDeadline(time.Now().Add(o.options.timeout * 3)) + if err != nil { + return err + } + } + } + return o.rpcClient.CallWithContext(ctx, method, args, reply) +} From 8ba88585c6d25d603b160b5dc5493afeec3aa26e Mon Sep 17 00:00:00 2001 From: Periyasamy Palanisamy Date: Thu, 9 Nov 2023 12:28:09 +0530 Subject: [PATCH 3/3] Set user timeout for tcp connection This commit sets TCP_USER_TIMEOUT socket option for tcp connection so that channel write doesn't block indefinitely on network disconnect. Signed-off-by: Periyasamy Palanisamy --- client/client.go | 21 +++++++++++++++++++-- client/options.go | 1 + internal/syscall_linux.go | 31 +++++++++++++++++++++++++++++++ internal/syscall_nonlinux.go | 14 ++++++++++++++ 4 files changed, 65 insertions(+), 2 deletions(-) create mode 100644 internal/syscall_linux.go create mode 100644 internal/syscall_nonlinux.go diff --git a/client/client.go b/client/client.go index 48de2026..efd76b44 100644 --- a/client/client.go +++ b/client/client.go @@ -18,6 +18,7 @@ import ( "github.com/cenkalti/rpc2/jsonrpc" "github.com/go-logr/logr" "github.com/ovn-kubernetes/libovsdb/cache" + syscall "github.com/ovn-kubernetes/libovsdb/internal" "github.com/ovn-kubernetes/libovsdb/mapper" "github.com/ovn-kubernetes/libovsdb/model" "github.com/ovn-kubernetes/libovsdb/ovsdb" @@ -345,7 +346,10 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) (string, erro return "", fmt.Errorf("failed to open connection: %w", err) } - o.createRPC2Client(c) + err = o.createRPC2Client(c) + if err != nil { + return "", err + } serverDBNames, err := o.listDbs(ctx) if err != nil { @@ -416,12 +420,24 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) (string, erro // createRPC2Client creates an rpcClient using the provided connection // It is also responsible for setting up go routines for client-side event handling // Should only be called when the mutex is held -func (o *ovsdbClient) createRPC2Client(conn net.Conn) { +func (o *ovsdbClient) createRPC2Client(conn net.Conn) error { o.stopCh = make(chan struct{}) if o.options.inactivityTimeout > 0 { o.trafficSeen = make(chan struct{}) } o.conn = conn + // set TCP_USER_TIMEOUT socket option for connection so that + // channel write doesn't block indefinitely on network disconnect. + var userTimeout time.Duration + if o.options.timeout > 0 { + userTimeout = o.options.timeout * 3 + } else { + userTimeout = defaultTimeout + } + err := syscall.SetTCPUserTimeout(conn, userTimeout) + if err != nil { + return err + } o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn)) o.rpcClient.SetBlocking(true) o.rpcClient.Handle("echo", func(_ *rpc2.Client, args []any, reply *[]any) error { @@ -437,6 +453,7 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) { return o.update3(args, reply) }) go o.rpcClient.Run() + return nil } // isEndpointLeader returns true if the currently connected endpoint is leader, diff --git a/client/options.go b/client/options.go index 6f9b7b17..66a60b5f 100644 --- a/client/options.go +++ b/client/options.go @@ -15,6 +15,7 @@ const ( defaultTCPEndpoint = "tcp:127.0.0.1:6640" defaultSSLEndpoint = "ssl:127.0.0.1:6640" defaultUnixEndpoint = "unix:/var/run/openvswitch/ovsdb.sock" + defaultTimeout = 60 * time.Second ) type options struct { diff --git a/internal/syscall_linux.go b/internal/syscall_linux.go new file mode 100644 index 00000000..5138f22d --- /dev/null +++ b/internal/syscall_linux.go @@ -0,0 +1,31 @@ +package internal + +import ( + "fmt" + "net" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +// SetTCPUserTimeout sets the TCP user timeout on a connection's socket +func SetTCPUserTimeout(conn net.Conn, timeout time.Duration) error { + tcpconn, ok := conn.(*net.TCPConn) + if !ok { + // not a TCP connection. exit early + return nil + } + rawConn, err := tcpconn.SyscallConn() + if err != nil { + return fmt.Errorf("error getting raw connection: %v", err) + } + err = rawConn.Control(func(fd uintptr) { + err = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_USER_TIMEOUT, int(timeout/time.Millisecond)) + }) + if err != nil { + return fmt.Errorf("error setting option on socket: %v", err) + } + + return nil +} diff --git a/internal/syscall_nonlinux.go b/internal/syscall_nonlinux.go new file mode 100644 index 00000000..6e6a26d6 --- /dev/null +++ b/internal/syscall_nonlinux.go @@ -0,0 +1,14 @@ +//go:build !linux +// +build !linux + +package internal + +import ( + "net" + "time" +) + +// SetTCPUserTimeout is a no-op function under non-linux environments. +func SetTCPUserTimeout(conn net.Conn, timeout time.Duration) error { + return nil +}