Skip to content

Commit

Permalink
client: add a mechanism for various endpoint selection mode
Browse files Browse the repository at this point in the history
Current etcd client library chooses a default destination node from
every member of a cluster in a random manner. However, requests of
write and read (for consistent results) need to be forwarded to the
leader node as the nature of Raft algorithm. If the chosen node is a
follower, additional network traffic will be caused by the forwarding
from follower to leader.

Mainly for reducing the forward traffic, this commit adds a new
mechanism for various endpoint selection mode to the client library
which can be configured with client.Config.SelectionMode.

Currently, two modes are provided:
 - EndpointSelectionRandom: default, same to existing behavior (pick
   a node in a random manner)
 - EndpointSelectionPrioritizeLeader: prioritize leader, for the above
   purpose

I evaluated the effectiveness of the EndpointSelectionPrioritizeLeader
with 4 t1.micro instances of AWS (3 nodes for etcd cluster and 1 node
for etcd client). Client executes this simple benchmark
(https://github.com/mitake/etcd-things/tree/master/prioritize-leader-bench),
just writes 10000 keys. When SelectionMode == EndpointSelectionRandom
(default), the benchmark needed 1 min and 32.102 sec to finish. When
SelectionMode == EndpointSelectionPrioritizeLeader, the benchmark
needed 1 min 4.760 sec.
  • Loading branch information
mitake committed Dec 22, 2015
1 parent 1a8bf58 commit 7616765
Showing 1 changed file with 55 additions and 4 deletions.
59 changes: 55 additions & 4 deletions client/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,19 @@ var DefaultTransport CancelableTransport = &http.Transport{
TLSHandshakeTimeout: 10 * time.Second,
}

type EndpointSelectionMode int

const (
// EndpointSelectionRandom is to pick an endpoint in a random manner.
EndpointSelectionRandom EndpointSelectionMode = iota

// EndpointSelectionPrioritizeLeader is to prioritize leader for reducing needless
// forward between follower and leader.
//
// This mode should be used with Client.AutoSync().
EndpointSelectionPrioritizeLeader
)

type Config struct {
// Endpoints defines a set of URLs (schemes, hosts and ports only)
// that can be used to communicate with a logical etcd cluster. For
Expand Down Expand Up @@ -104,6 +117,9 @@ type Config struct {
//
// A HeaderTimeoutPerRequest of zero means no timeout.
HeaderTimeoutPerRequest time.Duration

// SelectionMode specifies a way of selecting destination endpoint.
SelectionMode EndpointSelectionMode
}

func (cfg *Config) transport() CancelableTransport {
Expand Down Expand Up @@ -169,6 +185,7 @@ func New(cfg Config) (Client, error) {
c := &httpClusterClient{
clientFactory: newHTTPClientFactory(cfg.transport(), cfg.checkRedirect(), cfg.HeaderTimeoutPerRequest),
rand: rand.New(rand.NewSource(int64(time.Now().Nanosecond()))),
selectionMode: cfg.SelectionMode,
}
if cfg.Username != "" {
c.credentials = &credentials{
Expand Down Expand Up @@ -216,7 +233,18 @@ type httpClusterClient struct {
pinned int
credentials *credentials
sync.RWMutex
rand *rand.Rand
rand *rand.Rand
selectionMode EndpointSelectionMode
}

func (c *httpClusterClient) getLeaderEndpoint() (string, error) {
mAPI := NewMembersAPI(c)
leader, err := mAPI.Leader(context.Background())
if err != nil {
return "", err
}

return leader.ClientURLs[0], nil // TODO: how to handle multiple client URLs?
}

func (c *httpClusterClient) reset(eps []string) error {
Expand All @@ -233,9 +261,32 @@ func (c *httpClusterClient) reset(eps []string) error {
neps[i] = *u
}

c.endpoints = shuffleEndpoints(c.rand, neps)
// TODO: pin old endpoint if possible, and rebalance when new endpoint appears
c.pinned = 0
switch c.selectionMode {
case EndpointSelectionRandom:
c.endpoints = shuffleEndpoints(c.rand, neps)
c.pinned = 0
case EndpointSelectionPrioritizeLeader:
c.endpoints = neps
// TODO: should return ErrNoEndpoints in a case of getting leader fail?
lep, err := c.getLeaderEndpoint()
if err != nil {
return ErrNoEndpoints
}

lu, err := url.Parse(lep)
if err != nil {
return ErrNoEndpoints
}

for i := range c.endpoints {
if c.endpoints[i].String() == lu.String() {
c.pinned = i
break
}
}
default:
return errors.New(fmt.Sprintf("invalid mode of endpoint selection: %d", c.selectionMode))
}

return nil
}
Expand Down

0 comments on commit 7616765

Please sign in to comment.