diff --git a/example/test/main.sh b/example/test/main.sh index a89a7a67..fe97f6a7 100755 --- a/example/test/main.sh +++ b/example/test/main.sh @@ -1047,6 +1047,139 @@ test_self_deletion() { shutdown_systems } +test_failed_join() { + echo "Testing failed join" + + new_systems 4 --heartbeat 2s + + # Bootstrap initial cluster. + microctl --state-dir "${test_dir}/c1" init "c1" 127.0.0.1:9001 --bootstrap + + token_c2=$(microctl --state-dir "${test_dir}/c1" tokens add "c2") + token_c3=$(microctl --state-dir "${test_dir}/c1" tokens add "c3") + microctl --state-dir "${test_dir}/c2" init "c2" 127.0.0.1:9002 --token "${token_c2}" + microctl --state-dir "${test_dir}/c3" init "c3" 127.0.0.1:9003 --token "${token_c3}" + + # Wait for cluster to stabilize + while [[ -n "$(microctl --state-dir "${test_dir}/c1" cluster list -f yaml | yq '.[] | select(.role == "PENDING")')" ]]; do + sleep 2 + done + + # An invalid schema patch makes the join of c4 fail after dqlite has already added it to the cluster. + echo " -> Testing join of c4 with a failing schema update" + token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4") + echo "INTENTIONALLY INVALID SQL STATEMENT;" > "${test_dir}/c4/patch.global.sql" + if microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then + echo "ERROR: Member c4 should have failed to join" + exit 1 + fi + + # The remaining cluster members keep connecting to the failed joiner for a while, + # which must not crash its daemon after its dqlite app has been closed. + echo " -> Testing c4 daemon survives the failed join" + sleep 5 + if ! microctl --state-dir "${test_dir}/c4" waitready --timeout 30; then + echo "ERROR: Daemon of member c4 did not survive the failed join" + exit 1 + fi + + echo " -> Testing c4 is no longer present on the remaining cluster" + if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "3" ]; then + echo "ERROR: Member c4 is still present on the remaining cluster" + microctl --state-dir "${test_dir}/c1" cluster list + exit 1 + fi + + echo " -> Testing c4 can join after the failure is resolved" + rm -f "${test_dir}/c4/patch.global.sql" + token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4") + if ! microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then + echo "ERROR: Member c4 failed to join after the failure was resolved" + exit 1 + fi + + if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "4" ]; then + echo "ERROR: Member c4 is missing from the cluster after joining" + microctl --state-dir "${test_dir}/c1" cluster list + exit 1 + fi + + shutdown_systems +} + +test_failed_join_client_timeout() { + echo "Testing failed join after the client timed out" + + new_systems 4 --heartbeat 2s + + # Bootstrap initial cluster. + microctl --state-dir "${test_dir}/c1" init "c1" 127.0.0.1:9001 --bootstrap + + token_c2=$(microctl --state-dir "${test_dir}/c1" tokens add "c2") + token_c3=$(microctl --state-dir "${test_dir}/c1" tokens add "c3") + microctl --state-dir "${test_dir}/c2" init "c2" 127.0.0.1:9002 --token "${token_c2}" + microctl --state-dir "${test_dir}/c3" init "c3" 127.0.0.1:9003 --token "${token_c3}" + + # Wait for cluster to stabilize + while [[ -n "$(microctl --state-dir "${test_dir}/c1" cluster list -f yaml | yq '.[] | select(.role == "PENDING")')" ]]; do + sleep 2 + done + + # A slow and then invalid schema patch makes the join of c4 fail several seconds after + # dqlite has already added it to the cluster, and after the client below has given up. + echo " -> Testing join of c4 with a client that times out before the join fails" + token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4") + cat << EOF > "${test_dir}/c4/patch.global.sql" +CREATE TEMP TABLE slow AS WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x < 3000000) SELECT x FROM c; +INTENTIONALLY INVALID SQL STATEMENT; +EOF + + if curl -sS --max-time 1 --unix-socket "${test_dir}/c4/control.socket" \ + -X POST -H "Content-Type: application/json" \ + -d "{\"bootstrap\":false,\"join_token\":\"${token_c4}\",\"address\":\"127.0.0.1:9004\",\"name\":\"c4\",\"config\":{}}" \ + http://control.socket/core/control; then + echo "ERROR: Join request of c4 should have timed out" + exit 1 + fi + + # Make sure the join got far enough to change the cluster state before the client gave up. + if [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "4" ]; then + echo "ERROR: Member c4 was not added to the cluster before the join request timed out" + microctl --state-dir "${test_dir}/c1" cluster list + exit 1 + fi + + # The join failure cleanup must still run, even though the join request context has expired. + echo " -> Testing c4 is removed from the remaining cluster" + retries=0 + while [ "$(microctl --state-dir "${test_dir}/c1" cluster list -f csv | wc -l)" != "3" ]; do + retries=$((retries + 1)) + if [ "${retries}" -gt 30 ]; then + echo "ERROR: Member c4 is still present on the remaining cluster" + microctl --state-dir "${test_dir}/c1" cluster list + exit 1 + fi + + sleep 2 + done + + if ! microctl --state-dir "${test_dir}/c1" tokens add "c5" > /dev/null; then + echo "ERROR: Cluster membership is inconsistent after the failed join" + exit 1 + fi + + echo " -> Testing c4 can join after the failure is resolved" + microctl --state-dir "${test_dir}/c4" waitready --timeout 30 + rm -f "${test_dir}/c4/patch.global.sql" + token_c4=$(microctl --state-dir "${test_dir}/c1" tokens add "c4") + if ! microctl --state-dir "${test_dir}/c4" init "c4" 127.0.0.1:9004 --token "${token_c4}"; then + echo "ERROR: Member c4 failed to join after the failure was resolved" + exit 1 + fi + + shutdown_systems +} + # allow for running a specific set of tests TEST_RESULT="success" if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then @@ -1061,6 +1194,8 @@ if [ "${1:-"all"}" = "all" ] || [ "${1}" = "" ]; then run_test truststore_force_removal run_test parallel_joins run_test self_deletion + run_test failed_join + run_test failed_join_client_timeout elif [ "${1}" = "recover" ]; then run_test recover elif [ "${1}" = "tokens" ]; then @@ -1081,6 +1216,10 @@ elif [ "${1}" = "parallel-join" ]; then run_test parallel_joins elif [ "${1}" = "self-deletion" ]; then run_test self_deletion +elif [ "${1}" = "failed-join" ]; then + run_test failed_join +elif [ "${1}" = "failed-join-client-timeout" ]; then + run_test failed_join_client_timeout elif [ "${1}" = "daemon-config" ]; then run_test daemon_config_api else diff --git a/internal/db/dqlite.go b/internal/db/dqlite.go index 835f5ca6..c772ef16 100644 --- a/internal/db/dqlite.go +++ b/internal/db/dqlite.go @@ -68,7 +68,9 @@ const ( // Accept passes conn to dqlite. During normal operation the send completes // immediately. During a Restart the channel has no consumer yet, so the call // blocks for up to 10s, keeping clients "on hold" until the new dqlite -// instance is ready, making the restart invisible to them. +// instance is ready, making the restart invisible to them. If there is no +// dqlite instance at all (e.g. after a failed bootstrap or join), the +// connection is closed after the timeout. // acceptMu is held so that Restart can atomically rotate the channel. func (db *DqliteDB) Accept(conn net.Conn) { db.acceptMu.RLock() @@ -83,6 +85,23 @@ func (db *DqliteDB) Accept(conn net.Conn) { } } +// rotateAcceptCh swaps in a fresh accept channel and closes the old one, so that the +// go-dqlite reader ranging over it exits. This must be done before closing a dqlite app, +// as the reader otherwise keeps accepting connections and panics when it fails to dial +// the closed app's local socket. +// Accept() holds acceptMu for every send, so once the write lock is held there +// can be no remaining senders on the old channel. +func (db *DqliteDB) rotateAcceptCh() { + db.acceptMu.Lock() + oldAcceptCh := db.acceptCh + db.acceptCh = make(chan net.Conn) + db.acceptMu.Unlock() + + if oldAcceptCh != nil { + close(oldAcceptCh) + } +} + // NewDB creates an empty db struct with no dqlite connection. func NewDB(ctx context.Context, serverCert func() *shared.CertInfo, clusterCert func() *shared.CertInfo, memberName func() string, failureDomain func() uint64, os types.OS, heartbeatInterval time.Duration) (*DqliteDB, error) { shutdownCtx, shutdownCancel := context.WithCancel(ctx) @@ -179,6 +198,8 @@ func (db *DqliteDB) Bootstrap(extensions types.Extensions, addr *url.URL, cluste defer reverter.Fail() reverter.Add(func() { if db.dqlite != nil { + db.rotateAcceptCh() + closeErr := db.dqlite.Close() if closeErr != nil { db.log().Error("Failed to close dqlite app", slog.String("address", db.listenAddr.String()), slog.String("error", closeErr.Error())) @@ -232,6 +253,8 @@ func (db *DqliteDB) Join(extensions types.Extensions, addr *url.URL, joinAddress defer reverter.Fail() reverter.Add(func() { if db.dqlite != nil { + db.rotateAcceptCh() + closeErr := db.dqlite.Close() if closeErr != nil { db.log().Error("Failed to close dqlite app", slog.String("address", db.listenAddr.String()), slog.String("error", closeErr.Error())) @@ -522,18 +545,7 @@ func (db *DqliteDB) Restart(extensions types.Extensions, clusterMembers map[stri db.status = types.DatabaseOffline db.statusLock.Unlock() - // Swap in a fresh accept channel for the next dqlite instance and close the - // old one so any stale go-dqlite reader ranging over it can exit. - // Accept() holds acceptMu for every send, so once the write lock is held there - // can be no remaining senders on the old channel. - db.acceptMu.Lock() - oldAcceptCh := db.acceptCh - db.acceptCh = make(chan net.Conn) - db.acceptMu.Unlock() - - if oldAcceptCh != nil { - close(oldAcceptCh) - } + db.rotateAcceptCh() if db.db != nil { _ = db.db.Close() diff --git a/internal/rest/resources/cluster.go b/internal/rest/resources/cluster.go index 691e6ab2..7dea2ede 100644 --- a/internal/rest/resources/cluster.go +++ b/internal/rest/resources/cluster.go @@ -53,6 +53,9 @@ var clusterMemberCmd = types.Endpoint{ var clusterMemberInternalCmd = types.Endpoint{ Path: "cluster/{name}", + // Required to allow resetting a member whose database failed to start, e.g. after a failed join. + AllowedBeforeInit: true, + Put: types.EndpointAction{Handler: clusterMemberPut, AccessHandler: access.AllowAuthenticated}, } diff --git a/internal/rest/resources/control.go b/internal/rest/resources/control.go index 939fd144..b5843a89 100644 --- a/internal/rest/resources/control.go +++ b/internal/rest/resources/control.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "slices" + "time" "github.com/canonical/lxd/shared" "github.com/canonical/lxd/shared/api" @@ -91,8 +92,14 @@ func controlPost(state types.State, r *http.Request) types.Response { return } + // Use the daemon's context for cleanup so that it is not tied to the request + // context, which may already be expired (e.g. the join timed out), while + // still stopping the cleanup if the daemon shuts down. + cleanupCtx, cleanupCancel := context.WithTimeout(intState.Context, 2*time.Minute) + defer cleanupCancel() + // Run the pre-remove hook like we do for cluster node removals. - err := intState.Hooks.PreRemove(r.Context(), state, true) + err := intState.Hooks.PreRemove(cleanupCtx, state, true) if err != nil { logger.Error("Failed to run pre-remove hook on initialization error", slog.String("error", err.Error())) } @@ -101,6 +108,8 @@ func controlPost(state types.State, r *http.Request) types.Response { // As part of this request the cluster member gets re-executed. // If we don't send the request, re-exec the member manually. if joinInfo == nil || req.JoinToken == "" { + // Use the request context, as the re-exec waits for it to be done + // so that the response can be sent first. reExec, err := resetClusterMember(r.Context(), state, true) if err != nil { logger.Error("Failed to reset cluster member on bootstrap error", slog.String("error", err.Error())) @@ -114,13 +123,15 @@ func controlPost(state types.State, r *http.Request) types.Response { } url := api.NewURL().Scheme("https").Host(joinInfo.TrustedMember.Address.String()) - cert, err := shared.GetRemoteCertificate(r.Context(), url.String(), "") + cert, err := shared.GetRemoteCertificate(cleanupCtx, url.String(), "") if err != nil { + logger.Error("Failed to get certificate of cluster member for cleanup", slog.String("address", url.String()), slog.String("error", err.Error())) return } client, err := state.Connect().Member(&url.URL, false, cert) if err != nil { + logger.Error("Failed to create client for cluster member cleanup", slog.String("address", url.String()), slog.String("error", err.Error())) return }