OAS-10291 Allow to sync setup.json in default mode (no host and port specified) (#437) (#439)

Author: jwierzbo

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Use --net=host mode for Docker in tests. Extend restart tests.
 - Fix termination issue in Process tests
 - Bump Go version (1.22.8) and dependencies for CVE fixes
+- Allow to sync setup.json in default mode (no host and port specified)
 
 ## [v0.19.6](https://github.com/arangodb-helper/arangodb/tree/0.19.6) (2024-09-26)
 - Fix for member discovery during leader election

service/bootstrap_slave.go

@@ -98,7 +98,7 @@ func (s *Service) bootstrapSlave(peerAddress string, runner Runner, config Confi
 
 func RegisterPeer(log zerolog.Logger, masterURL string, req HelloRequest) ClusterConfig {
 	for {
-		log.Info().Msgf("Contacting leader %s...", masterURL)
+		log.Info().Msgf("Registering peer with master at %s", masterURL)
 
 		encoded, err := json.Marshal(req)
 		if err != nil {
@@ -157,6 +157,8 @@ func RegisterPeer(log zerolog.Logger, masterURL string, req HelloRequest) Cluste
 			return result
 		}
 
+		log.Info().Msgf("Successfully registered peer (ID: %s) with master at %s", req.SlaveID, masterURL)
+
 		return result
 	}
 }
@@ -173,3 +175,19 @@ func BuildHelloRequest(id string, slavePort int, isSecure bool, config Config, b
 		Coordinator:  copyBoolRef(bsCfg.StartCoordinator),
 	}
 }
+
+func BuildHelloRequestFromPeer(p Peer) HelloRequest {
+	return HelloRequest{
+		DataDir: p.DataDir,
+		SlaveID: p.ID,
+
+		// we can not change the address and port of the peer
+		SlaveAddress: p.Address,
+		SlavePort:    p.Port,
+
+		IsSecure:    p.IsSecure,
+		Agent:       &p.HasAgentFlag,
+		DBServer:    copyBoolRef(p.HasDBServerFlag),
+		Coordinator: copyBoolRef(p.HasCoordinatorFlag),
+	}
+}

service/cluster_config.go

@@ -81,6 +81,15 @@ func (p ClusterConfig) AllAgents() []Peer {
 	return result
 }
 
+func (p *ClusterConfig) IsPortOffsetInUse() bool {
+	for _, x := range p.AllPeers {
+		if x.PortOffset != 0 {
+			return true
+		}
+	}
+	return false
+}
+
 // Initialize a new cluster configuration
 func (p *ClusterConfig) Initialize(initialPeer Peer, agencySize int, storageEngine string, persistentOptions options.PersistentOptions) {
 	p.AllPeers = []Peer{initialPeer}

service/runtime_cluster_manager.go

@@ -25,6 +25,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"reflect"
 	"sync"
 	"time"
 
@@ -79,7 +80,7 @@ func (s *runtimeClusterManager) createAgencyAPI() (agency.Agency, error) {
 }
 
 // updateClusterConfiguration asks the master at given URL for the latest cluster configuration.
-func (s *runtimeClusterManager) updateClusterConfiguration(ctx context.Context, masterURL string, req HelloRequest) error {
+func (s *runtimeClusterManager) updateClusterConfiguration(ctx context.Context, masterURL string, myPeer string) error {
 	helloURL, err := getURLWithPath(masterURL, "/hello?update=1")
 	if err != nil {
 		return maskAny(err)
@@ -103,26 +104,25 @@ func (s *runtimeClusterManager) updateClusterConfiguration(ctx context.Context,
 	if err := json.Unmarshal(body, &clusterConfig); err != nil {
 		return maskAny(err)
 	}
-	// We've received a cluster config
-	err = s.runtimeContext.UpdateClusterConfig(clusterConfig)
-	if err != nil {
-		s.log.Error().Err(err).Msg("Failed to update cluster configuration, Trying to register peer again")
 
-		_, hostPort, err := s.runtimeContext.GetHTTPServerPort()
-		if err != nil {
-			s.log.Fatal().Err(err).Msg("Failed to get HTTP server port during runtime cluster manager start")
-		}
-		req.SlavePort = hostPort
+	latestPeerVersion, _ := s.myPeers.PeerByID(myPeer)
 
-		cfg := RegisterPeer(s.log, masterURL, req)
-		s.myPeers = cfg
+	myPeerFromMaster, exist := clusterConfig.PeerByID(myPeer)
+	if !exist {
+		s.log.Warn().Msgf("Leader responded with cluster config that does not contain this peer, re-registering. Local peer: %v", latestPeerVersion)
+		clusterConfig = RegisterPeer(s.log, masterURL, BuildHelloRequestFromPeer(latestPeerVersion))
+	}
 
-		// retry to update cluster configuration
-		err = s.runtimeContext.UpdateClusterConfig(cfg)
-		if err != nil {
-			s.log.Error().Err(err).Msg("Failed to update cluster configuration after registering peer")
-			return maskAny(err)
-		}
+	if !reflect.DeepEqual(latestPeerVersion, myPeerFromMaster) {
+		s.log.Warn().Msgf("Leader responded with cluster config that does contain different peer, re-registering. Peer from master: %v, Local peer: %v", myPeerFromMaster, latestPeerVersion)
+		clusterConfig = RegisterPeer(s.log, masterURL, BuildHelloRequestFromPeer(latestPeerVersion))
+	}
+
+	// We've received a cluster config - let's store it
+	err = s.runtimeContext.UpdateClusterConfig(clusterConfig)
+	if err != nil {
+		s.log.Warn().Err(err).Msg("Failed to update cluster configuration")
+		return err
 	}
 
 	return nil
@@ -182,16 +182,19 @@ func (s *runtimeClusterManager) runLeaderElection(ctx context.Context, myURL str
 		var masterURL string
 		var isMaster bool
 
-		s.log.Debug().
-			Str("myURL", myURL).
-			Str("masterURL", masterURL).
-			Msg("Updating leadership")
 		masterURL, isMaster, delay, err = le.Update(ctx, agencyClient, myURL)
 		if err != nil {
 			delay = 5 * time.Second
 			s.log.Error().Err(err).Msgf("Update leader election failed. Retrying in %s", delay)
 			continue
 		}
+
+		s.log.Debug().
+			Str("myURL", myURL).
+			Str("masterURL", masterURL).
+			Str("oldMasterURL", oldMasterURL).
+			Msg("Updating leadership")
+
 		if isMaster && masterURL != myURL {
 			s.log.Error().Msgf("Unexpected error: this peer is a master but URL differs. Should be %s got %s", myURL, masterURL)
 		}
@@ -225,7 +228,7 @@ func (s *runtimeClusterManager) updateMasterURL(masterURL string, isMaster bool)
 
 // Run keeps the cluster configuration up to date, either as master or as slave
 // during a running state.
-func (s *runtimeClusterManager) Run(ctx context.Context, log zerolog.Logger, runtimeContext runtimeClusterManagerContext, req HelloRequest) {
+func (s *runtimeClusterManager) Run(ctx context.Context, log zerolog.Logger, runtimeContext runtimeClusterManagerContext) {
 	s.log = log
 	s.runtimeContext = runtimeContext
 	s.interruptChan = make(chan struct{}, 32)
@@ -255,7 +258,7 @@ func (s *runtimeClusterManager) Run(ctx context.Context, log zerolog.Logger, run
 		if masterURL != "" && masterURL != ownURL {
 			log.Debug().Msgf("Updating cluster configuration master URL: %s", masterURL)
 			// We are a follower, try to update cluster configuration from leader
-			if err := s.updateClusterConfiguration(ctx, masterURL, req); err != nil {
+			if err := s.updateClusterConfiguration(ctx, masterURL, myPeer.ID); err != nil {
 				delay = time.Second * 5
 				log.Warn().Err(err).Msgf("Failed to load cluster configuration from %s", masterURL)
 			} else {

service/service.go

@@ -805,21 +805,26 @@ func (s *Service) HandleHello(ownAddress, remoteAddress string, req *HelloReques
 								break
 							}
 						}
-						// Slave address may not change in this case
-						if addrFoundInOtherPeer && p.Address != slaveAddr {
-							s.log.Warn().Msgf("Cannot change slave address while using an existing ID. Remote address: %s", remoteAddress)
-							return ClusterConfig{}, maskAny(client.NewBadRequestError("Cannot change slave address while using an existing ID."))
+						if addrFoundInOtherPeer && isLocalAddress(slaveAddr) && isLocalAddress(p.Address) && s.runtimeClusterManager.myPeers.IsPortOffsetInUse() {
+							// This is a default configuration, where host and port are not set. Keep offset.
+							s.log.Warn().Msgf("Updating slave with local address (%s). Offset (%d) will be kept. Peer id: %s",
+								p.Address, p.PortOffset, p.ID)
+						} else if addrFoundInOtherPeer && p.Address != slaveAddr {
+							msg := fmt.Sprintf("Cannot change slave address (%s) to an address that is already in use by another peer (id: %s)", slaveAddr, p.ID)
+							s.log.Warn().Msgf(msg)
+							return ClusterConfig{}, maskAny(client.NewBadRequestError(msg))
+						} else {
+							// We accept the new address (it might be the old one):
+							peer.Address = slaveAddr
+							// However, since we also accept the port, we must set the
+							// port offset of that replaced peer to 0 such that the AllPeers
+							// information actually contains the right port.
+							peer.PortOffset = 0
 						}
-						// We accept the new address (it might be the old one):
-						peer.Address = slaveAddr
-						// However, since we also accept the port, we must set the
-						// port ofset of that replaced peer to 0 such that the AllPeers
-						// information actually contains the right port.
-						peer.PortOffset = 0
+
 					}
 					peer.Port = req.SlavePort
 					peer.DataDir = req.DataDir
-
 					peer.HasAgentFlag = boolFromRef(req.Agent, peer.HasAgentFlag)
 					peer.HasCoordinatorFlag = utils.NotNilDefault(req.Coordinator, peer.HasCoordinatorFlag)
 					peer.HasDBServerFlag = utils.NotNilDefault(req.DBServer, peer.HasDBServerFlag)
@@ -867,6 +872,10 @@ func (s *Service) HandleHello(ownAddress, remoteAddress string, req *HelloReques
 	return s.runtimeClusterManager.myPeers, nil
 }
 
+func isLocalAddress(addr string) bool {
+	return addr == "127.0.0.1" || addr == "localhost"
+}
+
 // ChangeState alters the current state of the service
 func (s *Service) ChangeState(newState State) {
 	s.mutex.Lock()
@@ -989,8 +998,8 @@ func (s *Service) UpdateClusterConfig(newConfig ClusterConfig) error {
 	// Only update when changed
 	if !reflect.DeepEqual(s.runtimeClusterManager.myPeers, newConfig) {
 		s.runtimeClusterManager.myPeers = newConfig
-		s.saveSetup()
-		s.log.Debug().Msg("Updated cluster config")
+		s.log.Debug().Msgf("Updating cluster config - %v", newConfig)
+		return s.saveSetup()
 	} else {
 		s.log.Debug().Msg("Updating cluster config is not needed")
 	}
@@ -1030,7 +1039,7 @@ func (s *Service) GetHTTPServerPort() (containerPort, hostPort int, err error) {
 		if myPeer, ok := s.runtimeClusterManager.myPeers.PeerByID(s.id); ok {
 			containerPort += myPeer.PortOffset
 		} else {
-			return 0, 0, maskAny(fmt.Errorf("No peer information found for ID '%s'", s.id))
+			return 0, 0, maskAny(fmt.Errorf("no peer information found for ID '%s'", s.id))
 		}
 	}
 	if s.isNetHost {
@@ -1050,6 +1059,7 @@ func (s *Service) createHTTPServer(config Config) (srv *httpServer, containerPor
 	hostAddr = net.JoinHostPort(config.OwnAddress, strconv.Itoa(hostPort))
 
 	// Create HTTP server
+	s.log.Debug().Msgf("Creating HTTP server on %s", hostAddr)
 	return newHTTPServer(s.log, s, &s.runtimeServerManager, config, s.id), containerPort, hostAddr, containerAddr, nil
 }
 
@@ -1095,7 +1105,7 @@ func (s *Service) startRunning(runner Runner, config Config, bsCfg BootstrapConf
 	go func() {
 		defer wg.Done()
 
-		s.runtimeClusterManager.Run(s.stopPeer.ctx, s.log, s, BuildHelloRequest(s.id, 0, s.IsSecure(), config, bsCfg))
+		s.runtimeClusterManager.Run(s.stopPeer.ctx, s.log, s)
 	}()
 
 	// Start the upgrade manager
@@ -1228,7 +1238,7 @@ func (s *Service) adjustClusterConfigForRelaunch(bsCfg BootstrapConfig) {
 
 	s.runtimeClusterManager.myPeers.ForEachPeer(func(p Peer) Peer {
 		if bsCfg.ID == p.ID {
-			s.log.Debug().Msgf("Adjusting current memeber cluster config after restart (port: %d)", p.Port)
+			s.log.Debug().Msgf("Adjusting current memeber cluster config (locally) after restart (port: %d)", p.Port)
 			p.peerServers = preparePeerServers(s.mode, bsCfg, &p)
 		}
 		return p

test/process_cluster_local_test.go

@@ -40,7 +40,7 @@ func TestProcessClusterLocal(t *testing.T) {
 	child := Spawn(t, "${STARTER} --starter.local "+createEnvironmentStarterOptions())
 	defer child.Close()
 
-	if ok := WaitUntilStarterReady(t, whatCluster, 3, child); ok {
+	if ok := WaitUntilStarterReady(t, whatCluster, 1, child); ok {
 		t.Logf("Cluster start took %s", time.Since(start))
 		testCluster(t, insecureStarterEndpoint(0*portIncrement), false)
 		testCluster(t, insecureStarterEndpoint(1*portIncrement), false)
@@ -65,7 +65,7 @@ func TestProcessClusterLocalShutdownViaAPI(t *testing.T) {
 	child := Spawn(t, "${STARTER} --starter.local "+createEnvironmentStarterOptions())
 	defer child.Close()
 
-	if ok := WaitUntilStarterReady(t, whatCluster, 3, child); ok {
+	if ok := WaitUntilStarterReady(t, whatCluster, 1, child); ok {
 		t.Logf("Cluster start took %s", time.Since(start))
 		testCluster(t, insecureStarterEndpoint(0*portIncrement), false)
 		testCluster(t, insecureStarterEndpoint(1*portIncrement), false)
@@ -88,7 +88,7 @@ func TestOldProcessClusterLocal(t *testing.T) {
 	child := Spawn(t, "${STARTER} --local "+createEnvironmentStarterOptions())
 	defer child.Close()
 
-	if ok := WaitUntilStarterReady(t, whatCluster, 3, child); ok {
+	if ok := WaitUntilStarterReady(t, whatCluster, 1, child); ok {
 		t.Logf("Cluster start took %s", time.Since(start))
 		testCluster(t, insecureStarterEndpoint(0*portIncrement), false)
 		testCluster(t, insecureStarterEndpoint(1*portIncrement), false)