diff --git a/client/client.go b/client/client.go index 5d7217bb..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" @@ -84,6 +85,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 @@ -344,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 { @@ -415,11 +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 { @@ -435,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, @@ -741,7 +760,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 +775,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 +848,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 +881,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 +993,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 +1092,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 @@ -1190,72 +1209,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) } } @@ -1471,3 +1451,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) +} diff --git a/client/options.go b/client/options.go index 81ccffe2..66a60b5f 100644 --- a/client/options.go +++ b/client/options.go @@ -2,6 +2,7 @@ package client import ( "crypto/tls" + "errors" "net/url" "time" @@ -14,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 { @@ -120,6 +122,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 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 +}