handlers.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350
  1. package mq
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "time"
  6. mqtt "github.com/eclipse/paho.mqtt.golang"
  7. "github.com/gravitl/netmaker/database"
  8. "github.com/gravitl/netmaker/logger"
  9. "github.com/gravitl/netmaker/logic"
  10. "github.com/gravitl/netmaker/models"
  11. "github.com/gravitl/netmaker/netclient/ncutils"
  12. "github.com/gravitl/netmaker/servercfg"
  13. )
  14. // DefaultHandler default message queue handler -- NOT USED
  15. func DefaultHandler(client mqtt.Client, msg mqtt.Message) {
  16. logger.Log(0, "MQTT Message: Topic: ", string(msg.Topic()), " Message: ", string(msg.Payload()))
  17. }
  18. // Ping message Handler -- handles ping topic from client nodes
  19. func Ping(client mqtt.Client, msg mqtt.Message) {
  20. go func() {
  21. id, err := getID(msg.Topic())
  22. if err != nil {
  23. logger.Log(0, "error getting node.ID sent on ping topic ")
  24. return
  25. }
  26. node, err := logic.GetNodeByID(id)
  27. if err != nil {
  28. logger.Log(0, "mq-ping error getting node: ", err.Error())
  29. record, err := database.FetchRecord(database.NODES_TABLE_NAME, id)
  30. if err != nil {
  31. logger.Log(0, "error reading database ", err.Error())
  32. return
  33. }
  34. logger.Log(0, "record from database")
  35. logger.Log(0, record)
  36. return
  37. }
  38. decrypted, decryptErr := decryptMsg(&node, msg.Payload())
  39. if decryptErr != nil {
  40. logger.Log(0, "error decrypting when updating node ", node.ID.String(), decryptErr.Error())
  41. return
  42. }
  43. var checkin models.NodeCheckin
  44. if err := json.Unmarshal(decrypted, &checkin); err != nil {
  45. logger.Log(1, "error unmarshaling payload ", err.Error())
  46. return
  47. }
  48. host, err := logic.GetHost(node.HostID.String())
  49. if err != nil {
  50. logger.Log(0, "error retrieving host for node ", node.ID.String(), err.Error())
  51. return
  52. }
  53. node.SetLastCheckIn()
  54. host.Version = checkin.Version
  55. node.Connected = checkin.Connected
  56. host.Interfaces = checkin.Ifaces
  57. for i := range host.Interfaces {
  58. host.Interfaces[i].AddressString = host.Interfaces[i].Address.String()
  59. }
  60. if err := logic.UpdateNode(&node, &node); err != nil {
  61. logger.Log(0, "error updating node", node.ID.String(), " on checkin", err.Error())
  62. return
  63. }
  64. logger.Log(3, "ping processed for node", node.ID.String())
  65. // --TODO --set client version once feature is implemented.
  66. //node.SetClientVersion(msg.Payload())
  67. }()
  68. }
  69. // UpdateNode message Handler -- handles updates from client nodes
  70. func UpdateNode(client mqtt.Client, msg mqtt.Message) {
  71. go func() {
  72. id, err := getID(msg.Topic())
  73. if err != nil {
  74. logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
  75. return
  76. }
  77. currentNode, err := logic.GetNodeByID(id)
  78. if err != nil {
  79. logger.Log(1, "error getting node ", id, err.Error())
  80. return
  81. }
  82. decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
  83. if decryptErr != nil {
  84. logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error())
  85. return
  86. }
  87. var oldNode models.LegacyNode
  88. if err := json.Unmarshal(decrypted, &oldNode); err != nil {
  89. logger.Log(1, "error unmarshaling payload ", err.Error())
  90. return
  91. }
  92. _, newNode := oldNode.ConvertToNewNode()
  93. ifaceDelta := logic.IfaceDelta(&currentNode, newNode)
  94. if servercfg.Is_EE && ifaceDelta {
  95. if err = logic.EnterpriseResetAllPeersFailovers(currentNode.ID.String(), currentNode.Network); err != nil {
  96. logger.Log(1, "failed to reset failover list during node update", currentNode.ID.String(), currentNode.Network)
  97. }
  98. }
  99. newNode.SetLastCheckIn()
  100. if err := logic.UpdateNode(&currentNode, newNode); err != nil {
  101. logger.Log(1, "error saving node", err.Error())
  102. return
  103. }
  104. if ifaceDelta { // reduce number of unneeded updates, by only sending on iface changes
  105. if err = PublishPeerUpdate(currentNode.Network, true); err != nil {
  106. logger.Log(0, "error updating peers when node", currentNode.ID.String(), "informed the server of an interface change", err.Error())
  107. }
  108. }
  109. logger.Log(1, "updated node", id, newNode.ID.String())
  110. }()
  111. }
  112. // UpdateHost message Handler -- handles updates from client hosts
  113. func UpdateHost(client mqtt.Client, msg mqtt.Message) {
  114. go func() {
  115. id, err := getID(msg.Topic())
  116. if err != nil {
  117. logger.Log(1, "error getting host.ID sent on ", msg.Topic(), err.Error())
  118. return
  119. }
  120. currentHost, err := logic.GetHost(id)
  121. if err != nil {
  122. logger.Log(1, "error getting node ", id, err.Error())
  123. return
  124. }
  125. decrypted, decryptErr := decryptMsgWithHost(currentHost, msg.Payload())
  126. if decryptErr != nil {
  127. logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error())
  128. return
  129. }
  130. var newHost models.Host
  131. if err := json.Unmarshal(decrypted, &newHost); err != nil {
  132. logger.Log(1, "error unmarshaling payload ", err.Error())
  133. return
  134. }
  135. // ifaceDelta := logic.IfaceDelta(&currentHost, newNode)
  136. // if servercfg.Is_EE && ifaceDelta {
  137. // if err = logic.EnterpriseResetAllPeersFailovers(currentHost.ID.String(), currentHost.Network); err != nil {
  138. // logger.Log(1, "failed to reset failover list during node update", currentHost.ID.String(), currentHost.Network)
  139. // }
  140. // }
  141. logic.UpdateHost(&newHost, currentHost)
  142. if err := logic.UpsertHost(&newHost); err != nil {
  143. logger.Log(1, "error saving host", err.Error())
  144. return
  145. }
  146. logger.Log(1, "updated host", newHost.ID.String())
  147. }()
  148. }
  149. // UpdateMetrics message Handler -- handles updates from client nodes for metrics
  150. func UpdateMetrics(client mqtt.Client, msg mqtt.Message) {
  151. if servercfg.Is_EE {
  152. go func() {
  153. id, err := getID(msg.Topic())
  154. if err != nil {
  155. logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
  156. return
  157. }
  158. currentNode, err := logic.GetNodeByID(id)
  159. if err != nil {
  160. logger.Log(1, "error getting node ", id, err.Error())
  161. return
  162. }
  163. decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
  164. if decryptErr != nil {
  165. logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error())
  166. return
  167. }
  168. var newMetrics models.Metrics
  169. if err := json.Unmarshal(decrypted, &newMetrics); err != nil {
  170. logger.Log(1, "error unmarshaling payload ", err.Error())
  171. return
  172. }
  173. shouldUpdate := updateNodeMetrics(&currentNode, &newMetrics)
  174. if err = logic.UpdateMetrics(id, &newMetrics); err != nil {
  175. logger.Log(1, "faield to update node metrics", id, err.Error())
  176. return
  177. }
  178. if servercfg.IsMetricsExporter() {
  179. if err := pushMetricsToExporter(newMetrics); err != nil {
  180. logger.Log(2, fmt.Sprintf("failed to push node: [%s] metrics to exporter, err: %v",
  181. currentNode.ID, err))
  182. }
  183. }
  184. if newMetrics.Connectivity != nil {
  185. err := logic.EnterpriseFailoverFunc(&currentNode)
  186. if err != nil {
  187. logger.Log(0, "failed to failover for node", currentNode.ID.String(), "on network", currentNode.Network, "-", err.Error())
  188. }
  189. }
  190. if shouldUpdate {
  191. logger.Log(2, "updating peers after node", currentNode.ID.String(), currentNode.Network, "detected connectivity issues")
  192. host, err := logic.GetHost(currentNode.HostID.String())
  193. if err == nil {
  194. if err = PublishSingleHostUpdate(host); err != nil {
  195. logger.Log(0, "failed to publish update after failover peer change for node", currentNode.ID.String(), currentNode.Network)
  196. }
  197. }
  198. }
  199. logger.Log(1, "updated node metrics", id)
  200. }()
  201. }
  202. }
  203. // ClientPeerUpdate message handler -- handles updating peers after signal from client nodes
  204. func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) {
  205. go func() {
  206. id, err := getID(msg.Topic())
  207. if err != nil {
  208. logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
  209. return
  210. }
  211. currentNode, err := logic.GetNodeByID(id)
  212. if err != nil {
  213. logger.Log(1, "error getting node ", id, err.Error())
  214. return
  215. }
  216. decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
  217. if decryptErr != nil {
  218. logger.Log(1, "failed to decrypt message during client peer update for node ", id, decryptErr.Error())
  219. return
  220. }
  221. switch decrypted[0] {
  222. case ncutils.ACK:
  223. //do we still need this
  224. case ncutils.DONE:
  225. updateNodePeers(&currentNode)
  226. }
  227. logger.Log(1, "sent peer updates after signal received from", id)
  228. }()
  229. }
  230. func updateNodePeers(currentNode *models.Node) {
  231. if err := PublishPeerUpdate(currentNode.Network, false); err != nil {
  232. logger.Log(1, "error publishing peer update ", err.Error())
  233. return
  234. }
  235. }
  236. func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) bool {
  237. if newMetrics.FailoverPeers == nil {
  238. newMetrics.FailoverPeers = make(map[string]string)
  239. }
  240. oldMetrics, err := logic.GetMetrics(currentNode.ID.String())
  241. if err != nil {
  242. logger.Log(1, "error finding old metrics for node", currentNode.ID.String())
  243. return false
  244. }
  245. if oldMetrics.FailoverPeers == nil {
  246. oldMetrics.FailoverPeers = make(map[string]string)
  247. }
  248. var attachedClients []models.ExtClient
  249. if currentNode.IsIngressGateway {
  250. clients, err := logic.GetExtClientsByID(currentNode.ID.String(), currentNode.Network)
  251. if err == nil {
  252. attachedClients = clients
  253. }
  254. }
  255. if len(attachedClients) > 0 {
  256. // associate ext clients with IDs
  257. for i := range attachedClients {
  258. extMetric := newMetrics.Connectivity[attachedClients[i].PublicKey]
  259. if len(extMetric.NodeName) == 0 &&
  260. len(newMetrics.Connectivity[attachedClients[i].ClientID].NodeName) > 0 { // cover server clients
  261. extMetric = newMetrics.Connectivity[attachedClients[i].ClientID]
  262. if extMetric.TotalReceived > 0 && extMetric.TotalSent > 0 {
  263. extMetric.Connected = true
  264. }
  265. }
  266. extMetric.NodeName = attachedClients[i].ClientID
  267. extMetric.IsServer = "no"
  268. delete(newMetrics.Connectivity, attachedClients[i].PublicKey)
  269. newMetrics.Connectivity[attachedClients[i].ClientID] = extMetric
  270. }
  271. }
  272. // run through metrics for each peer
  273. for k := range newMetrics.Connectivity {
  274. currMetric := newMetrics.Connectivity[k]
  275. oldMetric := oldMetrics.Connectivity[k]
  276. currMetric.TotalTime += oldMetric.TotalTime
  277. currMetric.Uptime += oldMetric.Uptime // get the total uptime for this connection
  278. if currMetric.Uptime == 0 || currMetric.TotalTime == 0 {
  279. currMetric.PercentUp = 0
  280. } else {
  281. currMetric.PercentUp = 100.0 * (float64(currMetric.Uptime) / float64(currMetric.TotalTime))
  282. }
  283. totalUpMinutes := currMetric.Uptime * ncutils.CheckInInterval
  284. currMetric.ActualUptime = time.Duration(totalUpMinutes) * time.Minute
  285. delete(oldMetrics.Connectivity, k) // remove from old data
  286. newMetrics.Connectivity[k] = currMetric
  287. if oldProxyMetric, ok := oldMetrics.ProxyMetrics[k]; ok {
  288. newProxyMetric := newMetrics.ProxyMetrics[k]
  289. newProxyMetric.TrafficSent += oldProxyMetric.TrafficSent
  290. newProxyMetric.TrafficRecieved += oldProxyMetric.TrafficRecieved
  291. newMetrics.ProxyMetrics[k] = newProxyMetric
  292. }
  293. }
  294. // add nodes that need failover
  295. nodes, err := logic.GetNetworkNodes(currentNode.Network)
  296. if err != nil {
  297. logger.Log(0, "failed to retrieve nodes while updating metrics")
  298. return false
  299. }
  300. for _, node := range nodes {
  301. if !newMetrics.Connectivity[node.ID.String()].Connected &&
  302. len(newMetrics.Connectivity[node.ID.String()].NodeName) > 0 &&
  303. node.Connected == true &&
  304. len(node.FailoverNode) > 0 &&
  305. !node.Failover {
  306. newMetrics.FailoverPeers[node.ID.String()] = node.FailoverNode.String()
  307. }
  308. }
  309. shouldUpdate := len(oldMetrics.FailoverPeers) == 0 && len(newMetrics.FailoverPeers) > 0
  310. for k, v := range oldMetrics.FailoverPeers {
  311. if len(newMetrics.FailoverPeers[k]) > 0 && len(v) == 0 {
  312. shouldUpdate = true
  313. }
  314. if len(v) > 0 && len(newMetrics.FailoverPeers[k]) == 0 {
  315. newMetrics.FailoverPeers[k] = v
  316. }
  317. }
  318. for k := range oldMetrics.Connectivity { // cleanup any left over data, self healing
  319. delete(newMetrics.Connectivity, k)
  320. }
  321. return shouldUpdate
  322. }