NET-655

functions/daemon.go

@@ -187,6 +187,11 @@ func startGoRoutines(wg *sync.WaitGroup) context.CancelFunc {
 	go Checkin(ctx, wg)
 	wg.Add(1)
 	go networking.StartIfaceDetection(ctx, wg, config.Netclient().ListenPort)
+
+	// MQTT Fallback Goroutine
+	wg.Add(1)
+	go MQFallback(ctx, wg)
+
 	return cancel
 }
 
@@ -249,7 +254,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/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,86 @@ 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()
+			return
+		case <-mqFallbackTicker.C: // Execute pull every 30 seconds
+			if Mqclient == nil || !Mqclient.IsConnectionOpen() {
+				// Call netclient http config pull
+				response, err := Pull(false)
+				if err != nil {
+					slog.Error("pull failed", "error", err)
+				} else {
+					MQFallbackPull(response)
+				}
+			}
+		}
+	}
+}
+
+// MQTT Fallback Config Pull
+func MQFallbackPull(pullResponse models.HostPull) {
+	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)
+}

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.20231116105900-9c7e3dfec85e
 	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,10 @@ 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/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=