Merge pull request #1930 from gravitl/story/GRA-944

Story/gra 944

controllers/hosts.go

@@ -109,11 +109,9 @@ func updateHost(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	go func() {
-		if err := mq.PublishPeerUpdate(); err != nil {
-			logger.Log(0, "fail to publish peer update: ", err.Error())
-		}
-	}()
+	if err = mq.HostUpdate(newHost); err != nil {
+		logger.Log(0, "failed to send host update to host", newHost.ID.String(), newHost.Name, err.Error())
+	}
 
 	apiHostData := newHost.ConvertNMHostToAPI()
 	logger.Log(2, r.Header.Get("user"), "updated host", newHost.ID.String())

mq/handlers.go

@@ -128,7 +128,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
 		}
 		currentHost, err := logic.GetHost(id)
 		if err != nil {
-			logger.Log(1, "error getting node ", id, err.Error())
+			logger.Log(1, "error getting host ", id, err.Error())
 			return
 		}
 		decrypted, decryptErr := decryptMsgWithHost(currentHost, msg.Payload())
@@ -136,24 +136,11 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
 			logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error())
 			return
 		}
-		var newHost models.Host
-		if err := json.Unmarshal(decrypted, &newHost); err != nil {
-			logger.Log(1, "error unmarshaling payload ", err.Error())
-			return
-		}
-		// ifaceDelta := logic.IfaceDelta(&currentHost, newNode)
-		// if servercfg.Is_EE && ifaceDelta {
-		// 	if err = logic.EnterpriseResetAllPeersFailovers(currentHost.ID.String(), currentHost.Network); err != nil {
-		// 		logger.Log(1, "failed to reset failover list during node update", currentHost.ID.String(), currentHost.Network)
-		// 	}
-		// }
-		logic.UpdateHost(&newHost, currentHost)
-		if err := logic.UpsertHost(&newHost); err != nil {
-			logger.Log(1, "error saving host", err.Error())
-			return
-		}
 
-		logger.Log(1, "updated host", newHost.ID.String())
+		if decrypted[0] == byte(ncutils.ACK) {
+			logger.Log(1, "sending peer updates after completed host update", currentHost.ID.String(), currentHost.Name)
+			sendPeers(true)
+		}
 	}()
 }
 

mq/mq.go

@@ -125,7 +125,7 @@ func Keepalive(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-time.After(time.Second * KEEPALIVE_TIMEOUT):
-			sendPeers()
+			sendPeers(false)
 		}
 	}
 }

mq/publishers.go

@@ -113,35 +113,36 @@ func HostUpdate(host *models.Host) error {
 }
 
 // sendPeers - retrieve networks, send peer ports to all peers
-func sendPeers() {
+func sendPeers(skipServerCheck bool) {
 
 	hosts, err := logic.GetAllHosts()
 	if err != nil {
 		logger.Log(1, "error retrieving networks for keepalive", err.Error())
 	}
-
 	var force bool
-	peer_force_send++
-	if peer_force_send == 5 {
+	if !skipServerCheck {
+		peer_force_send++
+		if peer_force_send == 5 {
 
-		// run iptables update to ensure gateways work correctly and mq is forwarded if containerized
-		if servercfg.ManageIPTables() != "off" {
-			serverctl.InitIPTables(false)
-		}
-		servercfg.SetHost()
+			// run iptables update to ensure gateways work correctly and mq is forwarded if containerized
+			if servercfg.ManageIPTables() != "off" {
+				serverctl.InitIPTables(false)
+			}
+			servercfg.SetHost()
 
-		force = true
-		peer_force_send = 0
-		err := logic.TimerCheckpoint() // run telemetry & log dumps if 24 hours has passed..
-		if err != nil {
-			logger.Log(3, "error occurred on timer,", err.Error())
-		}
+			force = true
+			peer_force_send = 0
+			err := logic.TimerCheckpoint() // run telemetry & log dumps if 24 hours has passed..
+			if err != nil {
+				logger.Log(3, "error occurred on timer,", err.Error())
+			}
 
-		//collectServerMetrics(networks[:])
+			//collectServerMetrics(networks[:])
+		}
 	}
 
 	for _, host := range hosts {
-		if force {
+		if force || skipServerCheck {
 			logger.Log(2, "sending scheduled peer update (5 min)")
 			err = PublishSingleHostUpdate(&host)
 			if err != nil {