From 0fd4c7d5f94837abe6e977fcafe077d7842acea0 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 25 Jul 2024 18:09:26 +0200 Subject: [PATCH 1/8] healthcheck: improve logging of observers --- healthcheck/healthcheck.go | 34 +++++++++++++++++++++++----------- 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/healthcheck/healthcheck.go b/healthcheck/healthcheck.go index e15cbd4cb..ed4866678 100644 --- a/healthcheck/healthcheck.go +++ b/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,7 +282,7 @@ 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 } @@ -291,17 +290,18 @@ func (o *Observation) retryCheck(quit chan struct{}, // 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...) +} From 2c8d1c878ecce74eb502134d546ab50972ac3cea Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 25 Jul 2024 18:13:20 +0200 Subject: [PATCH 2/8] kvdb: make etcd calls timeout to ensure liveness Previously our RPC calls to etcd would hang even in the case of properly set dial timeouts and even if there was a network partition. To ensure liveness we need to make sure that calls fail correctly in case of system failure. To fix this we add a default timeout of 30 seconds to each etcd RPC call. --- kvdb/etcd/stm.go | 36 ++++++++++++++++++++++++++++++++---- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/kvdb/etcd/stm.go b/kvdb/etcd/stm.go index 5b3a9a977..3c933d5c4 100644 --- a/kvdb/etcd/stm.go +++ b/kvdb/etcd/stm.go @@ -8,12 +8,25 @@ import ( "fmt" "math" "strings" + "time" "github.com/google/btree" pb "go.etcd.io/etcd/api/v3/etcdserverpb" v3 "go.etcd.io/etcd/client/v3" ) +const ( + // rpcTimeout is the timeout for all RPC calls to etcd. It is set to 30 + // seconds to avoid blocking the server for too long but give reasonable + // time for etcd to respond. If any operations would take longer than 30 + // seconds that generally means there's a problem with the etcd server + // or the network resulting in degraded performance in which case we + // want LND to fail fast. Due to the underlying gRPC implementation in + // etcd calls without a timeout can hang indefinitely even in the case + // of network partitions or other critical failures. + rpcTimeout = time.Second * 30 +) + type CommitStats struct { Rset int Wset int @@ -609,8 +622,13 @@ func (s *stm) FetchRangePaginatedRaw(prefix string, limit int64, key := prefix for { + timeoutCtx, cancel := context.WithTimeout( + s.options.ctx, rpcTimeout, + ) + defer cancel() + resp, err := s.client.Get( - s.options.ctx, key, append(opts, s.getOpts...)..., + timeoutCtx, key, append(opts, s.getOpts...)..., ) if err != nil { return DatabaseError{ @@ -645,8 +663,12 @@ func (s *stm) FetchRangePaginatedRaw(prefix string, limit int64, // We'll also cache the returned key/value in the read set. func (s *stm) fetch(key string, opts ...v3.OpOption) ([]KV, error) { s.callCount++ + + timeoutCtx, cancel := context.WithTimeout(s.options.ctx, rpcTimeout) + defer cancel() + resp, err := s.client.Get( - s.options.ctx, key, append(opts, s.getOpts...)..., + timeoutCtx, key, append(opts, s.getOpts...)..., ) if err != nil { return nil, DatabaseError{ @@ -1049,7 +1071,10 @@ func (s *stm) Prefetch(keys []string, prefixes []string) { []v3.OpOption{v3.WithPrefix()}, s.getOpts..., ) - txn := s.client.Txn(s.options.ctx) + timeoutCtx, cancel := context.WithTimeout(s.options.ctx, rpcTimeout) + defer cancel() + + txn := s.client.Txn(timeoutCtx) ops := make([]v3.Op, 0, len(fetchKeys)+len(fetchPrefixes)) for _, key := range fetchKeys { @@ -1103,8 +1128,11 @@ func (s *stm) commit() (CommitStats, error) { // Create the compare set. cmps := append(rset, wset...) + // Create a transaction with the optional abort context. - txn := s.client.Txn(s.options.ctx) + timeoutCtx, cancel := context.WithTimeout(s.options.ctx, rpcTimeout) + defer cancel() + txn := s.client.Txn(timeoutCtx) // If the compare set holds, try executing the puts. txn = txn.If(cmps...) From 7784d6abf613655facd1f279f47260eabf2211a8 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 25 Jul 2024 18:20:13 +0200 Subject: [PATCH 3/8] build: pin healthcheck and kvdb modules temporarily This is to ensure that the added functionality works correctly and should be removed once these changes are merged and the packages are tagged. --- go.mod | 6 ++++++ go.sum | 4 ---- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/go.mod b/go.mod index 69e59f357..f3d7d1655 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index d371e4837..d8536ac54 100644 --- a/go.sum +++ b/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= From 8e0534f756de1725028de324fe17cfffc05f61b7 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 25 Jul 2024 18:21:47 +0200 Subject: [PATCH 4/8] multi: add leader check to the healthcheck monitor This commit extends our healtcheck with an optional leader check. This is to ensure that given network partition or other cluster wide failure we act as soon as possible to avoid a split-brain situation where a new leader is elected but we still hold onto our etcd client. --- cluster/etcd_elector.go | 22 +++++++++++++-- cluster/etcd_elector_test.go | 4 +-- cluster/interface.go | 5 +++- config.go | 17 ++++++++++++ lncfg/healthcheck.go | 2 ++ lnd.go | 26 +++++++++++++++-- lntest/node/config.go | 3 ++ server.go | 54 +++++++++++++++++++++++++++++++++--- 8 files changed, 121 insertions(+), 12 deletions(-) diff --git a/cluster/etcd_elector.go b/cluster/etcd_elector.go index 25b11af25..7e6c74d53 100644 --- a/cluster/etcd_elector.go +++ b/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) } diff --git a/cluster/etcd_elector_test.go b/cluster/etcd_elector_test.go index 999a3d791..f7fcebec3 100644 --- a/cluster/etcd_elector_test.go +++ b/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) diff --git a/cluster/interface.go b/cluster/interface.go index 8a317095e..098c061a6 100644 --- a/cluster/interface.go +++ b/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) } diff --git a/config.go b/config.go index 792c286ac..34d84720c 100644 --- a/config.go +++ b/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, diff --git a/lncfg/healthcheck.go b/lncfg/healthcheck.go index 92c415474..9a1403687 100644 --- a/lncfg/healthcheck.go +++ b/lncfg/healthcheck.go @@ -34,6 +34,8 @@ type HealthCheckConfig struct { TorConnection *CheckConfig `group:"torconnection" namespace:"torconnection"` RemoteSigner *CheckConfig `group:"remotesigner" namespace:"remotesigner"` + + LeaderCheck *CheckConfig `group:"leader" namespace:"leader"` } // Validate checks the values configured for our health checks. diff --git a/lnd.go b/lnd.go index 0a85e1bf3..bc74c9b96 100644 --- a/lnd.go +++ b/lnd.go @@ -24,6 +24,7 @@ import ( "github.com/lightningnetwork/lnd/build" "github.com/lightningnetwork/lnd/chanacceptor" "github.com/lightningnetwork/lnd/channeldb" + "github.com/lightningnetwork/lnd/cluster" "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lncfg" "github.com/lightningnetwork/lnd/lnrpc" @@ -56,6 +57,14 @@ const ( // admin macaroon unless the administrator explicitly allowed it. Thus // there's no harm allowing group read. adminMacaroonFilePermissions = 0640 + + // leaderResignTimeout is the timeout used when resigning from the + // leader role. This is kept short so LND can shut down quickly in case + // of a system failure or network partition making the cluster + // unresponsive. The cluster itself should ensure that the leader is not + // elected again until the previous leader has resigned or the leader + // election timeout has passed. + leaderResignTimeout = 5 * time.Second ) // AdminAuthOptions returns a list of DialOptions that can be used to @@ -381,6 +390,7 @@ func Main(cfg *Config, lisCfg ListenerCfg, implCfg *ImplementationCfg, // blocked until this instance is elected as the current leader or // shutting down. elected := false + var leaderElector cluster.LeaderElector if cfg.Cluster.EnableLeaderElection { electionCtx, cancelElection := context.WithCancel(ctx) @@ -392,7 +402,7 @@ func Main(cfg *Config, lisCfg ListenerCfg, implCfg *ImplementationCfg, ltndLog.Infof("Using %v leader elector", cfg.Cluster.LeaderElector) - leaderElector, err := cfg.Cluster.MakeLeaderElector( + leaderElector, err = cfg.Cluster.MakeLeaderElector( electionCtx, cfg.DB, ) if err != nil { @@ -407,7 +417,17 @@ func Main(cfg *Config, lisCfg ListenerCfg, implCfg *ImplementationCfg, ltndLog.Infof("Attempting to resign from leader role "+ "(%v)", cfg.Cluster.ID) - if err := leaderElector.Resign(); err != nil { + // Ensure that we don't block the shutdown process if + // the leader resigning process takes too long. The + // cluster will ensure that the leader is not elected + // again until the previous leader has resigned or the + // leader election timeout has passed. + timeoutCtx, cancel := context.WithTimeout( + ctx, leaderResignTimeout, + ) + defer cancel() + + if err := leaderElector.Resign(timeoutCtx); err != nil { ltndLog.Errorf("Leader elector failed to "+ "resign: %v", err) } @@ -579,7 +599,7 @@ func Main(cfg *Config, lisCfg ListenerCfg, implCfg *ImplementationCfg, server, err := newServer( cfg, cfg.Listeners, dbs, activeChainControl, &idKeyDesc, activeChainControl.Cfg.WalletUnlockParams.ChansToRestore, - multiAcceptor, torController, tlsManager, + multiAcceptor, torController, tlsManager, leaderElector, ) if err != nil { return mkErr("unable to create server: %v", err) diff --git a/lntest/node/config.go b/lntest/node/config.go index de1ea92ec..f0e3f7059 100644 --- a/lntest/node/config.go +++ b/lntest/node/config.go @@ -315,6 +315,9 @@ func ExtraArgsEtcd(etcdCfg *etcd.Config, name string, cluster bool, leaderSessionTTL), } extraArgs = append(extraArgs, clusterArgs...) + extraArgs = append( + extraArgs, "--healthcheck.leader.interval=10s", + ) } return extraArgs diff --git a/server.go b/server.go index 5d52cd399..ae05637bc 100644 --- a/server.go +++ b/server.go @@ -36,6 +36,7 @@ import ( "github.com/lightningnetwork/lnd/channeldb/models" "github.com/lightningnetwork/lnd/channelnotifier" "github.com/lightningnetwork/lnd/clock" + "github.com/lightningnetwork/lnd/cluster" "github.com/lightningnetwork/lnd/contractcourt" "github.com/lightningnetwork/lnd/discovery" "github.com/lightningnetwork/lnd/feature" @@ -484,8 +485,8 @@ func newServer(cfg *Config, listenAddrs []net.Addr, nodeKeyDesc *keychain.KeyDescriptor, chansToRestore walletunlocker.ChannelsToRecover, chanPredicate chanacceptor.ChannelAcceptor, - torController *tor.Controller, tlsManager *TLSManager) (*server, - error) { + torController *tor.Controller, tlsManager *TLSManager, + leaderElector cluster.LeaderElector) (*server, error) { var ( err error @@ -1674,7 +1675,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr, } // Create liveness monitor. - s.createLivenessMonitor(cfg, cc) + s.createLivenessMonitor(cfg, cc, leaderElector) // Create the connection manager which will be responsible for // maintaining persistent outbound connections and also accepting new @@ -1721,7 +1722,9 @@ func (s *server) signAliasUpdate(u *lnwire.ChannelUpdate) (*ecdsa.Signature, // // If a health check has been disabled by setting attempts to 0, our monitor // will not run it. -func (s *server) createLivenessMonitor(cfg *Config, cc *chainreg.ChainControl) { +func (s *server) createLivenessMonitor(cfg *Config, cc *chainreg.ChainControl, + leaderElector cluster.LeaderElector) { + chainBackendAttempts := cfg.HealthChecks.ChainCheck.Attempts if cfg.Bitcoin.Node == "nochainbackend" { srvrLog.Info("Disabling chain backend checks for " + @@ -1837,6 +1840,49 @@ func (s *server) createLivenessMonitor(cfg *Config, cc *chainreg.ChainControl) { checks = append(checks, remoteSignerConnectionCheck) } + // If we have a leader elector, we add a health check to ensure we are + // still the leader. During normal operation, we should always be the + // leader, but there are circumstances where this may change, such as + // when we lose network connectivity for long enough expiring out lease. + if leaderElector != nil { + leaderCheck := healthcheck.NewObservation( + "leader status", + func() error { + // Check if we are still the leader. Note that + // we don't need to use a timeout context here + // as the healthcheck observer will handle the + // timeout case for us. + timeoutCtx, cancel := context.WithTimeout( + context.Background(), + cfg.HealthChecks.LeaderCheck.Timeout, + ) + defer cancel() + + leader, err := leaderElector.IsLeader( + timeoutCtx, + ) + if err != nil { + return fmt.Errorf("unable to check if "+ + "still leader: %v", err) + } + + if !leader { + srvrLog.Debug("Not the current leader") + return fmt.Errorf("not the current " + + "leader") + } + + return nil + }, + cfg.HealthChecks.LeaderCheck.Interval, + cfg.HealthChecks.LeaderCheck.Timeout, + cfg.HealthChecks.LeaderCheck.Backoff, + cfg.HealthChecks.LeaderCheck.Attempts, + ) + + checks = append(checks, leaderCheck) + } + // If we have not disabled all of our health checks, we create a // liveness monitor with our configured checks. s.livenessMonitor = healthcheck.NewMonitor( From 8e49eb652d7d335f3de683b7063b3f302dae74ec Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 25 Jul 2024 18:23:48 +0200 Subject: [PATCH 5/8] itest: add itest covering the leader healthcheck --- itest/list_on_test.go | 4 + itest/lnd_etcd_failover_test.go | 202 +++++++++++++++++++++++ itest/lnd_no_etcd_dummy_failover_test.go | 4 + 3 files changed, 210 insertions(+) diff --git a/itest/list_on_test.go b/itest/list_on_test.go index adedfd541..f02036e7c 100644 --- a/itest/list_on_test.go +++ b/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, diff --git a/itest/lnd_etcd_failover_test.go b/itest/lnd_etcd_failover_test.go index c2a345889..53b1dd6b3 100644 --- a/itest/lnd_etcd_failover_test.go +++ b/itest/lnd_etcd_failover_test.go @@ -4,6 +4,11 @@ package itest import ( + "context" + "fmt" + "io" + "net" + "sync" "testing" "time" @@ -137,3 +142,200 @@ func testEtcdFailoverCase(ht *lntest.HarnessTest, kill bool) { // process. ht.Shutdown(carol2) } + +// Proxy is a simple TCP proxy that forwards all traffic between a local and a +// remote address. We use it to simulate a network partition in the leader +// health check test. +type Proxy struct { + listenAddr string + targetAddr string + cancel context.CancelFunc + wg sync.WaitGroup + stopped chan struct{} +} + +// NewProxy creates a new Proxy instance with a provided context. +func NewProxy(listenAddr, targetAddr string) *Proxy { + return &Proxy{ + listenAddr: listenAddr, + targetAddr: targetAddr, + stopped: make(chan struct{}), + } +} + +// Start starts the proxy. It listens on the listen address and forwards all +// traffic to the target address. +func (p *Proxy) Start(ctx context.Context, t *testing.T) { + listener, err := net.Listen("tcp", p.listenAddr) + require.NoError(t, err, "Failed to listen on %s", p.listenAddr) + t.Logf("Proxy is listening on %s", p.listenAddr) + + proxyCtx, cancel := context.WithCancel(ctx) + p.cancel = cancel + + p.wg.Add(1) + go func() { + defer func() { + close(p.stopped) + p.wg.Done() + }() + + for { + select { + case <-proxyCtx.Done(): + listener.Close() + return + default: + } + + conn, err := listener.Accept() + if err != nil { + if proxyCtx.Err() != nil { + // Context is done, exit the loop + return + } + t.Logf("Proxy failed to accept connection: %v", + err) + + continue + } + + p.wg.Add(1) + go p.handleConnection(proxyCtx, t, conn) + } + }() +} + +// handleConnection handles an accepted connection and forwards all traffic +// between the listener and target. +func (p *Proxy) handleConnection(ctx context.Context, t *testing.T, + conn net.Conn) { + + targetConn, err := net.Dial("tcp", p.targetAddr) + require.NoError(t, err, "Failed to connect to target %s", p.targetAddr) + + defer func() { + conn.Close() + targetConn.Close() + p.wg.Done() + }() + + done := make(chan struct{}) + + p.wg.Add(2) + go func() { + defer p.wg.Done() + // Ignore the copy error due to the connection being closed. + _, _ = io.Copy(targetConn, conn) + }() + + go func() { + defer p.wg.Done() + // Ignore the copy error due to the connection being closed. + _, _ = io.Copy(conn, targetConn) + close(done) + }() + + select { + case <-ctx.Done(): + case <-done: + } +} + +// Stop stops the proxy and waits for all connections to be closed and all +// goroutines to be stopped. +func (p *Proxy) Stop(t *testing.T) { + require.NotNil(t, p.cancel, "Proxy is not started") + + p.cancel() + p.wg.Wait() + <-p.stopped + + t.Log("Proxy stopped", time.Now()) +} + +// testLeaderHealthCheck tests that a node is properly shut down when the leader +// health check fails. +func testLeaderHealthCheck(ht *lntest.HarnessTest) { + clientPort := port.NextAvailablePort() + + // Let's start a test etcd instance that we'll redirect through a proxy. + etcdCfg, cleanup, err := kvdb.StartEtcdTestBackend( + ht.T.TempDir(), uint16(clientPort), + uint16(port.NextAvailablePort()), "", + ) + require.NoError(ht, err, "Failed to start etcd instance") + + // Make leader election session TTL 5 sec to make the test run fast. + const leaderSessionTTL = 5 + + // Create an election observer that we will use to monitor the leader + // election. + observer, err := cluster.MakeLeaderElector( + ht.Context(), cluster.EtcdLeaderElector, "observer", + lncfg.DefaultEtcdElectionPrefix, leaderSessionTTL, etcdCfg, + ) + require.NoError(ht, err, "Cannot start election observer") + + // Start a proxy that will forward all traffic to the etcd instance. + clientAddr := fmt.Sprintf("localhost:%d", clientPort) + proxyAddr := fmt.Sprintf("localhost:%d", port.NextAvailablePort()) + + ctx, cancel := context.WithCancel(ht.Context()) + defer cancel() + + proxy := NewProxy(proxyAddr, clientAddr) + proxy.Start(ctx, ht.T) + + // Copy the etcd config so that we can modify the host to point to the + // proxy. + proxyEtcdCfg := *etcdCfg + // With the proxy in place, we can now configure the etcd client to + // connect to the proxy instead of the etcd instance. + proxyEtcdCfg.Host = "http://" + proxyAddr + + defer cleanup() + + // Start Carol-1 with cluster support and connect to etcd through the + // proxy. + password := []byte("the quick brown fox jumps the lazy dog") + stateless := false + cluster := true + + carol, _, _ := ht.NewNodeWithSeedEtcd( + "Carol-1", &proxyEtcdCfg, password, stateless, cluster, + leaderSessionTTL, + ) + + // Make sure Carol-1 is indeed the leader. + assertLeader(ht, observer, "Carol-1") + + // At this point Carol-1 is the elected leader, while Carol-2 will wait + // to become the leader when Carol-1 releases the lease. Note that for + // Carol-2 we don't use the proxy as we want to simulate a network + // partition only for Carol-1. + carol2 := ht.NewNodeEtcd( + "Carol-2", etcdCfg, password, cluster, leaderSessionTTL, + ) + + // Stop the proxy so that we simulate a network partition which + // consequently will make the leader health check fail and force Carol + // to shut down. + proxy.Stop(ht.T) + + // Wait for Carol-1 to stop. If the health check wouldn't properly work + // this call would timeout and trigger a test failure. + require.NoError(ht.T, carol.WaitForProcessExit()) + + // Now that Carol-1 is shut down we should fail over to Carol-2. + failoverTimeout := time.Duration(2*leaderSessionTTL) * time.Second + + // Make sure that Carol-2 becomes the leader (reported by Carol-2). + err = carol2.WaitUntilLeader(failoverTimeout) + + require.NoError(ht, err, "Waiting for Carol-2 to become the leader "+ + "failed") + + // Make sure Carol-2 is indeed the leader (repoted by the observer). + assertLeader(ht, observer, "Carol-2") +} diff --git a/itest/lnd_no_etcd_dummy_failover_test.go b/itest/lnd_no_etcd_dummy_failover_test.go index 2d731febc..468eca3f6 100644 --- a/itest/lnd_no_etcd_dummy_failover_test.go +++ b/itest/lnd_no_etcd_dummy_failover_test.go @@ -8,3 +8,7 @@ import "github.com/lightningnetwork/lnd/lntest" // testEtcdFailover is an empty itest when LND is not compiled with etcd // support. func testEtcdFailover(ht *lntest.HarnessTest) {} + +// testLeaderHealthCheck is an empty itest when LND is not compiled with etcd +// support. +func testLeaderHealthCheck(ht *lntest.HarnessTest) {} From 91d9fb47648bd1ec19d85b2322b5fd2f4a8ef217 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 1 Aug 2024 17:14:13 +0200 Subject: [PATCH 6/8] lncfg: increase default leader session TTL to 90 seconds --- lncfg/cluster.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lncfg/cluster.go b/lncfg/cluster.go index 8b3b14f5f..e9e871a36 100644 --- a/lncfg/cluster.go +++ b/lncfg/cluster.go @@ -33,7 +33,7 @@ func DefaultCluster() *Cluster { return &Cluster{ LeaderElector: cluster.EtcdLeaderElector, EtcdElectionPrefix: DefaultEtcdElectionPrefix, - LeaderSessionTTL: 60, + LeaderSessionTTL: 90, ID: hostname, } } From f63bccb575bbaad71d377b5a0cd3e82d5722ef29 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Fri, 26 Jul 2024 17:04:04 +0200 Subject: [PATCH 7/8] config: update sample-lnd.conf --- sample-lnd.conf | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/sample-lnd.conf b/sample-lnd.conf index 70251ea10..c9fb0e4f8 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1130,6 +1130,24 @@ ; checks. This value must be >= 1m. ; healthcheck.remotesigner.interval=1m +; The number of times we should attempt to check the node's leader status +; before gracefully shutting down. Set this value to 0 to disable this health +; check. +; healthcheck.leader.attempts=1 + +; The amount of time after the leader check times out due to unanswered RPC. +; This value must be >= 1s. +; healthcheck.leader.timeout=5s + +; The amount of time we should backoff between failed attempts of leader checks. +; This value must be >= 1s. +; healthcheck.leader.backoff=5s + +; The amount of time we should wait between leader checks. +; This value must be >= 1m. +; healthcheck.leader.interval=1m + + [signrpc] @@ -1537,7 +1555,7 @@ ; The session TTL in seconds after which a new leader is elected if the old ; leader is shut down, crashed or becomes unreachable. -; cluster.leader-session-ttl=60 +; cluster.leader-session-ttl=90 [rpcmiddleware] From 037161ee6ce81eb37556ce3a9269618796b21b6e Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Thu, 1 Aug 2024 16:54:21 +0200 Subject: [PATCH 8/8] docs: add release notes for 0.18.3 --- docs/release-notes/release-notes-0.18.3.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/release-notes/release-notes-0.18.3.md b/docs/release-notes/release-notes-0.18.3.md index a4cbda9ec..fbe9c3924 100644 --- a/docs/release-notes/release-notes-0.18.3.md +++ b/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