cluster.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841
  1. // Copyright 2015 The etcd Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package membership
  15. import (
  16. "bytes"
  17. "context"
  18. "crypto/sha1"
  19. "encoding/binary"
  20. "encoding/json"
  21. "fmt"
  22. "path"
  23. "sort"
  24. "strings"
  25. "sync"
  26. "time"
  27. "go.etcd.io/etcd/etcdserver/api/v2store"
  28. "go.etcd.io/etcd/mvcc/backend"
  29. "go.etcd.io/etcd/pkg/netutil"
  30. "go.etcd.io/etcd/pkg/types"
  31. "go.etcd.io/etcd/raft"
  32. "go.etcd.io/etcd/raft/raftpb"
  33. "go.etcd.io/etcd/version"
  34. "github.com/coreos/go-semver/semver"
  35. "github.com/prometheus/client_golang/prometheus"
  36. "go.uber.org/zap"
  37. )
  38. const maxLearners = 1
  39. // RaftCluster is a list of Members that belong to the same raft cluster
  40. type RaftCluster struct {
  41. lg *zap.Logger
  42. localID types.ID
  43. cid types.ID
  44. token string
  45. v2store v2store.Store
  46. be backend.Backend
  47. sync.Mutex // guards the fields below
  48. version *semver.Version
  49. members map[types.ID]*Member
  50. // removed contains the ids of removed members in the cluster.
  51. // removed id cannot be reused.
  52. removed map[types.ID]bool
  53. }
  54. // ConfigChangeContext represents a context for confChange.
  55. type ConfigChangeContext struct {
  56. Member
  57. // IsPromote indicates if the config change is for promoting a learner member.
  58. // This flag is needed because both adding a new member and promoting a learner member
  59. // uses the same config change type 'ConfChangeAddNode'.
  60. IsPromote bool `json:"isPromote"`
  61. }
  62. // NewClusterFromURLsMap creates a new raft cluster using provided urls map. Currently, it does not support creating
  63. // cluster with raft learner member.
  64. func NewClusterFromURLsMap(lg *zap.Logger, token string, urlsmap types.URLsMap) (*RaftCluster, error) {
  65. c := NewCluster(lg, token)
  66. for name, urls := range urlsmap {
  67. m := NewMember(name, urls, token, nil)
  68. if _, ok := c.members[m.ID]; ok {
  69. return nil, fmt.Errorf("member exists with identical ID %v", m)
  70. }
  71. if uint64(m.ID) == raft.None {
  72. return nil, fmt.Errorf("cannot use %x as member id", raft.None)
  73. }
  74. c.members[m.ID] = m
  75. }
  76. c.genID()
  77. return c, nil
  78. }
  79. func NewClusterFromMembers(lg *zap.Logger, token string, id types.ID, membs []*Member) *RaftCluster {
  80. c := NewCluster(lg, token)
  81. c.cid = id
  82. for _, m := range membs {
  83. c.members[m.ID] = m
  84. }
  85. return c
  86. }
  87. func NewCluster(lg *zap.Logger, token string) *RaftCluster {
  88. return &RaftCluster{
  89. lg: lg,
  90. token: token,
  91. members: make(map[types.ID]*Member),
  92. removed: make(map[types.ID]bool),
  93. }
  94. }
  95. func (c *RaftCluster) ID() types.ID { return c.cid }
  96. func (c *RaftCluster) Members() []*Member {
  97. c.Lock()
  98. defer c.Unlock()
  99. var ms MembersByID
  100. for _, m := range c.members {
  101. ms = append(ms, m.Clone())
  102. }
  103. sort.Sort(ms)
  104. return []*Member(ms)
  105. }
  106. func (c *RaftCluster) Member(id types.ID) *Member {
  107. c.Lock()
  108. defer c.Unlock()
  109. return c.members[id].Clone()
  110. }
  111. func (c *RaftCluster) VotingMembers() []*Member {
  112. c.Lock()
  113. defer c.Unlock()
  114. var ms MembersByID
  115. for _, m := range c.members {
  116. if !m.IsLearner {
  117. ms = append(ms, m.Clone())
  118. }
  119. }
  120. sort.Sort(ms)
  121. return []*Member(ms)
  122. }
  123. // MemberByName returns a Member with the given name if exists.
  124. // If more than one member has the given name, it will panic.
  125. func (c *RaftCluster) MemberByName(name string) *Member {
  126. c.Lock()
  127. defer c.Unlock()
  128. var memb *Member
  129. for _, m := range c.members {
  130. if m.Name == name {
  131. if memb != nil {
  132. if c.lg != nil {
  133. c.lg.Panic("two member with same name found", zap.String("name", name))
  134. } else {
  135. plog.Panicf("two members with the given name %q exist", name)
  136. }
  137. }
  138. memb = m
  139. }
  140. }
  141. return memb.Clone()
  142. }
  143. func (c *RaftCluster) MemberIDs() []types.ID {
  144. c.Lock()
  145. defer c.Unlock()
  146. var ids []types.ID
  147. for _, m := range c.members {
  148. ids = append(ids, m.ID)
  149. }
  150. sort.Sort(types.IDSlice(ids))
  151. return ids
  152. }
  153. func (c *RaftCluster) IsIDRemoved(id types.ID) bool {
  154. c.Lock()
  155. defer c.Unlock()
  156. return c.removed[id]
  157. }
  158. // PeerURLs returns a list of all peer addresses.
  159. // The returned list is sorted in ascending lexicographical order.
  160. func (c *RaftCluster) PeerURLs() []string {
  161. c.Lock()
  162. defer c.Unlock()
  163. urls := make([]string, 0)
  164. for _, p := range c.members {
  165. urls = append(urls, p.PeerURLs...)
  166. }
  167. sort.Strings(urls)
  168. return urls
  169. }
  170. // ClientURLs returns a list of all client addresses.
  171. // The returned list is sorted in ascending lexicographical order.
  172. func (c *RaftCluster) ClientURLs() []string {
  173. c.Lock()
  174. defer c.Unlock()
  175. urls := make([]string, 0)
  176. for _, p := range c.members {
  177. urls = append(urls, p.ClientURLs...)
  178. }
  179. sort.Strings(urls)
  180. return urls
  181. }
  182. func (c *RaftCluster) String() string {
  183. c.Lock()
  184. defer c.Unlock()
  185. b := &bytes.Buffer{}
  186. fmt.Fprintf(b, "{ClusterID:%s ", c.cid)
  187. var ms []string
  188. for _, m := range c.members {
  189. ms = append(ms, fmt.Sprintf("%+v", m))
  190. }
  191. fmt.Fprintf(b, "Members:[%s] ", strings.Join(ms, " "))
  192. var ids []string
  193. for id := range c.removed {
  194. ids = append(ids, id.String())
  195. }
  196. fmt.Fprintf(b, "RemovedMemberIDs:[%s]}", strings.Join(ids, " "))
  197. return b.String()
  198. }
  199. func (c *RaftCluster) genID() {
  200. mIDs := c.MemberIDs()
  201. b := make([]byte, 8*len(mIDs))
  202. for i, id := range mIDs {
  203. binary.BigEndian.PutUint64(b[8*i:], uint64(id))
  204. }
  205. hash := sha1.Sum(b)
  206. c.cid = types.ID(binary.BigEndian.Uint64(hash[:8]))
  207. }
  208. func (c *RaftCluster) SetID(localID, cid types.ID) {
  209. c.localID = localID
  210. c.cid = cid
  211. }
  212. func (c *RaftCluster) SetStore(st v2store.Store) { c.v2store = st }
  213. func (c *RaftCluster) SetBackend(be backend.Backend) {
  214. c.be = be
  215. mustCreateBackendBuckets(c.be)
  216. }
  217. func (c *RaftCluster) Recover(onSet func(*zap.Logger, *semver.Version)) {
  218. c.Lock()
  219. defer c.Unlock()
  220. c.members, c.removed = membersFromStore(c.lg, c.v2store)
  221. c.version = clusterVersionFromStore(c.lg, c.v2store)
  222. mustDetectDowngrade(c.lg, c.version)
  223. onSet(c.lg, c.version)
  224. for _, m := range c.members {
  225. if c.lg != nil {
  226. c.lg.Info(
  227. "recovered/added member from store",
  228. zap.String("cluster-id", c.cid.String()),
  229. zap.String("local-member-id", c.localID.String()),
  230. zap.String("recovered-remote-peer-id", m.ID.String()),
  231. zap.Strings("recovered-remote-peer-urls", m.PeerURLs),
  232. )
  233. } else {
  234. plog.Infof("added member %s %v to cluster %s from store", m.ID, m.PeerURLs, c.cid)
  235. }
  236. }
  237. if c.version != nil {
  238. if c.lg != nil {
  239. c.lg.Info(
  240. "set cluster version from store",
  241. zap.String("cluster-version", version.Cluster(c.version.String())),
  242. )
  243. } else {
  244. plog.Infof("set the cluster version to %v from store", version.Cluster(c.version.String()))
  245. }
  246. }
  247. }
  248. // ValidateConfigurationChange takes a proposed ConfChange and
  249. // ensures that it is still valid.
  250. func (c *RaftCluster) ValidateConfigurationChange(cc raftpb.ConfChange) error {
  251. members, removed := membersFromStore(c.lg, c.v2store)
  252. id := types.ID(cc.NodeID)
  253. if removed[id] {
  254. return ErrIDRemoved
  255. }
  256. switch cc.Type {
  257. case raftpb.ConfChangeAddNode, raftpb.ConfChangeAddLearnerNode:
  258. confChangeContext := new(ConfigChangeContext)
  259. if err := json.Unmarshal(cc.Context, confChangeContext); err != nil {
  260. if c.lg != nil {
  261. c.lg.Panic("failed to unmarshal confChangeContext", zap.Error(err))
  262. } else {
  263. plog.Panicf("unmarshal confChangeContext should never fail: %v", err)
  264. }
  265. }
  266. if confChangeContext.IsPromote { // promoting a learner member to voting member
  267. if members[id] == nil {
  268. return ErrIDNotFound
  269. }
  270. if !members[id].IsLearner {
  271. return ErrMemberNotLearner
  272. }
  273. } else { // adding a new member
  274. if members[id] != nil {
  275. return ErrIDExists
  276. }
  277. urls := make(map[string]bool)
  278. for _, m := range members {
  279. for _, u := range m.PeerURLs {
  280. urls[u] = true
  281. }
  282. }
  283. for _, u := range confChangeContext.Member.PeerURLs {
  284. if urls[u] {
  285. return ErrPeerURLexists
  286. }
  287. }
  288. if confChangeContext.Member.IsLearner { // the new member is a learner
  289. numLearners := 0
  290. for _, m := range members {
  291. if m.IsLearner {
  292. numLearners++
  293. }
  294. }
  295. if numLearners+1 > maxLearners {
  296. return ErrTooManyLearners
  297. }
  298. }
  299. }
  300. case raftpb.ConfChangeRemoveNode:
  301. if members[id] == nil {
  302. return ErrIDNotFound
  303. }
  304. case raftpb.ConfChangeUpdateNode:
  305. if members[id] == nil {
  306. return ErrIDNotFound
  307. }
  308. urls := make(map[string]bool)
  309. for _, m := range members {
  310. if m.ID == id {
  311. continue
  312. }
  313. for _, u := range m.PeerURLs {
  314. urls[u] = true
  315. }
  316. }
  317. m := new(Member)
  318. if err := json.Unmarshal(cc.Context, m); err != nil {
  319. if c.lg != nil {
  320. c.lg.Panic("failed to unmarshal member", zap.Error(err))
  321. } else {
  322. plog.Panicf("unmarshal member should never fail: %v", err)
  323. }
  324. }
  325. for _, u := range m.PeerURLs {
  326. if urls[u] {
  327. return ErrPeerURLexists
  328. }
  329. }
  330. default:
  331. if c.lg != nil {
  332. c.lg.Panic("unknown ConfChange type", zap.String("type", cc.Type.String()))
  333. } else {
  334. plog.Panicf("ConfChange type should be either AddNode, RemoveNode or UpdateNode")
  335. }
  336. }
  337. return nil
  338. }
  339. // AddMember adds a new Member into the cluster, and saves the given member's
  340. // raftAttributes into the store. The given member should have empty attributes.
  341. // A Member with a matching id must not exist.
  342. func (c *RaftCluster) AddMember(m *Member) {
  343. c.Lock()
  344. defer c.Unlock()
  345. if c.v2store != nil {
  346. mustSaveMemberToStore(c.v2store, m)
  347. }
  348. if c.be != nil {
  349. mustSaveMemberToBackend(c.be, m)
  350. }
  351. c.members[m.ID] = m
  352. if c.lg != nil {
  353. c.lg.Info(
  354. "added member",
  355. zap.String("cluster-id", c.cid.String()),
  356. zap.String("local-member-id", c.localID.String()),
  357. zap.String("added-peer-id", m.ID.String()),
  358. zap.Strings("added-peer-peer-urls", m.PeerURLs),
  359. )
  360. } else {
  361. plog.Infof("added member %s %v to cluster %s", m.ID, m.PeerURLs, c.cid)
  362. }
  363. }
  364. // RemoveMember removes a member from the store.
  365. // The given id MUST exist, or the function panics.
  366. func (c *RaftCluster) RemoveMember(id types.ID) {
  367. c.Lock()
  368. defer c.Unlock()
  369. if c.v2store != nil {
  370. mustDeleteMemberFromStore(c.v2store, id)
  371. }
  372. if c.be != nil {
  373. mustDeleteMemberFromBackend(c.be, id)
  374. }
  375. m, ok := c.members[id]
  376. delete(c.members, id)
  377. c.removed[id] = true
  378. if c.lg != nil {
  379. if ok {
  380. c.lg.Info(
  381. "removed member",
  382. zap.String("cluster-id", c.cid.String()),
  383. zap.String("local-member-id", c.localID.String()),
  384. zap.String("removed-remote-peer-id", id.String()),
  385. zap.Strings("removed-remote-peer-urls", m.PeerURLs),
  386. )
  387. } else {
  388. c.lg.Warn(
  389. "skipped removing already removed member",
  390. zap.String("cluster-id", c.cid.String()),
  391. zap.String("local-member-id", c.localID.String()),
  392. zap.String("removed-remote-peer-id", id.String()),
  393. )
  394. }
  395. } else {
  396. plog.Infof("removed member %s from cluster %s", id, c.cid)
  397. }
  398. }
  399. func (c *RaftCluster) UpdateAttributes(id types.ID, attr Attributes) {
  400. c.Lock()
  401. defer c.Unlock()
  402. if m, ok := c.members[id]; ok {
  403. m.Attributes = attr
  404. if c.v2store != nil {
  405. mustUpdateMemberAttrInStore(c.v2store, m)
  406. }
  407. if c.be != nil {
  408. mustSaveMemberToBackend(c.be, m)
  409. }
  410. return
  411. }
  412. _, ok := c.removed[id]
  413. if !ok {
  414. if c.lg != nil {
  415. c.lg.Panic(
  416. "failed to update; member unknown",
  417. zap.String("cluster-id", c.cid.String()),
  418. zap.String("local-member-id", c.localID.String()),
  419. zap.String("unknown-remote-peer-id", id.String()),
  420. )
  421. } else {
  422. plog.Panicf("error updating attributes of unknown member %s", id)
  423. }
  424. }
  425. if c.lg != nil {
  426. c.lg.Warn(
  427. "skipped attributes update of removed member",
  428. zap.String("cluster-id", c.cid.String()),
  429. zap.String("local-member-id", c.localID.String()),
  430. zap.String("updated-peer-id", id.String()),
  431. )
  432. } else {
  433. plog.Warningf("skipped updating attributes of removed member %s", id)
  434. }
  435. }
  436. // PromoteMember marks the member's IsLearner RaftAttributes to false.
  437. func (c *RaftCluster) PromoteMember(id types.ID) {
  438. c.Lock()
  439. defer c.Unlock()
  440. c.members[id].RaftAttributes.IsLearner = false
  441. if c.v2store != nil {
  442. mustUpdateMemberInStore(c.v2store, c.members[id])
  443. }
  444. if c.be != nil {
  445. mustSaveMemberToBackend(c.be, c.members[id])
  446. }
  447. if c.lg != nil {
  448. c.lg.Info(
  449. "promote member",
  450. zap.String("cluster-id", c.cid.String()),
  451. zap.String("local-member-id", c.localID.String()),
  452. )
  453. } else {
  454. plog.Noticef("promote member %s in cluster %s", id, c.cid)
  455. }
  456. }
  457. func (c *RaftCluster) UpdateRaftAttributes(id types.ID, raftAttr RaftAttributes) {
  458. c.Lock()
  459. defer c.Unlock()
  460. c.members[id].RaftAttributes = raftAttr
  461. if c.v2store != nil {
  462. mustUpdateMemberInStore(c.v2store, c.members[id])
  463. }
  464. if c.be != nil {
  465. mustSaveMemberToBackend(c.be, c.members[id])
  466. }
  467. if c.lg != nil {
  468. c.lg.Info(
  469. "updated member",
  470. zap.String("cluster-id", c.cid.String()),
  471. zap.String("local-member-id", c.localID.String()),
  472. zap.String("updated-remote-peer-id", id.String()),
  473. zap.Strings("updated-remote-peer-urls", raftAttr.PeerURLs),
  474. )
  475. } else {
  476. plog.Noticef("updated member %s %v in cluster %s", id, raftAttr.PeerURLs, c.cid)
  477. }
  478. }
  479. func (c *RaftCluster) Version() *semver.Version {
  480. c.Lock()
  481. defer c.Unlock()
  482. if c.version == nil {
  483. return nil
  484. }
  485. return semver.Must(semver.NewVersion(c.version.String()))
  486. }
  487. func (c *RaftCluster) SetVersion(ver *semver.Version, onSet func(*zap.Logger, *semver.Version)) {
  488. c.Lock()
  489. defer c.Unlock()
  490. if c.version != nil {
  491. if c.lg != nil {
  492. c.lg.Info(
  493. "updated cluster version",
  494. zap.String("cluster-id", c.cid.String()),
  495. zap.String("local-member-id", c.localID.String()),
  496. zap.String("from", version.Cluster(c.version.String())),
  497. zap.String("from", version.Cluster(ver.String())),
  498. )
  499. } else {
  500. plog.Noticef("updated the cluster version from %v to %v", version.Cluster(c.version.String()), version.Cluster(ver.String()))
  501. }
  502. } else {
  503. if c.lg != nil {
  504. c.lg.Info(
  505. "set initial cluster version",
  506. zap.String("cluster-id", c.cid.String()),
  507. zap.String("local-member-id", c.localID.String()),
  508. zap.String("cluster-version", version.Cluster(ver.String())),
  509. )
  510. } else {
  511. plog.Noticef("set the initial cluster version to %v", version.Cluster(ver.String()))
  512. }
  513. }
  514. oldVer := c.version
  515. c.version = ver
  516. mustDetectDowngrade(c.lg, c.version)
  517. if c.v2store != nil {
  518. mustSaveClusterVersionToStore(c.v2store, ver)
  519. }
  520. if c.be != nil {
  521. mustSaveClusterVersionToBackend(c.be, ver)
  522. }
  523. if oldVer != nil {
  524. ClusterVersionMetrics.With(prometheus.Labels{"cluster_version": version.Cluster(oldVer.String())}).Set(0)
  525. }
  526. ClusterVersionMetrics.With(prometheus.Labels{"cluster_version": version.Cluster(ver.String())}).Set(1)
  527. onSet(c.lg, ver)
  528. }
  529. func (c *RaftCluster) IsReadyToAddVotingMember() bool {
  530. nmembers := 1
  531. nstarted := 0
  532. for _, member := range c.VotingMembers() {
  533. if member.IsStarted() {
  534. nstarted++
  535. }
  536. nmembers++
  537. }
  538. if nstarted == 1 && nmembers == 2 {
  539. // a case of adding a new node to 1-member cluster for restoring cluster data
  540. // https://github.com/etcd-io/etcd/blob/master/Documentation/v2/admin_guide.md#restoring-the-cluster
  541. if c.lg != nil {
  542. c.lg.Debug("number of started member is 1; can accept add member request")
  543. } else {
  544. plog.Debugf("The number of started member is 1. This cluster can accept add member request.")
  545. }
  546. return true
  547. }
  548. nquorum := nmembers/2 + 1
  549. if nstarted < nquorum {
  550. if c.lg != nil {
  551. c.lg.Warn(
  552. "rejecting member add; started member will be less than quorum",
  553. zap.Int("number-of-started-member", nstarted),
  554. zap.Int("quorum", nquorum),
  555. zap.String("cluster-id", c.cid.String()),
  556. zap.String("local-member-id", c.localID.String()),
  557. )
  558. } else {
  559. plog.Warningf("Reject add member request: the number of started member (%d) will be less than the quorum number of the cluster (%d)", nstarted, nquorum)
  560. }
  561. return false
  562. }
  563. return true
  564. }
  565. func (c *RaftCluster) IsReadyToRemoveVotingMember(id uint64) bool {
  566. nmembers := 0
  567. nstarted := 0
  568. for _, member := range c.VotingMembers() {
  569. if uint64(member.ID) == id {
  570. continue
  571. }
  572. if member.IsStarted() {
  573. nstarted++
  574. }
  575. nmembers++
  576. }
  577. nquorum := nmembers/2 + 1
  578. if nstarted < nquorum {
  579. if c.lg != nil {
  580. c.lg.Warn(
  581. "rejecting member remove; started member will be less than quorum",
  582. zap.Int("number-of-started-member", nstarted),
  583. zap.Int("quorum", nquorum),
  584. zap.String("cluster-id", c.cid.String()),
  585. zap.String("local-member-id", c.localID.String()),
  586. )
  587. } else {
  588. plog.Warningf("Reject remove member request: the number of started member (%d) will be less than the quorum number of the cluster (%d)", nstarted, nquorum)
  589. }
  590. return false
  591. }
  592. return true
  593. }
  594. func (c *RaftCluster) IsReadyToPromoteMember(id uint64) bool {
  595. nmembers := 1
  596. nstarted := 0
  597. for _, member := range c.VotingMembers() {
  598. if member.IsStarted() {
  599. nstarted++
  600. }
  601. nmembers++
  602. }
  603. nquorum := nmembers/2 + 1
  604. if nstarted < nquorum {
  605. if c.lg != nil {
  606. c.lg.Warn(
  607. "rejecting member promote; started member will be less than quorum",
  608. zap.Int("number-of-started-member", nstarted),
  609. zap.Int("quorum", nquorum),
  610. zap.String("cluster-id", c.cid.String()),
  611. zap.String("local-member-id", c.localID.String()),
  612. )
  613. } else {
  614. plog.Warningf("Reject promote member request: the number of started member (%d) will be less than the quorum number of the cluster (%d)", nstarted, nquorum)
  615. }
  616. return false
  617. }
  618. return true
  619. }
  620. func membersFromStore(lg *zap.Logger, st v2store.Store) (map[types.ID]*Member, map[types.ID]bool) {
  621. members := make(map[types.ID]*Member)
  622. removed := make(map[types.ID]bool)
  623. e, err := st.Get(StoreMembersPrefix, true, true)
  624. if err != nil {
  625. if isKeyNotFound(err) {
  626. return members, removed
  627. }
  628. if lg != nil {
  629. lg.Panic("failed to get members from store", zap.String("path", StoreMembersPrefix), zap.Error(err))
  630. } else {
  631. plog.Panicf("get storeMembers should never fail: %v", err)
  632. }
  633. }
  634. for _, n := range e.Node.Nodes {
  635. var m *Member
  636. m, err = nodeToMember(n)
  637. if err != nil {
  638. if lg != nil {
  639. lg.Panic("failed to nodeToMember", zap.Error(err))
  640. } else {
  641. plog.Panicf("nodeToMember should never fail: %v", err)
  642. }
  643. }
  644. members[m.ID] = m
  645. }
  646. e, err = st.Get(storeRemovedMembersPrefix, true, true)
  647. if err != nil {
  648. if isKeyNotFound(err) {
  649. return members, removed
  650. }
  651. if lg != nil {
  652. lg.Panic(
  653. "failed to get removed members from store",
  654. zap.String("path", storeRemovedMembersPrefix),
  655. zap.Error(err),
  656. )
  657. } else {
  658. plog.Panicf("get storeRemovedMembers should never fail: %v", err)
  659. }
  660. }
  661. for _, n := range e.Node.Nodes {
  662. removed[MustParseMemberIDFromKey(n.Key)] = true
  663. }
  664. return members, removed
  665. }
  666. func clusterVersionFromStore(lg *zap.Logger, st v2store.Store) *semver.Version {
  667. e, err := st.Get(path.Join(storePrefix, "version"), false, false)
  668. if err != nil {
  669. if isKeyNotFound(err) {
  670. return nil
  671. }
  672. if lg != nil {
  673. lg.Panic(
  674. "failed to get cluster version from store",
  675. zap.String("path", path.Join(storePrefix, "version")),
  676. zap.Error(err),
  677. )
  678. } else {
  679. plog.Panicf("unexpected error (%v) when getting cluster version from store", err)
  680. }
  681. }
  682. return semver.Must(semver.NewVersion(*e.Node.Value))
  683. }
  684. // ValidateClusterAndAssignIDs validates the local cluster by matching the PeerURLs
  685. // with the existing cluster. If the validation succeeds, it assigns the IDs
  686. // from the existing cluster to the local cluster.
  687. // If the validation fails, an error will be returned.
  688. func ValidateClusterAndAssignIDs(lg *zap.Logger, local *RaftCluster, existing *RaftCluster) error {
  689. ems := existing.Members()
  690. lms := local.Members()
  691. if len(ems) != len(lms) {
  692. return fmt.Errorf("member count is unequal")
  693. }
  694. sort.Sort(MembersByPeerURLs(ems))
  695. sort.Sort(MembersByPeerURLs(lms))
  696. ctx, cancel := context.WithTimeout(context.TODO(), 30*time.Second)
  697. defer cancel()
  698. for i := range ems {
  699. if ok, err := netutil.URLStringsEqual(ctx, lg, ems[i].PeerURLs, lms[i].PeerURLs); !ok {
  700. return fmt.Errorf("unmatched member while checking PeerURLs (%v)", err)
  701. }
  702. lms[i].ID = ems[i].ID
  703. }
  704. local.members = make(map[types.ID]*Member)
  705. for _, m := range lms {
  706. local.members[m.ID] = m
  707. }
  708. return nil
  709. }
  710. func mustDetectDowngrade(lg *zap.Logger, cv *semver.Version) {
  711. lv := semver.Must(semver.NewVersion(version.Version))
  712. // only keep major.minor version for comparison against cluster version
  713. lv = &semver.Version{Major: lv.Major, Minor: lv.Minor}
  714. if cv != nil && lv.LessThan(*cv) {
  715. if lg != nil {
  716. lg.Fatal(
  717. "invalid downgrade; server version is lower than determined cluster version",
  718. zap.String("current-server-version", version.Version),
  719. zap.String("determined-cluster-version", version.Cluster(cv.String())),
  720. )
  721. } else {
  722. plog.Fatalf("cluster cannot be downgraded (current version: %s is lower than determined cluster version: %s).", version.Version, version.Cluster(cv.String()))
  723. }
  724. }
  725. }
  726. // IsLocalMemberLearner returns if the local member is raft learner
  727. func (c *RaftCluster) IsLocalMemberLearner() bool {
  728. c.Lock()
  729. defer c.Unlock()
  730. localMember, ok := c.members[c.localID]
  731. if !ok {
  732. if c.lg != nil {
  733. c.lg.Panic(
  734. "failed to find local ID in cluster members",
  735. zap.String("cluster-id", c.cid.String()),
  736. zap.String("local-member-id", c.localID.String()),
  737. )
  738. } else {
  739. plog.Panicf("failed to find local ID %s in cluster %s", c.localID.String(), c.cid.String())
  740. }
  741. }
  742. return localMember.IsLearner
  743. }
  744. // IsMemberExist returns if the member with the given id exists in cluster.
  745. func (c *RaftCluster) IsMemberExist(id types.ID) bool {
  746. c.Lock()
  747. defer c.Unlock()
  748. _, ok := c.members[id]
  749. return ok
  750. }
  751. // VotingMemberIDs returns the ID of voting members in cluster.
  752. func (c *RaftCluster) VotingMemberIDs() []types.ID {
  753. c.Lock()
  754. defer c.Unlock()
  755. var ids []types.ID
  756. for _, m := range c.members {
  757. if !m.IsLearner {
  758. ids = append(ids, m.ID)
  759. }
  760. }
  761. sort.Sort(types.IDSlice(ids))
  762. return ids
  763. }