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

remove go routines for handling reciept of subscribed messages

Browse Source
This commit is contained in:
Matthew R. Kasun
2022-02-08 13:13:43 -05:00
parent 05ada0402a
commit 3b7f4e31b1
1 changed files with 129 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

264
netclient/functions/daemon.go
View File

@@ -224,157 +224,152 @@ var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
// NodeUpdate -- mqtt message handler for /update/<NodeID> topic // NodeUpdate -- mqtt message handler for /update/<NodeID> topic
func NodeUpdate(client mqtt.Client, msg mqtt.Message) { func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
//potentiall blocking i/o so do this in a go routine var newNode models.Node
go func() { var cfg config.ClientConfig
var newNode models.Node var network = parseNetworkFromTopic(msg.Topic())
var cfg config.ClientConfig cfg.Network = network
var network = parseNetworkFromTopic(msg.Topic()) cfg.ReadConfig()
cfg.Network = network
cfg.ReadConfig()
data, dataErr := decryptMsg(&cfg, msg.Payload()) data, dataErr := decryptMsg(&cfg, msg.Payload())
if dataErr != nil { if dataErr != nil {
return return
}
err := json.Unmarshal([]byte(data), &newNode)
if err != nil {
ncutils.Log("error unmarshalling node update data" + err.Error())
return
}
ncutils.Log("received message to update node " + newNode.Name)
// see if cache hit, if so skip
var currentMessage = read(newNode.Network, lastNodeUpdate)
if currentMessage == string(data) {
return
}
insert(newNode.Network, lastNodeUpdate, string(data)) // store new message in cache
//check if interface name has changed if so delete.
if cfg.Node.Interface != newNode.Interface {
if err = wireguard.RemoveConf(cfg.Node.Interface, true); err != nil {
ncutils.PrintLog("could not delete old interface "+cfg.Node.Interface+": "+err.Error(), 1)
} }
err := json.Unmarshal([]byte(data), &newNode) }
newNode.PullChanges = "no"
//ensure that OS never changes
newNode.OS = runtime.GOOS
// check if interface needs to delta
ifaceDelta := ncutils.IfaceDelta(&cfg.Node, &newNode)
shouldDNSChange := cfg.Node.DNSOn != newNode.DNSOn
cfg.Node = newNode
switch newNode.Action {
case models.NODE_DELETE:
if cancel, ok := networkcontext.Load(newNode.Network); ok {
ncutils.Log("cancelling message queue context for " + newNode.Network)
cancel.(context.CancelFunc)()
} else {
ncutils.Log("failed to kill go routines for network " + newNode.Network)
}
ncutils.PrintLog(fmt.Sprintf("received delete request for %s", cfg.Node.Name), 1)
if err = LeaveNetwork(cfg.Node.Network); err != nil {
if !strings.Contains("rpc error", err.Error()) {
ncutils.PrintLog(fmt.Sprintf("failed to leave, please check that local files for network %s were removed", cfg.Node.Network), 1)
}
}
ncutils.PrintLog(fmt.Sprintf("%s was removed", cfg.Node.Name), 1)
return
case models.NODE_UPDATE_KEY:
if err := UpdateKeys(&cfg, client); err != nil {
ncutils.PrintLog("err updating wireguard keys: "+err.Error(), 1)
}
ifaceDelta = true
case models.NODE_NOOP:
default:
}
// Save new config
cfg.Node.Action = models.NODE_NOOP
if err := config.Write(&cfg, cfg.Network); err != nil {
ncutils.PrintLog("error updating node configuration: "+err.Error(), 1)
}
nameserver := cfg.Server.CoreDNSAddr
privateKey, err := wireguard.RetrievePrivKey(newNode.Network)
if err != nil {
ncutils.Log("error reading PrivateKey " + err.Error())
return
}
file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
if err := wireguard.UpdateWgInterface(file, privateKey, nameserver, newNode); err != nil {
ncutils.Log("error updating wireguard config " + err.Error())
return
}
if ifaceDelta {
ncutils.Log("applying WG conf to " + file)
err = wireguard.ApplyConf(&cfg.Node, cfg.Node.Interface, file)
if err != nil { if err != nil {
ncutils.Log("error unmarshalling node update data" + err.Error()) ncutils.Log("error restarting wg after node update " + err.Error())
return return
} }
time.Sleep(time.Second >> 1)
ncutils.Log("received message to update node " + newNode.Name) if err = Resubscribe(client, &cfg); err != nil {
// see if cache hit, if so skip ncutils.Log("error resubscribing after interface change " + err.Error())
var currentMessage = read(newNode.Network, lastNodeUpdate)
if currentMessage == string(data) {
return return
} }
insert(newNode.Network, lastNodeUpdate, string(data)) // store new message in cache if newNode.DNSOn == "yes" {
for _, server := range newNode.NetworkSettings.DefaultServerAddrs {
//check if interface name has changed if so delete. if server.IsLeader {
if cfg.Node.Interface != newNode.Interface { go local.SetDNSWithRetry(newNode, server.Address)
if err = wireguard.RemoveConf(cfg.Node.Interface, true); err != nil { break
ncutils.PrintLog("could not delete old interface "+cfg.Node.Interface+": "+err.Error(), 1)
}
}
newNode.PullChanges = "no"
//ensure that OS never changes
newNode.OS = runtime.GOOS
// check if interface needs to delta
ifaceDelta := ncutils.IfaceDelta(&cfg.Node, &newNode)
shouldDNSChange := cfg.Node.DNSOn != newNode.DNSOn
cfg.Node = newNode
switch newNode.Action {
case models.NODE_DELETE:
if cancel, ok := networkcontext.Load(newNode.Network); ok {
ncutils.Log("cancelling message queue context for " + newNode.Network)
cancel.(context.CancelFunc)()
} else {
ncutils.Log("failed to kill go routines for network " + newNode.Network)
}
ncutils.PrintLog(fmt.Sprintf("received delete request for %s", cfg.Node.Name), 1)
if err = LeaveNetwork(cfg.Node.Network); err != nil {
if !strings.Contains("rpc error", err.Error()) {
ncutils.PrintLog(fmt.Sprintf("failed to leave, please check that local files for network %s were removed", cfg.Node.Network), 1)
}
}
ncutils.PrintLog(fmt.Sprintf("%s was removed", cfg.Node.Name), 1)
return
case models.NODE_UPDATE_KEY:
if err := UpdateKeys(&cfg, client); err != nil {
ncutils.PrintLog("err updating wireguard keys: "+err.Error(), 1)
}
ifaceDelta = true
case models.NODE_NOOP:
default:
}
// Save new config
cfg.Node.Action = models.NODE_NOOP
if err := config.Write(&cfg, cfg.Network); err != nil {
ncutils.PrintLog("error updating node configuration: "+err.Error(), 1)
}
nameserver := cfg.Server.CoreDNSAddr
privateKey, err := wireguard.RetrievePrivKey(newNode.Network)
if err != nil {
ncutils.Log("error reading PrivateKey " + err.Error())
return
}
file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
if err := wireguard.UpdateWgInterface(file, privateKey, nameserver, newNode); err != nil {
ncutils.Log("error updating wireguard config " + err.Error())
return
}
if ifaceDelta {
ncutils.Log("applying WG conf to " + file)
err = wireguard.ApplyConf(&cfg.Node, cfg.Node.Interface, file)
if err != nil {
ncutils.Log("error restarting wg after node update " + err.Error())
return
}
time.Sleep(time.Second >> 1)
if err = Resubscribe(client, &cfg); err != nil {
ncutils.Log("error resubscribing after interface change " + err.Error())
return
}
if newNode.DNSOn == "yes" {
for _, server := range newNode.NetworkSettings.DefaultServerAddrs {
if server.IsLeader {
go local.SetDNSWithRetry(newNode, server.Address)
break
}
} }
} }
} }
//deal with DNS }
if newNode.DNSOn != "yes" && shouldDNSChange && cfg.Node.Interface != "" { //deal with DNS
ncutils.Log("settng DNS off") if newNode.DNSOn != "yes" && shouldDNSChange && cfg.Node.Interface != "" {
_, err := ncutils.RunCmd("/usr/bin/resolvectl revert "+cfg.Node.Interface, true) ncutils.Log("settng DNS off")
if err != nil { _, err := ncutils.RunCmd("/usr/bin/resolvectl revert "+cfg.Node.Interface, true)
ncutils.Log("error applying dns" + err.Error()) if err != nil {
} ncutils.Log("error applying dns" + err.Error())
} }
}() }
} }
// UpdatePeers -- mqtt message handler for peers/<Network>/<NodeID> topic // UpdatePeers -- mqtt message handler for peers/<Network>/<NodeID> topic
func UpdatePeers(client mqtt.Client, msg mqtt.Message) { func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
go func() { var peerUpdate models.PeerUpdate
var peerUpdate models.PeerUpdate var network = parseNetworkFromTopic(msg.Topic())
var network = parseNetworkFromTopic(msg.Topic()) var cfg = config.ClientConfig{}
var cfg = config.ClientConfig{} cfg.Network = network
cfg.Network = network cfg.ReadConfig()
cfg.ReadConfig()
data, dataErr := decryptMsg(&cfg, msg.Payload()) data, dataErr := decryptMsg(&cfg, msg.Payload())
if dataErr != nil { if dataErr != nil {
return return
} }
err := json.Unmarshal([]byte(data), &peerUpdate) err := json.Unmarshal([]byte(data), &peerUpdate)
if err != nil { if err != nil {
ncutils.Log("error unmarshalling peer data") ncutils.Log("error unmarshalling peer data")
return return
} }
// see if cached hit, if so skip // see if cached hit, if so skip
var currentMessage = read(peerUpdate.Network, lastPeerUpdate) var currentMessage = read(peerUpdate.Network, lastPeerUpdate)
if currentMessage == string(data) { if currentMessage == string(data) {
return return
} }
insert(peerUpdate.Network, lastPeerUpdate, string(data)) insert(peerUpdate.Network, lastPeerUpdate, string(data))
file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf" file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
err = wireguard.UpdateWgPeers(file, peerUpdate.Peers) err = wireguard.UpdateWgPeers(file, peerUpdate.Peers)
if err != nil { if err != nil {
ncutils.Log("error updating wireguard peers" + err.Error()) ncutils.Log("error updating wireguard peers" + err.Error())
return return
} }
//err = wireguard.SyncWGQuickConf(cfg.Node.Interface, file) //err = wireguard.SyncWGQuickConf(cfg.Node.Interface, file)
err = wireguard.SetPeers(cfg.Node.Interface, cfg.Node.Address, cfg.Node.PersistentKeepalive, peerUpdate.Peers) err = wireguard.SetPeers(cfg.Node.Interface, cfg.Node.Address, cfg.Node.PersistentKeepalive, peerUpdate.Peers)
if err != nil { if err != nil {
ncutils.Log("error syncing wg after peer update " + err.Error()) ncutils.Log("error syncing wg after peer update " + err.Error())
return return
} }
ncutils.Log(fmt.Sprintf("received peer update on network, %s", cfg.Network)) ncutils.Log(fmt.Sprintf("received peer update on network, %s", cfg.Network))
}()
} }
// MonitorKeepalive - checks time last server keepalive received. If more than 3+ minutes, notify and resubscribe // MonitorKeepalive - checks time last server keepalive received. If more than 3+ minutes, notify and resubscribe
@@ -456,7 +451,6 @@ func Resubscribe(client mqtt.Client, cfg *config.ClientConfig) error {
// UpdateKeys -- updates private key and returns new publickey // UpdateKeys -- updates private key and returns new publickey
func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) error { func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) error {
ncutils.Log("received message to update keys") ncutils.Log("received message to update keys")
//potentiall blocking i/o so do this in a go routine
key, err := wgtypes.GeneratePrivateKey() key, err := wgtypes.GeneratePrivateKey()
if err != nil { if err != nil {
ncutils.Log("error generating privatekey " + err.Error()) ncutils.Log("error generating privatekey " + err.Error())