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())
 		}

functions/daemon.go

@@ -74,6 +74,11 @@ func Daemon() {
 	httpWg := sync.WaitGroup{}
 	httpWg.Add(1)
 	go HttpServer(httpctx, &httpWg)
+
+	// MQTT Fallback Goroutine
+	httpWg.Add(1)
+	go mqFallback(httpctx, &httpWg)
+
 	for {
 		select {
 		case <-quit:
@@ -151,7 +156,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())
 	}
@@ -187,6 +192,7 @@ func startGoRoutines(wg *sync.WaitGroup) context.CancelFunc {
 	go Checkin(ctx, wg)
 	wg.Add(1)
 	go networking.StartIfaceDetection(ctx, wg, config.Netclient().ListenPort)
+
 	return cancel
 }
 
@@ -249,7 +255,6 @@ func setupMQTT(server *config.Server) error {
 				nil,
 			)
 		}
-
 		// 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})
 	}

functions/mqhandlers.go

@@ -1,11 +1,13 @@
 package functions
 
 import (
+	"context"
 	"encoding/json"
 	"log"
 	"net"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -26,6 +28,9 @@ import (
 // MQTimeout - time out for mqtt connections
 const MQTimeout = 30
 
+// MQTT Fallback Routine
+var mqFallbackTicker *time.Ticker = time.NewTicker(time.Second * 30)
+
 // All -- mqtt message hander for all ('#') topics
 var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 	slog.Info("default message handler -- received message but not handling", "topic", msg.Topic())
@@ -573,3 +578,120 @@ func handleFwUpdate(server string, payload *models.FwUpdate) {
 	}
 
 }
+
+// MQTT Fallback Mechanism
+func mqFallback(ctx context.Context, wg *sync.WaitGroup) {
+	defer wg.Done()
+	for {
+		select {
+		case <-ctx.Done():
+			mqFallbackTicker.Stop()
+			slog.Info("mqfallback routine stop")
+			return
+		case <-mqFallbackTicker.C: // Execute pull every 30 seconds
+			if config.CurrServer == "" {
+				// skip tick if there is no server
+				continue
+			}
+			if Mqclient == nil || !Mqclient.IsConnectionOpen() {
+				// Call netclient http config pull
+				slog.Info("mqfallback routine execute")
+				response, resetInterface, err := Pull(false)
+				if err != nil {
+					slog.Error("pull failed", "error", err)
+				} else {
+					mqFallbackPull(response, resetInterface)
+					server := config.GetServer(config.CurrServer)
+					if server == nil {
+						continue
+					}
+					slog.Info("re-attempt mqtt connection after pull")
+					if Mqclient != nil {
+						Mqclient.Disconnect(0)
+					}
+					if err := setupMQTT(server); err != nil {
+						slog.Error("unable to connect to broker", "server", server.Broker, "error", err)
+					}
+				}
+			}
+		}
+	}
+}
+
+// MQTT Fallback Config Pull
+func mqFallbackPull(pullResponse models.HostPull, resetInterface bool) {
+	serverName := config.CurrServer
+	server := config.GetServer(serverName)
+	if server == nil {
+		slog.Error("server not found in config", "server", serverName)
+		return
+	}
+	if server.UseTurn {
+		turn.ResetCh <- struct{}{}
+	}
+	if pullResponse.ServerConfig.Version != config.Version {
+		slog.Warn("server/client version mismatch", "server", pullResponse.ServerConfig.Version, "client", config.Version)
+		vlt, err := versionLessThan(config.Version, pullResponse.ServerConfig.Version)
+		if err != nil {
+			slog.Error("error checking version less than", "error", err)
+			return
+		}
+		if vlt && config.Netclient().Host.AutoUpdate {
+			slog.Info("updating client to server's version", "version", pullResponse.ServerConfig.Version)
+			if err := UseVersion(pullResponse.ServerConfig.Version, false); err != nil {
+				slog.Error("error updating client to server's version", "error", err)
+			} else {
+				slog.Info("updated client to server's version", "version", pullResponse.ServerConfig.Version)
+				daemon.HardRestart()
+			}
+		}
+	}
+	if pullResponse.ServerConfig.Version != server.Version {
+		slog.Info("updating server version", "server", serverName, "version", pullResponse.ServerConfig.Version)
+		server.Version = pullResponse.ServerConfig.Version
+		config.WriteServerConfig()
+	}
+	gwDetected := config.GW4PeerDetected || config.GW6PeerDetected
+	currentGW4 := config.GW4Addr
+	currentGW6 := config.GW6Addr
+	isInetGW := config.UpdateHostPeers(pullResponse.Peers)
+	_ = config.WriteNetclientConfig()
+	_ = wireguard.SetPeers(false)
+	if len(pullResponse.EgressRoutes) > 0 {
+		wireguard.SetEgressRoutes(pullResponse.EgressRoutes)
+	}
+	if err := routes.SetNetmakerPeerEndpointRoutes(config.Netclient().DefaultInterface); err != nil {
+		slog.Warn("error when setting peer routes on mq fallback pull", "error", err)
+	}
+	gwDelta := (currentGW4.IP != nil && !currentGW4.IP.Equal(config.GW4Addr.IP)) ||
+		(currentGW6.IP != nil && !currentGW6.IP.Equal(config.GW6Addr.IP))
+	originalGW := currentGW4
+	if originalGW.IP != nil {
+		originalGW = currentGW6
+	}
+	handlePeerInetGateways(
+		gwDetected,
+		isInetGW,
+		gwDelta,
+		&originalGW,
+	)
+	go handleEndpointDetection(pullResponse.Peers, pullResponse.HostNetworkInfo)
+	handleFwUpdate(serverName, &pullResponse.FwUpdate)
+
+	if resetInterface {
+		nc := wireguard.GetInterface()
+		nc.Close()
+		nc = wireguard.NewNCIface(config.Netclient(), config.GetNodes())
+		nc.Create()
+		if err := nc.Configure(); err != nil {
+			slog.Error("could not configure netmaker interface", "error", err)
+			return
+		}
+		if err := wireguard.SetPeers(false); err == nil {
+			if err = routes.SetNetmakerPeerEndpointRoutes(config.Netclient().DefaultInterface); err != nil {
+				slog.Error("error when setting peer routes after host update", "error", err)
+			}
+		}
+		slog.Info("mqfallback reset interface")
+	}
+}

functions/pull.go

@@ -10,21 +10,22 @@ import (
 	"github.com/gravitl/netclient/auth"
 	"github.com/gravitl/netclient/config"
 	"github.com/gravitl/netclient/daemon"
+	"github.com/gravitl/netclient/wireguard"
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/models"
 )
 
 // Pull - pulls the latest config from the server, if manual it will overwrite
-func Pull(restart bool) (models.HostPull, error) {
-
+func Pull(restart bool) (models.HostPull, bool, error) {
+	resetInterface := false
 	serverName := config.CurrServer
 	server := config.GetServer(serverName)
 	if server == nil {
-		return models.HostPull{}, errors.New("server config not found")
+		return models.HostPull{}, resetInterface, errors.New("server config not found")
 	}
 	token, err := auth.Authenticate(server, config.Netclient())
 	if err != nil {
-		return models.HostPull{}, err
+		return models.HostPull{}, resetInterface, err
 	}
 	endpoint := httpclient.JSONEndpoint[models.HostPull, models.ErrorResponse]{
 		URL:           "https://" + server.API,
@@ -39,8 +40,28 @@ func Pull(restart bool) (models.HostPull, error) {
 		if errors.Is(err, httpclient.ErrStatus) {
 			logger.Log(0, "error pulling server", serverName, strconv.Itoa(errData.Code), errData.Message)
 		}
-		return models.HostPull{}, err
+		return models.HostPull{}, resetInterface, err
 	}
+
+	// MQTT Fallback Reset Interface
+	var hostPullNode config.Node
+	for _, pullNode := range pullResponse.Nodes {
+		hostPullNode.CommonNode = pullNode.CommonNode
+		nodeMap := config.GetNodes()
+		currNode, ok := nodeMap[pullNode.Network]
+		if !ok {
+			resetInterface = true
+			break
+		}
+		if wireguard.IfaceDelta(&hostPullNode, &currNode) {
+			resetInterface = true
+			break
+		}
+	}
+	if len(config.GetNodes()) != len(pullResponse.Nodes) {
+		resetInterface = true
+	}
+
 	_ = config.UpdateHostPeers(pullResponse.Peers)
 	pullResponse.ServerConfig.MQPassword = server.MQPassword // pwd can't change currently
 	config.UpdateServerConfig(&pullResponse.ServerConfig)
@@ -52,7 +73,7 @@ func Pull(restart bool) (models.HostPull, error) {
 	_ = config.WriteNodeConfig()
 	if restart {
 		logger.Log(3, "restarting daemon")
-		return models.HostPull{}, daemon.Restart()
+		return models.HostPull{}, resetInterface, daemon.Restart()
 	}
-	return pullResponse, nil
+	return pullResponse, resetInterface, nil
 }

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/google/nftables v0.1.0
 	github.com/google/uuid v1.4.0
 	github.com/gorilla/websocket v1.5.0
-	github.com/gravitl/netmaker v0.21.2-0.20231031062230-896041e44236
+	github.com/gravitl/netmaker v0.21.2-0.20231122125326-b17aeff3ef3d
 	github.com/gravitl/tcping v0.1.2-0.20230801110928-546055ebde06
 	github.com/gravitl/txeh v0.0.0-20230509181318-3778c58bd69f
 	github.com/guumaster/hostctl v1.1.4

go.sum

@@ -216,6 +216,12 @@ github.com/gravitl/netmaker v0.21.1-0.20231002153631-3f1211795c2c h1:kmjMZcY8HNt
 github.com/gravitl/netmaker v0.21.1-0.20231002153631-3f1211795c2c/go.mod h1:99ewh2N2nkrsqCr3l6s6t8RrPF7emACjE6LPPy4w2pM=
 github.com/gravitl/netmaker v0.21.2-0.20231031062230-896041e44236 h1:OnXwXGbeMR2f0qJja2zo/axyUouJDEyXCcg0rc2qqWk=
 github.com/gravitl/netmaker v0.21.2-0.20231031062230-896041e44236/go.mod h1:XSuEtCr4ZBT2mZlTHVqTZhCPlXWq6W+iEILwC5yuLtw=
+github.com/gravitl/netmaker v0.21.2-0.20231115080052-d9a708840602 h1:w3dfpS0mD8/0Df4oyOf0ybuT5yRhkv+om12nvsA2Iqw=
+github.com/gravitl/netmaker v0.21.2-0.20231115080052-d9a708840602/go.mod h1:KaMjQ32quN84tGTPmN9qIVZSEQFpd1AFROFgMRnlUss=
+github.com/gravitl/netmaker v0.21.2-0.20231116105900-9c7e3dfec85e h1:Q54TZkH3AOubMXnsqF7tsppcDc7QNCVcaNbqvwXO1IA=
+github.com/gravitl/netmaker v0.21.2-0.20231116105900-9c7e3dfec85e/go.mod h1:KaMjQ32quN84tGTPmN9qIVZSEQFpd1AFROFgMRnlUss=
+github.com/gravitl/netmaker v0.21.2-0.20231122125326-b17aeff3ef3d h1:jVKGRmEC3NXWPDU8v4X1icaFc4RvnSIdZTbgYAVa+Dc=
+github.com/gravitl/netmaker v0.21.2-0.20231122125326-b17aeff3ef3d/go.mod h1:KaMjQ32quN84tGTPmN9qIVZSEQFpd1AFROFgMRnlUss=
 github.com/gravitl/tcping v0.1.2-0.20230801110928-546055ebde06 h1:g2fBXRNT9eiQohyHcoME3SVmeG7OKoJPWrs7A+009kU=
 github.com/gravitl/tcping v0.1.2-0.20230801110928-546055ebde06/go.mod h1:12iViYKWAzRPj5/oEGAaD7Wje+Nuz8M9eDJbV7qhKAA=
 github.com/gravitl/txeh v0.0.0-20230509181318-3778c58bd69f h1:XzsYovKdrDvj2z2HEHoeHU67+JIEFMHQKHU6oU+1fVE=

gui/exported.go

@@ -240,7 +240,7 @@ func (app *App) GoWriteToClipboard(data string) (any, error) {
 
 // App.GoPullLatestNodeConfig pulls the latest node config from the server and returns the network config
 func (app *App) GoPullLatestNodeConfig(network string) (Network, error) {
-	_, err := functions.Pull(true)
+	_, _, err := functions.Pull(true)
 	if err != nil {
 		return Network{}, err
 	}