Merge pull request #1302 from gravitl/bugfix_v0.14.5_extra_peer_update

reduce number of peer updates sent

controllers/dns.go

@@ -127,7 +127,7 @@ func createDNS(w http.ResponseWriter, r *http.Request) {
 			if err = logic.ServerUpdate(&serverNode, false); err != nil {
 				logger.Log(1, "failed to update server node after DNS update on", entry.Network)
 			}
-			if err = mq.PublishPeerUpdate(&serverNode); err != nil {
+			if err = mq.PublishPeerUpdate(&serverNode, false); err != nil {
 				logger.Log(0, "failed to publish peer update after ACL update on", entry.Network)
 			}
 		}

controllers/network.go

@@ -255,7 +255,7 @@ func updateNetworkACL(w http.ResponseWriter, r *http.Request) {
 			if err = logic.ServerUpdate(&serverNode, false); err != nil {
 				logger.Log(1, "failed to update server node after ACL update on", netname)
 			}
-			if err = mq.PublishPeerUpdate(&serverNode); err != nil {
+			if err = mq.PublishPeerUpdate(&serverNode, false); err != nil {
 				logger.Log(0, "failed to publish peer update after ACL update on", netname)
 			}
 		}

controllers/node.go

@@ -501,7 +501,7 @@ func createNode(w http.ResponseWriter, r *http.Request) {
 	logger.Log(1, r.Header.Get("user"), "created new node", node.Name, "on network", node.Network)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(response)
-	runForceServerUpdate(&node)
+	runForceServerUpdate(&node, true)
 }
 
 // Takes node out of pending state
@@ -722,7 +722,7 @@ func deleteNode(w http.ResponseWriter, r *http.Request) {
 
 	logger.Log(1, r.Header.Get("user"), "Deleted node", nodeid, "from network", params["network"])
 	runUpdates(&node, false)
-	runForceServerUpdate(&node)
+	runForceServerUpdate(&node, false)
 }
 
 func runUpdates(node *models.Node, ifaceDelta bool) {
@@ -751,7 +751,7 @@ func runServerUpdate(node *models.Node, ifaceDelta bool) error {
 	}
 
 	if ifaceDelta && logic.IsLeader(&currentServerNode) {
-		if err := mq.PublishPeerUpdate(&currentServerNode); err != nil {
+		if err := mq.PublishPeerUpdate(&currentServerNode, false); err != nil {
 			logger.Log(1, "failed to publish peer update "+err.Error())
 		}
 	}
@@ -763,9 +763,9 @@ func runServerUpdate(node *models.Node, ifaceDelta bool) error {
 	return nil
 }
 
-func runForceServerUpdate(node *models.Node) {
+func runForceServerUpdate(node *models.Node, publishPeerUpdateToNode bool) {
 	go func() {
-		if err := mq.PublishPeerUpdate(node); err != nil {
+		if err := mq.PublishPeerUpdate(node, publishPeerUpdateToNode); err != nil {
 			logger.Log(1, "failed a peer update after creation of node", node.Name)
 		}
 

mq/handlers.go

@@ -133,7 +133,7 @@ func updateNodePeers(currentNode *models.Node) {
 		return
 	}
 	if logic.IsLeader(&currentServerNode) {
-		if err := PublishPeerUpdate(currentNode); err != nil {
+		if err := PublishPeerUpdate(currentNode, false); err != nil {
 			logger.Log(1, "error publishing peer update ", err.Error())
 			return
 		}

mq/publishers.go

@@ -12,7 +12,7 @@ import (
 )
 
 // PublishPeerUpdate --- deterines and publishes a peer update to all the peers of a node
-func PublishPeerUpdate(newNode *models.Node) error {
+func PublishPeerUpdate(newNode *models.Node, publishToSelf bool) error {
 	if !servercfg.IsMessageQueueBackend() {
 		return nil
 	}
@@ -26,6 +26,10 @@ func PublishPeerUpdate(newNode *models.Node) error {
 		if node.IsServer == "yes" {
 			continue
 		}
+		if !publishToSelf && newNode.ID == node.ID {
+			//skip self
+			continue
+		}
 		peerUpdate, err := logic.GetPeerUpdate(&node)
 		if err != nil {
 			logger.Log(1, "error getting peer update for node", node.ID, err.Error())
@@ -70,7 +74,7 @@ func PublishExtPeerUpdate(node *models.Node) error {
 	if err = publish(node, fmt.Sprintf("peers/%s/%s", node.Network, node.ID), data); err != nil {
 		return err
 	}
-	go PublishPeerUpdate(node)
+	go PublishPeerUpdate(node, false)
 	return nil
 }
 
@@ -132,7 +136,7 @@ func sendPeers() {
 					if force {
 						logger.Log(2, "sending scheduled peer update (5 min)")
 					}
-					err = PublishPeerUpdate(&serverNode)
+					err = PublishPeerUpdate(&serverNode, false)
 					if err != nil {
 						logger.Log(1, "error publishing udp port updates for network", network.NetID)
 						logger.Log(1, errN.Error())