gravitl/netmaker
1
1
Fork 0
mirror of https://github.com/gravitl/netmaker.git synced 2025-11-01 00:49:04 +08:00
Code Issues Projects Releases Packages Wiki Activity

Net 262 - structured logging (#2366)

Browse source 
* structured logging infra

* structured logging mq handlers
This commit is contained in:
Matthew R Kasun 2023-06-06 13:47:16 -04:00 • committed by GitHub
parent c6b1ec494c
commit f94cda11f6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 73 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
go.mod
View file

@ -36,7 +36,7 @@ require (
github.com/coreos/go-oidc/v3 v3.6.0
github.com/gorilla/websocket v1.5.0
github.com/pkg/errors v0.9.1
golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e
golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1
gortc.io/stun v1.23.0
)

4
go.sum
View file

@ -123,8 +123,8 @@ golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5y
golang.org/x/crypto v0.0.0-20220208050332-20e1d8d225ab/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e h1:+WEEuIdZHnUeJJmEUjyYC2gfUMj69yZXw17EnHg/otA=
golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=
golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1 h1:k/i9J1pBpvlfR+9QsetwPyERsqu1GIbi967PQMq3Ivc=
golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=

21
main.go
View file

@ -7,6 +7,7 @@ import (
"fmt"
"os"
"os/signal"
"path/filepath"
"runtime/debug"
"sync"
"syscall"
@ -26,6 +27,7 @@ import (
"github.com/gravitl/netmaker/servercfg"
"github.com/gravitl/netmaker/serverctl"
stunserver "github.com/gravitl/netmaker/stun-server"
"golang.org/x/exp/slog"
)
var version = "v0.20.2"
@ -179,6 +181,25 @@ func runMessageQueue(wg *sync.WaitGroup, ctx context.Context) {
func setVerbosity() {
verbose := int(servercfg.GetVerbosity())
logger.Verbosity = verbose
logLevel := &slog.LevelVar{}
replace := func(groups []string, a slog.Attr) slog.Attr {
if a.Key == slog.SourceKey {
a.Value = slog.StringValue(filepath.Base(a.Value.String()))
}
return a
}
logger := slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{AddSource: true, ReplaceAttr: replace, Level: logLevel}))
slog.SetDefault(logger)
switch verbose {
case 4:
logLevel.Set(slog.LevelDebug)
case 3:
logLevel.Set(slog.LevelInfo)
case 2:
logLevel.Set(slog.LevelWarn)
default:
logLevel.Set(slog.LevelError)
}
}
func setGarbageCollection() {

100
mq/handlers.go
View file

@ -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(&currentNode, 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(&currentNode, &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(&currentNode, &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(&currentNode, 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(&currentNode, &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(&currentNode)
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(&currentNode, 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
}