NET-655

cmd/pull.go

@@ -16,7 +16,7 @@ var pullCmd = &cobra.Command{
 	Short: "get the latest host configuration",
 	Long:  `get the latest host configuration and peers from all connected servers`,
 	Run: func(cmd *cobra.Command, args []string) {
-		_, err := functions.Pull(true)
+		_, _, _, err := functions.Pull(true)
 		if err != nil {
 			logger.Log(0, "failed to pull", err.Error())
 		}

config/config.go

@@ -10,6 +10,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -57,7 +58,8 @@ func (i InitType) String() string {
 }
 
 var (
-	netclient Config // netclient contains the netclient config
+	netclientCfgMutex = &sync.RWMutex{}
+	netclient         Config // netclient contains the netclient config
 	// Version - default version string
 	Version = "dev"
 	// FwClose - firewall manager shutdown func
@@ -85,10 +87,14 @@ func init() {
 	Nodes = make(map[string]Node)
 }
 
-// UpdateNetcllient updates the in memory version of the host configuration
+// UpdateNetclient updates the in memory version of the host configuration
 func UpdateNetclient(c Config) {
+	netclientCfgMutex.Lock()
+	defer netclientCfgMutex.Unlock()
+	if c.Verbosity != logger.Verbosity {
+		logger.Log(3, "Logging verbosity updated to", strconv.Itoa(logger.Verbosity))
+	}
 	logger.Verbosity = c.Verbosity
-	logger.Log(3, "Logging verbosity updated to", strconv.Itoa(logger.Verbosity))
 	netclient = c
 }
 
@@ -130,21 +136,28 @@ func UpdateHost(host *models.Host) (resetInterface, restart, sendHostUpdate bool
 
 // Netclient returns a pointer to the im memory version of the host configuration
 func Netclient() *Config {
+	netclientCfgMutex.RLock()
+	defer netclientCfgMutex.RUnlock()
 	return &netclient
 }
 
 // UpdateHostPeers - updates host peer map in the netclient config
 func UpdateHostPeers(peers []wgtypes.PeerConfig) {
+	netclientCfgMutex.Lock()
+	defer netclientCfgMutex.Unlock()
 	netclient.HostPeers = peers
 }
 
 // DeleteServerHostPeerCfg - deletes the host peers for the server
 func DeleteServerHostPeerCfg() {
+	netclientCfgMutex.Lock()
+	defer netclientCfgMutex.Unlock()
 	netclient.HostPeers = []wgtypes.PeerConfig{}
 }
 
 // RemoveServerHostPeerCfg - sets remove flag for all peers on the given server peers
 func RemoveServerHostPeerCfg() {
+	netclient := Netclient()
 	if netclient.HostPeers == nil {
 		netclient.HostPeers = []wgtypes.PeerConfig{}
 		return
@@ -156,6 +169,7 @@ func RemoveServerHostPeerCfg() {
 		peers[i] = peer
 	}
 	netclient.HostPeers = peers
+	UpdateNetclient(*netclient)
 	_ = WriteNetclientConfig()
 }
 
@@ -177,6 +191,8 @@ func ReadNetclientConfig() (*Config, error) {
 		return nil, err
 	}
 	defer f.Close()
+	netclientCfgMutex.Lock()
+	defer netclientCfgMutex.Unlock()
 	netclient = Config{}
 	if err := yaml.NewDecoder(f).Decode(&netclient); err != nil {
 		return nil, err
@@ -209,6 +225,8 @@ func WriteNetclientConfig() error {
 		return err
 	}
 	defer f.Close()
+	netclientCfgMutex.Lock()
+	defer netclientCfgMutex.Unlock()
 	err = yaml.NewEncoder(f).Encode(netclient)
 	if err != nil {
 		return err

config/node.go

@@ -7,13 +7,16 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"sync"
 
 	"github.com/gravitl/netclient/ncutils"
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/models"
 	"gopkg.in/yaml.v3"
 )
 
+var nodeMutex = &sync.RWMutex{}
+
 // NodeMap is an in memory map of the all nodes indexed by network name
 type NodeMap map[string]Node
 
@@ -41,9 +44,9 @@ func ReadNodeConfig() error {
 		return err
 	}
 	defer f.Close()
-	for k := range Nodes {
-		delete(Nodes, k)
-	}
+	nodeMutex.Lock()
+	defer nodeMutex.Unlock()
+	Nodes = make(NodeMap)
 	if err := yaml.NewDecoder(f).Decode(&Nodes); err != nil {
 		return err
 	}
@@ -52,11 +55,15 @@ func ReadNodeConfig() error {
 
 // GetNodes returns a copy of the NodeMap
 func GetNodes() NodeMap {
+	nodeMutex.RLock()
+	defer nodeMutex.RUnlock()
 	return Nodes
 }
 
 // GetNode returns returns the node configuation of the specified network name
 func GetNode(k string) Node {
+	nodeMutex.RLock()
+	defer nodeMutex.RUnlock()
 	if node, ok := Nodes[k]; ok {
 		return node
 	}
@@ -65,6 +72,8 @@ func GetNode(k string) Node {
 
 // SetNodes - sets server nodes in client config
 func SetNodes(nodes []models.Node) {
+	nodeMutex.Lock()
+	defer nodeMutex.Unlock()
 	Nodes = make(NodeMap)
 	for _, node := range nodes {
 		Nodes[node.Network] = Node{
@@ -75,11 +84,15 @@ func SetNodes(nodes []models.Node) {
 
 // UpdateNodeMap updates the in memory nodemap for the specified network
 func UpdateNodeMap(k string, value Node) {
+	nodeMutex.Lock()
+	defer nodeMutex.Unlock()
 	Nodes[k] = value
 }
 
 // DeleteNode deletes the node from the nodemap for the specified network
 func DeleteNode(k string) {
+	nodeMutex.Lock()
+	defer nodeMutex.Unlock()
 	delete(Nodes, k)
 }
 
@@ -116,6 +129,8 @@ func WriteNodeConfig() error {
 		return err
 	}
 	defer f.Close()
+	nodeMutex.RLock()
+	defer nodeMutex.RUnlock()
 	err = yaml.NewEncoder(f).Encode(Nodes)
 	if err != nil {
 		return err

config/server.go

@@ -5,13 +5,16 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/google/uuid"
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/models"
 	"gopkg.in/yaml.v3"
 )
 
+var serverMutex = &sync.RWMutex{}
+
 var serverCtxFile = ".serverctx"
 
 // CurrServer - holds the value of current server of client
@@ -65,9 +68,9 @@ func ReadServerConf() error {
 		return err
 	}
 	defer f.Close()
-	for k := range Servers {
-		delete(Servers, k)
-	}
+	serverMutex.Lock()
+	defer serverMutex.Unlock()
+	Servers = make(map[string]Server)
 	if err := yaml.NewDecoder(f).Decode(&Servers); err != nil {
 		return err
 	}
@@ -99,6 +102,8 @@ func WriteServerConfig() error {
 		return err
 	}
 	defer f.Close()
+	serverMutex.Lock()
+	defer serverMutex.Unlock()
 	err = yaml.NewEncoder(f).Encode(Servers)
 	if err != nil {
 		return err
@@ -108,17 +113,23 @@ func WriteServerConfig() error {
 
 // SaveServer updates the server map with current server struct and writes map to disk
 func SaveServer(name string, server Server) error {
+	serverMutex.Lock()
 	Servers[name] = server
+	serverMutex.Unlock()
 	return WriteServerConfig()
 }
 
 // UpdateServer updates the in-memory server map
 func UpdateServer(name string, server Server) {
+	serverMutex.Lock()
+	defer serverMutex.Unlock()
 	Servers[name] = server
 }
 
 // GetServer returns the server struct for the given server name
 func GetServer(name string) *Server {
+	serverMutex.RLock()
+	defer serverMutex.RUnlock()
 	if server, ok := Servers[name]; ok {
 		return &server
 	}
@@ -127,6 +138,8 @@ func GetServer(name string) *Server {
 
 // GetServers - gets all the server names host has registered to.
 func GetServers() (servers []string) {
+	serverMutex.RLock()
+	defer serverMutex.RUnlock()
 	for _, server := range Servers {
 		servers = append(servers, server.Name)
 	}
@@ -173,6 +186,8 @@ func SetServerCtx() {
 
 // DeleteServer deletes the specified server name from the server map
 func DeleteServer(k string) {
+	serverMutex.Lock()
+	defer serverMutex.Unlock()
 	delete(Servers, k)
 }
 
@@ -194,6 +209,8 @@ func ConvertServerCfg(cfg *OldNetmakerServerConfig) *Server {
 
 // UpdateServerConfig updates the in memory server map with values provided from netmaker server
 func UpdateServerConfig(cfg *models.ServerConfig) {
+	serverMutex.Lock()
+	defer serverMutex.Unlock()
 	if cfg == nil {
 		return
 	}
@@ -205,6 +222,5 @@ func UpdateServerConfig(cfg *models.ServerConfig) {
 	server.Name = cfg.Server
 	server.MQID = netclient.ID
 	server.ServerConfig = *cfg
-
 	Servers[cfg.Server] = server
 }

functions/daemon.go

@@ -70,19 +70,20 @@ func Daemon() {
 	}
 	cancel := startGoRoutines(&wg)
 	//start httpserver on its own -- doesn't need to restart on reset
-	httpctx, httpCancel := context.WithCancel(context.Background())
-	httpWg := sync.WaitGroup{}
-	httpWg.Add(1)
-	go HttpServer(httpctx, &httpWg)
+	ctx0, cancel0 := context.WithCancel(context.Background())
+	wg0 := sync.WaitGroup{}
+	wg0.Add(1)
+	go HttpServer(ctx0, &wg0)
+
 	for {
 		select {
 		case <-quit:
 			slog.Info("shutting down netclient daemon")
 			closeRoutines([]context.CancelFunc{
 				cancel,
 			}, &wg)
-			httpCancel()
-			httpWg.Wait()
+			cancel0()
+			wg0.Wait()
 			config.FwClose()
 			slog.Info("shutdown complete")
 			return
@@ -151,7 +152,7 @@ func startGoRoutines(wg *sync.WaitGroup) context.CancelFunc {
 		}
 	}
 	slog.Info("configuring netmaker wireguard interface")
-	pullresp, pullErr := Pull(false)
+	pullresp, _, _, pullErr := Pull(false)
 	if pullErr != nil {
 		slog.Error("fail to pull config from server", "error", pullErr.Error())
 	}
@@ -181,6 +182,8 @@ func startGoRoutines(wg *sync.WaitGroup) context.CancelFunc {
 		wg.Add(1)
 		go watchPeerConnections(ctx, wg)
 	}
+	wg.Add(1)
+	go mqFallback(ctx, wg)
 
 	return cancel
 }
@@ -208,8 +211,13 @@ func messageQueue(ctx context.Context, wg *sync.WaitGroup, server *config.Server
 func setupMQTT(server *config.Server) error {
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(server.Broker)
-	opts.SetUsername(server.MQUserName)
-	opts.SetPassword(server.MQPassword)
+	if server.BrokerType == "emqx" {
+		opts.SetUsername(config.Netclient().ID.String())
+		opts.SetPassword(config.Netclient().HostPass)
+	} else {
+		opts.SetUsername(server.MQUserName)
+		opts.SetPassword(server.MQPassword)
+	}
 	//opts.SetClientID(ncutils.MakeRandomString(23))
 	opts.SetClientID(server.MQID.String())
 	opts.SetAutoReconnect(true)
@@ -231,7 +239,6 @@ func setupMQTT(server *config.Server) error {
 	opts.SetResumeSubs(true)
 	opts.SetConnectionLostHandler(func(c mqtt.Client, e error) {
 		slog.Warn("detected broker connection lost for", "server", server.Broker)
-
 		// restart daemon for new udp hole punch if MQTT connection is lost (can happen on network change)
 		if !config.Netclient().IsStatic {
 			daemon.Restart()

functions/httpserver.go

@@ -220,7 +220,7 @@ func uninstall(c *gin.Context) {
 
 func pull(c *gin.Context) {
 	net := c.Params.ByName("net")
-	_, err := Pull(true)
+	_, _, _, err := Pull(true)
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err})
 	}