cherry pick tikv#3305 to release-5.0-rc

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>
ti-srebot · Dec 28, 2020 · 5a5e578 · 5a5e578
1 parent fa65de5
commit 5a5e578
Show file tree

Hide file tree

Showing 3 changed files with 58 additions and 8 deletions.
diff --git a/server/id/id.go b/server/id/id.go
@@ -55,12 +55,11 @@ func (alloc *AllocatorImpl) Alloc() (uint64, error) {
  defer alloc.mu.Unlock()
 
  if alloc.base == alloc.end {
- end, err := alloc.generate()
+ err := alloc.Generate()
  if err != nil {
  return 0, err
  }
 
- alloc.end = end
  alloc.base = alloc.end - allocStep
  }
 
@@ -69,11 +68,12 @@ func (alloc *AllocatorImpl) Alloc() (uint64, error) {
  return alloc.base, nil
 }
 
-func (alloc *AllocatorImpl) generate() (uint64, error) {
+// Generate synchronizes and generates id range.
+func (alloc *AllocatorImpl) Generate() error {
  key := alloc.getAllocIDPath()
  value, err := etcdutil.GetValue(alloc.client, key)
  if err != nil {
- return 0, err
+ return err
  }
 
  var (
@@ -88,7 +88,7 @@ func (alloc *AllocatorImpl) generate() (uint64, error) {
  // update the key
  end, err = typeutil.BytesToUint64(value)
  if err != nil {
- return 0, err
+ return err
  }
 
  cmp = clientv3.Compare(clientv3.Value(key), "=", string(value))
@@ -101,15 +101,16 @@ func (alloc *AllocatorImpl) generate() (uint64, error) {
  t := txn.If(append([]clientv3.Cmp{cmp}, clientv3.Compare(clientv3.Value(leaderPath), "=", alloc.member))...)
  resp, err := t.Then(clientv3.OpPut(key, string(value))).Commit()
  if err != nil {
- return 0, errs.ErrEtcdTxn.Wrap(err).GenWithStackByArgs()
+ return errs.ErrEtcdTxn.Wrap(err).GenWithStackByArgs()
  }
  if !resp.Succeeded {
- return 0, errs.ErrEtcdTxn.FastGenByArgs()
+ return errs.ErrEtcdTxn.FastGenByArgs()
  }
 
  log.Info("idAllocator allocates a new id", zap.Uint64("alloc-id", end))
  idGauge.WithLabelValues("idalloc").Set(float64(end))
- return end, nil
+ alloc.end = end
+ return nil
 }
 
 func (alloc *AllocatorImpl) getAllocIDPath() string {

diff --git a/server/server.go b/server/server.go
@@ -1223,6 +1223,10 @@ func (s *Server) campaignLeader() {
  log.Error("failed to load persistOptions from etcd", errs.ZapError(err))
  return
  }
+ if err := s.idAllocator.Generate(); err != nil {
+ log.Error("failed to sync id from etcd", errs.ZapError(err))
+ return
+ }
  s.member.EnableLeader()
 
  CheckPDVersion(s.persistOptions)

diff --git a/tests/server/id/id_test.go b/tests/server/id/id_test.go
@@ -119,3 +119,48 @@ func (s *testAllocIDSuite) TestCommand(c *C) {
  last = resp.GetId()
  }
 }
+
+func (s *testAllocIDSuite) TestMonotonicID(c *C) {
+ var err error
+ cluster, err := tests.NewTestCluster(s.ctx, 2)
+ defer cluster.Destroy()
+ c.Assert(err, IsNil)
+
+ err = cluster.RunInitialServers()
+ c.Assert(err, IsNil)
+ cluster.WaitLeader()
+
+ leaderServer := cluster.GetServer(cluster.GetLeader())
+ var last1 uint64
+ for i := uint64(0); i < 10; i++ {
+ id, err := leaderServer.GetAllocator().Alloc()
+ c.Assert(err, IsNil)
+ c.Assert(id, Greater, last1)
+ last1 = id
+ }
+ err = cluster.ResignLeader()
+ c.Assert(err, IsNil)
+ cluster.WaitLeader()
+ leaderServer = cluster.GetServer(cluster.GetLeader())
+ var last2 uint64
+ for i := uint64(0); i < 10; i++ {
+ id, err := leaderServer.GetAllocator().Alloc()
+ c.Assert(err, IsNil)
+ c.Assert(id, Greater, last2)
+ last2 = id
+ }
+ err = cluster.ResignLeader()
+ c.Assert(err, IsNil)
+ cluster.WaitLeader()
+ leaderServer = cluster.GetServer(cluster.GetLeader())
+ id, err := leaderServer.GetAllocator().Alloc()
+ c.Assert(err, IsNil)
+ c.Assert(id, Greater, last2)
+ var last3 uint64
+ for i := uint64(0); i < 1000; i++ {
+ id, err := leaderServer.GetAllocator().Alloc()
+ c.Assert(err, IsNil)
+ c.Assert(id, Greater, last3)
+ last3 = id
+ }
+}