Merge pull request #8938 from bhandras/etcd-leader-election-fixups

multi: check leader status with our health checker to correctly shut down LND if network partitions

Author: Roasbeef

cluster/etcd_elector.go

@@ -99,9 +99,27 @@ func (e *etcdLeaderElector) Leader(ctx context.Context) (string, error) {
 		return "", err
 	}
 
+	if resp == nil || len(resp.Kvs) == 0 {
+		return "", nil
+	}
+
 	return string(resp.Kvs[0].Value), nil
 }
 
+// IsLeader returns true if the caller is the leader.
+func (e *etcdLeaderElector) IsLeader(ctx context.Context) (bool, error) {
+	resp, err := e.election.Leader(ctx)
+	if err != nil {
+		return false, err
+	}
+
+	if resp == nil || len(resp.Kvs) == 0 {
+		return false, nil
+	}
+
+	return string(resp.Kvs[0].Value) == e.id, nil
+}
+
 // Campaign will start a new leader election campaign. Campaign will block until
 // the elector context is canceled or the caller is elected as the leader.
 func (e *etcdLeaderElector) Campaign(ctx context.Context) error {
@@ -110,6 +128,6 @@ func (e *etcdLeaderElector) Campaign(ctx context.Context) error {
 
 // Resign resigns the leader role allowing other election members to take
 // the place.
-func (e *etcdLeaderElector) Resign() error {
-	return e.election.Resign(context.Background())
+func (e *etcdLeaderElector) Resign(ctx context.Context) error {
+	return e.election.Resign(ctx)
 }

cluster/etcd_elector_test.go

@@ -87,12 +87,12 @@ func TestEtcdElector(t *testing.T) {
 	tmp := <-ch
 	first, err := tmp.Leader(ctxb)
 	require.NoError(t, err)
-	require.NoError(t, tmp.Resign())
+	require.NoError(t, tmp.Resign(ctxb))
 
 	tmp = <-ch
 	second, err := tmp.Leader(ctxb)
 	require.NoError(t, err)
-	require.NoError(t, tmp.Resign())
+	require.NoError(t, tmp.Resign(ctxb))
 
 	require.Contains(t, []string{id1, id2}, first)
 	require.Contains(t, []string{id1, id2}, second)

cluster/interface.go

@@ -19,8 +19,11 @@ type LeaderElector interface {
 
 	// Resign resigns from the leader role, allowing other election members
 	// to take on leadership.
-	Resign() error
+	Resign(ctx context.Context) error
 
 	// Leader returns the leader value for the current election.
 	Leader(ctx context.Context) (string, error)
+
+	// IsLeader returns true if the caller is the leader.
+	IsLeader(ctx context.Context) (bool, error)
 }

config.go

@@ -169,6 +169,17 @@ const (
 	defaultRSBackoff  = time.Second * 30
 	defaultRSAttempts = 1
 
+	// Set defaults for a health check which ensures that the leader
+	// election is functioning correctly. Although this check is off by
+	// default (as etcd leader election is only used in a clustered setup),
+	// we still set the default values so that the health check can be
+	// easily enabled with sane defaults. Note that by default we only run
+	// this check once, as it is critical for the node's operation.
+	defaultLeaderCheckInterval = time.Minute
+	defaultLeaderCheckTimeout  = time.Second * 5
+	defaultLeaderCheckBackoff  = time.Second * 5
+	defaultLeaderCheckAttempts = 1
+
 	// defaultRemoteMaxHtlcs specifies the default limit for maximum
 	// concurrent HTLCs the remote party may add to commitment transactions.
 	// This value can be overridden with --default-remote-max-htlcs.
@@ -672,6 +683,12 @@ func DefaultConfig() Config {
 				Attempts: defaultRSAttempts,
 				Backoff:  defaultRSBackoff,
 			},
+			LeaderCheck: &lncfg.CheckConfig{
+				Interval: defaultLeaderCheckInterval,
+				Timeout:  defaultLeaderCheckTimeout,
+				Attempts: defaultLeaderCheckAttempts,
+				Backoff:  defaultLeaderCheckBackoff,
+			},
 		},
 		Gossip: &lncfg.Gossip{
 			MaxChannelUpdateBurst: discovery.DefaultMaxChannelUpdateBurst,

docs/release-notes/release-notes-0.18.3.md

@@ -150,6 +150,10 @@ commitment when the channel was force closed.
 * [Fixed](https://github.com/lightningnetwork/lnd/pull/8854) pagination issues
   in SQL invoicedb queries.
 
+* [Check](https://github.com/lightningnetwork/lnd/pull/8938) leader status with
+  our health checker to correctly shut down LND if network partitioning occurs
+  towards the etcd cluster.
+
 ## Code Health
 
 * [Move graph building and

go.mod

@@ -207,6 +207,12 @@ replace google.golang.org/protobuf => github.com/lightninglabs/protobuf-go-hex-d
 // Temporary replace until the next version of sqldb is tagged.
 replace github.com/lightningnetwork/lnd/sqldb => ./sqldb
 
+// Temporary replace until the next version of healthcheck is tagged.
+replace github.com/lightningnetwork/lnd/healthcheck => ./healthcheck
+
+// Temporary replace until the next version of kvdb is tagged.
+replace github.com/lightningnetwork/lnd/kvdb => ./kvdb
+
 // If you change this please also update .github/pull_request_template.md and
 // docs/INSTALL.md.
 go 1.21.4

go.sum

@@ -452,10 +452,6 @@ github.com/lightningnetwork/lnd/clock v1.1.1 h1:OfR3/zcJd2RhH0RU+zX/77c0ZiOnIMsD
 github.com/lightningnetwork/lnd/clock v1.1.1/go.mod h1:mGnAhPyjYZQJmebS7aevElXKTFDuO+uNFFfMXK1W8xQ=
 github.com/lightningnetwork/lnd/fn v1.2.0 h1:YTb2m8NN5ZiJAskHeBZAmR1AiPY8SXziIYPAX1VI/ZM=
 github.com/lightningnetwork/lnd/fn v1.2.0/go.mod h1:SyFohpVrARPKH3XVAJZlXdVe+IwMYc4OMAvrDY32kw0=
-github.com/lightningnetwork/lnd/healthcheck v1.2.4 h1:lLPLac+p/TllByxGSlkCwkJlkddqMP5UCoawCj3mgFQ=
-github.com/lightningnetwork/lnd/healthcheck v1.2.4/go.mod h1:G7Tst2tVvWo7cx6mSBEToQC5L1XOGxzZTPB29g9Rv2I=
-github.com/lightningnetwork/lnd/kvdb v1.4.8 h1:xH0a5Vi1yrcZ5BEeF2ba3vlKBRxrL9uYXlWTjOjbNTY=
-github.com/lightningnetwork/lnd/kvdb v1.4.8/go.mod h1:J2diNABOoII9UrMnxXS5w7vZwP7CA1CStrl8MnIrb3A=
 github.com/lightningnetwork/lnd/queue v1.1.1 h1:99ovBlpM9B0FRCGYJo6RSFDlt8/vOkQQZznVb18iNMI=
 github.com/lightningnetwork/lnd/queue v1.1.1/go.mod h1:7A6nC1Qrm32FHuhx/mi1cieAiBZo5O6l8IBIoQxvkz4=
 github.com/lightningnetwork/lnd/ticker v1.1.1 h1:J/b6N2hibFtC7JLV77ULQp++QLtCwT6ijJlbdiZFbSM=

healthcheck/healthcheck.go

@@ -234,14 +234,13 @@ func (o *Observation) monitor(shutdown shutdownFunc, quit chan struct{}) {
 			// the max attempts are reached. In that case we will
 			// stop the ticker and quit.
 			if o.retryCheck(quit, shutdown) {
-				log.Debugf("Health check: max attempts " +
-					"failed, monitor exiting")
+				o.Debugf("max attempts failed, monitor exiting")
 				return
 			}
 
 		// Exit if we receive the instruction to shutdown.
 		case <-quit:
-			log.Debug("Health check: monitor quit")
+			o.Debugf("monitor quit")
 			return
 		}
 	}
@@ -270,7 +269,7 @@ func (o *Observation) retryCheck(quit chan struct{},
 			// so we'll invoke our success callback if defined and
 			// then exit.
 			if err == nil {
-				log.Debug("invoking success callback")
+				o.Debugf("invoking success callback")
 
 				// Invoke the success callback.
 				o.OnSuccess()
@@ -283,25 +282,26 @@ func (o *Observation) retryCheck(quit chan struct{},
 				"%v", o, o.Timeout)
 
 		case <-quit:
-			log.Debug("Health check: monitor quit")
+			o.Debugf("monitor quit")
 			return false
 		}
 
 		// If we have reached our allowed number of attempts, this
 		// check has failed so we'll fire the on failure callback
 		// and request shutdown.
 		if count == o.Attempts {
-			log.Debug("invoking failure callback")
+			o.Debugf("invoking failure callback")
 
 			o.OnFailure()
 
-			shutdown("Health check: %v failed after %v "+
-				"calls", o, o.Attempts)
+			shutdown("Health check: %v failed after %v calls", o,
+				o.Attempts)
+
 			return true
 		}
 
-		log.Infof("Health check: %v, call: %v failed with: %v, "+
-			"backing off for: %v", o, count, err, o.Backoff)
+		o.Infof("failed with: %v, attempts: %v backing off for: %v",
+			err, count, o.Backoff)
 
 		// If we are still within the number of calls allowed for this
 		// check, we wait for our back off period to elapse, or exit if
@@ -310,10 +310,22 @@ func (o *Observation) retryCheck(quit chan struct{},
 		case <-time.After(o.Backoff):
 
 		case <-quit:
-			log.Debug("Health check: monitor quit")
+			o.Debugf("monitor quit")
 			return false
 		}
 	}
 
 	return false
 }
+
+// Infof logs an info message for an observation prefixed with the health check
+// name.
+func (o *Observation) Infof(format string, params ...interface{}) {
+	log.Debugf(fmt.Sprintf("Health check: %v ", o)+format, params...)
+}
+
+// Debugf logs a debug message for an observation prefixed with the health check
+// name.
+func (o *Observation) Debugf(format string, params ...interface{}) {
+	log.Debugf(fmt.Sprintf("Health check: %v ", o)+format, params...)
+}

itest/list_on_test.go

@@ -233,6 +233,10 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "etcd failover",
 		TestFunc: testEtcdFailover,
 	},
+	{
+		Name:     "leader health check",
+		TestFunc: testLeaderHealthCheck,
+	},
 	{
 		Name:     "hold invoice force close",
 		TestFunc: testHoldInvoiceForceClose,