work in progress for message queue

netclient/functions/daemon.go

@@ -1,329 +1,214 @@
 package functions
 
 import (
-	"context"
 	"encoding/json"
+	"fmt"
 	"log"
-	"os"
-	"os/signal"
-	"runtime"
-	"strings"
-	"syscall"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
-	"github.com/gravitl/netmaker/models"
+	"github.com/go-ping/ping"
 	"github.com/gravitl/netmaker/netclient/config"
 	"github.com/gravitl/netmaker/netclient/ncutils"
-	"github.com/gravitl/netmaker/netclient/wireguard"
 	"golang.zx2c4.com/wireguard/wgctrl"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
-// Daemon runs netclient daemon from command line
 func Daemon() error {
-	ctx, cancel := context.WithCancel(context.Background())
 	networks, err := ncutils.GetSystemNetworks()
 	if err != nil {
-		cancel()
 		return err
 	}
 	for _, network := range networks {
-		go Netclient(ctx, network)
+		go Netclient(network)
+	}
+	for {
 	}
-	quit := make(chan os.Signal, 1)
-	signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
-	<-quit
-	cancel()
-	ncutils.Log("all done")
 	return nil
 }
 
-// SetupMQTT creates a connection to broker and return client
-func SetupMQTT(cfg config.ClientConfig) mqtt.Client {
+func Netclient(network string) {
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
+	ncutils.Log("daemon started for network:" + network)
+	//setup MQTT
 	opts := mqtt.NewClientOptions()
 	ncutils.Log("setting broker to " + cfg.Server.CoreDNSAddr + ":1883")
 	opts.AddBroker(cfg.Server.CoreDNSAddr + ":1883")
 	opts.SetDefaultPublishHandler(All)
+	opts.SetClientID("netclient-mqtt")
 	client := mqtt.NewClient(opts)
 	if token := client.Connect(); token.Wait() && token.Error() != nil {
 		log.Fatal(token.Error())
 	}
-	return client
-}
-
-// Netclient sets up Message Queue and subsribes/publishes updates to/from server
-func Netclient(ctx context.Context, network string) {
-	ncutils.Log("netclient go routine started for " + network)
-	var cfg config.ClientConfig
-	cfg.Network = network
-	cfg.ReadConfig()
-	//fix NodeID to remove ### so NodeID can be used as message topic
-	//remove with GRA-73
-	cfg.Node.ID = strings.Replace(cfg.Node.ID, "###", "-", 1)
-	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)
-	//handle key updates in node update
-	//client.AddRoute("update/keys/"+cfg.Node.ID, UpdateKeys)
+	client.AddRoute("update/"+network+"/"+cfg.Node.MacAddress, NodeUpdate)
+	client.AddRoute("update/"+network+"/peers", UpdatePeers)
+	client.AddRoute("update/"+network+"/keys", UpdateKeys)
+	client.AddRoute("update/"+network+"/keys/"+cfg.Node.MacAddress, UpdateKeys)
 	defer client.Disconnect(250)
-	go Checkin(ctx, cfg, network)
-	go Metrics(ctx, cfg, network)
-	<-ctx.Done()
-	ncutils.Log("shutting down daemon")
+	go Checkin(client, network)
+	//go Metrics(client, network)
+	//go Connectivity(client, network)
+	for {
+	}
 }
 
-// All -- mqtt message hander for all ('#') topics
 var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 	ncutils.Log("Topic: " + string(msg.Topic()))
 	ncutils.Log("Message: " + string(msg.Payload()))
 }
 
-// NodeUpdate -- mqtt message handler for /update/<NodeID> topic
 var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 	ncutils.Log("received message to update node " + string(msg.Payload()))
-	//potentiall blocking i/o so do this in a go routine
-	go func() {
-		var newNode models.Node
-		var cfg config.ClientConfig
-		cfg.Network = newNode.Network
-		cfg.ReadConfig()
-		err := json.Unmarshal(msg.Payload(), &newNode)
-		if err != nil {
-			ncutils.Log("error unmarshalling node update data" + err.Error())
-			return
-		}
-		//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)
-			}
-		}
-		newNode.PullChanges = "no"
-		//ensure that OS never changes
-		newNode.OS = runtime.GOOS
-		cfg.Node = newNode
-		switch newNode.Action {
-		case models.NODE_DELETE:
-			if err := RemoveLocalInstance(&cfg, cfg.Network); err != nil {
-				ncutils.PrintLog("error deleting local instance: "+err.Error(), 1)
-				return
-			}
-		case models.NODE_UPDATE_KEY:
-			UpdateKeys(&cfg, client)
-		case models.NODE_NOOP:
-		default:
-		}
-		//Save new config
-		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
-		}
-		if err := wireguard.UpdateWgInterface(cfg.Node.Interface, privateKey, nameserver, newNode); err != nil {
-			ncutils.Log("error updating wireguard config " + err.Error())
-			return
-		}
-		// path hardcoded for now... should be updated
-		err = wireguard.ApplyWGQuickConf("/etc/netclient/config/" + cfg.Node.Interface + ".conf")
-		if err != nil {
-			ncutils.Log("error restarting wg after node update " + err.Error())
-			return
-		}
-	}()
 }
 
-// UpdatePeers -- mqtt message handler for /update/peers/<NodeID> topic
 var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 	ncutils.Log("received message to update peers " + string(msg.Payload()))
-	go func() {
-		var peerUpdate models.PeerUpdate
-		err := json.Unmarshal(msg.Payload(), &peerUpdate)
-		if err != nil {
-			ncutils.Log("error unmarshalling peer data")
-			return
-		}
-		var cfg config.ClientConfig
-		cfg.Network = peerUpdate.Network
-		cfg.ReadConfig()
-		peers, err := CalculatePeers(cfg.Node, peerUpdate.Nodes, cfg.Node.IsDualStack, cfg.Node.IsEgressGateway, cfg.Node.IsServer)
-		if err != nil {
-			ncutils.Log("error calculating Peers " + err.Error())
-			return
-		}
-		extpeers, err := CalculateExtPeers(cfg.Node, peerUpdate.ExtPeers)
-		if err != nil {
-			ncutils.Log("error updated external wireguard peers " + err.Error())
-		}
-		peers = append(peers, extpeers...)
-		err = wireguard.UpdateWgPeers(cfg.Node.Interface, peers)
-		if err != nil {
-			ncutils.Log("error updating wireguard peers" + err.Error())
-			return
-		}
-		// path hardcoded for now... should be updated
-		err = wireguard.ApplyWGQuickConf("/etc/netclient/config/" + cfg.Node.Interface + ".conf")
-		if err != nil {
-			ncutils.Log("error restarting wg after peer update " + err.Error())
-			return
-		}
-	}()
 }
 
-// UpdateKeys -- updates private key and returns new publickey
-func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) (*config.ClientConfig, error) {
-	ncutils.Log("received message to update keys")
-	//potentiall blocking i/o so do this in a go routine
-	key, err := wgtypes.GeneratePrivateKey()
-	if err != nil {
-		ncutils.Log("error generating privatekey " + err.Error())
-		return cfg, err
-	}
-	if err := wireguard.UpdatePrivateKey(cfg.Node.Interface, key.String()); err != nil {
-		ncutils.Log("error updating wireguard key " + err.Error())
-		return cfg, err
-	}
-	publicKey := key.PublicKey()
-	if token := client.Publish("update/publickey/"+cfg.Node.ID, 0, false, publicKey.String()); token.Wait() && token.Error() != nil {
-		ncutils.Log("error publishing publickey update " + token.Error().Error())
-		client.Disconnect(250)
-		return cfg, err
-	}
-	if err := config.ModConfig(&cfg.Node); err != nil {
-		ncutils.Log("error updating local config " + err.Error())
-	}
-	return cfg, nil
+var UpdateKeys mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("received message to update keys " + string(msg.Payload()))
 }
 
-// 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
-func Checkin(ctx context.Context, cfg config.ClientConfig, network string) {
+func Checkin(client mqtt.Client, network string) {
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
 	for {
-		select {
-		case <-ctx.Done():
-			ncutils.Log("Checkin cancelled")
-			return
-			//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
-		case <-time.After(time.Second * 10):
-			ncutils.Log("Checkin running")
-			//read latest config
-			cfg.ReadConfig()
-			//fix NodeID to remove ### so NodeID can be used as message topic
-			//remove with GRA-73
-			cfg.Node.ID = strings.Replace(cfg.Node.ID, "###", "-", 1)
-			if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
-				extIP, err := ncutils.GetPublicIP()
-				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)
-					UpdateEndpoint(cfg, network, extIP)
-				}
-				intIP, err := getPrivateAddr()
-				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)
-					UpdateLocalAddress(cfg, network, intIP)
-				}
-			} else {
-				localIP, err := ncutils.GetLocalIP(cfg.Node.LocalRange)
-				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)
-					UpdateEndpoint(cfg, network, localIP)
-				}
+		time.Sleep(time.Duration(cfg.Node.NetworkSettings.DefaultCheckInInterval) * time.Second)
+		ncutils.Log("Checkin running")
+		if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
+			extIP, err := ncutils.GetPublicIP()
+			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)
+				UpdateEndpoint(client, network, extIP)
+			}
+			intIP, err := getPrivateAddr()
+			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)
+				UpdateLocalAddress(client, network, intIP)
+			}
+		} else {
+			localIP, err := ncutils.GetLocalIP(cfg.Node.LocalRange)
+			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)
+				UpdateEndpoint(client, network, localIP)
 			}
-			Hello(cfg, network)
-			ncutils.Log("Checkin complete")
 		}
+		Ping(client, network)
 	}
 }
 
-// UpdateEndpoint -- publishes an endpoint update to broker
-func UpdateEndpoint(cfg config.ClientConfig, network, ip string) {
-	ncutils.Log("Updating endpoint")
-	client := SetupMQTT(cfg)
-	if token := client.Publish("update/ip/"+cfg.Node.ID, 0, false, ip); token.Wait() && token.Error() != nil {
-		ncutils.Log("error publishing endpoint update " + token.Error().Error())
-	}
-	cfg.Node.Endpoint = ip
-	if err := config.Write(&cfg, cfg.Network); err != nil {
-		ncutils.Log("error updating local config " + err.Error())
-	}
-	client.Disconnect(250)
-}
-
-// UpdateLocalAddress -- publishes a local address update to broker
-func UpdateLocalAddress(cfg config.ClientConfig, network, ip string) {
-	ncutils.Log("Updating local address")
-	client := SetupMQTT(cfg)
-	if token := client.Publish("update/localaddress/"+cfg.Node.ID, 0, false, ip); token.Wait() && token.Error() != nil {
-		ncutils.Log("error publishing local address update " + token.Error().Error())
-	}
-	cfg.Node.LocalAddress = ip
-	ncutils.Log("updating local address in local config to: " + cfg.Node.LocalAddress)
-	if err := config.Write(&cfg, cfg.Network); err != nil {
-		ncutils.Log("error updating local config " + err.Error())
-	}
-	client.Disconnect(250)
-}
-
-// Hello -- ping the broker to let server know node is alive and doing fine
-func Hello(cfg config.ClientConfig, network string) {
-	client := SetupMQTT(cfg)
-	ncutils.Log("sending ping " + cfg.Node.ID)
-	if token := client.Publish("ping/"+cfg.Node.ID, 2, false, "hello world!"); token.Wait() && token.Error() != nil {
+func Ping(client mqtt.Client, network string) {
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
+	if token := client.Publish("ping/"+network+"/"+cfg.Node.ID, 0, false, []byte("ping")); token.Wait() && token.Error() != nil {
 		ncutils.Log("error publishing ping " + token.Error().Error())
 	}
-	client.Disconnect(250)
 }
 
-// Metics --  go routine that collects wireguard metrics and publishes to broker
-func Metrics(ctx context.Context, cfg config.ClientConfig, network string) {
+func Metrics(client mqtt.Client, network string) {
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
 	for {
-		select {
-		case <-ctx.Done():
-			ncutils.Log("Metrics collection cancelled")
-			return
-			//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
-		case <-time.After(time.Second * 60):
-			ncutils.Log("Metrics collection running")
-			ncutils.Log("Metrics running")
-			wg, err := wgctrl.New()
+		time.Sleep(time.Second * 60)
+		ncutils.Log("Metrics running")
+		wg, err := wgctrl.New()
+		if err != nil {
+			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
+		}
+		if token := client.Publish("metrics/"+network+"/"+cfg.Node.ID, 1, false, bytes); token.Wait() && token.Error() != nil {
+			ncutils.Log("error publishing metrics " + token.Error().Error())
+			break
+		}
+		wg.Close()
+	}
+}
+
+type PingStat struct {
+	Name      string
+	Reachable bool
+}
+
+func Connectivity(client mqtt.Client, network string) {
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
+	for {
+		time.Sleep(time.Duration(cfg.NetworkSettings.DefaultCheckInInterval) * time.Second)
+		ncutils.Log("Connectivity running")
+		var pingStats []PingStat
+		peers, err := ncutils.GetPeers(cfg.Node.Interface)
+		if err != nil {
+			ncutils.Log("error retriving peers " + err.Error())
+			break
+		}
+		for _, peer := range peers {
+			var pingStat PingStat
+			pingStat.Name = peer.PublicKey.String()
+			pingStat.Reachable = true
+			ip := peer.Endpoint.IP.String()
+			fmt.Println("----------", peer.Endpoint.IP, ip)
+			pinger, err := ping.NewPinger(ip)
 			if err != nil {
-				ncutils.Log("error getting devices " + err.Error())
+				ncutils.Log("error creating pinger " + err.Error())
 				break
 			}
-			device, err := wg.Device(cfg.Node.Interface)
-			if err != nil {
-				ncutils.Log("error readind wg device " + err.Error())
-				break
+			pinger.Timeout = 2 * time.Second
+			pinger.Run()
+			stats := pinger.Statistics()
+			if stats.PacketLoss == 100 {
+				pingStat.Reachable = false
 			}
-			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")
+			pingStats = append(pingStats, pingStat)
+		}
+		bytes, err := json.Marshal(pingStats)
+		if err != nil {
+			ncutils.Log("error marshaling stats" + err.Error())
+			break
+		}
+		if token := client.Publish("connectivity/"+network+"/"+cfg.Node.ID, 1, false, bytes); token.Wait() && token.Error() != nil {
+			ncutils.Log("error publishing ping stats " + token.Error().Error())
+			break
 		}
 	}
 }
+
+func UpdateEndpoint(client mqtt.Client, network, ip string) {
+	ncutils.Log("Updating endpoint")
+}
+
+func UpdateLocalAddress(client mqtt.Client, network, ip string) {
+	ncutils.Log("Updating local address")
+}