peer_server.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766
  1. package server
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "fmt"
  6. "io/ioutil"
  7. "math/rand"
  8. "net/http"
  9. "net/url"
  10. "sort"
  11. "strconv"
  12. "time"
  13. "github.com/coreos/etcd/third_party/github.com/goraft/raft"
  14. "github.com/coreos/etcd/third_party/github.com/gorilla/mux"
  15. "github.com/coreos/etcd/discovery"
  16. etcdErr "github.com/coreos/etcd/error"
  17. "github.com/coreos/etcd/log"
  18. "github.com/coreos/etcd/metrics"
  19. "github.com/coreos/etcd/store"
  20. )
  21. const (
  22. // ThresholdMonitorTimeout is the time between log notifications that the
  23. // Raft heartbeat is too close to the election timeout.
  24. ThresholdMonitorTimeout = 5 * time.Second
  25. // ActiveMonitorTimeout is the time between checks on the active size of
  26. // the cluster. If the active size is different than the actual size then
  27. // etcd attempts to promote/demote to bring it to the correct number.
  28. ActiveMonitorTimeout = 1 * time.Second
  29. // PeerActivityMonitorTimeout is the time between checks for dead nodes in
  30. // the cluster.
  31. PeerActivityMonitorTimeout = 1 * time.Second
  32. )
  33. const (
  34. peerModeFlag = 0
  35. proxyModeFlag = 1
  36. )
  37. type PeerServerConfig struct {
  38. Name string
  39. Scheme string
  40. URL string
  41. SnapshotCount int
  42. RetryTimes int
  43. RetryInterval float64
  44. }
  45. type PeerServer struct {
  46. Config PeerServerConfig
  47. clusterConfig *ClusterConfig
  48. raftServer raft.Server
  49. server *Server
  50. joinIndex uint64
  51. followersStats *raftFollowersStats
  52. serverStats *raftServerStats
  53. registry *Registry
  54. store store.Store
  55. snapConf *snapshotConf
  56. mode Mode
  57. closeChan chan bool
  58. timeoutThresholdChan chan interface{}
  59. proxyPeerURL string
  60. proxyClientURL string
  61. metrics *metrics.Bucket
  62. }
  63. // TODO: find a good policy to do snapshot
  64. type snapshotConf struct {
  65. // Etcd will check if snapshot is need every checkingInterval
  66. checkingInterval time.Duration
  67. // The index when the last snapshot happened
  68. lastIndex uint64
  69. // If the incremental number of index since the last snapshot
  70. // exceeds the snapshot Threshold, etcd will do a snapshot
  71. snapshotThr uint64
  72. }
  73. func NewPeerServer(psConfig PeerServerConfig, registry *Registry, store store.Store, mb *metrics.Bucket, followersStats *raftFollowersStats, serverStats *raftServerStats) *PeerServer {
  74. s := &PeerServer{
  75. Config: psConfig,
  76. clusterConfig: NewClusterConfig(),
  77. registry: registry,
  78. store: store,
  79. followersStats: followersStats,
  80. serverStats: serverStats,
  81. timeoutThresholdChan: make(chan interface{}, 1),
  82. metrics: mb,
  83. }
  84. return s
  85. }
  86. func (s *PeerServer) SetRaftServer(raftServer raft.Server) {
  87. s.snapConf = &snapshotConf{
  88. checkingInterval: time.Second * 3,
  89. // this is not accurate, we will update raft to provide an api
  90. lastIndex: raftServer.CommitIndex(),
  91. snapshotThr: uint64(s.Config.SnapshotCount),
  92. }
  93. raftServer.AddEventListener(raft.StateChangeEventType, s.raftEventLogger)
  94. raftServer.AddEventListener(raft.LeaderChangeEventType, s.raftEventLogger)
  95. raftServer.AddEventListener(raft.TermChangeEventType, s.raftEventLogger)
  96. raftServer.AddEventListener(raft.AddPeerEventType, s.raftEventLogger)
  97. raftServer.AddEventListener(raft.RemovePeerEventType, s.raftEventLogger)
  98. raftServer.AddEventListener(raft.HeartbeatIntervalEventType, s.raftEventLogger)
  99. raftServer.AddEventListener(raft.ElectionTimeoutThresholdEventType, s.raftEventLogger)
  100. raftServer.AddEventListener(raft.HeartbeatEventType, s.recordMetricEvent)
  101. s.raftServer = raftServer
  102. }
  103. // Mode retrieves the current mode of the server.
  104. func (s *PeerServer) Mode() Mode {
  105. return s.mode
  106. }
  107. // SetMode updates the current mode of the server.
  108. // Switching to a peer mode will start the Raft server.
  109. // Switching to a proxy mode will stop the Raft server.
  110. func (s *PeerServer) setMode(mode Mode) {
  111. s.mode = mode
  112. switch mode {
  113. case PeerMode:
  114. if !s.raftServer.Running() {
  115. s.raftServer.Start()
  116. }
  117. case ProxyMode:
  118. if s.raftServer.Running() {
  119. s.raftServer.Stop()
  120. }
  121. }
  122. }
  123. // ClusterConfig retrieves the current cluster configuration.
  124. func (s *PeerServer) ClusterConfig() *ClusterConfig {
  125. return s.clusterConfig
  126. }
  127. // SetClusterConfig updates the current cluster configuration.
  128. // Adjusting the active size will cause the PeerServer to demote peers or
  129. // promote proxies to match the new size.
  130. func (s *PeerServer) SetClusterConfig(c *ClusterConfig) {
  131. // Set minimums.
  132. if c.ActiveSize < MinActiveSize {
  133. c.ActiveSize = MinActiveSize
  134. }
  135. if c.PromoteDelay < MinPromoteDelay {
  136. c.PromoteDelay = MinPromoteDelay
  137. }
  138. s.clusterConfig = c
  139. }
  140. // Helper function to do discovery and return results in expected format
  141. func (s *PeerServer) handleDiscovery(discoverURL string) (peers []string, err error) {
  142. peers, err = discovery.Do(discoverURL, s.Config.Name, s.Config.URL)
  143. // Warn about errors coming from discovery, this isn't fatal
  144. // since the user might have provided a peer list elsewhere,
  145. // or there is some log in data dir.
  146. if err != nil {
  147. log.Warnf("Discovery encountered an error: %v", err)
  148. return
  149. }
  150. for i := range peers {
  151. // Strip the scheme off of the peer if it has one
  152. // TODO(bp): clean this up!
  153. purl, err := url.Parse(peers[i])
  154. if err == nil {
  155. peers[i] = purl.Host
  156. }
  157. }
  158. log.Infof("Discovery fetched back peer list: %v", peers)
  159. return
  160. }
  161. // Try all possible ways to find clusters to join
  162. // Include -discovery, -peers and log data in -data-dir
  163. //
  164. // Peer discovery follows this order:
  165. // 1. -discovery
  166. // 2. -peers
  167. // 3. previous peers in -data-dir
  168. // RaftServer should be started as late as possible. Current implementation
  169. // to start it is not that good, and will be refactored in #627.
  170. func (s *PeerServer) findCluster(discoverURL string, peers []string) {
  171. // Attempt cluster discovery
  172. toDiscover := discoverURL != ""
  173. if toDiscover {
  174. discoverPeers, discoverErr := s.handleDiscovery(discoverURL)
  175. // It is registered in discover url
  176. if discoverErr == nil {
  177. // start as a leader in a new cluster
  178. if len(discoverPeers) == 0 {
  179. log.Debug("This peer is starting a brand new cluster based on discover URL.")
  180. s.startAsLeader()
  181. } else {
  182. s.startAsFollower(discoverPeers)
  183. }
  184. return
  185. }
  186. }
  187. hasPeerList := len(peers) > 0
  188. // if there is log in data dir, append previous peers to peers in config
  189. // to find cluster
  190. prevPeers := s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name)
  191. for i := 0; i < len(prevPeers); i++ {
  192. u, err := url.Parse(prevPeers[i])
  193. if err != nil {
  194. log.Debug("rejoin cannot parse url: ", err)
  195. }
  196. prevPeers[i] = u.Host
  197. }
  198. peers = append(peers, prevPeers...)
  199. // Remove its own peer address from the peer list to join
  200. u, err := url.Parse(s.Config.URL)
  201. if err != nil {
  202. log.Fatalf("cannot parse peer address %v: %v", s.Config.URL, err)
  203. }
  204. filteredPeers := make([]string, 0)
  205. for _, v := range peers {
  206. if v != u.Host {
  207. filteredPeers = append(filteredPeers, v)
  208. }
  209. }
  210. peers = filteredPeers
  211. // if there is backup peer lists, use it to find cluster
  212. if len(peers) > 0 {
  213. ok := s.joinCluster(peers)
  214. if !ok {
  215. log.Warn("No living peers are found!")
  216. } else {
  217. s.raftServer.Start()
  218. log.Debugf("%s restart as a follower based on peers[%v]", s.Config.Name)
  219. return
  220. }
  221. }
  222. if !s.raftServer.IsLogEmpty() {
  223. log.Debug("Entire cluster is down! %v will restart the cluster.", s.Config.Name)
  224. s.raftServer.Start()
  225. return
  226. }
  227. if toDiscover {
  228. log.Fatalf("Discovery failed, no available peers in backup list, and no log data")
  229. }
  230. if hasPeerList {
  231. log.Fatalf("No available peers in backup list, and no log data")
  232. }
  233. log.Infof("This peer is starting a brand new cluster now.")
  234. s.startAsLeader()
  235. }
  236. // Start the raft server
  237. func (s *PeerServer) Start(snapshot bool, discoverURL string, peers []string) error {
  238. // LoadSnapshot
  239. if snapshot {
  240. err := s.raftServer.LoadSnapshot()
  241. if err == nil {
  242. log.Debugf("%s finished load snapshot", s.Config.Name)
  243. } else {
  244. log.Debug(err)
  245. }
  246. }
  247. s.raftServer.Init()
  248. s.findCluster(discoverURL, peers)
  249. s.closeChan = make(chan bool)
  250. go s.monitorSync()
  251. go s.monitorTimeoutThreshold(s.closeChan)
  252. go s.monitorActiveSize(s.closeChan)
  253. go s.monitorPeerActivity(s.closeChan)
  254. // open the snapshot
  255. if snapshot {
  256. go s.monitorSnapshot()
  257. }
  258. return nil
  259. }
  260. func (s *PeerServer) Stop() {
  261. if s.closeChan != nil {
  262. close(s.closeChan)
  263. s.closeChan = nil
  264. }
  265. s.raftServer.Stop()
  266. }
  267. func (s *PeerServer) HTTPHandler() http.Handler {
  268. router := mux.NewRouter()
  269. // internal commands
  270. router.HandleFunc("/name", s.NameHttpHandler)
  271. router.HandleFunc("/version", s.VersionHttpHandler)
  272. router.HandleFunc("/version/{version:[0-9]+}/check", s.VersionCheckHttpHandler)
  273. router.HandleFunc("/upgrade", s.UpgradeHttpHandler)
  274. router.HandleFunc("/join", s.JoinHttpHandler)
  275. router.HandleFunc("/promote", s.PromoteHttpHandler).Methods("POST")
  276. router.HandleFunc("/remove/{name:.+}", s.RemoveHttpHandler)
  277. router.HandleFunc("/vote", s.VoteHttpHandler)
  278. router.HandleFunc("/log", s.GetLogHttpHandler)
  279. router.HandleFunc("/log/append", s.AppendEntriesHttpHandler)
  280. router.HandleFunc("/snapshot", s.SnapshotHttpHandler)
  281. router.HandleFunc("/snapshotRecovery", s.SnapshotRecoveryHttpHandler)
  282. router.HandleFunc("/etcdURL", s.EtcdURLHttpHandler)
  283. router.HandleFunc("/v2/admin/config", s.getClusterConfigHttpHandler).Methods("GET")
  284. router.HandleFunc("/v2/admin/config", s.setClusterConfigHttpHandler).Methods("PUT")
  285. router.HandleFunc("/v2/admin/machines", s.getMachinesHttpHandler).Methods("GET")
  286. router.HandleFunc("/v2/admin/machines/{name}", s.getMachineHttpHandler).Methods("GET")
  287. router.HandleFunc("/v2/admin/machines/{name}", s.addMachineHttpHandler).Methods("PUT")
  288. router.HandleFunc("/v2/admin/machines/{name}", s.removeMachineHttpHandler).Methods("DELETE")
  289. return router
  290. }
  291. // Retrieves the underlying Raft server.
  292. func (s *PeerServer) RaftServer() raft.Server {
  293. return s.raftServer
  294. }
  295. // Associates the client server with the peer server.
  296. func (s *PeerServer) SetServer(server *Server) {
  297. s.server = server
  298. }
  299. func (s *PeerServer) startAsLeader() {
  300. s.raftServer.Start()
  301. // leader need to join self as a peer
  302. for {
  303. c := &JoinCommandV1{
  304. MinVersion: store.MinVersion(),
  305. MaxVersion: store.MaxVersion(),
  306. Name: s.raftServer.Name(),
  307. RaftURL: s.Config.URL,
  308. EtcdURL: s.server.URL(),
  309. }
  310. _, err := s.raftServer.Do(c)
  311. if err == nil {
  312. break
  313. }
  314. }
  315. log.Debugf("%s start as a leader", s.Config.Name)
  316. }
  317. func (s *PeerServer) startAsFollower(cluster []string) {
  318. // start as a follower in a existing cluster
  319. for i := 0; i < s.Config.RetryTimes; i++ {
  320. ok := s.joinCluster(cluster)
  321. if ok {
  322. s.raftServer.Start()
  323. return
  324. }
  325. log.Warnf("%v is unable to join the cluster using any of the peers %v at %dth time. Retrying in %.1f seconds", s.Config.Name, cluster, i, s.Config.RetryInterval)
  326. time.Sleep(time.Second * time.Duration(s.Config.RetryInterval))
  327. }
  328. log.Fatalf("Cannot join the cluster via given peers after %x retries", s.Config.RetryTimes)
  329. }
  330. // getVersion fetches the peer version of a cluster.
  331. func getVersion(t *transporter, versionURL url.URL) (int, error) {
  332. resp, _, err := t.Get(versionURL.String())
  333. if err != nil {
  334. return 0, err
  335. }
  336. defer resp.Body.Close()
  337. body, err := ioutil.ReadAll(resp.Body)
  338. if err != nil {
  339. return 0, err
  340. }
  341. // Parse version number.
  342. version, _ := strconv.Atoi(string(body))
  343. return version, nil
  344. }
  345. // Upgradable checks whether all peers in a cluster support an upgrade to the next store version.
  346. func (s *PeerServer) Upgradable() error {
  347. nextVersion := s.store.Version() + 1
  348. for _, peerURL := range s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name) {
  349. u, err := url.Parse(peerURL)
  350. if err != nil {
  351. return fmt.Errorf("PeerServer: Cannot parse URL: '%s' (%s)", peerURL, err)
  352. }
  353. t, _ := s.raftServer.Transporter().(*transporter)
  354. checkURL := (&url.URL{Host: u.Host, Scheme: s.Config.Scheme, Path: fmt.Sprintf("/version/%d/check", nextVersion)}).String()
  355. resp, _, err := t.Get(checkURL)
  356. if err != nil {
  357. return fmt.Errorf("PeerServer: Cannot check version compatibility: %s", u.Host)
  358. }
  359. if resp.StatusCode != 200 {
  360. return fmt.Errorf("PeerServer: Version %d is not compatible with peer: %s", nextVersion, u.Host)
  361. }
  362. }
  363. return nil
  364. }
  365. func (s *PeerServer) joinCluster(cluster []string) bool {
  366. for _, peer := range cluster {
  367. if len(peer) == 0 {
  368. continue
  369. }
  370. err := s.joinByPeer(s.raftServer, peer, s.Config.Scheme)
  371. if err == nil {
  372. log.Debugf("%s joined the cluster via peer %s", s.Config.Name, peer)
  373. return true
  374. }
  375. if _, ok := err.(etcdErr.Error); ok {
  376. log.Fatal(err)
  377. }
  378. log.Warnf("Attempt to join via %s failed: %s", peer, err)
  379. }
  380. return false
  381. }
  382. // Send join requests to peer.
  383. func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string) error {
  384. // t must be ok
  385. t, _ := server.Transporter().(*transporter)
  386. // Our version must match the leaders version
  387. versionURL := url.URL{Host: peer, Scheme: scheme, Path: "/version"}
  388. version, err := getVersion(t, versionURL)
  389. if err != nil {
  390. return fmt.Errorf("Error during join version check: %v", err)
  391. }
  392. if version < store.MinVersion() || version > store.MaxVersion() {
  393. return fmt.Errorf("Unable to join: cluster version is %d; version compatibility is %d - %d", version, store.MinVersion(), store.MaxVersion())
  394. }
  395. var b bytes.Buffer
  396. c := &JoinCommandV2{
  397. MinVersion: store.MinVersion(),
  398. MaxVersion: store.MaxVersion(),
  399. Name: server.Name(),
  400. PeerURL: s.Config.URL,
  401. ClientURL: s.server.URL(),
  402. }
  403. json.NewEncoder(&b).Encode(c)
  404. joinURL := url.URL{Host: peer, Scheme: scheme, Path: "/v2/admin/machines/" + server.Name()}
  405. log.Infof("Send Join Request to %s", joinURL.String())
  406. req, _ := http.NewRequest("PUT", joinURL.String(), &b)
  407. resp, err := t.client.Do(req)
  408. for {
  409. if err != nil {
  410. return fmt.Errorf("Unable to join: %v", err)
  411. }
  412. if resp != nil {
  413. defer resp.Body.Close()
  414. log.Infof("»»»» %d", resp.StatusCode)
  415. if resp.StatusCode == http.StatusOK {
  416. var msg joinMessageV2
  417. if err := json.NewDecoder(resp.Body).Decode(&msg); err != nil {
  418. log.Debugf("Error reading join response: %v", err)
  419. return err
  420. }
  421. s.joinIndex = msg.CommitIndex
  422. s.setMode(msg.Mode)
  423. if msg.Mode == ProxyMode {
  424. s.proxyClientURL = resp.Header.Get("X-Leader-Client-URL")
  425. s.proxyPeerURL = resp.Header.Get("X-Leader-Peer-URL")
  426. }
  427. return nil
  428. }
  429. if resp.StatusCode == http.StatusTemporaryRedirect {
  430. address := resp.Header.Get("Location")
  431. log.Debugf("Send Join Request to %s", address)
  432. c := &JoinCommandV1{
  433. MinVersion: store.MinVersion(),
  434. MaxVersion: store.MaxVersion(),
  435. Name: server.Name(),
  436. RaftURL: s.Config.URL,
  437. EtcdURL: s.server.URL(),
  438. }
  439. json.NewEncoder(&b).Encode(c)
  440. resp, _, err = t.Post(address, &b)
  441. } else if resp.StatusCode == http.StatusBadRequest {
  442. log.Debug("Reach max number peers in the cluster")
  443. decoder := json.NewDecoder(resp.Body)
  444. err := &etcdErr.Error{}
  445. decoder.Decode(err)
  446. return *err
  447. } else {
  448. return fmt.Errorf("Unable to join")
  449. }
  450. }
  451. }
  452. }
  453. func (s *PeerServer) Stats() []byte {
  454. s.serverStats.LeaderInfo.Uptime = time.Now().Sub(s.serverStats.LeaderInfo.startTime).String()
  455. // TODO: register state listener to raft to change this field
  456. // rather than compare the state each time Stats() is called.
  457. if s.RaftServer().State() == raft.Leader {
  458. s.serverStats.LeaderInfo.Name = s.RaftServer().Name()
  459. }
  460. queue := s.serverStats.sendRateQueue
  461. s.serverStats.SendingPkgRate, s.serverStats.SendingBandwidthRate = queue.Rate()
  462. queue = s.serverStats.recvRateQueue
  463. s.serverStats.RecvingPkgRate, s.serverStats.RecvingBandwidthRate = queue.Rate()
  464. b, _ := json.Marshal(s.serverStats)
  465. return b
  466. }
  467. func (s *PeerServer) PeerStats() []byte {
  468. if s.raftServer.State() == raft.Leader {
  469. b, _ := json.Marshal(s.followersStats)
  470. return b
  471. }
  472. return nil
  473. }
  474. // raftEventLogger converts events from the Raft server into log messages.
  475. func (s *PeerServer) raftEventLogger(event raft.Event) {
  476. value := event.Value()
  477. prevValue := event.PrevValue()
  478. if value == nil {
  479. value = "<nil>"
  480. }
  481. if prevValue == nil {
  482. prevValue = "<nil>"
  483. }
  484. switch event.Type() {
  485. case raft.StateChangeEventType:
  486. log.Infof("%s: state changed from '%v' to '%v'.", s.Config.Name, prevValue, value)
  487. case raft.TermChangeEventType:
  488. log.Infof("%s: term #%v started.", s.Config.Name, value)
  489. case raft.LeaderChangeEventType:
  490. log.Infof("%s: leader changed from '%v' to '%v'.", s.Config.Name, prevValue, value)
  491. case raft.AddPeerEventType:
  492. log.Infof("%s: peer added: '%v'", s.Config.Name, value)
  493. case raft.RemovePeerEventType:
  494. log.Infof("%s: peer removed: '%v'", s.Config.Name, value)
  495. case raft.HeartbeatIntervalEventType:
  496. var name = "<unknown>"
  497. if peer, ok := value.(*raft.Peer); ok {
  498. name = peer.Name
  499. }
  500. log.Infof("%s: warning: heartbeat timed out: '%v'", s.Config.Name, name)
  501. case raft.ElectionTimeoutThresholdEventType:
  502. select {
  503. case s.timeoutThresholdChan <- value:
  504. default:
  505. }
  506. }
  507. }
  508. func (s *PeerServer) recordMetricEvent(event raft.Event) {
  509. name := fmt.Sprintf("raft.event.%s", event.Type())
  510. value := event.Value().(time.Duration)
  511. (*s.metrics).Timer(name).Update(value)
  512. }
  513. // logSnapshot logs about the snapshot that was taken.
  514. func (s *PeerServer) logSnapshot(err error, currentIndex, count uint64) {
  515. info := fmt.Sprintf("%s: snapshot of %d events at index %d", s.Config.Name, count, currentIndex)
  516. if err != nil {
  517. log.Infof("%s attempted and failed: %v", info, err)
  518. } else {
  519. log.Infof("%s completed", info)
  520. }
  521. }
  522. func (s *PeerServer) monitorSnapshot() {
  523. for {
  524. time.Sleep(s.snapConf.checkingInterval)
  525. currentIndex := s.RaftServer().CommitIndex()
  526. count := currentIndex - s.snapConf.lastIndex
  527. if uint64(count) > s.snapConf.snapshotThr {
  528. err := s.raftServer.TakeSnapshot()
  529. s.logSnapshot(err, currentIndex, count)
  530. s.snapConf.lastIndex = currentIndex
  531. }
  532. }
  533. }
  534. func (s *PeerServer) monitorSync() {
  535. ticker := time.Tick(time.Millisecond * 500)
  536. for {
  537. select {
  538. case now := <-ticker:
  539. if s.raftServer.State() == raft.Leader {
  540. s.raftServer.Do(s.store.CommandFactory().CreateSyncCommand(now))
  541. }
  542. }
  543. }
  544. }
  545. // monitorTimeoutThreshold groups timeout threshold events together and prints
  546. // them as a single log line.
  547. func (s *PeerServer) monitorTimeoutThreshold(closeChan chan bool) {
  548. for {
  549. select {
  550. case value := <-s.timeoutThresholdChan:
  551. log.Infof("%s: warning: heartbeat near election timeout: %v", s.Config.Name, value)
  552. case <-closeChan:
  553. return
  554. }
  555. time.Sleep(ThresholdMonitorTimeout)
  556. }
  557. }
  558. // monitorActiveSize has the leader periodically check the status of cluster
  559. // nodes and swaps them out for proxies as needed.
  560. func (s *PeerServer) monitorActiveSize(closeChan chan bool) {
  561. for {
  562. select {
  563. case <-time.After(ActiveMonitorTimeout):
  564. case <-closeChan:
  565. return
  566. }
  567. // Ignore while this peer is not a leader.
  568. if s.raftServer.State() != raft.Leader {
  569. continue
  570. }
  571. // Retrieve target active size and actual active size.
  572. activeSize := s.ClusterConfig().ActiveSize
  573. peerCount := s.registry.PeerCount()
  574. proxies := s.registry.Proxies()
  575. peers := s.registry.Peers()
  576. if index := sort.SearchStrings(peers, s.Config.Name); index < len(peers) && peers[index] == s.Config.Name {
  577. peers = append(peers[:index], peers[index+1:]...)
  578. }
  579. // If we have more active nodes than we should then demote.
  580. if peerCount > activeSize {
  581. peer := peers[rand.Intn(len(peers))]
  582. log.Infof("%s: demoting: %v", s.Config.Name, peer)
  583. if _, err := s.raftServer.Do(&DemoteCommand{Name: peer}); err != nil {
  584. log.Infof("%s: warning: demotion error: %v", s.Config.Name, err)
  585. }
  586. continue
  587. }
  588. // If we don't have enough active nodes then try to promote a proxy.
  589. if peerCount < activeSize && len(proxies) > 0 {
  590. loop:
  591. for _, i := range rand.Perm(len(proxies)) {
  592. proxy := proxies[i]
  593. proxyPeerURL, _ := s.registry.ProxyPeerURL(proxy)
  594. log.Infof("%s: attempting to promote: %v (%s)", s.Config.Name, proxy, proxyPeerURL)
  595. // Notify proxy to promote itself.
  596. client := &http.Client{
  597. Transport: &http.Transport{
  598. DisableKeepAlives: false,
  599. ResponseHeaderTimeout: ActiveMonitorTimeout,
  600. },
  601. }
  602. resp, err := client.Post(fmt.Sprintf("%s/promote", proxyPeerURL), "application/json", nil)
  603. if err != nil {
  604. log.Infof("%s: warning: promotion error: %v", s.Config.Name, err)
  605. continue
  606. } else if resp.StatusCode != http.StatusOK {
  607. log.Infof("%s: warning: promotion failure: %v", s.Config.Name, resp.StatusCode)
  608. continue
  609. }
  610. break loop
  611. }
  612. }
  613. }
  614. }
  615. // monitorPeerActivity has the leader periodically for dead nodes and demotes them.
  616. func (s *PeerServer) monitorPeerActivity(closeChan chan bool) {
  617. for {
  618. select {
  619. case <-time.After(PeerActivityMonitorTimeout):
  620. case <-closeChan:
  621. return
  622. }
  623. // Ignore while this peer is not a leader.
  624. if s.raftServer.State() != raft.Leader {
  625. continue
  626. }
  627. // Check last activity for all peers.
  628. now := time.Now()
  629. promoteDelay := time.Duration(s.ClusterConfig().PromoteDelay) * time.Second
  630. peers := s.raftServer.Peers()
  631. for _, peer := range peers {
  632. // If the last response from the peer is longer than the promote delay
  633. // then automatically demote the peer.
  634. if !peer.LastActivity().IsZero() && now.Sub(peer.LastActivity()) > promoteDelay {
  635. log.Infof("%s: demoting node: %v; last activity %v ago", s.Config.Name, peer.Name, now.Sub(peer.LastActivity()))
  636. if _, err := s.raftServer.Do(&DemoteCommand{Name: peer.Name}); err != nil {
  637. log.Infof("%s: warning: autodemotion error: %v", s.Config.Name, err)
  638. }
  639. continue
  640. }
  641. }
  642. }
  643. }
  644. // Mode represents whether the server is an active peer or if the server is
  645. // simply acting as a proxy.
  646. type Mode string
  647. const (
  648. // PeerMode is when the server is an active node in Raft.
  649. PeerMode = Mode("peer")
  650. // ProxyMode is when the server is an inactive, request-forwarding node.
  651. ProxyMode = Mode("proxy")
  652. )