api: update leader when peer url updates (#3319)

* update leader when peer url updates

Signed-off-by: Ryan Leung <rleungx@gmail.com>

* add the unit test

Signed-off-by: Ryan Leung <rleungx@gmail.com>

Co-authored-by: Ti Prow Robot <71242396+ti-community-prow-bot@users.noreply.github.com>

server/api/member_test.go

@@ -14,9 +14,12 @@
 package api
 
 import (
+ "bytes"
  "encoding/json"
+ "fmt"
  "io/ioutil"
  "math/rand"
+ "net/http"
  "sort"
  "strings"
 
@@ -96,3 +99,46 @@ func (s *testMemberAPISuite) TestMemberLeader(c *C) {
  c.Assert(got.GetClientUrls(), DeepEquals, leader.GetClientUrls())
  c.Assert(got.GetMemberId(), Equals, leader.GetMemberId())
 }
+
+func (s *testMemberAPISuite) TestChangeLeaderPeerUrls(c *C) {
+ leader := s.servers[0].GetLeader()
+ addr := s.cfgs[rand.Intn(len(s.cfgs))].ClientUrls + apiPrefix + "/api/v1/leader"
+ resp, err := testDialClient.Get(addr)
+ c.Assert(err, IsNil)
+ defer resp.Body.Close()
+ buf, err := ioutil.ReadAll(resp.Body)
+ c.Assert(err, IsNil)
+
+ var got pdpb.Member
+ c.Assert(json.Unmarshal(buf, &got), IsNil)
+ id := got.GetMemberId()
+ peerUrls := got.GetPeerUrls()
+
+ newPeerUrls := []string{"http://127.0.0.1:1111"}
+ changeLeaderPeerUrls(c, leader, id, newPeerUrls)
+ addr = s.cfgs[rand.Intn(len(s.cfgs))].ClientUrls + apiPrefix + "/api/v1/members"
+ resp, err = testDialClient.Get(addr)
+ c.Assert(err, IsNil)
+ buf, err = ioutil.ReadAll(resp.Body)
+ c.Assert(err, IsNil)
+ resp.Body.Close()
+ got1 := make(map[string]*pdpb.Member)
+ json.Unmarshal(buf, &got1)
+ c.Assert(got1["leader"].GetPeerUrls(), DeepEquals, newPeerUrls)
+ c.Assert(got1["etcd_leader"].GetPeerUrls(), DeepEquals, newPeerUrls)
+
+ // reset
+ changeLeaderPeerUrls(c, leader, id, peerUrls)
+}
+
+func changeLeaderPeerUrls(c *C, leader *pdpb.Member, id uint64, urls []string) {
+ data := map[string][]string{"peerURLs": urls}
+ postData, err := json.Marshal(data)
+ c.Assert(err, IsNil)
+ req, err := http.NewRequest("PUT", fmt.Sprintf("%s/v2/members/%s", leader.GetClientUrls()[0], fmt.Sprintf("%x", id)), bytes.NewBuffer(postData))
+ c.Assert(err, IsNil)
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := testDialClient.Do(req)
+ c.Assert(err, IsNil)
+ c.Assert(resp.StatusCode, Equals, 204)
+}

server/grpc_service.go

@@ -59,7 +59,7 @@ func (s *Server) GetMembers(context.Context, *pdpb.GetMembersRequest) (*pdpb.Get
  return nil, status.Errorf(codes.Unknown, err.Error())
  }
 
- var etcdLeader *pdpb.Member
+ var etcdLeader, pdLeader *pdpb.Member
  leadID := s.member.GetEtcdLeader()
  for _, m := range members {
  if m.MemberId == leadID {
@@ -74,10 +74,18 @@ func (s *Server) GetMembers(context.Context, *pdpb.GetMembersRequest) (*pdpb.Get
  return nil, status.Errorf(codes.Unknown, err.Error())
  }
 
+ leader := s.member.GetLeader()
+ for _, m := range members {
+ if m.MemberId == leader.GetMemberId() {
+ pdLeader = m
+ break
+ }
+ }
+
  return &pdpb.GetMembersResponse{
  Header: s.header(),
  Members: members,
- Leader: s.member.GetLeader(),
+ Leader: pdLeader,
  EtcdLeader: etcdLeader,
  TsoAllocatorLeaders: tsoAllocatorLeaders,
  }, nil