ovsdb: implement background echo loop in Client

Author: mdlayher

ovsdb/client.go

@@ -23,12 +23,15 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/digitalocean/go-openvswitch/ovsdb/internal/jsonrpc"
 )
 
-// A Client is an OVSDB client.
+// A Client is an OVSDB client.  Clients can be customized by using OptionFuncs
+// in the Dial and New functions.
 type Client struct {
+	// The RPC connection, and its logger.
 	c  *jsonrpc.Conn
 	ll *log.Logger
 
@@ -39,7 +42,12 @@ type Client struct {
 	cbMu      sync.RWMutex
 	callbacks map[int]callback
 
-	wg *sync.WaitGroup
+	// Interval at which echo RPCs should occur in the background.
+	echoInterval time.Duration
+
+	// Track and clean up background goroutines.
+	cancel func()
+	wg     *sync.WaitGroup
 }
 
 // An OptionFunc is a function which can configure a Client.
@@ -53,6 +61,18 @@ func Debug(ll *log.Logger) OptionFunc {
 	}
 }
 
+// EchoInterval specifies an interval at which the Client will send
+// echo RPCs to an OVSDB server to keep the connection alive.  If this
+// option is not used, the Client will not send any echo RPCs on its own.
+//
+// Specify a duration of 0 to disable sending background echo RPCs.
+func EchoInterval(d time.Duration) OptionFunc {
+	return func(c *Client) error {
+		c.echoInterval = d
+		return nil
+	}
+}
+
 // Dial dials a connection to an OVSDB server and returns a Client.
 func Dial(network, addr string, options ...OptionFunc) (*Client, error) {
 	conn, err := net.Dial(network, addr)
@@ -82,10 +102,22 @@ func New(conn net.Conn, options ...OptionFunc) (*Client, error) {
 	// Set up callbacks.
 	client.callbacks = make(map[int]callback)
 
-	// Start up any background routines.
+	// Start up any background routines, and enable canceling them via context.
+	ctx, cancel := context.WithCancel(context.Background())
+	client.cancel = cancel
+
 	var wg sync.WaitGroup
 	wg.Add(1)
 
+	// If configured, send echo RPCs in the background at a fixed interval.
+	if d := client.echoInterval; d != 0 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			client.echoLoop(ctx, d)
+		}()
+	}
+
 	// Handle all incoming RPC responses and notifications.
 	go func() {
 		defer wg.Done()
@@ -102,9 +134,10 @@ func (c *Client) requestID() int {
 	return int(atomic.AddInt64(c.rpcID, 1))
 }
 
-// Close closes a Client's connection.
+// Close closes a Client's connection and cleans up its resources.
 func (c *Client) Close() error {
 	err := c.c.Close()
+	c.cancel()
 	c.wg.Wait()
 	return err
 }
@@ -229,6 +262,25 @@ func (c *Client) listen() {
 	}
 }
 
+// echoLoop starts a loop that sends echo RPCs at the interval defined by d.
+func (c *Client) echoLoop(ctx context.Context, d time.Duration) {
+	t := time.NewTicker(d)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		}
+
+		// No feasible way to handle errors here.  In the future, it may be
+		// possible to do something like re-establishing the connection.
+		// TOOD(mdlayher): improve error handling for echo loop.
+		_ = c.Echo(ctx)
+	}
+}
+
 // A callback can be used to send a message back to a caller, or
 // allow the caller to cancel waiting for a message.
 type callback struct {

ovsdb/client_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -182,12 +183,61 @@ func TestClientLeakCallbacks(t *testing.T) {
 	}
 }
 
-func testClient(t *testing.T, fn jsonrpc.TestFunc) (*ovsdb.Client, chan<- *jsonrpc.Response, func()) {
+func TestClientEchoLoop(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping during short test run")
+	}
+
+	// Count the number of requests sent to the server.
+	echo := ovsdb.EchoInterval(50 * time.Millisecond)
+	var reqID int64
+
+	_, _, done := testClient(t, func(req jsonrpc.Request) jsonrpc.Response {
+		if diff := cmp.Diff("echo", req.Method); diff != "" {
+			panicf("unexpected RPC method (-want +got):\n%s", diff)
+		}
+
+		// Keep incrementing the request ID to match the client.
+		id := int(atomic.AddInt64(&reqID, 1))
+		return jsonrpc.Response{
+			ID:     &id,
+			Result: mustMarshalJSON(t, req.Params),
+		}
+	}, echo)
+	defer done()
+
+	// Fail the test if the RPCs don't fire.
+	timer := time.AfterFunc(2*time.Second, func() {
+		panicf("took too long to wait for echo RPCs")
+	})
+	defer timer.Stop()
+
+	// Ensure that background echo RPCs are being sent.
+	tick := time.NewTicker(100 * time.Millisecond)
+	defer tick.Stop()
+
+	for {
+		// Just wait for a handful of RPCs to be sent before success.
+		<-tick.C
+		if atomic.LoadInt64(&reqID) > 5 {
+			break
+		}
+	}
+}
+
+func testClient(t *testing.T, fn jsonrpc.TestFunc, options ...ovsdb.OptionFunc) (*ovsdb.Client, chan<- *jsonrpc.Response, func()) {
 	t.Helper()
 
+	// Prepend a verbose logger so the caller can override it easily.
+	if testing.Verbose() {
+		options = append([]ovsdb.OptionFunc{
+			ovsdb.Debug(log.New(os.Stderr, "", 0)),
+		}, options...)
+	}
+
 	conn, notifC, done := jsonrpc.TestNetConn(t, fn)
 
-	c, err := ovsdb.New(conn, ovsdb.Debug(log.New(os.Stderr, "", 0)))
+	c, err := ovsdb.New(conn, options...)
 	if err != nil {
 		t.Fatalf("failed to dial: %v", err)
 	}