gravitl/netmaker
1
1
Fork 0
mirror of https://github.com/gravitl/netmaker.git synced 2024-09-20 15:26:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

structured logging mq handlers

Browse source
This commit is contained in:
Matthew R Kasun 2023-06-05 08:26:46 -04:00
parent 9637dd5b25
commit 56f360dfe4
1 changed files with 49 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

100
mq/handlers.go
View file

@ -10,86 +10,86 @@ import (
mqtt "github.com/eclipse/paho.mqtt.golang" mqtt "github.com/eclipse/paho.mqtt.golang"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/gravitl/netmaker/database" "github.com/gravitl/netmaker/database"
"github.com/gravitl/netmaker/logger"
"github.com/gravitl/netmaker/logic" "github.com/gravitl/netmaker/logic"
"github.com/gravitl/netmaker/logic/hostactions" "github.com/gravitl/netmaker/logic/hostactions"
"github.com/gravitl/netmaker/models" "github.com/gravitl/netmaker/models"
"github.com/gravitl/netmaker/netclient/ncutils" "github.com/gravitl/netmaker/netclient/ncutils"
"github.com/gravitl/netmaker/servercfg" "github.com/gravitl/netmaker/servercfg"
"golang.org/x/exp/slog"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes" "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
) )
// DefaultHandler default message queue handler -- NOT USED // DefaultHandler default message queue handler -- NOT USED
func DefaultHandler(client mqtt.Client, msg mqtt.Message) { func DefaultHandler(client mqtt.Client, msg mqtt.Message) {
logger.Log(0, "MQTT Message: Topic: ", string(msg.Topic()), " Message: ", string(msg.Payload())) slog.Info("mqtt default handler", "topic", msg.Topic(), "message", msg.Payload())
} }
// UpdateNode message Handler -- handles updates from client nodes // UpdateNode message Handler -- handles updates from client nodes
func UpdateNode(client mqtt.Client, msg mqtt.Message) { func UpdateNode(client mqtt.Client, msg mqtt.Message) {
id, err := getID(msg.Topic()) id, err := getID(msg.Topic())
if err != nil { if err != nil {
logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error()) slog.Error("error getting node.ID ", "topic", msg.Topic(), "error", err)
return return
} }
currentNode, err := logic.GetNodeByID(id) currentNode, err := logic.GetNodeByID(id)
if err != nil { if err != nil {
logger.Log(1, "error getting node ", id, err.Error()) slog.Error("error getting node", "id", id, "error", err)
return return
} }
decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload()) decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
if decryptErr != nil { if decryptErr != nil {
logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error()) slog.Error("failed to decrypt message for node", "id", id, "error", decryptErr)
return return
} }
var newNode models.Node var newNode models.Node
if err := json.Unmarshal(decrypted, &newNode); err != nil { if err := json.Unmarshal(decrypted, &newNode); err != nil {
logger.Log(1, "error unmarshaling payload ", err.Error()) slog.Error("error unmarshaling payload", "error", err)
return return
} }
ifaceDelta := logic.IfaceDelta(&currentNode, &newNode) ifaceDelta := logic.IfaceDelta(&currentNode, &newNode)
if servercfg.Is_EE && ifaceDelta { if servercfg.Is_EE && ifaceDelta {
if err = logic.EnterpriseResetAllPeersFailovers(currentNode.ID, currentNode.Network); err != nil { if err = logic.EnterpriseResetAllPeersFailovers(currentNode.ID, currentNode.Network); err != nil {
logger.Log(1, "failed to reset failover list during node update", currentNode.ID.String(), currentNode.Network) slog.Warn("failed to reset failover list during node update", "nodeid", currentNode.ID, "network", currentNode.Network)
} }
} }
newNode.SetLastCheckIn() newNode.SetLastCheckIn()
if err := logic.UpdateNode(&currentNode, &newNode); err != nil { if err := logic.UpdateNode(&currentNode, &newNode); err != nil {
logger.Log(1, "error saving node", err.Error()) slog.Error("error saving node", "id", id, "error", err)
return return
} }
if ifaceDelta { // reduce number of unneeded updates, by only sending on iface changes if ifaceDelta { // reduce number of unneeded updates, by only sending on iface changes
if err = PublishPeerUpdate(); err != nil { if err = PublishPeerUpdate(); err != nil {
logger.Log(0, "error updating peers when node", currentNode.ID.String(), "informed the server of an interface change", err.Error()) slog.Warn("error updating peers when node informed the server of an interface change", "nodeid", currentNode.ID, "error", err)
} }
} }
logger.Log(1, "updated node", id, newNode.ID.String()) slog.Info("updated node", "id", id, "newnodeid", newNode.ID)
} }
// UpdateHost message Handler -- handles host updates from clients // UpdateHost message Handler -- handles host updates from clients
func UpdateHost(client mqtt.Client, msg mqtt.Message) { func UpdateHost(client mqtt.Client, msg mqtt.Message) {
id, err := getID(msg.Topic()) id, err := getID(msg.Topic())
if err != nil { if err != nil {
logger.Log(1, "error getting host.ID sent on ", msg.Topic(), err.Error()) slog.Error("error getting host.ID sent on ", "topic", msg.Topic(), "error", err)
return return
} }
currentHost, err := logic.GetHost(id) currentHost, err := logic.GetHost(id)
if err != nil { if err != nil {
logger.Log(1, "error getting host ", id, err.Error()) slog.Error("error getting host", "id", id, "error", err)
return return
} }
decrypted, decryptErr := decryptMsgWithHost(currentHost, msg.Payload()) decrypted, decryptErr := decryptMsgWithHost(currentHost, msg.Payload())
if decryptErr != nil { if decryptErr != nil {
logger.Log(1, "failed to decrypt message for host ", id, decryptErr.Error()) slog.Error("failed to decrypt message for host", "id", id, "error", decryptErr)
return return
} }
var hostUpdate models.HostUpdate var hostUpdate models.HostUpdate
if err := json.Unmarshal(decrypted, &hostUpdate); err != nil { if err := json.Unmarshal(decrypted, &hostUpdate); err != nil {
logger.Log(1, "error unmarshaling payload ", err.Error()) slog.Error("error unmarshaling payload", "error", err)
return return
} }
logger.Log(3, fmt.Sprintf("recieved host update: %s\n", hostUpdate.Host.ID.String())) slog.Info("recieved host update", "name", hostUpdate.Host.Name, "id", hostUpdate.Host.ID)
var sendPeerUpdate bool var sendPeerUpdate bool
switch hostUpdate.Action { switch hostUpdate.Action {
case models.CheckIn: case models.CheckIn:
@ -98,21 +98,21 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
hu := hostactions.GetAction(currentHost.ID.String()) hu := hostactions.GetAction(currentHost.ID.String())
if hu != nil { if hu != nil {
if err = HostUpdate(hu); err != nil { if err = HostUpdate(hu); err != nil {
logger.Log(0, "failed to send new node to host", hostUpdate.Host.Name, currentHost.ID.String(), err.Error()) slog.Error("failed to send new node to host", "name", hostUpdate.Host.Name, "id", currentHost.ID, "error", err)
return return
} else { } else {
if servercfg.GetBrokerType() == servercfg.EmqxBrokerType { if servercfg.GetBrokerType() == servercfg.EmqxBrokerType {
if err = AppendNodeUpdateACL(hu.Host.ID.String(), hu.Node.Network, hu.Node.ID.String(), servercfg.GetServer()); err != nil { if err = AppendNodeUpdateACL(hu.Host.ID.String(), hu.Node.Network, hu.Node.ID.String(), servercfg.GetServer()); err != nil {
logger.Log(0, "failed to add ACLs for EMQX node", err.Error()) slog.Error("failed to add ACLs for EMQX node", "error", err)
return return
} }
} }
if err = PublishSingleHostPeerUpdate(context.Background(), currentHost, nil, nil); err != nil { if err = PublishSingleHostPeerUpdate(context.Background(), currentHost, nil, nil); err != nil {
logger.Log(0, "failed peers publish after join acknowledged", hostUpdate.Host.Name, currentHost.ID.String(), err.Error()) slog.Error("failed peers publish after join acknowledged", "name", hostUpdate.Host.Name, "id", currentHost.ID, "error", err)
return return
} }
if err = handleNewNodeDNS(&hu.Host, &hu.Node); err != nil { if err = handleNewNodeDNS(&hu.Host, &hu.Node); err != nil {
logger.Log(0, "failed to send dns update after node,", hu.Node.ID.String(), ", added to host", hu.Host.Name, err.Error()) slog.Error("failed to send dns update after node added to host", "name", hostUpdate.Host.Name, "id", currentHost.ID, "error", err)
return return
} }
} }
@ -131,7 +131,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
} }
data, err := json.Marshal(&peerUpdate) data, err := json.Marshal(&peerUpdate)
if err != nil { if err != nil {
logger.Log(2, "json error", err.Error()) slog.Error("failed to marshal peer update", "error", err)
} }
hosts := logic.GetRelatedHosts(hostUpdate.Host.ID.String()) hosts := logic.GetRelatedHosts(hostUpdate.Host.ID.String())
server := servercfg.GetServer() server := servercfg.GetServer()
@ -143,23 +143,23 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
sendPeerUpdate = logic.UpdateHostFromClient(&hostUpdate.Host, currentHost) sendPeerUpdate = logic.UpdateHostFromClient(&hostUpdate.Host, currentHost)
err := logic.UpsertHost(currentHost) err := logic.UpsertHost(currentHost)
if err != nil { if err != nil {
logger.Log(0, "failed to update host: ", currentHost.ID.String(), err.Error()) slog.Error("failed to update host", "id", currentHost.ID, "error", err)
return return
} }
case models.DeleteHost: case models.DeleteHost:
if servercfg.GetBrokerType() == servercfg.EmqxBrokerType { if servercfg.GetBrokerType() == servercfg.EmqxBrokerType {
// delete EMQX credentials for host // delete EMQX credentials for host
if err := DeleteEmqxUser(currentHost.ID.String()); err != nil { if err := DeleteEmqxUser(currentHost.ID.String()); err != nil {
logger.Log(0, "failed to remove host credentials from EMQX: ", currentHost.ID.String(), err.Error()) slog.Error("failed to remove host credentials from EMQX", "id", currentHost.ID, "error", err)
return return
} }
} }
if err := logic.DisassociateAllNodesFromHost(currentHost.ID.String()); err != nil { if err := logic.DisassociateAllNodesFromHost(currentHost.ID.String()); err != nil {
logger.Log(0, "failed to delete all nodes of host: ", currentHost.ID.String(), err.Error()) slog.Error("failed to delete all nodes of host", "id", currentHost.ID, "error", err)
return return
} }
if err := logic.RemoveHostByID(currentHost.ID.String()); err != nil { if err := logic.RemoveHostByID(currentHost.ID.String()); err != nil {
logger.Log(0, "failed to delete host: ", currentHost.ID.String(), err.Error()) slog.Error("failed to delete host", "id", currentHost.ID, "error", err)
return return
} }
sendPeerUpdate = true sendPeerUpdate = true
@ -167,7 +167,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
if servercfg.IsUsingTurn() { if servercfg.IsUsingTurn() {
err = logic.RegisterHostWithTurn(hostUpdate.Host.ID.String(), hostUpdate.Host.HostPass) err = logic.RegisterHostWithTurn(hostUpdate.Host.ID.String(), hostUpdate.Host.HostPass)
if err != nil { if err != nil {
logger.Log(0, "failed to register host with turn server: ", err.Error()) slog.Error("failed to register host with turn server", "id", currentHost.ID, "error", err)
return return
} }
} }
@ -177,7 +177,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) {
if sendPeerUpdate { if sendPeerUpdate {
err := PublishPeerUpdate() err := PublishPeerUpdate()
if err != nil { if err != nil {
logger.Log(0, "failed to pulish peer update: ", err.Error()) slog.Error("failed to publish peer update", "error", err)
} }
} }
// if servercfg.Is_EE && ifaceDelta { // if servercfg.Is_EE && ifaceDelta {
@ -192,57 +192,55 @@ func UpdateMetrics(client mqtt.Client, msg mqtt.Message) {
if servercfg.Is_EE { if servercfg.Is_EE {
id, err := getID(msg.Topic()) id, err := getID(msg.Topic())
if err != nil { if err != nil {
logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error()) slog.Error("error getting ID sent on ", "topic", msg.Topic(), "error", err)
return return
} }
currentNode, err := logic.GetNodeByID(id) currentNode, err := logic.GetNodeByID(id)
if err != nil { if err != nil {
logger.Log(1, "error getting node ", id, err.Error()) slog.Error("error getting node", "id", id, "error", err)
return return
} }
decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload()) decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
if decryptErr != nil { if decryptErr != nil {
logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error()) slog.Error("failed to decrypt message for node", "id", id, "error", decryptErr)
return return
} }
var newMetrics models.Metrics var newMetrics models.Metrics
if err := json.Unmarshal(decrypted, &newMetrics); err != nil { if err := json.Unmarshal(decrypted, &newMetrics); err != nil {
logger.Log(1, "error unmarshaling payload ", err.Error()) slog.Error("error unmarshaling payload", "error", err)
return return
} }
shouldUpdate := updateNodeMetrics(&currentNode, &newMetrics) shouldUpdate := updateNodeMetrics(&currentNode, &newMetrics)
if err = logic.UpdateMetrics(id, &newMetrics); err != nil { if err = logic.UpdateMetrics(id, &newMetrics); err != nil {
logger.Log(1, "faield to update node metrics", id, err.Error()) slog.Error("failed to update node metrics", "id", id, "error", err)
return return
} }
if servercfg.IsMetricsExporter() { if servercfg.IsMetricsExporter() {
if err := pushMetricsToExporter(newMetrics); err != nil { if err := pushMetricsToExporter(newMetrics); err != nil {
logger.Log(2, fmt.Sprintf("failed to push node: [%s] metrics to exporter, err: %v", slog.Error("failed to push node metrics to exporter", "id", currentNode.ID, "error", err)
currentNode.ID, err))
} }
} }
if newMetrics.Connectivity != nil { if newMetrics.Connectivity != nil {
err := logic.EnterpriseFailoverFunc(&currentNode) err := logic.EnterpriseFailoverFunc(&currentNode)
if err != nil { if err != nil {
logger.Log(0, "failed to failover for node", currentNode.ID.String(), "on network", currentNode.Network, "-", err.Error()) slog.Error("failed to failover for node", "id", currentNode.ID, "network", currentNode.Network, "error", err)
} }
} }
if shouldUpdate { if shouldUpdate {
logger.Log(2, "updating peers after node", currentNode.ID.String(), currentNode.Network, "detected connectivity issues") slog.Info("updating peers after node detected connectivity issues", "id", currentNode.ID, "network", currentNode.Network)
host, err := logic.GetHost(currentNode.HostID.String()) host, err := logic.GetHost(currentNode.HostID.String())
if err == nil { if err == nil {
if err = PublishSingleHostPeerUpdate(context.Background(), host, nil, nil); err != nil { if err = PublishSingleHostPeerUpdate(context.Background(), host, nil, nil); err != nil {
logger.Log(0, "failed to publish update after failover peer change for node", currentNode.ID.String(), currentNode.Network) slog.Warn("failed to publish update after failover peer change for node", "id", currentNode.ID, "network", currentNode.Network, "error", err)
} }
} }
} }
slog.Info("updated node metrics", "id", id)
logger.Log(1, "updated node metrics", id)
} }
} }
@ -250,17 +248,17 @@ func UpdateMetrics(client mqtt.Client, msg mqtt.Message) {
func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) { func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) {
id, err := getID(msg.Topic()) id, err := getID(msg.Topic())
if err != nil { if err != nil {
logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error()) slog.Error("error getting node.ID sent on ", "topic", msg.Topic(), "error", err)
return return
} }
currentNode, err := logic.GetNodeByID(id) currentNode, err := logic.GetNodeByID(id)
if err != nil { if err != nil {
logger.Log(1, "error getting node ", id, err.Error()) slog.Error("error getting node", "id", id, "error", err)
return return
} }
decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload()) decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
if decryptErr != nil { if decryptErr != nil {
logger.Log(1, "failed to decrypt message during client peer update for node ", id, decryptErr.Error()) slog.Error("failed to decrypt message for node", "id", id, "error", decryptErr)
return return
} }
switch decrypted[0] { switch decrypted[0] {
@ -268,12 +266,12 @@ func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) {
// do we still need this // do we still need this
case ncutils.DONE: case ncutils.DONE:
if err = PublishPeerUpdate(); err != nil { if err = PublishPeerUpdate(); err != nil {
logger.Log(1, "error publishing peer update for node", currentNode.ID.String(), err.Error()) slog.Error("error publishing peer update for node", "id", currentNode.ID, "error", err)
return return
} }
} }
logger.Log(1, "sent peer updates after signal received from", id) slog.Info("sent peer updates after signal received from", "id", id)
} }
func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) bool { func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) bool {
@ -282,7 +280,7 @@ func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) boo
} }
oldMetrics, err := logic.GetMetrics(currentNode.ID.String()) oldMetrics, err := logic.GetMetrics(currentNode.ID.String())
if err != nil { if err != nil {
logger.Log(1, "error finding old metrics for node", currentNode.ID.String()) slog.Error("error finding old metrics for node", "id", currentNode.ID, "error", err)
return false return false
} }
if oldMetrics.FailoverPeers == nil { if oldMetrics.FailoverPeers == nil {
@ -349,7 +347,7 @@ func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) boo
// add nodes that need failover // add nodes that need failover
nodes, err := logic.GetNetworkNodes(currentNode.Network) nodes, err := logic.GetNetworkNodes(currentNode.Network)
if err != nil { if err != nil {
logger.Log(0, "failed to retrieve nodes while updating metrics") slog.Error("failed to retrieve nodes while updating metrics", "error", err)
return false return false
} }
for _, node := range nodes { for _, node := range nodes {
@ -415,13 +413,13 @@ func handleHostCheckin(h, currentHost *models.Host) bool {
fakeNode.Action = models.NODE_DELETE fakeNode.Action = models.NODE_DELETE
fakeNode.PendingDelete = true fakeNode.PendingDelete = true
if err := NodeUpdate(&fakeNode); err != nil { if err := NodeUpdate(&fakeNode); err != nil {
logger.Log(0, "failed to inform host", currentHost.Name, currentHost.ID.String(), "to remove node", currNodeID, err.Error()) slog.Warn("failed to inform host to remove node", "host", currentHost.Name, "hostid", currentHost.ID, "nodeid", currNodeID, "error", err)
} }
} }
continue continue
} }
if err := logic.UpdateNodeCheckin(&node); err != nil { if err := logic.UpdateNodeCheckin(&node); err != nil {
logger.Log(0, "error updating node", node.ID.String(), " on checkin", err.Error()) slog.Warn("failed to update node on checkin", "nodeid", node.ID, "error", err)
} }
} }
@ -433,7 +431,7 @@ func handleHostCheckin(h, currentHost *models.Host) bool {
currentHost.FirewallInUse = h.FirewallInUse currentHost.FirewallInUse = h.FirewallInUse
currentHost.Version = h.Version currentHost.Version = h.Version
if err := logic.UpsertHost(currentHost); err != nil { if err := logic.UpsertHost(currentHost); err != nil {
logger.Log(0, "failed to update host after check-in", h.Name, h.ID.String(), err.Error()) slog.Error("failed to update host after check-in", "name", h.Name, "id", h.ID, "error", err)
return false return false
} }
} }
@ -447,12 +445,12 @@ func handleHostCheckin(h, currentHost *models.Host) bool {
currentHost.DefaultInterface = h.DefaultInterface currentHost.DefaultInterface = h.DefaultInterface
currentHost.NatType = h.NatType currentHost.NatType = h.NatType
if err := logic.UpsertHost(currentHost); err != nil { if err := logic.UpsertHost(currentHost); err != nil {
logger.Log(0, "failed to update host after check-in", h.Name, h.ID.String(), err.Error()) slog.Error("failed to update host after check-in", "name", h.Name, "id", h.ID, "error", err)
return false return false
} }
logger.Log(1, "updated host after check-in", currentHost.Name, currentHost.ID.String()) slog.Info("updated host after check-in", "name", currentHost.Name, "id", currentHost.ID)
} }
logger.Log(2, "check-in processed for host", h.Name, h.ID.String()) slog.Info("check-in processed for host", "name", h.Name, "id", h.ID)
return ifaceDelta return ifaceDelta
} }