2022-01-13 05:23:34 +08:00
|
|
|
package mq
|
|
|
|
|
|
|
|
import (
|
2022-01-16 07:16:44 +08:00
|
|
|
"encoding/json"
|
2022-01-13 05:23:34 +08:00
|
|
|
"errors"
|
|
|
|
"strings"
|
|
|
|
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
|
"github.com/gravitl/netmaker/database"
|
|
|
|
"github.com/gravitl/netmaker/logger"
|
|
|
|
"github.com/gravitl/netmaker/logic"
|
|
|
|
"github.com/gravitl/netmaker/models"
|
2022-01-19 06:32:12 +08:00
|
|
|
"github.com/gravitl/netmaker/servercfg"
|
2022-01-13 05:23:34 +08:00
|
|
|
)
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// default message handler - only called in GetDebug == true
|
2022-01-13 05:23:34 +08:00
|
|
|
var DefaultHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
|
|
|
logger.Log(0, "MQTT Message: Topic: "+string(msg.Topic())+" Message: "+string(msg.Payload()))
|
|
|
|
}
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// Ping message Handler -- handles ping topic from client nodes
|
2022-01-13 05:23:34 +08:00
|
|
|
var Ping mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
2022-01-15 03:30:26 +08:00
|
|
|
logger.Log(0, "Ping Handler: "+msg.Topic())
|
2022-01-13 05:23:34 +08:00
|
|
|
go func() {
|
2022-01-15 04:35:15 +08:00
|
|
|
id, err := GetID(msg.Topic())
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
|
|
|
logger.Log(0, "error getting node.ID sent on ping topic ")
|
|
|
|
return
|
|
|
|
}
|
2022-01-15 04:35:15 +08:00
|
|
|
node, err := logic.GetNodeByID(id)
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
|
|
|
logger.Log(0, "mq-ping error getting node: "+err.Error())
|
2022-01-15 04:35:15 +08:00
|
|
|
record, err := database.FetchRecord(database.NODES_TABLE_NAME, id)
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
|
|
|
logger.Log(0, "error reading database ", err.Error())
|
|
|
|
return
|
|
|
|
}
|
|
|
|
logger.Log(0, "record from database")
|
|
|
|
logger.Log(0, record)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
node.SetLastCheckIn()
|
2022-01-19 22:44:35 +08:00
|
|
|
if err := logic.UpdateNode(&node, &node); err != nil {
|
|
|
|
logger.Log(0, "error updating node "+err.Error())
|
2022-01-19 06:32:12 +08:00
|
|
|
}
|
2022-01-15 03:30:26 +08:00
|
|
|
logger.Log(0, "ping processed")
|
2022-01-13 05:23:34 +08:00
|
|
|
// --TODO --set client version once feature is implemented.
|
|
|
|
//node.SetClientVersion(msg.Payload())
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// UpdateNode message Handler -- handles updates from client nodes
|
|
|
|
var UpdateNode mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
2022-01-13 05:23:34 +08:00
|
|
|
go func() {
|
2022-01-15 04:35:15 +08:00
|
|
|
id, err := GetID(msg.Topic())
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(1, "error getting node.ID sent on "+msg.Topic()+" "+err.Error())
|
|
|
|
return
|
2022-01-15 03:30:26 +08:00
|
|
|
}
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(1, "Update Node Handler"+id)
|
|
|
|
var newNode models.Node
|
|
|
|
if err := json.Unmarshal(msg.Payload(), &newNode); err != nil {
|
|
|
|
logger.Log(1, "error unmarshaling payload "+err.Error())
|
2022-01-13 05:23:34 +08:00
|
|
|
return
|
|
|
|
}
|
2022-01-21 18:19:17 +08:00
|
|
|
currentNode, err := logic.GetNodeByID(newNode.ID)
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(1, "error getting node "+newNode.ID+" "+err.Error())
|
2022-01-13 05:23:34 +08:00
|
|
|
return
|
|
|
|
}
|
2022-01-21 18:19:17 +08:00
|
|
|
if err := logic.UpdateNode(¤tNode, &newNode); err != nil {
|
|
|
|
logger.Log(1, "error saving node"+err.Error())
|
2022-01-19 06:32:12 +08:00
|
|
|
}
|
2022-01-21 18:19:17 +08:00
|
|
|
if logic.ShouldPeersUpdate(¤tNode, &newNode) {
|
|
|
|
if err := PublishPeerUpdate(client, &newNode); err != nil {
|
|
|
|
logger.Log(1, "error publishing peer update "+err.Error())
|
|
|
|
return
|
|
|
|
}
|
2022-01-15 03:30:26 +08:00
|
|
|
}
|
2022-01-13 05:23:34 +08:00
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// PublishPeerUpdate --- deterines and publishes a peer update to all the peers of a node
|
|
|
|
func PublishPeerUpdate(client mqtt.Client, newNode *models.Node) error {
|
|
|
|
networkNodes, err := logic.GetNetworkNodes(newNode.Network)
|
2022-01-13 05:23:34 +08:00
|
|
|
if err != nil {
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(1, "err getting Network Nodes"+err.Error())
|
2022-01-16 07:16:44 +08:00
|
|
|
return err
|
2022-01-15 03:30:26 +08:00
|
|
|
}
|
2022-01-19 22:44:35 +08:00
|
|
|
for _, node := range networkNodes {
|
2022-01-21 18:19:17 +08:00
|
|
|
peerUpdate, err := logic.GetPeerUpdate(&node)
|
|
|
|
if err != nil {
|
|
|
|
logger.Log(1, "error getting peer update for node "+node.ID+" "+err.Error())
|
|
|
|
continue
|
2022-01-19 22:44:35 +08:00
|
|
|
}
|
2022-01-19 06:32:12 +08:00
|
|
|
data, err := json.Marshal(&peerUpdate)
|
2022-01-16 07:16:44 +08:00
|
|
|
if err != nil {
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(2, "error marshaling peer update "+err.Error())
|
2022-01-16 07:16:44 +08:00
|
|
|
return err
|
|
|
|
}
|
2022-01-21 20:07:38 +08:00
|
|
|
if token := client.Publish("/update/peers/"+node.ID, 0, false, data); token.Wait() && token.Error() != nil {
|
2022-01-21 18:19:17 +08:00
|
|
|
logger.Log(2, "error publishing peer update to peer "+node.ID+" "+token.Error().Error())
|
2022-01-16 07:16:44 +08:00
|
|
|
return err
|
2022-01-13 05:23:34 +08:00
|
|
|
}
|
|
|
|
}
|
2022-01-15 03:30:26 +08:00
|
|
|
return nil
|
2022-01-13 05:23:34 +08:00
|
|
|
}
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// GetID -- decodes a message queue topic and returns the embedded node.ID
|
2022-01-13 05:23:34 +08:00
|
|
|
func GetID(topic string) (string, error) {
|
|
|
|
parts := strings.Split(topic, "/")
|
|
|
|
count := len(parts)
|
|
|
|
if count == 1 {
|
|
|
|
return "", errors.New("invalid topic")
|
|
|
|
}
|
2022-01-15 04:35:15 +08:00
|
|
|
//the last part of the topic will be the node.ID
|
2022-01-15 06:22:11 +08:00
|
|
|
return parts[count-1], nil
|
2022-01-13 05:23:34 +08:00
|
|
|
}
|
2022-01-19 06:32:12 +08:00
|
|
|
|
2022-01-22 06:22:56 +08:00
|
|
|
// UpdateNode -- publishes a node update
|
|
|
|
func NodeUpdate(node *models.Node) error {
|
2022-01-22 22:04:11 +08:00
|
|
|
logger.Log(3, "publishing node update to "+node.Name)
|
2022-01-22 06:22:56 +08:00
|
|
|
opts := mqtt.NewClientOptions()
|
|
|
|
broker := servercfg.GetMessageQueueEndpoint()
|
|
|
|
logger.Log(0, "broker: "+broker)
|
|
|
|
opts.AddBroker(broker)
|
|
|
|
client := mqtt.NewClient(opts)
|
|
|
|
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
|
|
|
return token.Error()
|
|
|
|
}
|
|
|
|
data, err := json.Marshal(node)
|
|
|
|
if err != nil {
|
|
|
|
logger.Log(2, "error marshalling node update "+err.Error())
|
|
|
|
return err
|
|
|
|
}
|
2022-01-22 22:22:19 +08:00
|
|
|
if token := client.Publish("update/"+node.ID, 0, false, data); token.Wait() && token.Error() != nil {
|
2022-01-22 06:22:56 +08:00
|
|
|
logger.Log(2, "error publishing peer update to peer "+node.ID+" "+token.Error().Error())
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-01-21 18:19:17 +08:00
|
|
|
// NewPeer -- publishes a peer update to all the peers of a newNode
|
2022-01-19 06:32:12 +08:00
|
|
|
func NewPeer(node models.Node) error {
|
|
|
|
opts := mqtt.NewClientOptions()
|
|
|
|
broker := servercfg.GetMessageQueueEndpoint()
|
|
|
|
logger.Log(0, "broker: "+broker)
|
|
|
|
opts.AddBroker(broker)
|
|
|
|
client := mqtt.NewClient(opts)
|
|
|
|
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
|
|
|
return token.Error()
|
|
|
|
}
|
2022-01-21 18:19:17 +08:00
|
|
|
if err := PublishPeerUpdate(client, &node); err != nil {
|
2022-01-19 06:32:12 +08:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|