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") + } +}