Merge pull request #1136 from gravitl/feature_v0.14.2_MQ_port

configurable mq ports

config/config.go

@@ -66,6 +66,7 @@ type ServerConfig struct {
 	PortForwardServices   string `yaml:"portforwardservices"`
 	HostNetwork           string `yaml:"hostnetwork"`
 	MQPort                string `yaml:"mqport"`
+	MQServerPort          string `yaml:"mqserverport"`
 	Server                string `yaml:"server"`
 }
 

netclient/functions/daemon.go

@@ -25,6 +25,7 @@ import (
 	"github.com/gravitl/netmaker/netclient/local"
 	"github.com/gravitl/netmaker/netclient/ncutils"
 	"github.com/gravitl/netmaker/netclient/wireguard"
+	"github.com/gravitl/netmaker/servercfg"
 	ssl "github.com/gravitl/netmaker/tls"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
@@ -209,7 +210,7 @@ func NewTLSConfig(server string) *tls.Config {
 func setupMQTT(cfg *config.ClientConfig, publish bool) (mqtt.Client, error) {
 	opts := mqtt.NewClientOptions()
 	server := cfg.Server.Server
-	opts.AddBroker("ssl://" + server + ":8883") // TODO get the appropriate port of the comms mq server
+	opts.AddBroker("ssl://" + server + ":" + servercfg.GetMQPort())
 	opts.SetTLSConfig(NewTLSConfig(server))
 	opts.SetClientID(ncutils.MakeRandomString(23))
 	opts.SetDefaultPublishHandler(All)

netclient/functions/mqpublish.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"strconv"
 	"sync"
 	"time"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/gravitl/netmaker/netclient/auth"
 	"github.com/gravitl/netmaker/netclient/config"
 	"github.com/gravitl/netmaker/netclient/ncutils"
+	"github.com/gravitl/netmaker/servercfg"
 	"github.com/gravitl/netmaker/tls"
 )
 
@@ -170,10 +172,14 @@ func checkBroker(broker string) error {
 		return errors.New("nslookup failed for broker ... check dns records")
 	}
 	pinger := ping.NewTCPing()
+	port, err := strconv.Atoi(servercfg.GetMQPort())
+	if err != nil {
+		port = 8883
+	}
 	pinger.SetTarget(&ping.Target{
 		Protocol: ping.TCP,
 		Host:     broker,
-		Port:     8883,
+		Port:     port,
 		Counter:  3,
 		Interval: 1 * time.Second,
 		Timeout:  2 * time.Second,

servercfg/serverconf.go

@@ -196,13 +196,24 @@ func GetCoreDNSAddr() string {
 
 // GetMQPort - gets the mq port
 func GetMQPort() string {
-	mqport := "1883"
+	port := "8883" //default
 	if os.Getenv("MQ_PORT") != "" {
-		mqport = os.Getenv("MQ_PORT")
+		port = os.Getenv("MQ_PORT")
 	} else if config.Config.Server.MQPort != "" {
-		mqport = config.Config.Server.MQPort
+		port = config.Config.Server.MQPort
 	}
-	return mqport
+	return port
+}
+
+// GetMQServerPort - get mq port for server
+func GetMQServerPort() string {
+	port := "1883" //default
+	if os.Getenv("MQ_SERVER_PORT") != "" {
+		port = os.Getenv("MQ_SERVER_PORT")
+	} else if config.Config.Server.MQServerPort != "" {
+		port = config.Config.Server.MQServerPort
+	}
+	return port
 }
 
 // GetMessageQueueEndpoint - gets the message queue endpoint
@@ -213,8 +224,7 @@ func GetMessageQueueEndpoint() string {
 	} else if config.Config.Server.MQHOST != "" {
 		host = config.Config.Server.MQHOST
 	}
-	//Do we want MQ port configurable???
-	return host + ":1883"
+	return host + ":" + GetMQServerPort()
 }
 
 // GetMasterKey - gets the configured master key of server

serverctl/iptables.go

@@ -46,7 +46,7 @@ func portForwardServices(force bool) error {
 	for _, service := range services {
 		switch service {
 		case "mq":
-			err = iptablesPortForward("mq", "1883", "1883", false, force)
+			err = iptablesPortForward("mq", servercfg.GetMQServerPort(), servercfg.GetMQServerPort(), false, force)
 		case "dns":
 			err = iptablesPortForward("coredns", "53", "53", false, force)
 		case "ssh":