election.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. // Copyright 2016 CoreOS, Inc.
  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 recipe
  15. import (
  16. "github.com/coreos/etcd/clientv3"
  17. "github.com/coreos/etcd/etcdserver"
  18. "github.com/coreos/etcd/storage/storagepb"
  19. )
  20. type Election struct {
  21. client *clientv3.Client
  22. keyPrefix string
  23. leaderKey *EphemeralKV
  24. }
  25. // NewElection returns a new election on a given key prefix.
  26. func NewElection(client *clientv3.Client, keyPrefix string) *Election {
  27. return &Election{client, keyPrefix, nil}
  28. }
  29. // Volunteer puts a value as eligible for the election. It blocks until
  30. // it is elected or an error occurs (cannot withdraw candidacy)
  31. func (e *Election) Volunteer(val string) error {
  32. if e.leaderKey != nil {
  33. return e.leaderKey.Put(val)
  34. }
  35. myKey, err := NewUniqueEphemeralKV(e.client, e.keyPrefix, val)
  36. if err != nil {
  37. return err
  38. }
  39. if err = e.waitLeadership(myKey); err != nil {
  40. return err
  41. }
  42. e.leaderKey = myKey
  43. return nil
  44. }
  45. // Resign lets a leader start a new election.
  46. func (e *Election) Resign() (err error) {
  47. if e.leaderKey != nil {
  48. err = e.leaderKey.Delete()
  49. e.leaderKey = nil
  50. }
  51. return err
  52. }
  53. // Leader returns the leader value for the current election.
  54. func (e *Election) Leader() (string, error) {
  55. resp, err := NewRange(e.client, e.keyPrefix).FirstCreate()
  56. if err != nil {
  57. return "", err
  58. } else if len(resp.Kvs) == 0 {
  59. // no leader currently elected
  60. return "", etcdserver.ErrNoLeader
  61. }
  62. return string(resp.Kvs[0].Value), nil
  63. }
  64. // Wait waits for a leader to be elected, returning the leader value.
  65. func (e *Election) Wait() (string, error) {
  66. resp, err := NewRange(e.client, e.keyPrefix).FirstCreate()
  67. if err != nil {
  68. return "", err
  69. } else if len(resp.Kvs) != 0 {
  70. // leader already exists
  71. return string(resp.Kvs[0].Value), nil
  72. }
  73. _, err = WaitPrefixEvents(
  74. e.client,
  75. e.keyPrefix,
  76. resp.Header.Revision,
  77. []storagepb.Event_EventType{storagepb.PUT})
  78. if err != nil {
  79. return "", err
  80. }
  81. return e.Wait()
  82. }
  83. func (e *Election) waitLeadership(tryKey *EphemeralKV) error {
  84. resp, err := NewRangeRev(
  85. e.client,
  86. e.keyPrefix,
  87. tryKey.Revision()-1).LastCreate()
  88. if err != nil {
  89. return err
  90. } else if len(resp.Kvs) == 0 {
  91. // nothing before tryKey => have leadership
  92. return nil
  93. }
  94. _, err = WaitEvents(
  95. e.client,
  96. string(resp.Kvs[0].Key),
  97. tryKey.Revision(),
  98. []storagepb.Event_EventType{storagepb.DELETE})
  99. return err
  100. }