From ff15b6c7b0d22b6e2ed03f569913e8f6c9f8838f Mon Sep 17 00:00:00 2001 From: Joseph Robert Torsella Date: Wed, 30 Sep 2026 10:47:48 -0400 Subject: [PATCH] cmd/microcloud: Issue each peer's join tokens just before its join This PR prevents an issue where every service join token was issued before the first join such that if there were many joins to do the tokens for the later joiners could expire before their joins had a chance to start (or finish). This changes the behavior to issue tokens prior to each join rather than all at once at the beginning. Fixes canonical/microcloud#1565 Signed-off-by: Joseph Robert Torsella --- cmd/microcloud/main_init.go | 29 ++++- cmd/microcloud/main_init_test.go | 200 +++++++++++++++++++++++++++++++ 2 files changed, 224 insertions(+), 5 deletions(-) diff --git a/cmd/microcloud/main_init.go b/cmd/microcloud/main_init.go index c576cac31..f5ef1eaef 100644 --- a/cmd/microcloud/main_init.go +++ b/cmd/microcloud/main_init.go @@ -480,7 +480,7 @@ func (c *initConfig) addPeers(sh *service.Handler) (revert.Hook, error) { } // Concurrently issue a token for each joiner. - for peer := range c.systems { + issueTokens := func(peer string) error { mut := sync.Mutex{} err := sh.RunConcurrent("", "", func(s service.Service) error { // Skip MicroCloud as the cluster is already formed. @@ -527,13 +527,22 @@ func (c *initConfig) addPeers(sh *service.Handler) (revert.Hook, error) { return nil }) if err != nil { - return nil, err + return err } + + return nil } fmt.Println("Awaiting cluster formation ...") + // Peers join one at a time, so issue each peer's tokens right before its join. + // Issuing them all up front lets the tokens of the last peers expire on large clusters. // If the local node needs to join an existing cluster, do it first so we can proceed as normal. + err := issueTokens(sh.Name) + if err != nil { + return nil, err + } + if len(joinConfig[sh.Name].Tokens) > 0 { cfg := joinConfig[sh.Name] err := waitForJoin(sh, clusterSize, sh.Name, nil, cfg) @@ -544,13 +553,23 @@ func (c *initConfig) addPeers(sh *service.Handler) (revert.Hook, error) { fmt.Println(tui.SummarizeResult("Peer %s has joined the cluster", sh.Name)) } - for peer, cfg := range joinConfig { - if len(cfg.Tokens) == 0 || peer == sh.Name { + for peer := range joinConfig { + if peer == sh.Name { + continue + } + + err := issueTokens(peer) + if err != nil { + return nil, err + } + + cfg := joinConfig[peer] + if len(cfg.Tokens) == 0 { continue } logger.Debug("Initiating sequential request for cluster join", logger.Ctx{"peer": peer}) - err := waitForJoin(sh, clusterSize, peer, nil, cfg) + err = waitForJoin(sh, clusterSize, peer, nil, cfg) if err != nil { return nil, err } diff --git a/cmd/microcloud/main_init_test.go b/cmd/microcloud/main_init_test.go index 4cc8360f9..9906fef5a 100644 --- a/cmd/microcloud/main_init_test.go +++ b/cmd/microcloud/main_init_test.go @@ -1,10 +1,20 @@ package main import ( + "context" + "crypto/tls" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync" "testing" + "github.com/canonical/lxd/shared" lxdAPI "github.com/canonical/lxd/shared/api" + microTypes "github.com/canonical/microcluster/v3/microcluster/types" + "github.com/canonical/microcloud/microcloud/api/types" "github.com/canonical/microcloud/microcloud/multicast" "github.com/canonical/microcloud/microcloud/service" ) @@ -226,3 +236,193 @@ func TestValidateSystemsMultiSystem(t *testing.T) { t.Fatalf("sys4 with conflicting management IP and ipv6.ovn.ranges passed validation") } } + +// joinTracker records the tokens issued to peers and the peers that have joined. +// It checks that a peer's tokens are only issued once every peer before it has joined. +type joinTracker struct { + t *testing.T + + mu sync.Mutex + tokens map[string]string // Token to the peer it was issued for. + issued map[string]int // Peer to the number of tokens issued for it. + joined map[string]bool +} + +// issue records a token for the peer and returns it. +func (j *joinTracker) issue(serviceType types.ServiceType, peer string) string { + j.mu.Lock() + defer j.mu.Unlock() + + if j.joined[peer] { + j.t.Errorf("Issued %s token for peer %q after it joined", serviceType, peer) + } + + token := fmt.Sprintf("%s-%s", serviceType, peer) + j.tokens[token] = peer + j.issued[peer]++ + + return token +} + +// members returns the local system and the peers that have joined. +func (j *joinTracker) members() map[string]string { + j.mu.Lock() + defer j.mu.Unlock() + + members := map[string]string{"local": "10.0.0.1"} + for peer := range j.joined { + members[peer] = peer + } + + return members +} + +// ServeHTTP answers the join request that waitForJoin sends to a peer. +// The peer is told apart by the tokens it was asked to join with. +func (j *joinTracker) ServeHTTP(w http.ResponseWriter, r *http.Request) { + fail := func(format string, args ...any) { + j.t.Errorf(format, args...) + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(lxdAPI.Response{Type: lxdAPI.ErrorResponse, Code: http.StatusBadRequest, Error: fmt.Sprintf(format, args...)}) + } + + path := "/" + string(types.APIVersion) + "/services" + if r.Method != http.MethodPut || r.URL.Path != path { + fail("Unexpected join request %s %s", r.Method, r.URL.Path) + return + } + + var cfg types.ServicesPut + err := json.NewDecoder(r.Body).Decode(&cfg) + if err != nil { + fail("Failed to decode join request: %v", err) + return + } + + j.mu.Lock() + defer j.mu.Unlock() + + peer := "" + for _, token := range cfg.Tokens { + tokenPeer, ok := j.tokens[token.JoinToken] + if !ok || (peer != "" && tokenPeer != peer) { + fail("Join request carried %s token %q that was not issued for the joining peer", token.Service, token.JoinToken) + return + } + + peer = tokenPeer + } + + if peer == "" { + fail("Join request carried no tokens") + return + } + + if j.joined[peer] { + fail("Peer %q joined twice", peer) + return + } + + // Only the joining peer and the peers that already joined may have tokens. + for tokenPeer := range j.issued { + if tokenPeer != peer && !j.joined[tokenPeer] { + j.t.Errorf("Token for peer %q was issued before peer %q joined", tokenPeer, peer) + } + } + + j.joined[peer] = true + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(lxdAPI.Response{Type: lxdAPI.SyncResponse, Status: "Success", StatusCode: http.StatusOK, Metadata: json.RawMessage("null")}) +} + +// tokenTestService is a service whose tokens and cluster members are kept by a joinTracker. +type tokenTestService struct { + service.Service + + serviceType types.ServiceType + tracker *joinTracker +} + +func (s *tokenTestService) Type() types.ServiceType { + return s.serviceType +} + +func (s *tokenTestService) IssueToken(ctx context.Context, peer string) (string, error) { + return s.tracker.issue(s.serviceType, peer), nil +} + +func (s *tokenTestService) DeleteToken(ctx context.Context, tokenName string, address string) error { + return nil +} + +func (s *tokenTestService) ClusterMembers(ctx context.Context) (map[string]string, error) { + return s.tracker.members(), nil +} + +func TestAddPeersIssuesTokensBeforeEachJoin(t *testing.T) { + serviceTypes := []types.ServiceType{types.LXD, types.MicroCeph, types.MicroOVN} + peers := []string{"peer1", "peer2", "peer3", "peer4", "peer5"} + + tracker := &joinTracker{t: t, tokens: map[string]string{}, issued: map[string]int{}, joined: map[string]bool{}} + + // The peers answer join requests from a server presenting the cluster certificate in the state directory, + // which is what MicroCloud verifies peers against once they are members. + stateDir := t.TempDir() + clusterCert, err := shared.KeyPairAndCA(stateDir, string(microTypes.ClusterCertificateName), shared.CertServer, shared.CertOptions{}) + if err != nil { + t.Fatalf("Failed to generate cluster certificate: %v", err) + } + + server := httptest.NewUnstartedServer(tracker) + server.TLS = &tls.Config{Certificates: []tls.Certificate{clusterCert.KeyPair()}} + server.StartTLS() + t.Cleanup(server.Close) + + cloud, err := service.NewCloudService("local", "10.0.0.1", stateDir) + if err != nil { + t.Fatalf("Failed to create MicroCloud service: %v", err) + } + + handler := &service.Handler{Name: "local", Services: map[types.ServiceType]service.Service{types.MicroCloud: cloud}} + for _, serviceType := range serviceTypes { + handler.Services[serviceType] = &tokenTestService{serviceType: serviceType, tracker: tracker} + } + + // The local system has bootstrapped every service, and the peers have already joined MicroCloud. + localServices := map[types.ServiceType]map[string]string{types.MicroCloud: {"local": "10.0.0.1"}} + for _, serviceType := range serviceTypes { + localServices[serviceType] = map[string]string{"local": "10.0.0.1"} + } + + cfg := initConfig{ + systems: map[string]InitSystem{"local": {ServerInfo: multicast.ServerInfo{Name: "local", Address: "10.0.0.1"}}}, + state: map[string]service.SystemInformation{"local": {ExistingServices: localServices}}, + } + + // Every peer is reached at the test server, which tells them apart by their tokens. + address := server.Listener.Addr().String() + for _, peer := range peers { + localServices[types.MicroCloud][peer] = address + cfg.systems[peer] = InitSystem{ServerInfo: multicast.ServerInfo{Name: peer, Address: address}} + } + + _, err = cfg.addPeers(handler) + if err != nil { + t.Fatalf("Failed to add peers: %v", err) + } + + for _, peer := range peers { + if !tracker.joined[peer] { + t.Errorf("Peer %q did not join", peer) + } + + if tracker.issued[peer] != len(serviceTypes) { + t.Errorf("Issued %d tokens for peer %q, expected %d", tracker.issued[peer], peer, len(serviceTypes)) + } + } + + if tracker.joined["local"] || tracker.issued["local"] != 0 { + t.Errorf("Local system should not join or receive tokens") + } +}