handlers.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311
  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. // UpdateMetrics message Handler -- handles updates from client nodes for metrics
  113. func UpdateMetrics(client mqtt.Client, msg mqtt.Message) {
  114. if servercfg.Is_EE {
  115. go func() {
  116. id, err := getID(msg.Topic())
  117. if err != nil {
  118. logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
  119. return
  120. }
  121. currentNode, err := logic.GetNodeByID(id)
  122. if err != nil {
  123. logger.Log(1, "error getting node ", id, err.Error())
  124. return
  125. }
  126. decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
  127. if decryptErr != nil {
  128. logger.Log(1, "failed to decrypt message for node ", id, decryptErr.Error())
  129. return
  130. }
  131. var newMetrics models.Metrics
  132. if err := json.Unmarshal(decrypted, &newMetrics); err != nil {
  133. logger.Log(1, "error unmarshaling payload ", err.Error())
  134. return
  135. }
  136. shouldUpdate := updateNodeMetrics(&currentNode, &newMetrics)
  137. if err = logic.UpdateMetrics(id, &newMetrics); err != nil {
  138. logger.Log(1, "faield to update node metrics", id, err.Error())
  139. return
  140. }
  141. if servercfg.IsMetricsExporter() {
  142. if err := pushMetricsToExporter(newMetrics); err != nil {
  143. logger.Log(2, fmt.Sprintf("failed to push node: [%s] metrics to exporter, err: %v",
  144. currentNode.ID, err))
  145. }
  146. }
  147. if newMetrics.Connectivity != nil {
  148. err := logic.EnterpriseFailoverFunc(&currentNode)
  149. if err != nil {
  150. logger.Log(0, "failed to failover for node", currentNode.ID.String(), "on network", currentNode.Network, "-", err.Error())
  151. }
  152. }
  153. if shouldUpdate {
  154. logger.Log(2, "updating peers after node", currentNode.ID.String(), currentNode.Network, "detected connectivity issues")
  155. host, err := logic.GetHost(currentNode.HostID.String())
  156. if err == nil {
  157. if err = PublishSingleHostUpdate(host); err != nil {
  158. logger.Log(0, "failed to publish update after failover peer change for node", currentNode.ID.String(), currentNode.Network)
  159. }
  160. }
  161. }
  162. logger.Log(1, "updated node metrics", id)
  163. }()
  164. }
  165. }
  166. // ClientPeerUpdate message handler -- handles updating peers after signal from client nodes
  167. func ClientPeerUpdate(client mqtt.Client, msg mqtt.Message) {
  168. go func() {
  169. id, err := getID(msg.Topic())
  170. if err != nil {
  171. logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
  172. return
  173. }
  174. currentNode, err := logic.GetNodeByID(id)
  175. if err != nil {
  176. logger.Log(1, "error getting node ", id, err.Error())
  177. return
  178. }
  179. decrypted, decryptErr := decryptMsg(&currentNode, msg.Payload())
  180. if decryptErr != nil {
  181. logger.Log(1, "failed to decrypt message during client peer update for node ", id, decryptErr.Error())
  182. return
  183. }
  184. switch decrypted[0] {
  185. case ncutils.ACK:
  186. //do we still need this
  187. case ncutils.DONE:
  188. updateNodePeers(&currentNode)
  189. }
  190. logger.Log(1, "sent peer updates after signal received from", id)
  191. }()
  192. }
  193. func updateNodePeers(currentNode *models.Node) {
  194. if err := PublishPeerUpdate(currentNode.Network, false); err != nil {
  195. logger.Log(1, "error publishing peer update ", err.Error())
  196. return
  197. }
  198. }
  199. func updateNodeMetrics(currentNode *models.Node, newMetrics *models.Metrics) bool {
  200. if newMetrics.FailoverPeers == nil {
  201. newMetrics.FailoverPeers = make(map[string]string)
  202. }
  203. oldMetrics, err := logic.GetMetrics(currentNode.ID.String())
  204. if err != nil {
  205. logger.Log(1, "error finding old metrics for node", currentNode.ID.String())
  206. return false
  207. }
  208. if oldMetrics.FailoverPeers == nil {
  209. oldMetrics.FailoverPeers = make(map[string]string)
  210. }
  211. var attachedClients []models.ExtClient
  212. if currentNode.IsIngressGateway {
  213. clients, err := logic.GetExtClientsByID(currentNode.ID.String(), currentNode.Network)
  214. if err == nil {
  215. attachedClients = clients
  216. }
  217. }
  218. if len(attachedClients) > 0 {
  219. // associate ext clients with IDs
  220. for i := range attachedClients {
  221. extMetric := newMetrics.Connectivity[attachedClients[i].PublicKey]
  222. if len(extMetric.NodeName) == 0 &&
  223. len(newMetrics.Connectivity[attachedClients[i].ClientID].NodeName) > 0 { // cover server clients
  224. extMetric = newMetrics.Connectivity[attachedClients[i].ClientID]
  225. if extMetric.TotalReceived > 0 && extMetric.TotalSent > 0 {
  226. extMetric.Connected = true
  227. }
  228. }
  229. extMetric.NodeName = attachedClients[i].ClientID
  230. extMetric.IsServer = "no"
  231. delete(newMetrics.Connectivity, attachedClients[i].PublicKey)
  232. newMetrics.Connectivity[attachedClients[i].ClientID] = extMetric
  233. }
  234. }
  235. // run through metrics for each peer
  236. for k := range newMetrics.Connectivity {
  237. currMetric := newMetrics.Connectivity[k]
  238. oldMetric := oldMetrics.Connectivity[k]
  239. currMetric.TotalTime += oldMetric.TotalTime
  240. currMetric.Uptime += oldMetric.Uptime // get the total uptime for this connection
  241. if currMetric.Uptime == 0 || currMetric.TotalTime == 0 {
  242. currMetric.PercentUp = 0
  243. } else {
  244. currMetric.PercentUp = 100.0 * (float64(currMetric.Uptime) / float64(currMetric.TotalTime))
  245. }
  246. totalUpMinutes := currMetric.Uptime * ncutils.CheckInInterval
  247. currMetric.ActualUptime = time.Duration(totalUpMinutes) * time.Minute
  248. delete(oldMetrics.Connectivity, k) // remove from old data
  249. newMetrics.Connectivity[k] = currMetric
  250. if oldProxyMetric, ok := oldMetrics.ProxyMetrics[k]; ok {
  251. newProxyMetric := newMetrics.ProxyMetrics[k]
  252. newProxyMetric.TrafficSent += oldProxyMetric.TrafficSent
  253. newProxyMetric.TrafficRecieved += oldProxyMetric.TrafficRecieved
  254. newMetrics.ProxyMetrics[k] = newProxyMetric
  255. }
  256. }
  257. // add nodes that need failover
  258. nodes, err := logic.GetNetworkNodes(currentNode.Network)
  259. if err != nil {
  260. logger.Log(0, "failed to retrieve nodes while updating metrics")
  261. return false
  262. }
  263. for _, node := range nodes {
  264. if !newMetrics.Connectivity[node.ID.String()].Connected &&
  265. len(newMetrics.Connectivity[node.ID.String()].NodeName) > 0 &&
  266. node.Connected == true &&
  267. len(node.FailoverNode) > 0 &&
  268. !node.Failover {
  269. newMetrics.FailoverPeers[node.ID.String()] = node.FailoverNode.String()
  270. }
  271. }
  272. shouldUpdate := len(oldMetrics.FailoverPeers) == 0 && len(newMetrics.FailoverPeers) > 0
  273. for k, v := range oldMetrics.FailoverPeers {
  274. if len(newMetrics.FailoverPeers[k]) > 0 && len(v) == 0 {
  275. shouldUpdate = true
  276. }
  277. if len(v) > 0 && len(newMetrics.FailoverPeers[k]) == 0 {
  278. newMetrics.FailoverPeers[k] = v
  279. }
  280. }
  281. for k := range oldMetrics.Connectivity { // cleanup any left over data, self healing
  282. delete(newMetrics.Connectivity, k)
  283. }
  284. return shouldUpdate
  285. }