work in progress for message queue

go.mod

@@ -32,7 +32,9 @@ require (
 	cloud.google.com/go v0.34.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eclipse/paho.mqtt.golang v1.3.5 // indirect
 	github.com/felixge/httpsnoop v1.0.1 // indirect
+	github.com/go-ping/ping v0.0.0-20211130115550-779d1e919534 // indirect
 	github.com/go-playground/locales v0.14.0 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/google/go-cmp v0.5.5 // indirect
@@ -44,5 +46,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/russross/blackfriday/v2 v2.0.1 // indirect
 	github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	google.golang.org/appengine v1.4.0 // indirect
 )

go.sum

@@ -24,6 +24,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -34,6 +36,8 @@ github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8S
 github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+github.com/go-ping/ping v0.0.0-20211130115550-779d1e919534 h1:dhy9OQKGBh4zVXbjwbxxHjRxMJtLXj3zfgpBYQaR4Q4=
+github.com/go-ping/ping v0.0.0-20211130115550-779d1e919534/go.mod h1:xIFjORFzTxqIV/tDVGO4eDy/bLuSyawEeojSm3GfRGk=
 github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
 github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.14.0 h1:u50s323jtVGugKlcYeyzC0etD1HifMjqmJqb8WugfUU=
@@ -76,6 +80,8 @@ github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH
 github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
@@ -180,6 +186,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191007182048-72f939374954/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
@@ -187,6 +194,7 @@ golang.org/x/net v0.0.0-20201216054612-986b41b23924/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
 golang.org/x/net v0.0.0-20210504132125-bbd867fde50d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985 h1:4CSI6oo7cOjJKajidEljs9h+uP0rRZBPPPhcCbj5mw8=
 golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
@@ -197,6 +205,8 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -218,6 +228,7 @@ golang.org/x/sys v0.0.0-20210123111255-9b0068b26619/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210216163648-f7da38b97c65/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210309040221-94ec62e08169/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210403161142-5e06dd20ab57/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210503173754-0981d6026fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

netclient/cli_options/cmds.go

@@ -118,5 +118,14 @@ func GetCommands(cliFlags []cli.Flag) []*cli.Command {
 				return err
 			},
 		},
+		{
+			Name:  "daemon",
+			Usage: "run netclient as daemon",
+			Flags: cliFlags,
+			Action: func(c *cli.Context) error {
+				err := command.Daemon()
+				return err
+			},
+		},
 	}
 }

netclient/command/commands.go

@@ -210,3 +210,8 @@ func Uninstall() error {
 	ncutils.PrintLog("uninstalled netclient", 0)
 	return err
 }
+
+func Daemon() error {
+	err := functions.Daemon()
+	return err
+}

netclient/functions/daemon.go

@@ -0,0 +1,214 @@
+package functions
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/go-ping/ping"
+	"github.com/gravitl/netmaker/netclient/config"
+	"github.com/gravitl/netmaker/netclient/ncutils"
+	"golang.zx2c4.com/wireguard/wgctrl"
+)
+
+func Daemon() error {
+	networks, err := ncutils.GetSystemNetworks()
+	if err != nil {
+		return err
+	}
+	for _, network := range networks {
+		go Netclient(network)
+	}
+	for {
+	}
+	return nil
+}
+
+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-mqttt")
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	if token := client.Subscribe("#", 0, nil); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	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(client, network)
+	//go Metrics(client, network)
+	//go Connectivity(client, network)
+	for {
+	}
+}
+
+var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("Topic: " + string(msg.Topic()))
+	ncutils.Log("Message: " + string(msg.Payload()))
+}
+
+var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("received message to update node " + string(msg.Payload()))
+}
+
+var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("received message to update peers " + string(msg.Payload()))
+}
+
+var UpdateKeys mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("received message to update keys " + string(msg.Payload()))
+}
+
+func Checkin(client mqtt.Client, network string) {
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
+	for {
+		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)
+			}
+		}
+		Ping(client, network)
+	}
+}
+
+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())
+	}
+}
+
+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 {
+		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 creating pinger " + err.Error())
+				break
+			}
+			pinger.Timeout = 2 * time.Second
+			pinger.Run()
+			stats := pinger.Statistics()
+			if stats.PacketLoss == 100 {
+				pingStat.Reachable = false
+			}
+			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")
+}