Repository: ovn-org/libovsdb
Pull request #280 (2022-01-17 17:33:34)

client: don't reconnect to same db after leader change
If the active endpoint changed leadership and the client is following
leadership, don't reconnect to the same endpoint (which is no longer
the leader) until we've exhausted all other options. Doing so
prolongs the reconnect process for two reasons:

a) the same endpoint is no longer leader so we'll just disconnect
immediately anyway
b) the same endpoint may be compacting its database, which is why it
transferred leadership, and won't respond to our connection attempt
until it's done

@jcaamano @dave-tucker 

coveralls: ## Pull Request Test Coverage Report for [Build 1709187775](https://coveralls.io/builds/45689950)

* **13** of **20**   **(65.0%)**  changed or added relevant lines in **1** file are covered.
* **75** unchanged lines in **1** file lost coverage.
* Overall coverage decreased (**-0.04%**) to **72.449%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [client/client.go](https://coveralls.io/builds/45689950/source?filename=client%2Fclient.go#L218) | 13 | 20 | 65.0%
<!-- | **Total:** | **13** | **20** | **65.0%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/client.go](https://coveralls.io/builds/45689950/source?filename=client%2Fclient.go#L257) | 75 | 64.64% |
<!-- | **Total:** | **75** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/45689950/badge)](https://coveralls.io/builds/45689950) |
| :-- | --: |
| Change from base [Build 1698121798](https://coveralls.io/builds/45634724): |  -0.04% |
| Covered Lines: | 4289 |
| Relevant Lines: | 5920 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/client/client.go b/client/client.go
index b2cc81d..fc81f98 100644
--- a/client/client.go
+++ b/client/client.go
@@ -74,14 +74,21 @@ type bufferedUpdate struct {
 	lastTxnID string
 }
 
+type epInfo struct {
+	address  string
+	serverID string
+}
+
 // ovsdbClient is an OVSDB client
 type ovsdbClient struct {
-	options        *options
-	metrics        metrics
-	connected      bool
-	rpcClient      *rpc2.Client
-	rpcMutex       sync.RWMutex
-	activeEndpoint string
+	options   *options
+	metrics   metrics
+	connected bool
+	rpcClient *rpc2.Client
+	rpcMutex  sync.RWMutex
+	// endpoints contains all possible endpoints; the first element is
+	// the active endpoint if connected=true
+	endpoints []*epInfo
 
 	// The name of the "primary" database - that is to say, the DB
 	// that the user expects to interact with.
@@ -146,6 +153,9 @@ func newOVSDBClient(clientDBModel model.ClientDBModel, opts ...Option) (*ovsdbCl
 	if err != nil {
 		return nil, err
 	}
+	for _, address := range ovs.options.endpoints {
+		ovs.endpoints = append(ovs.endpoints, &epInfo{address: address})
+	}
 
 	if ovs.options.logger == nil {
 		// create a new logger to log to stdout
@@ -199,6 +209,28 @@ func (o *ovsdbClient) Connect(ctx context.Context) error {
 	return nil
 }
 
+// moveEndpointFirst makes the endpoint requested by active the first element
+// in the endpoints slice, indicating it is the active endpoint
+func (o *ovsdbClient) moveEndpointFirst(i int) {
+	firstEp := o.endpoints[i]
+	othereps := append(o.endpoints[:i], o.endpoints[i+1:]...)
+	o.endpoints = append([]*epInfo{firstEp}, othereps...)
+}
+
+// moveEndpointLast moves the requested endpoint to the end of the list
+func (o *ovsdbClient) moveEndpointLast(i int) {
+	lastEp := o.endpoints[i]
+	othereps := append(o.endpoints[:i], o.endpoints[i+1:]...)
+	o.endpoints = append(othereps, lastEp)
+}
+
+func (o *ovsdbClient) resetRPCClient() {
+	if o.rpcClient != nil {
+		o.rpcClient.Close()
+		o.rpcClient = nil
+	}
+}
+
 func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 	o.rpcMutex.Lock()
 	defer o.rpcMutex.Unlock()
@@ -208,18 +240,20 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 
 	connected := false
 	connectErrors := []error{}
-	for _, endpoint := range o.options.endpoints {
-		u, err := url.Parse(endpoint)
+	for i, endpoint := range o.endpoints {
+		u, err := url.Parse(endpoint.address)
 		if err != nil {
 			return err
 		}
-		if err := o.tryEndpoint(ctx, u); err != nil {
+		if sid, err := o.tryEndpoint(ctx, u); err != nil {
+			o.resetRPCClient()
 			connectErrors = append(connectErrors,
-				fmt.Errorf("failed to connect to %s: %w", endpoint, err))
+				fmt.Errorf("failed to connect to %s: %w", endpoint.address, err))
 			continue
 		} else {
-			o.logger.V(3).Info("successfully connected", "endpoint", endpoint)
-			o.activeEndpoint = endpoint
+			o.logger.V(3).Info("successfully connected", "endpoint", endpoint.address, "sid", sid)
+			endpoint.serverID = sid
+			o.moveEndpointFirst(i)
 			connected = true
 			break
 		}
@@ -246,8 +280,7 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 			for id, request := range db.monitors {
 				err := o.monitor(ctx, MonitorCookie{DatabaseName: dbName, ID: id}, true, request)
 				if err != nil {
-					o.rpcClient.Close()
-					o.rpcClient = nil
+					o.resetRPCClient()
 					return err
 				}
 			}
@@ -264,7 +297,9 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 	return nil
 }
 
-func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
+// tryEndpoint connects to a single database endpoint. Returns the
+// server ID (if clustered) on success, or an error.
+func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) (string, error) {
 	o.logger.V(5).Info("trying to connect", "endpoint", fmt.Sprintf("%v", u))
 	var dialer net.Dialer
 	var err error
@@ -283,18 +318,15 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 	default:
 		err = fmt.Errorf("unknown network protocol %s", u.Scheme)
 	}
-
 	if err != nil {
-		return fmt.Errorf("failed to open connection: %w", err)
+		return "", fmt.Errorf("failed to open connection: %w", err)
 	}
 
 	o.createRPC2Client(c)
 
 	serverDBNames, err := o.listDbs(ctx)
 	if err != nil {
-		o.rpcClient.Close()
-		o.rpcClient = nil
-		return err
+		return "", err
 	}
 
 	// for every requested database, ensure the DB exists in the server and
@@ -309,18 +341,13 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 			}
 		}
 		if !found {
-			err = fmt.Errorf("target database %s not found", dbName)
-			o.rpcClient.Close()
-			o.rpcClient = nil
-			return err
+			return "", fmt.Errorf("target database %s not found", dbName)
 		}
 
 		// load and validate the schema
 		schema, err := o.getSchema(ctx, dbName)
 		if err != nil {
-			o.rpcClient.Close()
-			o.rpcClient = nil
-			return err
+			return "", err
 		}
 
 		db.modelMutex.Lock()
@@ -332,11 +359,8 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 			for _, err := range errors {
 				combined = append(combined, err.Error())
 			}
-			err = fmt.Errorf("database %s validation error (%d): %s", dbName, len(errors),
-				strings.Join(combined, ". "))
-			o.rpcClient.Close()
-			o.rpcClient = nil
-			return err
+			return "", fmt.Errorf("database %s validation error (%d): %s",
+				dbName, len(errors), strings.Join(combined, ". "))
 		}
 
 		db.cacheMutex.Lock()
@@ -344,9 +368,7 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 			db.cache, err = cache.NewTableCache(db.model, nil, o.logger)
 			if err != nil {
 				db.cacheMutex.Unlock()
-				o.rpcClient.Close()
-				o.rpcClient = nil
-				return err
+				return "", err
 			}
 			db.api = newAPI(db.cache, o.logger)
 		} else {
@@ -356,22 +378,18 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 	}
 
 	// check that this is the leader
+	var sid string
 	if o.options.leaderOnly {
 		var leader bool
-		leader, err = o.isEndpointLeader(ctx)
+		leader, sid, err = o.isEndpointLeader(ctx)
 		if err != nil {
-			o.rpcClient.Close()
-			o.rpcClient = nil
-			return err
+			return "", err
 		}
 		if !leader {
-			err = fmt.Errorf("endpoint is not leader")
-			o.rpcClient.Close()
-			o.rpcClient = nil
-			return err
+			return "", fmt.Errorf("endpoint is not leader")
 		}
 	}
-	return nil
+	return sid, nil
 }
 
 // createRPC2Client creates an rpcClient using the provided connection
@@ -396,9 +414,11 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) {
 	go o.rpcClient.Run()
 }
 
-// isEndpointLeader returns true if the currently connected endpoint is leader.
-// assumes rpcMutex is held
-func (o *ovsdbClient) isEndpointLeader(ctx context.Context) (bool, error) {
+// isEndpointLeader returns true if the currently connected endpoint is leader,
+// otherwise false or an error. If the currently connected endpoint is the leader
+// and the database is clustered, also returns the database's Server ID.
+// Assumes rpcMutex is held.
+func (o *ovsdbClient) isEndpointLeader(ctx context.Context) (bool, string, error) {
 	op := ovsdb.Operation{
 		Op:      ovsdb.OperationSelect,
 		Table:   "Database",
@@ -406,21 +426,21 @@ func (o *ovsdbClient) isEndpointLeader(ctx context.Context) (bool, error) {
 	}
 	results, err := o.transact(ctx, serverDB, op)
 	if err != nil {
-		return false, fmt.Errorf("could not check if server was leader: %w", err)
+		return false, "", fmt.Errorf("could not check if server was leader: %w", err)
 	}
 	// for now, if no rows are returned, just accept this server
 	if len(results) != 1 {
-		return true, nil
+		return true, "", nil
 	}
 	result := results[0]
 	if len(result.Rows) == 0 {
-		return true, nil
+		return true, "", nil
 	}
 
 	for _, row := range result.Rows {
 		dbName, ok := row["name"].(string)
 		if !ok {
-			return false, fmt.Errorf("could not parse name")
+			return false, "", fmt.Errorf("could not parse name")
 		}
 		if dbName != o.primaryDBName {
 			continue
@@ -428,26 +448,33 @@ func (o *ovsdbClient) isEndpointLeader(ctx context.Context) (bool, error) {
 
 		model, ok := row["model"].(string)
 		if !ok {
-			return false, fmt.Errorf("could not parse model")
+			return false, "", fmt.Errorf("could not parse model")
 		}
 
 		// the database reports whether or not it is part of a cluster via the
 		// "model" column. If it's not clustered, it is by definition leader.
 		if model != serverdb.DatabaseModelClustered {
-			return true, nil
+			return true, "", nil
+		}
+
+		// Clustered database must have a Server ID
+		sid, ok := row["sid"].(ovsdb.UUID)
+		if !ok {
+			return false, "", fmt.Errorf("could not parse server id")
 		}
 
 		leader, ok := row["leader"].(bool)
 		if !ok {
-			return false, fmt.Errorf("could not parse leader")
+			return false, "", fmt.Errorf("could not parse leader")
 		}
-		return leader, nil
+
+		return leader, sid.GoUUID, nil
 	}
 
 	// Extremely unlikely: there is no _Server row for the desired DB (which we made sure existed)
 	// for now, just continue
 	o.logger.V(3).Info("Couldn't find a row in _Server for our database. Continuing without leader detection", "database", o.primaryDBName)
-	return true, nil
+	return true, "", nil
 }
 
 func (o *ovsdbClient) primaryDB() *database {
@@ -497,7 +524,7 @@ func (o *ovsdbClient) CurrentEndpoint() string {
 	if o.rpcClient == nil {
 		return ""
 	}
-	return o.activeEndpoint
+	return o.endpoints[0].address
 }
 
 // DisconnectNotify returns a channel which will notify the caller when the
@@ -983,10 +1010,37 @@ func (o *ovsdbClient) watchForLeaderChange() error {
 				continue
 			}
 
-			if dbInfo.Model == serverdb.DatabaseModelClustered && !dbInfo.Leader && o.Connected() {
-				o.logger.V(3).Info("endpoint lost leader, reconnecting", "endpoint", o.activeEndpoint)
-				o.Disconnect()
+			// Only handle leadership changes for clustered databases
+			if dbInfo.Model != serverdb.DatabaseModelClustered {
+				continue
+			}
+
+			// Clustered database servers must have a valid Server ID
+			var sid string
+			if dbInfo.Sid != nil {
+				sid = *dbInfo.Sid
 			}
+			if sid == "" {
+				o.logger.V(3).Info("clustered database update contained invalid server ID")
+				continue
+			}
+
+			o.rpcMutex.Lock()
+			if !dbInfo.Leader && o.connected {
+				activeEndpoint := o.endpoints[0]
+				if sid == activeEndpoint.serverID {
+					o.logger.V(3).Info("endpoint lost leader, reconnecting",
+						"endpoint", activeEndpoint.address, "sid", sid)
+					// don't immediately reconnect to the active endpoint since it's no longer leader
+					o.moveEndpointLast(0)
+					o._disconnect()
+				} else {
+					o.logger.V(3).Info("endpoint lost leader but had unexpected server ID",
+						"endpoint", activeEndpoint.address,
+						"expected", activeEndpoint.serverID, "found", sid)
+				}
+			}
+			o.rpcMutex.Unlock()
 		}
 	}()
 	return nil
@@ -1051,7 +1105,7 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 			suppressionCounter++
 			return err
 		}
-		o.logger.V(3).Info("connection lost, reconnecting", "endpoint", o.activeEndpoint)
+		o.logger.V(3).Info("connection lost, reconnecting", "endpoint", o.endpoints[0].address)
 		err := backoff.Retry(connect, o.options.backoff)
 		if err != nil {
 			// TODO: We should look at passing this back to the
@@ -1096,12 +1150,10 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 	}
 }
 
-// Disconnect will close the connection to the OVSDB server
+// _disconnect will close the connection to the OVSDB server
 // If the client was created with WithReconnect then the client
-// will reconnect afterwards
-func (o *ovsdbClient) Disconnect() {
-	o.rpcMutex.Lock()
-	defer o.rpcMutex.Unlock()
+// will reconnect afterwards. Assumes rpcMutex is held.
+func (o *ovsdbClient) _disconnect() {
 	o.connected = false
 	if o.rpcClient == nil {
 		return
@@ -1109,6 +1161,15 @@ func (o *ovsdbClient) Disconnect() {
 	o.rpcClient.Close()
 }
 
+// Disconnect will close the connection to the OVSDB server
+// If the client was created with WithReconnect then the client
+// will reconnect afterwards
+func (o *ovsdbClient) Disconnect() {
+	o.rpcMutex.Lock()
+	defer o.rpcMutex.Unlock()
+	o._disconnect()
+}
+
 // Close will close the connection to the OVSDB server
 // It will remove all stored state ready for the next connection
 // Even If the client was created with WithReconnect it will not reconnect afterwards
diff --git a/server/server.go b/server/server.go
index be5a9e5..88c0e03 100644
--- a/server/server.go
+++ b/server/server.go
@@ -71,6 +71,11 @@ func NewOvsdbServer(db Database, models ...model.DatabaseModel) (*OvsdbServer, e
 	return o, nil
 }
 
+// OnConnect registers a function to run when a client connects.
+func (o *OvsdbServer) OnConnect(f func(*rpc2.Client)) {
+	o.srv.OnConnect(f)
+}
+
 // Serve starts the OVSDB server on the given path and protocol
 func (o *OvsdbServer) Serve(protocol string, path string) error {
 	var err error


Test patch:
diff --git a/client/client_test.go b/client/client_test.go
index fe6ca06..2850268 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -3,14 +3,23 @@ package client
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"math/rand"
+	"os"
 	"reflect"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/cenkalti/rpc2"
+	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
+	"github.com/ovn-org/libovsdb/ovsdb/serverdb"
+	"github.com/ovn-org/libovsdb/server"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -855,3 +864,152 @@ func TestSetOption(t *testing.T) {
 	err = o.SetOption(WithEndpoint("tcp::6641"))
 	assert.EqualError(t, err, "cannot set option when client is connected")
 }
+
+func newOVSDBServer(t *testing.T, dbModel model.ClientDBModel, schema ovsdb.DatabaseSchema) (*server.OvsdbServer, string) {
+	serverDBModel, err := serverdb.FullDatabaseModel()
+	require.NoError(t, err)
+	serverSchema := serverdb.Schema()
+
+	db := server.NewInMemoryDatabase(map[string]model.ClientDBModel{
+		schema.Name:       dbModel,
+		serverSchema.Name: serverDBModel,
+	})
+
+	dbMod, errs := model.NewDatabaseModel(schema, dbModel)
+	require.Empty(t, errs)
+
+	servMod, errs := model.NewDatabaseModel(serverSchema, serverDBModel)
+	require.Empty(t, errs)
+
+	server, err := server.NewOvsdbServer(db, dbMod, servMod)
+	require.NoError(t, err)
+
+	tmpfile := fmt.Sprintf("/tmp/ovsdb-%d.sock", rand.Intn(10000))
+	t.Cleanup(func() {
+		os.Remove(tmpfile)
+	})
+	go func() {
+		if err := server.Serve("unix", tmpfile); err != nil {
+			t.Error(err)
+		}
+	}()
+	t.Cleanup(server.Close)
+	require.Eventually(t, func() bool {
+		return server.Ready()
+	}, 1*time.Second, 10*time.Millisecond)
+
+	return server, tmpfile
+}
+
+func newClientServerPair(t *testing.T, connectCounter *int32, isLeader bool) (Client, *serverdb.Database, string) {
+	var defSchema ovsdb.DatabaseSchema
+	err := json.Unmarshal([]byte(schema), &defSchema)
+	require.NoError(t, err)
+
+	serverDBModel, err := serverdb.FullDatabaseModel()
+	require.NoError(t, err)
+
+	// Create server
+	s, sock := newOVSDBServer(t, defDB, defSchema)
+	s.OnConnect(func(_ *rpc2.Client) {
+		atomic.AddInt32(connectCounter, 1)
+	})
+
+	// Create client for this server's Server database
+	endpoint := fmt.Sprintf("unix:%s", sock)
+	cli, err := newOVSDBClient(serverDBModel, WithEndpoint(endpoint))
+	require.NoError(t, err)
+	err = cli.Connect(context.Background())
+	require.NoError(t, err)
+	t.Cleanup(cli.Close)
+
+	// Populate the _Server database table
+	sid := fmt.Sprintf("%04x", rand.Uint32())
+	row := &serverdb.Database{
+		UUID:      uuid.NewString(),
+		Name:      defDB.Name(),
+		Connected: true,
+		Leader:    isLeader,
+		Model:     serverdb.DatabaseModelClustered,
+		Sid:       &sid,
+	}
+	ops, err := cli.Create(row)
+	require.Nil(t, err)
+	reply, err := cli.Transact(context.Background(), ops...)
+	assert.Nil(t, err)
+	opErr, err := ovsdb.CheckOperationResults(reply, ops)
+	assert.NoErrorf(t, err, "%+v", opErr)
+
+	row.UUID = reply[0].UUID.GoUUID
+	return cli, row, endpoint
+}
+
+func setLeader(t *testing.T, cli Client, row *serverdb.Database, isLeader bool) {
+	row.Leader = isLeader
+	ops, err := cli.Where(row).Update(row, &row.Leader)
+	require.Nil(t, err)
+	reply, err := cli.Transact(context.Background(), ops...)
+	require.Nil(t, err)
+	opErr, err := ovsdb.CheckOperationResults(reply, ops)
+	assert.NoErrorf(t, err, "%+v", opErr)
+}
+
+func TestClientReconnectLeaderOnly(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+
+	var connected1, connected2 int32
+	cli1, row1, endpoint1 := newClientServerPair(t, &connected1, true)
+	cli2, row2, endpoint2 := newClientServerPair(t, &connected2, false)
+
+	// Create client to test reconnection for
+	ovs, err := newOVSDBClient(defDB,
+		WithLeaderOnly(true),
+		WithReconnect(5*time.Second, &backoff.ZeroBackOff{}),
+		WithEndpoint(endpoint1),
+		WithEndpoint(endpoint2))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
+	t.Cleanup(ovs.Close)
+
+	// Server1 should have 2 connections: cli1 and ovs
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt32(&connected1) == 2
+	}, 2*time.Second, 10*time.Millisecond)
+
+	// Server2 should have 1 connection: cli2
+	require.Never(t, func() bool {
+		return atomic.LoadInt32(&connected2) > 1
+	}, 2*time.Second, 10*time.Millisecond)
+
+	// First leadership change
+	setLeader(t, cli2, row2, true)
+	setLeader(t, cli1, row1, false)
+
+	// Server2 should have 2 connections: cli2 and ovs
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt32(&connected2) == 2
+	}, 2*time.Second, 10*time.Millisecond)
+
+	// Server1 should still only have 2 total connections; eg the
+	// client under test should not have reconnected
+	require.Never(t, func() bool {
+		return atomic.LoadInt32(&connected1) > 2
+	}, 2*time.Second, 10*time.Millisecond)
+
+	// Second leadership change
+	setLeader(t, cli1, row1, true)
+	setLeader(t, cli2, row2, false)
+
+	// Server1 should now have 3 total connections: cli1, original ovs,
+	// and second ovs
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt32(&connected1) == 3
+	}, 2*time.Second, 10*time.Millisecond)
+
+	// Server2 should still only have 2 total connections; eg the
+	// client under test should not have reconnected
+	require.Never(t, func() bool {
+		return atomic.LoadInt32(&connected2) > 2
+	}, 2*time.Second, 10*time.Millisecond)
+}