clientv3/*: add 'MoveLeader' method to 'Maintenance'

Signed-off-by: Gyu-Ho Lee <gyuhox@gmail.com>

clientv3/integration/maintenance_test.go

@@ -0,0 +1,53 @@
+// Copyright 2017 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package integration
+
+import (
+	"context"
+	"testing"
+
+	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+	"github.com/coreos/etcd/integration"
+	"github.com/coreos/etcd/pkg/testutil"
+)
+
+func TestMaintenanceMoveLeader(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
+	oldLeadIdx := clus.WaitLeader(t)
+	targetIdx := (oldLeadIdx + 1) % 3
+	target := uint64(clus.Members[targetIdx].ID())
+
+	cli := clus.Client(targetIdx)
+	_, err := cli.MoveLeader(context.Background(), target)
+	if err != rpctypes.ErrNotLeader {
+		t.Fatalf("error expected %v, got %v", rpctypes.ErrNotLeader, err)
+	}
+
+	cli = clus.Client(oldLeadIdx)
+	_, err = cli.MoveLeader(context.Background(), target)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	leadIdx := clus.WaitLeader(t)
+	lead := uint64(clus.Members[leadIdx].ID())
+	if target != lead {
+		t.Fatalf("new leader expected %d, got %d", target, lead)
+	}
+}

clientv3/maintenance.go

@@ -28,6 +28,7 @@ type (
 	AlarmResponse      pb.AlarmResponse
 	AlarmMember        pb.AlarmMember
 	StatusResponse     pb.StatusResponse
+	MoveLeaderResponse pb.MoveLeaderResponse
 )
 
 type Maintenance interface {
@@ -51,6 +52,10 @@ type Maintenance interface {
 
 	// Snapshot provides a reader for a snapshot of a backend.
 	Snapshot(ctx context.Context) (io.ReadCloser, error)
+
+	// MoveLeader requests current leader to transfer its leadership to the transferee.
+	// Request must be made to the leader.
+	MoveLeader(ctx context.Context, transfereeID uint64) (*MoveLeaderResponse, error)
 }
 
 type maintenance struct {
@@ -180,3 +185,8 @@ func (m *maintenance) Snapshot(ctx context.Context) (io.ReadCloser, error) {
 	}()
 	return pr, nil
 }
+
+func (m *maintenance) MoveLeader(ctx context.Context, transfereeID uint64) (*MoveLeaderResponse, error) {
+	resp, err := m.remote.MoveLeader(ctx, &pb.MoveLeaderRequest{TargetID: transfereeID}, grpc.FailFast(false))
+	return (*MoveLeaderResponse)(resp), toErr(ctx, err)
+}

integration/cluster.go

@@ -578,6 +578,8 @@ func (m *member) electionTimeout() time.Duration {
 	return time.Duration(m.s.Cfg.ElectionTicks) * time.Millisecond
 }
 
+func (m *member) ID() types.ID { return m.s.ID() }
+
 func (m *member) DropConnections()    { m.grpcBridge.Reset() }
 func (m *member) PauseConnections()   { m.grpcBridge.Pause() }
 func (m *member) UnpauseConnections() { m.grpcBridge.Unpause() }