mqttSetup refactor

Signed-off-by: Matthew R. Kasun <[email protected]>

go.mod

@@ -34,6 +34,7 @@ require (
 require (
 	github.com/go-ping/ping v0.0.0-20211130115550-779d1e919534
 	github.com/guumaster/hostctl v1.1.2
+	github.com/kr/pretty v0.3.0
 	github.com/posthog/posthog-go v0.0.0-20211028072449-93c17c49e2b0
 )
 
@@ -53,6 +54,7 @@ require (
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/josharian/native v1.0.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/mdlayher/genetlink v1.2.0 // indirect
 	github.com/mdlayher/netlink v1.6.0 // indirect
@@ -61,6 +63,7 @@ require (
 	github.com/opencontainers/image-spec v1.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.8.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/spf13/afero v1.3.2 // indirect
 	github.com/xtgo/uuid v0.0.0-20140804021211-a0b114877d4c // indirect

models/node.go

@@ -393,6 +393,9 @@ func (newNode *Node) Fill(currentNode *Node) {
 	if newNode.IsHub == "" {
 		newNode.IsHub = currentNode.IsHub
 	}
+	if newNode.Server == "" {
+		newNode.Server = currentNode.Server
+	}
 }
 
 // StringWithCharset - returns random string inside defined charset

netclient/functions/daemon.go

@@ -20,6 +20,7 @@ import (
 	"github.com/gravitl/netmaker/netclient/daemon"
 	"github.com/gravitl/netmaker/netclient/ncutils"
 	"github.com/gravitl/netmaker/netclient/wireguard"
+	"github.com/kr/pretty"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
@@ -42,6 +43,7 @@ func Daemon() error {
 	networks, _ := ncutils.GetSystemNetworks()
 	for _, network := range networks {
 		serverSet[network] = exists
+		pretty.Println(serverSet)
 
 		//temporary code --- remove in version v0.13.0
 		removeHostDNS(network, ncutils.IsWindows())
@@ -52,9 +54,9 @@ func Daemon() error {
 		initialPull(cfg.Network)
 	}
 
-	// == subscribe to all nodes on each comms network on machine ==
+	// == subscribe to all nodes for each on machine ==
 	for server := range serverSet {
-		logger.Log(1, "started daemon for server , ", server)
+		logger.Log(1, "started daemon for server ", server)
 		ctx, cancel := context.WithCancel(context.Background())
 		networkcontext.Store(server, cancel)
 		go messageQueue(ctx, server)
@@ -66,7 +68,7 @@ func Daemon() error {
 	wg.Add(1)
 	go Checkin(ctx, &wg, serverSet)
 	quit := make(chan os.Signal, 1)
-	signal.Notify(quit, syscall.SIGTERM, os.Interrupt, os.Kill)
+	signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
 	<-quit
 	for server := range serverSet {
 		if cancel, ok := networkcontext.Load(server); ok {
@@ -106,7 +108,9 @@ func UpdateKeys(nodeCfg *config.ClientConfig, client mqtt.Client) error {
 // PingServer -- checks if server is reachable
 // use commsCfg only*
 func PingServer(cfg *config.ClientConfig) error {
-	pinger, err := ping.NewPinger(cfg.Node.Server)
+	fmt.Println("ping server ", cfg.Server.Server)
+	pretty.Println(cfg)
+	pinger, err := ping.NewPinger(cfg.Server.Server)
 	if err != nil {
 		return err
 	}
@@ -116,6 +120,7 @@ func PingServer(cfg *config.ClientConfig) error {
 	if stats.PacketLoss == 100 {
 		return errors.New("ping error")
 	}
+	logger.Log(3, "ping of server", cfg.Server.Server, "was successful")
 	return nil
 }
 
@@ -136,16 +141,12 @@ func setSubscriptions(client mqtt.Client, nodeCfg *config.ClientConfig) {
 		logger.Log(0, token.Error().Error())
 		return
 	}
-	if nodeCfg.DebugOn {
-		logger.Log(0, fmt.Sprintf("subscribed to node updates for node %s update/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
-	}
+	logger.Log(3, fmt.Sprintf("subscribed to node updates for node %s update/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
 	if token := client.Subscribe(fmt.Sprintf("peers/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(UpdatePeers)); token.Wait() && token.Error() != nil {
 		logger.Log(0, token.Error().Error())
 		return
 	}
-	if nodeCfg.DebugOn {
-		logger.Log(0, fmt.Sprintf("subscribed to peer updates for node %s peers/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
-	}
+	logger.Log(3, fmt.Sprintf("subscribed to peer updates for node %s peers/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
 }
 
 // on a delete usually, pass in the nodecfg to unsubscribe client broker communications
@@ -167,16 +168,102 @@ func unsubscribeNode(client mqtt.Client, nodeCfg *config.ClientConfig) {
 }
 
 // sets up Message Queue and subsribes/publishes updates to/from server
-// the client should subscribe to ALL nodes that exist on unique comms network locally
-func messageQueue(ctx context.Context, commsNet string) {
-	var commsCfg config.ClientConfig
-	commsCfg.Network = commsNet
-	commsCfg.ReadConfig()
-	logger.Log(0, "netclient daemon started for network: ", commsNet)
-	client := setupMQTT(&commsCfg, false)
+// the client should subscribe to ALL nodes that exist on server locally
+func messageQueue(ctx context.Context, server string) {
+	logger.Log(0, "netclient daemon started for server: ", server)
+	client := setupMQTTSub(server)
 	defer client.Disconnect(250)
 	<-ctx.Done()
-	logger.Log(0, "shutting down daemon for comms network ", commsNet)
+	logger.Log(0, "shutting down daemon for server ", server)
+}
+
+// setupMQTTSub creates a connection to broker and subscribes to topic
+// utilizes comms client configs to setup connections
+func setupMQTTSub(server string) mqtt.Client {
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(server + ":1883")             // TODO get the appropriate port of the comms mq server
+	opts.ClientID = ncutils.MakeRandomString(23) // helps avoid id duplication on broker
+	opts.SetDefaultPublishHandler(All)
+	opts.SetAutoReconnect(true)
+	opts.SetConnectRetry(true)
+	opts.SetConnectRetryInterval(time.Second << 2)
+	opts.SetKeepAlive(time.Minute >> 1)
+	opts.SetWriteTimeout(time.Minute)
+	opts.SetOnConnectHandler(func(client mqtt.Client) {
+		networks, err := ncutils.GetSystemNetworks()
+		if err != nil {
+			logger.Log(0, "error retriving networks ", err.Error())
+		}
+		for _, network := range networks {
+			var currNodeCfg config.ClientConfig
+			currNodeCfg.Network = network
+			currNodeCfg.ReadConfig()
+			if currNodeCfg.Server.Server == server {
+				setSubscriptions(client, &currNodeCfg)
+			}
+		}
+	})
+	opts.SetOrderMatters(true)
+	opts.SetResumeSubs(true)
+	opts.SetConnectionLostHandler(func(c mqtt.Client, e error) {
+		logger.Log(0, "detected broker connection lost, running pull for all nodes")
+		networks, err := ncutils.GetSystemNetworks()
+		if err != nil {
+			logger.Log(0, "error retriving networks ", err.Error())
+		}
+		for _, network := range networks {
+			var cfg config.ClientConfig
+			cfg.Network = network
+			cfg.ReadConfig()
+			_, err := Pull(cfg.Node.Network, true)
+			if err != nil {
+				logger.Log(0, "could not run pull, server unreachable: ", err.Error())
+				logger.Log(0, "waiting to retry...")
+			}
+		}
+		// don't think following log message is accurate
+		//logger.Log(0, "connection re-established with mqtt server")
+	})
+
+	client := mqtt.NewClient(opts)
+	tperiod := time.Now().Add(12 * time.Second)
+	for {
+		//if after 12 seconds, try a gRPC pull on the last try
+		if time.Now().After(tperiod) {
+			networks, err := ncutils.GetSystemNetworks()
+			if err != nil {
+				logger.Log(0, "error retriving networks ", err.Error())
+			}
+			for _, network := range networks {
+				var cfg config.ClientConfig
+				cfg.Network = network
+				cfg.ReadConfig()
+				if cfg.Server.Server == server {
+					_, err := Pull(cfg.Node.Network, true)
+					if err != nil {
+						logger.Log(0, "could not run pull, exiting ", cfg.Node.Network, " setup: ", err.Error())
+						return client
+					}
+				}
+			}
+			time.Sleep(time.Second)
+		}
+		if token := client.Connect(); token.Wait() && token.Error() != nil {
+			logger.Log(0, "unable to connect to broker, retrying ...")
+			if time.Now().After(tperiod) {
+				logger.Log(0, "could not connect to broker, exiting ", server, " setup: ", token.Error().Error())
+				if strings.Contains(token.Error().Error(), "connectex") || strings.Contains(token.Error().Error(), "i/o timeout") {
+					logger.Log(0, "connection issue detected.. restarting daemon")
+					daemon.Restart()
+				}
+				return client
+			}
+		} else {
+			break
+		}
+		time.Sleep(2 * time.Second)
+	}
+	return client
 }
 
 // setupMQTT creates a connection to broker and return client

netclient/functions/mqpublish.go

@@ -106,6 +106,7 @@ func Hello(nodeCfg *config.ClientConfig) {
 			logger.Log(0, "could not run pull on "+nodeCfg.Node.Network+", error: "+err.Error())
 		}
 	}
+	logger.Log(3, "server checkin complete")
 }
 
 // requires the commscfg in which to send traffic over and nodecfg of node that is publish the message