gravitl/netmaker
1
1
Fork 0
mirror of https://github.com/gravitl/netmaker.git synced 2025-10-10 05:45:52 +08:00
Code Issues Projects Releases Packages Wiki Activity

go routines now run forever like they are supposed to

Browse source
This commit is contained in:
Matthew R Kasun 2022-01-03 22:21:52 +00:00
parent efd8764de3
commit 8768a2c390
1 changed files with 90 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

176
netclient/functions/daemon.go
View file

@ -30,6 +30,7 @@ func Daemon() error {
signal.Notify(quit, syscall.SIGTERM, os.Interrupt) signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
<-quit <-quit
cancel() cancel()
ncutils.Log("all done")
return nil return nil
} }
@ -48,29 +49,28 @@ func SetupMQTT(cfg config.ClientConfig) mqtt.Client {
//Netclient sets up Message Queue and subsribes/publishes updates to/from server //Netclient sets up Message Queue and subsribes/publishes updates to/from server
func Netclient(ctx context.Context, network string) { func Netclient(ctx context.Context, network string) {
select { ncutils.Log("netclient go routine started for " + network)
case <-ctx.Done(): var cfg config.ClientConfig
ncutils.Log("shutting down daemon") cfg.Network = network
return cfg.ReadConfig()
default: //fix NodeID to remove ### so NodeID can be used as message topic
var cfg config.ClientConfig //remove with GRA-73
cfg.Network = network cfg.Node.ID = strings.ReplaceAll(cfg.Node.ID, "###", "-")
cfg.ReadConfig() ncutils.Log("daemon started for network:" + network)
//fix NodeID to remove ### so NodeID can be used as message topic client := SetupMQTT(cfg)
//remove with GRA-73 if token := client.Subscribe("#", 0, nil); token.Wait() && token.Error() != nil {
cfg.Node.ID = strings.ReplaceAll(cfg.Node.ID, "###", "-") log.Fatal(token.Error())
ncutils.Log("daemon started for network:" + network)
client := SetupMQTT(cfg)
if token := client.Subscribe("#", 0, nil); token.Wait() && token.Error() != nil {
log.Fatal(token.Error())
}
client.AddRoute("update/"+cfg.Node.ID, NodeUpdate)
client.AddRoute("update/peers/"+cfg.Node.ID, UpdatePeers)
client.AddRoute("update/keys/"+cfg.Node.ID, UpdateKeys)
defer client.Disconnect(250)
go Checkin(ctx, cfg, network)
go Metrics(ctx, cfg, network)
} }
client.AddRoute("update/"+cfg.Node.ID, NodeUpdate)
client.AddRoute("update/peers/"+cfg.Node.ID, UpdatePeers)
client.AddRoute("update/keys/"+cfg.Node.ID, UpdateKeys)
defer client.Disconnect(250)
go Checkin(ctx, cfg, network)
go Metrics(ctx, cfg, network)
<-ctx.Done()
ncutils.Log("shutting down daemon")
return
ncutils.Log("netclient go routine ended for " + network)
} }
//All -- mqtt message hander for all ('#') topics //All -- mqtt message hander for all ('#') topics
@ -97,42 +97,44 @@ var UpdateKeys mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
//Checkin -- go routine that checks for public or local ip changes, publishes changes //Checkin -- go routine that checks for public or local ip changes, publishes changes
// if there are no updates, simply "pings" the server as a checkin // if there are no updates, simply "pings" the server as a checkin
func Checkin(ctx context.Context, cfg config.ClientConfig, network string) { func Checkin(ctx context.Context, cfg config.ClientConfig, network string) {
select { for {
case <-ctx.Done(): select {
ncutils.Log("Checkin cancelled") case <-ctx.Done():
return ncutils.Log("Checkin cancelled")
//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ?? return
case <-time.After(time.Second * 10): //delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
ncutils.Log("Checkin running") case <-time.After(time.Second * 10):
if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" { ncutils.Log("Checkin running")
extIP, err := ncutils.GetPublicIP() if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
if err != nil { extIP, err := ncutils.GetPublicIP()
ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1) if err != nil {
} ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
if cfg.Node.Endpoint != extIP && extIP != "" { }
ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+extIP, 1) if cfg.Node.Endpoint != extIP && extIP != "" {
UpdateEndpoint(cfg, network, extIP) ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+extIP, 1)
} UpdateEndpoint(cfg, network, extIP)
intIP, err := getPrivateAddr() }
if err != nil { intIP, err := getPrivateAddr()
ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1) if err != nil {
} ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
if cfg.Node.LocalAddress != intIP && intIP != "" { }
ncutils.PrintLog("local Address has changed from "+cfg.Node.LocalAddress+" to "+intIP, 1) if cfg.Node.LocalAddress != intIP && intIP != "" {
UpdateLocalAddress(cfg, network, intIP) ncutils.PrintLog("local Address has changed from "+cfg.Node.LocalAddress+" to "+intIP, 1)
} UpdateLocalAddress(cfg, network, intIP)
} else { }
localIP, err := ncutils.GetLocalIP(cfg.Node.LocalRange) } else {
if err != nil { localIP, err := ncutils.GetLocalIP(cfg.Node.LocalRange)
ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1) if err != nil {
} ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
if cfg.Node.Endpoint != localIP && localIP != "" { }
ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+localIP, 1) if cfg.Node.Endpoint != localIP && localIP != "" {
UpdateEndpoint(cfg, network, localIP) ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+localIP, 1)
UpdateEndpoint(cfg, network, localIP)
}
} }
Hello(cfg, network)
ncutils.Log("Checkin complete")
} }
Hello(cfg, network)
ncutils.Log("Checkin complete")
} }
} }
@ -159,7 +161,7 @@ func UpdateLocalAddress(cfg config.ClientConfig, network, ip string) {
//Hello -- ping the broker to let server know node is alive and doing fine //Hello -- ping the broker to let server know node is alive and doing fine
func Hello(cfg config.ClientConfig, network string) { func Hello(cfg config.ClientConfig, network string) {
client := SetupMQTT(cfg) client := SetupMQTT(cfg)
if token := client.Publish("ping/"+network+"/"+cfg.Node.ID, 0, false, "hello world!"); token.Wait() && token.Error() != nil { if token := client.Publish("ping/"+cfg.Node.ID, 0, false, "hello world!"); token.Wait() && token.Error() != nil {
ncutils.Log("error publishing ping " + token.Error().Error()) ncutils.Log("error publishing ping " + token.Error().Error())
} }
client.Disconnect(250) client.Disconnect(250)
@ -167,35 +169,37 @@ func Hello(cfg config.ClientConfig, network string) {
//Metics -- go routine that collects wireguard metrics and publishes to broker //Metics -- go routine that collects wireguard metrics and publishes to broker
func Metrics(ctx context.Context, cfg config.ClientConfig, network string) { func Metrics(ctx context.Context, cfg config.ClientConfig, network string) {
select { for {
case <-ctx.Done(): select {
ncutils.Log("Metrics collection cancelled") case <-ctx.Done():
return ncutils.Log("Metrics collection cancelled")
//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ?? return
case <-time.After(time.Second * 60): //delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
ncutils.Log("Metrics collection running") case <-time.After(time.Second * 60):
ncutils.Log("Metrics running") ncutils.Log("Metrics collection running")
wg, err := wgctrl.New() ncutils.Log("Metrics running")
if err != nil { wg, err := wgctrl.New()
ncutils.Log("error getting devices " + err.Error()) if err != nil {
break ncutils.Log("error getting devices " + err.Error())
break
}
device, err := wg.Device(cfg.Node.Interface)
if err != nil {
ncutils.Log("error readind wg device " + err.Error())
break
}
bytes, err := json.Marshal(device.Peers)
if err != nil {
ncutils.Log("error marshaling peers " + err.Error())
break
}
client := SetupMQTT(cfg)
if token := client.Publish("metrics/"+cfg.Node.ID, 1, false, bytes); token.Wait() && token.Error() != nil {
ncutils.Log("error publishing metrics " + token.Error().Error())
}
wg.Close()
client.Disconnect(250)
ncutils.Log("metrics collection complete")
} }
device, err := wg.Device(cfg.Node.Interface)
if err != nil {
ncutils.Log("error readind wg device " + err.Error())
break
}
bytes, err := json.Marshal(device.Peers)
if err != nil {
ncutils.Log("error marshaling peers " + err.Error())
break
}
client := SetupMQTT(cfg)
if token := client.Publish("metrics/"+cfg.Node.ID, 1, false, bytes); token.Wait() && token.Error() != nil {
ncutils.Log("error publishing metrics " + token.Error().Error())
}
wg.Close()
client.Disconnect(250)
ncutils.Log("metrics collection complete")
} }
} }