health.go 2.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102
  1. package client
  2. import (
  3. "container/heap"
  4. "net"
  5. "net/http"
  6. "strings"
  7. "time"
  8. "ehang.io/nps/lib/conn"
  9. "ehang.io/nps/lib/file"
  10. "ehang.io/nps/lib/sheap"
  11. "github.com/astaxie/beego/logs"
  12. "github.com/pkg/errors"
  13. )
  14. var isStart bool
  15. var serverConn *conn.Conn
  16. func heathCheck(healths []*file.Health, c *conn.Conn) bool {
  17. serverConn = c
  18. if isStart {
  19. for _, v := range healths {
  20. v.HealthMap = make(map[string]int)
  21. }
  22. return true
  23. }
  24. isStart = true
  25. h := &sheap.IntHeap{}
  26. for _, v := range healths {
  27. if v.HealthMaxFail > 0 && v.HealthCheckTimeout > 0 && v.HealthCheckInterval > 0 {
  28. v.HealthNextTime = time.Now().Add(time.Duration(v.HealthCheckInterval) * time.Second)
  29. heap.Push(h, v.HealthNextTime.Unix())
  30. v.HealthMap = make(map[string]int)
  31. }
  32. }
  33. go session(healths, h)
  34. return true
  35. }
  36. func session(healths []*file.Health, h *sheap.IntHeap) {
  37. for {
  38. if h.Len() == 0 {
  39. logs.Error("health check error")
  40. break
  41. }
  42. rs := heap.Pop(h).(int64) - time.Now().Unix()
  43. if rs <= 0 {
  44. continue
  45. }
  46. timer := time.NewTimer(time.Duration(rs) * time.Second)
  47. select {
  48. case <-timer.C:
  49. for _, v := range healths {
  50. if v.HealthNextTime.Before(time.Now()) {
  51. v.HealthNextTime = time.Now().Add(time.Duration(v.HealthCheckInterval) * time.Second)
  52. //check
  53. go check(v)
  54. //reset time
  55. heap.Push(h, v.HealthNextTime.Unix())
  56. }
  57. }
  58. }
  59. }
  60. }
  61. // work when just one port and many target
  62. func check(t *file.Health) {
  63. arr := strings.Split(t.HealthCheckTarget, ",")
  64. var err error
  65. var rs *http.Response
  66. for _, v := range arr {
  67. if t.HealthCheckType == "tcp" {
  68. var c net.Conn
  69. c, err = net.DialTimeout("tcp", v, time.Duration(t.HealthCheckTimeout)*time.Second)
  70. if err == nil {
  71. c.Close()
  72. }
  73. } else {
  74. client := &http.Client{}
  75. client.Timeout = time.Duration(t.HealthCheckTimeout) * time.Second
  76. rs, err = client.Get("http://" + v + t.HttpHealthUrl)
  77. if err == nil && rs.StatusCode != 200 {
  78. err = errors.New("status code is not match")
  79. }
  80. }
  81. t.Lock()
  82. if err != nil {
  83. t.HealthMap[v] += 1
  84. } else if t.HealthMap[v] >= t.HealthMaxFail {
  85. //send recovery add
  86. serverConn.SendHealthInfo(v, "1")
  87. t.HealthMap[v] = 0
  88. }
  89. if t.HealthMap[v] > 0 && t.HealthMap[v]%t.HealthMaxFail == 0 {
  90. //send fail remove
  91. serverConn.SendHealthInfo(v, "0")
  92. }
  93. t.Unlock()
  94. }
  95. }