NM-271: Scalability Improvements

Dockerfile

@@ -17,4 +17,5 @@ RUN mkdir -p /etc/netclient/config
 COPY --from=builder /app/netmaker .
 COPY --from=builder /app/config config
 EXPOSE 8081
+EXPOSE 6060
 ENTRYPOINT ["./netmaker"]

compose/docker-compose.pro.yml

@@ -4,6 +4,7 @@ services:
     container_name: prometheus
     image: prom/prometheus:latest
     environment:
+      - METRICS_USERNAME=${METRICS_USERNAME:-netmaker}
       - METRICS_SECRET=${METRICS_SECRET}
       - NETMAKER_METRICS_TARGET=${NETMAKER_METRICS_TARGET:-http://netmaker-exporter:8085}
       - PROMETHEUS_RETENTION_TIME=${PROMETHEUS_RETENTION_TIME:-15d}
@@ -15,18 +16,19 @@ services:
     command:
       - -c
       - |
-        cp /etc/prometheus/prometheus.yml.tmpl /tmp/prometheus.yml
+        cp /etc/prometheus/prometheus.yml.tmpl /etc/prometheus/prometheus.rendered.yml
         url="$${NETMAKER_METRICS_TARGET}"
         if echo "$${url}" | grep -q '://'; then
           scheme="$${url%%://*}"; target="$${url#*://}"
         else
           scheme="https"; target="$${url}"
         fi
-        sed -i "s~NETMAKER_METRICS_TARGET~$${target}~g" /tmp/prometheus.yml
-        sed -i "s~__SCHEME__~$${scheme}~g" /tmp/prometheus.yml
-        sed -i "s~METRICS_SECRET~$${METRICS_SECRET}~g" /tmp/prometheus.yml
+        sed -i "s~NETMAKER_METRICS_TARGET~$${target}~g" /etc/prometheus/prometheus.rendered.yml
+        sed -i "s~__SCHEME__~$${scheme}~g" /etc/prometheus/prometheus.rendered.yml
+        sed -i "s~METRICS_USERNAME~$${METRICS_USERNAME}~g" /etc/prometheus/prometheus.rendered.yml
+        sed -i "s~METRICS_SECRET~$${METRICS_SECRET}~g" /etc/prometheus/prometheus.rendered.yml
         exec /bin/prometheus \
-          --config.file=/tmp/prometheus.yml \
+          --config.file=/etc/prometheus/prometheus.rendered.yml \
           --storage.tsdb.path=/prometheus \
           --storage.tsdb.retention.time=$${PROMETHEUS_RETENTION_TIME:-15d} \
           --storage.tsdb.retention.size=$${PROMETHEUS_RETENTION_SIZE:-0}

config/config.go

@@ -47,7 +47,7 @@ type ServerConfig struct {
 	MasterKey                  string        `yaml:"masterkey"`
 	DNSKey                     string        `yaml:"dnskey"`
 	AllowedOrigin              string        `yaml:"allowedorigin"`
-	NodeID                     string        `yaml:"nodeid"`
+	HostName                   string        `yaml:"host_name"`
 	RestBackend                string        `yaml:"restbackend"`
 	MessageQueueBackend        string        `yaml:"messagequeuebackend"`
 	DNSMode                    string        `yaml:"dnsmode"`

controllers/gateway.go

@@ -104,40 +104,14 @@ func createGateway(w http.ResponseWriter, r *http.Request) {
 			logic.UpsertHost(host)
 		}
 	}
-	for _, relayedNodeID := range relayNode.RelayedNodes {
-		relayedNode, err := logic.GetNodeByID(relayedNodeID)
-		if err == nil {
-			if relayedNode.FailedOverBy != uuid.Nil {
-				go logic.ResetFailedOverPeer(&relayedNode)
-			}
-			if len(relayedNode.AutoRelayedPeers) > 0 {
-				go logic.ResetAutoRelayedPeer(&relayedNode)
-			}
 
-		}
-	}
 	if len(req.InetNodeClientIDs) > 0 {
 		logic.SetInternetGw(&node, req.InetNodeReq)
-		if servercfg.IsPro {
-			if _, exists := logic.FailOverExists(node.Network); exists {
-				go func() {
-					logic.ResetFailedOverPeer(&node)
-					mq.PublishPeerUpdate(false)
-				}()
-			}
-
-			go func() {
-				logic.ResetAutoRelayedPeer(&node)
-				mq.PublishPeerUpdate(false)
-			}()
-
-		}
 		if node.IsGw && node.IngressDNS == "" {
 			node.IngressDNS = "1.1.1.1"
 		}
 		logic.UpsertNode(&node)
 	}
-
 	logger.Log(
 		1,
 		r.Header.Get("user"),
@@ -168,6 +142,18 @@ func createGateway(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(apiNode)
 	go func() {
+		for _, relayedNodeID := range relayNode.RelayedNodes {
+			relayedNode, err := logic.GetNodeByID(relayedNodeID)
+			if err == nil {
+				if relayedNode.FailedOverBy != uuid.Nil {
+					logic.ResetFailedOverPeer(&relayedNode)
+				}
+				if len(relayedNode.AutoRelayedPeers) > 0 {
+					logic.ResetAutoRelayedPeer(&relayedNode)
+				}
+			}
+		}
+		logic.ResetAutoRelayedPeer(&node)
 		if err := mq.NodeUpdate(&node); err != nil {
 			slog.Error("error publishing node update to node", "node", node.ID, "error", err)
 		}
@@ -352,12 +338,7 @@ func assignGw(w http.ResponseWriter, r *http.Request) {
 		autoAssignGw = false
 	}
 	if autoAssignGw {
-		if node.FailedOverBy != uuid.Nil {
-			go logic.ResetFailedOverPeer(&node)
-		}
-		if len(node.AutoRelayedPeers) > 0 {
-			go logic.ResetAutoRelayedPeer(&node)
-		}
+
 		if node.RelayedBy != "" {
 			gatewayNode, err := logic.GetNodeByID(node.RelayedBy)
 			if err == nil {
@@ -384,6 +365,12 @@ func assignGw(w http.ResponseWriter, r *http.Request) {
 		logic.UpsertNode(&node)
 		logic.GetNodeStatus(&node, false)
 		go func() {
+			if node.FailedOverBy != uuid.Nil {
+				logic.ResetFailedOverPeer(&node)
+			}
+			if len(node.AutoRelayedPeers) > 0 {
+				logic.ResetAutoRelayedPeer(&node)
+			}
 			if err := mq.NodeUpdate(&node); err != nil {
 				slog.Error("error publishing node update to node", "node", node.ID, "error", err)
 			}
@@ -408,18 +395,16 @@ func assignGw(w http.ResponseWriter, r *http.Request) {
 		logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("node %s is not a gateway", nodeid), "badrequest"))
 		return
 	}
-
-	if node.FailedOverBy != uuid.Nil {
-		go logic.ResetFailedOverPeer(&node)
-	}
-	if len(node.AutoRelayedPeers) > 0 {
-		go logic.ResetAutoRelayedPeer(&node)
-	}
 	newNodes := []string{node.ID.String()}
 	newNodes = append(newNodes, gatewayNode.RelayedNodes...)
 	newNodes = logic.UniqueStrings(newNodes)
 	logic.UpdateRelayNodes(gatewayNode.ID.String(), gatewayNode.RelayedNodes, newNodes)
 
+	node, err = logic.GetNodeByID(node.ID.String())
+	if err != nil {
+		logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
+		return
+	}
 	host := &schema.Host{
 		ID: node.HostID,
 	}
@@ -453,6 +438,13 @@ func assignGw(w http.ResponseWriter, r *http.Request) {
 	apiNode := node.ConvertToAPINode()
 
 	go func() {
+
+		if node.FailedOverBy != uuid.Nil {
+			logic.ResetFailedOverPeer(&node)
+		}
+		if len(node.AutoRelayedPeers) > 0 {
+			logic.ResetAutoRelayedPeer(&node)
+		}
 		if err := mq.NodeUpdate(&node); err != nil {
 			slog.Error("error publishing node update to node", "node", node.ID, "error", err)
 		}

controllers/hosts.go

@@ -301,43 +301,34 @@ func pull(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	sendPeerUpdate := false
-	for _, nodeID := range host.Nodes {
-		node, err := logic.GetNodeByID(nodeID)
-		if err != nil {
-			//slog.Error("failed to get node:", "id", node.ID, "error", err)
-			continue
-		}
-		if r.URL.Query().Get("reset_failovered") == "true" {
+	resetFailovered := r.URL.Query().Get("reset_failovered") == "true"
+	if resetFailovered {
+		for _, nodeID := range host.Nodes {
+			node, err := logic.GetNodeByID(nodeID)
+			if err != nil {
+				continue
+			}
 			logic.ResetFailedOverPeer(&node)
 			logic.ResetAutoRelayedPeer(&node)
-			sendPeerUpdate = true
-		}
-	}
-	if sendPeerUpdate {
-		if err := mq.PublishPeerUpdate(false); err != nil {
-			logger.Log(0, "fail to publish peer update: ", err.Error())
 		}
-	}
-	allNodes, err := logic.GetAllNodes()
-	if err != nil {
-		logger.Log(0, "failed to get nodes: ", hostIDStr)
-		logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
-		return
-	}
-	hPU, err := logic.GetPeerUpdateForHost("", host, allNodes, nil, nil)
-	if err != nil {
-		logger.Log(0, "could not pull peers for host", hostIDStr, err.Error())
-		logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
-		return
+		go mq.PublishPeerUpdate(false)
 	}
 
-	portChanged := logic.CheckHostPorts(host)
-	if portChanged {
-		// Save the port change to database immediately to prevent conflicts
-		if err := logic.UpsertHost(host); err != nil {
-			slog.Error("failed to save host port change", "host", host.Name, "error", err)
+	hPU, ok := logic.GetCachedHostPeerUpdate(hostID.String())
+	if !ok || resetFailovered {
+		allNodes, err := logic.GetAllNodes()
+		if err != nil {
+			logger.Log(0, "failed to get nodes: ", hostID.String())
+			logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
+			return
+		}
+		hPU, err = logic.GetPeerUpdateForHost("", host, allNodes, nil, nil)
+		if err != nil {
+			logger.Log(0, "could not pull peers for host", hostID.String(), err.Error())
+			logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
+			return
 		}
+		logic.StoreHostPeerUpdate(hostID.String(), hPU)
 	}
 
 	response := models.HostPull{
@@ -515,10 +506,6 @@ func hostUpdateFallback(w http.ResponseWriter, r *http.Request) {
 	switch hostUpdate.Action {
 	case models.CheckIn:
 		sendPeerUpdate = mq.HandleHostCheckin(&hostUpdate.Host, currentHost)
-		changed := logic.CheckHostPorts(currentHost)
-		if changed {
-			mq.HostUpdate(&models.HostUpdate{Action: models.UpdateHost, Host: *currentHost})
-		}
 	case models.UpdateHost:
 		if hostUpdate.Host.PublicKey != currentHost.PublicKey {
 			//remove old peer entry

controllers/node.go

@@ -666,15 +666,19 @@ func updateNode(w http.ResponseWriter, r *http.Request) {
 		},
 		Origin: schema.Dashboard,
 	})
+	ipChanged := currentNode.Address.String() != newNode.Address.String() ||
+		currentNode.Address6.String() != newNode.Address6.String()
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(apiNode)
 	go func(aclUpdate, relayupdate bool, newNode *models.Node) {
 		if err := mq.NodeUpdate(newNode); err != nil {
 			slog.Error("error publishing node update to node", "node", newNode.ID, "error", err)
 		}
-		// if !newNode.Connected {
-		// 	mq.HostUpdate(&models.HostUpdate{Host: *host, Action: models.SignalPull})
-		// }
+		if ipChanged {
+			if err := mq.HostUpdate(&models.HostUpdate{Action: models.RequestPull, Host: *host}); err != nil {
+				slog.Error("error sending sync pull to host on ip change", "host", host.ID, "error", err)
+			}
+		}
 		allNodes, err := logic.GetAllNodes()
 		if err == nil {
 			mq.PublishSingleHostPeerUpdate(host, allNodes, nil, nil, false, nil)

database/database.go

@@ -2,7 +2,6 @@ package database
 
 import (
 	"errors"
-	"sync"
 	"time"
 
 	"github.com/gravitl/netmaker/logger"
@@ -98,8 +97,6 @@ const (
 	isConnected = "isconnected"
 )
 
-var dbMutex sync.RWMutex
-
 var Tables = []string{
 	NODES_TABLE_NAME,
 	CERTS_TABLE_NAME,
@@ -176,8 +173,6 @@ func CreateTable(tableName string) error {
 
 // Insert - inserts object into db
 func Insert(key string, value string, tableName string) error {
-	dbMutex.Lock()
-	defer dbMutex.Unlock()
 	if key != "" && value != "" {
 		return getCurrentDB()[INSERT].(func(string, string, string) error)(key, value, tableName)
 	} else {
@@ -187,37 +182,25 @@ func Insert(key string, value string, tableName string) error {
 
 // DeleteRecord - deletes a record from db
 func DeleteRecord(tableName string, key string) error {
-	dbMutex.Lock()
-	defer dbMutex.Unlock()
 	return getCurrentDB()[DELETE].(func(string, string) error)(tableName, key)
 }
 
 // DeleteAllRecords - removes a table and remakes
 func DeleteAllRecords(tableName string) error {
-	dbMutex.Lock()
-	defer dbMutex.Unlock()
 	err := getCurrentDB()[DELETE_ALL].(func(string) error)(tableName)
 	if err != nil {
 		return err
 	}
-	err = CreateTable(tableName)
-	if err != nil {
-		return err
-	}
-	return nil
+	return CreateTable(tableName)
 }
 
 // FetchRecord - fetches a single record by key
 func FetchRecord(tableName string, key string) (string, error) {
-	dbMutex.RLock()
-	defer dbMutex.RUnlock()
 	return getCurrentDB()[FETCH_ONE].(func(string, string) (string, error))(tableName, key)
 }
 
 // FetchRecords - fetches all records in given table
 func FetchRecords(tableName string) (map[string]string, error) {
-	dbMutex.RLock()
-	defer dbMutex.RUnlock()
 	return getCurrentDB()[FETCH_ALL].(func(string) (map[string]string, error))(tableName)
 }
 

db/sqlite.go

@@ -49,7 +49,8 @@ func (s *sqliteConnector) connect() (*gorm.DB, error) {
 		}
 	}
 
-	db, err := gorm.Open(sqlite.Open(dbFilePath), &gorm.Config{
+	dsn := dbFilePath + "?_journal_mode=WAL&_busy_timeout=5000"
+	db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{
 		Logger: logger.Default.LogMode(logger.Silent),
 	})
 	if err != nil {
@@ -61,6 +62,7 @@ func (s *sqliteConnector) connect() (*gorm.DB, error) {
 		return nil, err
 	}
 
+	sqlDB.SetMaxOpenConns(1)
 	sqlDB.SetMaxIdleConns(1)
 
 	return db, nil

go.mod

@@ -46,6 +46,7 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/goombaio/namegenerator v0.0.0-20181006234301-989e774b106e
 	github.com/guumaster/tablewriter v0.0.10
+	github.com/klauspost/compress v1.18.3
 	github.com/matryer/is v1.4.1
 	github.com/okta/okta-sdk-golang/v5 v5.0.6
 	github.com/pquerna/otp v1.5.0
@@ -91,7 +92,6 @@ require (
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/kelseyhightower/envconfig v1.4.0 // indirect
-	github.com/klauspost/compress v1.18.3 // indirect
 	github.com/lestrrat-go/backoff/v2 v2.0.8 // indirect
 	github.com/lestrrat-go/blackmagic v1.0.2 // indirect
 	github.com/lestrrat-go/httpcc v1.0.1 // indirect

logic/extpeers.go

@@ -420,13 +420,11 @@ func SaveExtClient(extclient *models.ExtClient) error {
 	}
 	if servercfg.CacheEnabled() {
 		storeExtClientInCache(key, *extclient)
-		if _, ok := allocatedIpMap[extclient.Network]; ok {
-			if extclient.Address != "" {
-				AddIpToAllocatedIpMap(extclient.Network, net.ParseIP(extclient.Address))
-			}
-			if extclient.Address6 != "" {
-				AddIpToAllocatedIpMap(extclient.Network, net.ParseIP(extclient.Address6))
-			}
+		if extclient.Address != "" {
+			AddIpToAllocatedIpMap(extclient.Network, net.ParseIP(extclient.Address))
+		}
+		if extclient.Address6 != "" {
+			AddIpToAllocatedIpMap(extclient.Network, net.ParseIP(extclient.Address6))
 		}
 	}