From 56f360dfe404f47b1da60a4db6533db7f70b22a1 Mon Sep 17 00:00:00 2001 From: Matthew R Kasun Date: Mon, 5 Jun 2023 08:26:46 -0400 Subject: [PATCH] structured logging mq handlers --- mq/handlers.go | 100 ++++++++++++++++++++++++------------------------- 1 file changed, 49 insertions(+), 51 deletions(-) diff --git a/mq/handlers.go b/mq/handlers.go index 89cb1c0d..cf56f31d 100644 --- a/mq/handlers.go +++ b/mq/handlers.go @@ -10,86 +10,86 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/google/uuid" "github.com/gravitl/netmaker/database" - "github.com/gravitl/netmaker/logger" "github.com/gravitl/netmaker/logic" "github.com/gravitl/netmaker/logic/hostactions" "github.com/gravitl/netmaker/models" "github.com/gravitl/netmaker/netclient/ncutils" "github.com/gravitl/netmaker/servercfg" + "golang.org/x/exp/slog" "golang.zx2c4.com/wireguard/wgctrl/wgtypes" ) // DefaultHandler default message queue handler -- NOT USED 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 func UpdateNode(client mqtt.Client, msg mqtt.Message) { id, err := getID(msg.Topic()) 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 } currentNode, err := logic.GetNodeByID(id) if err != nil { - logger.Log(1, "error getting node ", id, err.Error()) + slog.Error("error getting node", "id", id, "error", err) return } decrypted, decryptErr := decryptMsg(¤tNode, msg.Payload()) 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 } var newNode models.Node if err := json.Unmarshal(decrypted, &newNode); err != nil { - logger.Log(1, "error unmarshaling payload ", err.Error()) + slog.Error("error unmarshaling payload", "error", err) return } ifaceDelta := logic.IfaceDelta(¤tNode, &newNode) if servercfg.Is_EE && ifaceDelta { 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() if err := logic.UpdateNode(¤tNode, &newNode); err != nil { - logger.Log(1, "error saving node", err.Error()) + slog.Error("error saving node", "id", id, "error", err) return } if ifaceDelta { // reduce number of unneeded updates, by only sending on iface changes 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 func UpdateHost(client mqtt.Client, msg mqtt.Message) { id, err := getID(msg.Topic()) 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 } currentHost, err := logic.GetHost(id) if err != nil { - logger.Log(1, "error getting host ", id, err.Error()) + slog.Error("error getting host", "id", id, "error", err) return } decrypted, decryptErr := decryptMsgWithHost(currentHost, msg.Payload()) 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 } var hostUpdate models.HostUpdate if err := json.Unmarshal(decrypted, &hostUpdate); err != nil { - logger.Log(1, "error unmarshaling payload ", err.Error()) + slog.Error("error unmarshaling payload", "error", err) 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 switch hostUpdate.Action { case models.CheckIn: @@ -98,21 +98,21 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) { hu := hostactions.GetAction(currentHost.ID.String()) if hu != 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 } else { if servercfg.GetBrokerType() == servercfg.EmqxBrokerType { 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 } } 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 } 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 } } @@ -131,7 +131,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) { } data, err := json.Marshal(&peerUpdate) 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()) server := servercfg.GetServer() @@ -143,23 +143,23 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) { sendPeerUpdate = logic.UpdateHostFromClient(&hostUpdate.Host, currentHost) err := logic.UpsertHost(currentHost) 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 } case models.DeleteHost: if servercfg.GetBrokerType() == servercfg.EmqxBrokerType { // delete EMQX credentials for host 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 } } 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 } 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 } sendPeerUpdate = true @@ -167,7 +167,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) { if servercfg.IsUsingTurn() { err = logic.RegisterHostWithTurn(hostUpdate.Host.ID.String(), hostUpdate.Host.HostPass) 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 } } @@ -177,7 +177,7 @@ func UpdateHost(client mqtt.Client, msg mqtt.Message) { if sendPeerUpdate { err := PublishPeerUpdate() 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 { @@ -192,57 +192,55 @@ func UpdateMetrics(client mqtt.Client, msg mqtt.Message) { if servercfg.Is_EE { id, err := getID(msg.Topic()) 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 } currentNode, err := logic.GetNodeByID(id) if err != nil { - logger.Log(1, "error getting node ", id, err.Error()) + slog.Error("error getting node", "id", id, "error", err) return } decrypted, decryptErr := decryptMsg(¤tNode, msg.Payload()) 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 } var newMetrics models.Metrics if err := json.Unmarshal(decrypted, &newMetrics); err != nil { - logger.Log(1, "error unmarshaling payload ", err.Error()) + slog.Error("error unmarshaling payload", "error", err) return } shouldUpdate := updateNodeMetrics(¤tNode, &newMetrics) 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 } if servercfg.IsMetricsExporter() { if err := pushMetricsToExporter(newMetrics); err != nil { - logger.Log(2, fmt.Sprintf("failed to push node: [%s] metrics to exporter, err: %v", - currentNode.ID, err)) + slog.Error("failed to push node metrics to exporter", "id", currentNode.ID, "error", err) } } if newMetrics.Connectivity != nil { err := logic.EnterpriseFailoverFunc(¤tNode) 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 { - 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()) if 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) } } } - - logger.Log(1, "updated node metrics", id) + slog.Info("updated node metrics", "id", id) } } @@ -250,17 +248,17 @@ func UpdateMetrics(client mqtt.Client, msg mqtt.Message) { func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) { id, err := getID(msg.Topic()) 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 } currentNode, err := logic.GetNodeByID(id) if err != nil { - logger.Log(1, "error getting node ", id, err.Error()) + slog.Error("error getting node", "id", id, "error", err) return } decrypted, decryptErr := decryptMsg(¤tNode, msg.Payload()) 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 } switch decrypted[0] { @@ -268,12 +266,12 @@ func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) { // do we still need this case ncutils.DONE: 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 } } - 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 { @@ -282,7 +280,7 @@ func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) boo } oldMetrics, err := logic.GetMetrics(currentNode.ID.String()) 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 } if oldMetrics.FailoverPeers == nil { @@ -349,7 +347,7 @@ func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) boo // add nodes that need failover nodes, err := logic.GetNetworkNodes(currentNode.Network) 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 } for _, node := range nodes { @@ -415,13 +413,13 @@ func handleHostCheckin(h, currentHost *models.Host) bool { fakeNode.Action = models.NODE_DELETE fakeNode.PendingDelete = true 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 } 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.Version = h.Version 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 } } @@ -447,12 +445,12 @@ func handleHostCheckin(h, currentHost *models.Host) bool { currentHost.DefaultInterface = h.DefaultInterface currentHost.NatType = h.NatType 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 } - 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 }