Merge pull request #1029 from gravitl/feature_v0.13.0_btr_serv_reset

added initial notification to clients on server startup

main.go

@@ -119,6 +119,12 @@ func initialize() { // Client Mode Prereq Check
 			logger.FatalLog(err.Error())
 		}
 	}
+
+	if servercfg.IsMessageQueueBackend() {
+		if err = mq.ServerStartNotify(); err != nil {
+			logger.Log(0, "error occurred when notifying nodes of startup", err.Error())
+		}
+	}
 	genCerts()
 }
 

models/node.go

@@ -17,7 +17,7 @@ const (
 	TEN_YEARS_IN_SECONDS = 300000000
 	// MAX_NAME_LENGTH - max name length of node
 	MAX_NAME_LENGTH = 62
-	// == ACTIONS == (can only be set by GRPC)
+	// == ACTIONS == (can only be set by server)
 	// NODE_UPDATE_KEY - action to update key
 	NODE_UPDATE_KEY = "updatekey"
 	// NODE_DELETE - delete node action
@@ -26,6 +26,8 @@ const (
 	NODE_IS_PENDING = "pending"
 	// NODE_NOOP - node no op action
 	NODE_NOOP = "noop"
+	// NODE_FORCE_UPDATE - indicates a node should pull all changes
+	NODE_FORCE_UPDATE = "force"
 )
 
 var seededRand *rand.Rand = rand.New(

mq/publishers.go

@@ -141,3 +141,18 @@ func sendPeers() {
 		}
 	}
 }
+
+// ServerStartNotify - notifies all non server nodes to pull changes after a restart
+func ServerStartNotify() error {
+	nodes, err := logic.GetAllNodes()
+	if err != nil {
+		return err
+	}
+	for i := range nodes {
+		nodes[i].Action = models.NODE_FORCE_UPDATE
+		if err = NodeUpdate(&nodes[i]); err != nil {
+			logger.Log(1, "error when notifying node", nodes[i].Name, " - ", nodes[i].ID, "of a server startup")
+		}
+	}
+	return nil
+}

netclient/functions/mqhandlers.go

@@ -87,6 +87,8 @@ func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
 			}
 		}
 		ifaceDelta = true
+	case models.NODE_FORCE_UPDATE:
+		ifaceDelta = true
 	case models.NODE_NOOP:
 	default:
 	}