2022-02-17 04:40:51 +08:00
package mq
import (
"encoding/json"
2022-09-14 03:25:56 +08:00
"errors"
2022-02-17 04:40:51 +08:00
"fmt"
2022-09-14 03:25:56 +08:00
"time"
2022-02-17 04:40:51 +08:00
2022-12-28 00:26:55 +08:00
proxy_models "github.com/gravitl/netclient/nmproxy/models"
2022-02-17 04:40:51 +08:00
"github.com/gravitl/netmaker/logger"
"github.com/gravitl/netmaker/logic"
"github.com/gravitl/netmaker/models"
"github.com/gravitl/netmaker/servercfg"
2022-02-23 05:14:23 +08:00
"github.com/gravitl/netmaker/serverctl"
2022-02-17 04:40:51 +08:00
)
2023-01-06 23:57:40 +08:00
// PublishPeerUpdate --- determines and publishes a peer update to all the hosts
2023-01-10 13:19:15 +08:00
func PublishPeerUpdate ( ) error {
2022-02-17 04:40:51 +08:00
if ! servercfg . IsMessageQueueBackend ( ) {
return nil
}
2023-01-04 12:35:05 +08:00
2023-01-06 19:28:41 +08:00
hosts , err := logic . GetAllHosts ( )
2022-02-17 04:40:51 +08:00
if err != nil {
2023-01-04 12:35:05 +08:00
logger . Log ( 1 , "err getting all hosts" , err . Error ( ) )
2022-02-17 04:40:51 +08:00
return err
}
2023-01-04 12:35:05 +08:00
for _ , host := range hosts {
err = PublishSingleHostUpdate ( & host )
2022-02-17 04:40:51 +08:00
if err != nil {
2023-01-04 12:35:05 +08:00
logger . Log ( 1 , "failed to publish peer update to host" , host . ID . String ( ) , ": " , err . Error ( ) )
2022-02-17 04:40:51 +08:00
}
}
2022-09-30 02:49:44 +08:00
return err
}
2023-01-06 23:57:40 +08:00
// PublishSingleHostUpdate --- determines and publishes a peer update to one host
2023-01-04 12:35:05 +08:00
func PublishSingleHostUpdate ( host * models . Host ) error {
2022-11-20 12:05:29 +08:00
2023-01-04 12:35:05 +08:00
peerUpdate , err := logic . GetPeerUpdateForHost ( host )
2022-09-30 02:49:44 +08:00
if err != nil {
return err
}
2022-12-22 19:42:33 +08:00
if host . ProxyEnabled {
2023-01-08 21:20:02 +08:00
proxyUpdate , err := logic . GetProxyUpdateForHost ( host )
if err != nil {
return err
}
proxyUpdate . Action = proxy_models . ProxyUpdate
peerUpdate . ProxyUpdate = proxyUpdate
2022-11-07 03:55:04 +08:00
}
2022-11-20 12:05:29 +08:00
2022-09-30 02:49:44 +08:00
data , err := json . Marshal ( & peerUpdate )
if err != nil {
return err
}
2023-01-04 13:33:13 +08:00
return publish ( host , fmt . Sprintf ( "peers/host/%s/%s" , host . ID . String ( ) , servercfg . GetServer ( ) ) , data )
2022-02-17 04:40:51 +08:00
}
// PublishPeerUpdate --- publishes a peer update to all the peers of a node
2022-12-21 04:29:09 +08:00
func PublishExtPeerUpdate ( node * models . Node ) error {
2022-12-13 13:15:57 +08:00
2023-01-10 13:19:15 +08:00
go PublishPeerUpdate ( )
2022-03-28 21:00:02 +08:00
return nil
2022-02-17 04:40:51 +08:00
}
// NodeUpdate -- publishes a node update
2022-12-21 04:29:09 +08:00
func NodeUpdate ( node * models . Node ) error {
2022-12-24 02:41:17 +08:00
host , err := logic . GetHost ( node . HostID . String ( ) )
2022-12-22 19:42:33 +08:00
if err != nil {
return nil
}
2022-12-21 04:29:09 +08:00
if ! servercfg . IsMessageQueueBackend ( ) {
2022-02-17 04:40:51 +08:00
return nil
}
2022-12-21 04:29:09 +08:00
logger . Log ( 3 , "publishing node update to " + node . ID . String ( ) )
2022-09-02 16:52:28 +08:00
2022-12-21 04:29:09 +08:00
//if len(node.NetworkSettings.AccessKeys) > 0 {
//node.NetworkSettings.AccessKeys = []models.AccessKey{} // not to be sent (don't need to spread access keys around the network; we need to know how to reach other nodes, not become them)
//}
2022-09-02 16:52:28 +08:00
2022-12-02 20:05:41 +08:00
data , err := json . Marshal ( node )
if err != nil {
logger . Log ( 2 , "error marshalling node update " , err . Error ( ) )
return err
}
2023-01-04 12:35:05 +08:00
if err = publish ( host , fmt . Sprintf ( "update/%s/%s" , node . Network , node . ID ) , data ) ; err != nil {
2022-12-21 04:29:09 +08:00
logger . Log ( 2 , "error publishing node update to peer " , node . ID . String ( ) , err . Error ( ) )
2022-12-02 20:05:41 +08:00
return err
}
2022-11-20 12:05:29 +08:00
2022-02-17 04:40:51 +08:00
return nil
}
2023-01-17 13:47:13 +08:00
// HostUpdate -- publishes a host update to clients
func HostUpdate ( hostUpdate * models . HostUpdate ) error {
2023-01-11 23:33:43 +08:00
if ! servercfg . IsMessageQueueBackend ( ) {
return nil
}
2023-01-17 13:47:13 +08:00
logger . Log ( 3 , "publishing host update to " + hostUpdate . Host . ID . String ( ) )
2023-01-11 23:33:43 +08:00
2023-01-17 13:47:13 +08:00
data , err := json . Marshal ( hostUpdate )
2023-01-11 23:33:43 +08:00
if err != nil {
logger . Log ( 2 , "error marshalling node update " , err . Error ( ) )
return err
}
2023-01-17 13:47:13 +08:00
if err = publish ( & hostUpdate . Host , fmt . Sprintf ( "host/update/%s/%s" , hostUpdate . Host . ID . String ( ) , servercfg . GetServer ( ) ) , data ) ; err != nil {
logger . Log ( 2 , "error publishing host update to" , hostUpdate . Host . ID . String ( ) , err . Error ( ) )
2023-01-11 23:33:43 +08:00
return err
}
return nil
}
2022-02-17 04:40:51 +08:00
// sendPeers - retrieve networks, send peer ports to all peers
func sendPeers ( ) {
2022-02-23 03:53:40 +08:00
2023-01-06 20:22:17 +08:00
hosts , err := logic . GetAllHosts ( )
2022-09-14 03:25:56 +08:00
if err != nil {
logger . Log ( 1 , "error retrieving networks for keepalive" , err . Error ( ) )
}
2022-02-17 04:40:51 +08:00
var force bool
peer_force_send ++
if peer_force_send == 5 {
2022-03-21 23:52:00 +08:00
// run iptables update to ensure gateways work correctly and mq is forwarded if containerized
if servercfg . ManageIPTables ( ) != "off" {
2022-03-28 22:36:23 +08:00
serverctl . InitIPTables ( false )
2022-03-21 23:52:00 +08:00
}
2022-05-27 22:52:47 +08:00
servercfg . SetHost ( )
2022-03-21 23:52:00 +08:00
2022-02-17 04:40:51 +08:00
force = true
peer_force_send = 0
2022-02-23 03:53:40 +08:00
err := logic . TimerCheckpoint ( ) // run telemetry & log dumps if 24 hours has passed..
if err != nil {
logger . Log ( 3 , "error occurred on timer," , err . Error ( ) )
}
2022-09-14 03:25:56 +08:00
2022-12-21 04:29:09 +08:00
//collectServerMetrics(networks[:])
2022-02-17 04:40:51 +08:00
}
2022-02-23 03:53:40 +08:00
2023-01-06 20:22:17 +08:00
for _ , host := range hosts {
2022-12-21 04:29:09 +08:00
if force {
logger . Log ( 2 , "sending scheduled peer update (5 min)" )
2023-01-06 20:22:17 +08:00
err = PublishSingleHostUpdate ( & host )
2022-12-21 04:29:09 +08:00
if err != nil {
2023-01-06 20:22:17 +08:00
logger . Log ( 1 , "error publishing peer updates for host: " , host . ID . String ( ) , " Err: " , err . Error ( ) )
2022-11-20 12:05:29 +08:00
}
2022-02-17 04:40:51 +08:00
}
}
}
2022-04-25 23:12:49 +08:00
// ServerStartNotify - notifies all non server nodes to pull changes after a restart
func ServerStartNotify ( ) error {
nodes , err := logic . GetAllNodes ( )
if err != nil {
return err
}
for i := range nodes {
nodes [ i ] . Action = models . NODE_FORCE_UPDATE
if err = NodeUpdate ( & nodes [ i ] ) ; err != nil {
2022-12-21 04:29:09 +08:00
logger . Log ( 1 , "error when notifying node" , nodes [ i ] . ID . String ( ) , "of a server startup" )
2022-04-25 23:12:49 +08:00
}
}
return nil
}
2022-09-14 03:25:56 +08:00
// function to collect and store metrics for server nodes
2022-12-21 04:29:09 +08:00
//func collectServerMetrics(networks []models.Network) {
// if !servercfg.Is_EE {
// return
// }
// if len(networks) > 0 {
// for i := range networks {
// currentNetworkNodes, err := logic.GetNetworkNodes(networks[i].NetID)
// if err != nil {
// continue
// }
// currentServerNodes := logic.GetServerNodes(networks[i].NetID)
// if len(currentServerNodes) > 0 {
// for i := range currentServerNodes {
// if logic.IsLocalServer(¤tServerNodes[i]) {
// serverMetrics := logic.CollectServerMetrics(currentServerNodes[i].ID, currentNetworkNodes)
// if serverMetrics != nil {
// serverMetrics.NodeName = currentServerNodes[i].Name
// serverMetrics.NodeID = currentServerNodes[i].ID
// serverMetrics.IsServer = "yes"
// serverMetrics.Network = currentServerNodes[i].Network
// if err = metrics.GetExchangedBytesForNode(¤tServerNodes[i], serverMetrics); err != nil {
// logger.Log(1, fmt.Sprintf("failed to update exchanged bytes info for server: %s, err: %v",
// currentServerNodes[i].Name, err))
// }
// updateNodeMetrics(¤tServerNodes[i], serverMetrics)
// if err = logic.UpdateMetrics(currentServerNodes[i].ID, serverMetrics); err != nil {
// logger.Log(1, "failed to update metrics for server node", currentServerNodes[i].ID)
// }
// if servercfg.IsMetricsExporter() {
// logger.Log(2, "-------------> SERVER METRICS: ", fmt.Sprintf("%+v", serverMetrics))
// if err := pushMetricsToExporter(*serverMetrics); err != nil {
// logger.Log(2, "failed to push server metrics to exporter: ", err.Error())
// }
// }
// }
// }
// }
// }
// }
// }
//}
2022-09-14 03:25:56 +08:00
func pushMetricsToExporter ( metrics models . Metrics ) error {
logger . Log ( 2 , "----> Pushing metrics to exporter" )
data , err := json . Marshal ( metrics )
if err != nil {
return errors . New ( "failed to marshal metrics: " + err . Error ( ) )
}
2022-09-30 21:01:57 +08:00
if token := mqclient . Publish ( "metrics_exporter" , 2 , true , data ) ; ! token . WaitTimeout ( MQ_TIMEOUT * time . Second ) || token . Error ( ) != nil {
2022-09-14 03:25:56 +08:00
var err error
if token . Error ( ) == nil {
err = errors . New ( "connection timeout" )
} else {
err = token . Error ( )
}
return err
}
return nil
}