api: update leader when peer url updates (tikv#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