version: bump to 3.1.4

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

auth/simple_token.go

@@ -21,33 +21,33 @@ import (
 	"crypto/rand"
 	"math/big"
 	"strings"
+	"sync"
 	"time"
 )
 
 const (
 	letters                  = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
 	defaultSimpleTokenLength = 16
+)
+
+// var for testing purposes
+var (
 	simpleTokenTTL           = 5 * time.Minute
 	simpleTokenTTLResolution = 1 * time.Second
 )
 
 type simpleTokenTTLKeeper struct {
-	tokens              map[string]time.Time
-	addSimpleTokenCh    chan string
-	resetSimpleTokenCh  chan string
-	deleteSimpleTokenCh chan string
-	stopCh              chan chan struct{}
-	deleteTokenFunc     func(string)
+	tokensMu        sync.Mutex
+	tokens          map[string]time.Time
+	stopCh          chan chan struct{}
+	deleteTokenFunc func(string)
 }
 
 func NewSimpleTokenTTLKeeper(deletefunc func(string)) *simpleTokenTTLKeeper {
 	stk := &simpleTokenTTLKeeper{
-		tokens:              make(map[string]time.Time),
-		addSimpleTokenCh:    make(chan string, 1),
-		resetSimpleTokenCh:  make(chan string, 1),
-		deleteSimpleTokenCh: make(chan string, 1),
-		stopCh:              make(chan chan struct{}),
-		deleteTokenFunc:     deletefunc,
+		tokens:          make(map[string]time.Time),
+		stopCh:          make(chan chan struct{}),
+		deleteTokenFunc: deletefunc,
 	}
 	go stk.run()
 	return stk
@@ -61,37 +61,34 @@ func (tm *simpleTokenTTLKeeper) stop() {
 }
 
 func (tm *simpleTokenTTLKeeper) addSimpleToken(token string) {
-	tm.addSimpleTokenCh <- token
+	tm.tokens[token] = time.Now().Add(simpleTokenTTL)
 }
 
 func (tm *simpleTokenTTLKeeper) resetSimpleToken(token string) {
-	tm.resetSimpleTokenCh <- token
+	if _, ok := tm.tokens[token]; ok {
+		tm.tokens[token] = time.Now().Add(simpleTokenTTL)
+	}
 }
 
 func (tm *simpleTokenTTLKeeper) deleteSimpleToken(token string) {
-	tm.deleteSimpleTokenCh <- token
+	delete(tm.tokens, token)
 }
+
 func (tm *simpleTokenTTLKeeper) run() {
 	tokenTicker := time.NewTicker(simpleTokenTTLResolution)
 	defer tokenTicker.Stop()
 	for {
 		select {
-		case t := <-tm.addSimpleTokenCh:
-			tm.tokens[t] = time.Now().Add(simpleTokenTTL)
-		case t := <-tm.resetSimpleTokenCh:
-			if _, ok := tm.tokens[t]; ok {
-				tm.tokens[t] = time.Now().Add(simpleTokenTTL)
-			}
-		case t := <-tm.deleteSimpleTokenCh:
-			delete(tm.tokens, t)
 		case <-tokenTicker.C:
 			nowtime := time.Now()
+			tm.tokensMu.Lock()
 			for t, tokenendtime := range tm.tokens {
 				if nowtime.After(tokenendtime) {
 					tm.deleteTokenFunc(t)
 					delete(tm.tokens, t)
 				}
 			}
+			tm.tokensMu.Unlock()
 		case waitCh := <-tm.stopCh:
 			tm.tokens = make(map[string]time.Time)
 			waitCh <- struct{}{}
@@ -116,6 +113,7 @@ func (as *authStore) GenSimpleToken() (string, error) {
 }
 
 func (as *authStore) assignSimpleTokenToUser(username, token string) {
+	as.simpleTokenKeeper.tokensMu.Lock()
 	as.simpleTokensMu.Lock()
 
 	_, ok := as.simpleTokens[token]
@@ -126,16 +124,21 @@ func (as *authStore) assignSimpleTokenToUser(username, token string) {
 	as.simpleTokens[token] = username
 	as.simpleTokenKeeper.addSimpleToken(token)
 	as.simpleTokensMu.Unlock()
+	as.simpleTokenKeeper.tokensMu.Unlock()
 }
 
 func (as *authStore) invalidateUser(username string) {
+	if as.simpleTokenKeeper == nil {
+		return
+	}
+	as.simpleTokenKeeper.tokensMu.Lock()
 	as.simpleTokensMu.Lock()
-	defer as.simpleTokensMu.Unlock()
-
 	for token, name := range as.simpleTokens {
 		if strings.Compare(name, username) == 0 {
 			delete(as.simpleTokens, token)
 			as.simpleTokenKeeper.deleteSimpleToken(token)
 		}
 	}
+	as.simpleTokensMu.Unlock()
+	as.simpleTokenKeeper.tokensMu.Unlock()
 }

auth/store.go

@@ -168,13 +168,13 @@ type authStore struct {
 
 	rangePermCache map[string]*unifiedRangePermissions // username -> unifiedRangePermissions
 
-	simpleTokensMu    sync.RWMutex
-	simpleTokens      map[string]string // token -> username
-	simpleTokenKeeper *simpleTokenTTLKeeper
-
 	revision uint64
 
-	indexWaiter func(uint64) <-chan struct{}
+	// tokenSimple in v3.2+
+	indexWaiter       func(uint64) <-chan struct{}
+	simpleTokenKeeper *simpleTokenTTLKeeper
+	simpleTokensMu    sync.Mutex
+	simpleTokens      map[string]string // token -> username
 }
 
 func newDeleterFunc(as *authStore) func(string) {
@@ -646,13 +646,16 @@ func (as *authStore) RoleAdd(r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse,
 }
 
 func (as *authStore) AuthInfoFromToken(token string) (*AuthInfo, bool) {
-	as.simpleTokensMu.RLock()
-	defer as.simpleTokensMu.RUnlock()
-	t, ok := as.simpleTokens[token]
+	// same as '(t *tokenSimple) info' in v3.2+
+	as.simpleTokenKeeper.tokensMu.Lock()
+	as.simpleTokensMu.Lock()
+	username, ok := as.simpleTokens[token]
 	if ok {
 		as.simpleTokenKeeper.resetSimpleToken(token)
 	}
-	return &AuthInfo{Username: t, Revision: as.revision}, ok
+	as.simpleTokensMu.Unlock()
+	as.simpleTokenKeeper.tokensMu.Unlock()
+	return &AuthInfo{Username: username, Revision: as.revision}, ok
 }
 
 type permSlice []*authpb.Permission
@@ -764,6 +767,9 @@ func (as *authStore) IsAdminPermitted(authInfo *AuthInfo) error {
 	if !as.isAuthEnabled() {
 		return nil
 	}
+	if authInfo == nil {
+		return ErrUserEmpty
+	}
 
 	tx := as.be.BatchTx()
 	tx.Lock()
@@ -911,7 +917,9 @@ func NewAuthStore(be backend.Backend, indexWaiter func(uint64) <-chan struct{})
 		as.simpleTokenKeeper = NewSimpleTokenTTLKeeper(newDeleterFunc(as))
 	}
 
-	as.commitRevision(tx)
+	if as.revision == 0 {
+		as.commitRevision(tx)
+	}
 
 	tx.Unlock()
 	be.ForceCommit()

auth/store_test.go

@@ -34,31 +34,30 @@ func dummyIndexWaiter(index uint64) <-chan struct{} {
 	return ch
 }
 
-func TestUserAdd(t *testing.T) {
+// TestNewAuthStoreRevision ensures newly auth store
+// keeps the old revision when there are no changes.
+func TestNewAuthStoreRevision(t *testing.T) {
 	b, tPath := backend.NewDefaultTmpBackend()
-	defer func() {
-		b.Close()
-		os.Remove(tPath)
-	}()
+	defer os.Remove(tPath)
 
 	as := NewAuthStore(b, dummyIndexWaiter)
-	ua := &pb.AuthUserAddRequest{Name: "foo"}
-	_, err := as.UserAdd(ua) // add a non-existing user
+	err := enableAuthAndCreateRoot(as)
 	if err != nil {
 		t.Fatal(err)
 	}
-	_, err = as.UserAdd(ua) // add an existing user
-	if err == nil {
-		t.Fatalf("expected %v, got %v", ErrUserAlreadyExist, err)
-	}
-	if err != ErrUserAlreadyExist {
-		t.Fatalf("expected %v, got %v", ErrUserAlreadyExist, err)
-	}
+	old := as.Revision()
+	b.Close()
+	as.Close()
 
-	ua = &pb.AuthUserAddRequest{Name: ""}
-	_, err = as.UserAdd(ua) // add a user with empty name
-	if err != ErrUserEmpty {
-		t.Fatal(err)
+	// no changes to commit
+	b2 := backend.NewDefaultBackend(tPath)
+	as = NewAuthStore(b2, dummyIndexWaiter)
+	new := as.Revision()
+	b2.Close()
+	as.Close()
+
+	if old != new {
+		t.Fatalf("expected revision %d, got %d", old, new)
 	}
 }
 

clientv3/integration/lease_test.go

@@ -156,6 +156,30 @@ func TestLeaseKeepAlive(t *testing.T) {
 	}
 }
 
+func TestLeaseKeepAliveOneSecond(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	cli := clus.Client(0)
+
+	resp, err := cli.Grant(context.Background(), 1)
+	if err != nil {
+		t.Errorf("failed to create lease %v", err)
+	}
+	rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
+	if kerr != nil {
+		t.Errorf("failed to keepalive lease %v", kerr)
+	}
+
+	for i := 0; i < 3; i++ {
+		if _, ok := <-rc; !ok {
+			t.Errorf("chan is closed, want not closed")
+		}
+	}
+}
+
 // TODO: add a client that can connect to all the members of cluster via unix sock.
 // TODO: test handle more complicated failures.
 func TestLeaseKeepAliveHandleFailure(t *testing.T) {

clientv3/lease.go

@@ -416,7 +416,7 @@ func (l *lessor) recvKeepAlive(resp *pb.LeaseKeepAliveResponse) {
 	}
 
 	// send update to all channels
-	nextKeepAlive := time.Now().Add(time.Duration(karesp.TTL+2) / 3 * time.Second)
+	nextKeepAlive := time.Now().Add((time.Duration(karesp.TTL) * time.Second) / 3.0)
 	ka.deadline = time.Now().Add(time.Duration(karesp.TTL) * time.Second)
 	for _, ch := range ka.chs {
 		select {

clientv3/watch.go

@@ -694,6 +694,10 @@ func (w *watchGrpcStream) waitCancelSubstreams(stopc <-chan struct{}) <-chan str
 		go func(ws *watcherStream) {
 			defer wg.Done()
 			if ws.closing {
+				if ws.initReq.ctx.Err() != nil && ws.outc != nil {
+					close(ws.outc)
+					ws.outc = nil
+				}
 				return
 			}
 			select {

discovery/srv.go

@@ -74,7 +74,7 @@ func SRVGetCluster(name, dns string, defaultToken string, apurls types.URLs) (st
 			shortHost := strings.TrimSuffix(srv.Target, ".")
 			urlHost := net.JoinHostPort(shortHost, port)
 			stringParts = append(stringParts, fmt.Sprintf("%s=%s://%s", n, scheme, urlHost))
-			plog.Noticef("got bootstrap from DNS for %s at %s%s", service, scheme, urlHost)
+			plog.Noticef("got bootstrap from DNS for %s at %s://%s", service, scheme, urlHost)
 			if ok && url.Scheme != scheme {
 				plog.Errorf("bootstrap at %s from DNS for %s has scheme mismatch with expected peer %s", scheme+"://"+urlHost, service, url.String())
 			}

embed/config.go

@@ -55,20 +55,12 @@ var (
 	DefaultInitialAdvertisePeerURLs = "http://localhost:2380"
 	DefaultAdvertiseClientURLs      = "http://localhost:2379"
 
-	defaultHostname   string = "localhost"
+	defaultHostname   string
 	defaultHostStatus error
 )
 
 func init() {
-	ip, err := netutil.GetDefaultHost()
-	if err != nil {
-		defaultHostStatus = err
-		return
-	}
-	// found default host, advertise on it
-	DefaultInitialAdvertisePeerURLs = "http://" + ip + ":2380"
-	DefaultAdvertiseClientURLs = "http://" + ip + ":2379"
-	defaultHostname = ip
+	defaultHostname, defaultHostStatus = netutil.GetDefaultHost()
 }
 
 // Config holds the arguments for configuring an etcd server.
@@ -237,6 +229,9 @@ func (cfg *configYAML) configFromFile(path string) error {
 		cfg.ACUrls = []url.URL(u)
 	}
 
+	if (cfg.Durl != "" || cfg.DNSCluster != "") && cfg.InitialCluster == cfg.InitialClusterFromName(cfg.Name) {
+		cfg.InitialCluster = ""
+	}
 	if cfg.ClusterState == "" {
 		cfg.ClusterState = ClusterStateFlagNew
 	}
@@ -346,34 +341,52 @@ func (cfg Config) InitialClusterFromName(name string) (ret string) {
 func (cfg Config) IsNewCluster() bool { return cfg.ClusterState == ClusterStateFlagNew }
 func (cfg Config) ElectionTicks() int { return int(cfg.ElectionMs / cfg.TickMs) }
 
-// IsDefaultHost returns the default hostname, if used, and the error, if any,
-// from getting the machine's default host.
-func (cfg Config) IsDefaultHost() (string, error) {
-	if len(cfg.APUrls) == 1 && cfg.APUrls[0].String() == DefaultInitialAdvertisePeerURLs {
-		return defaultHostname, defaultHostStatus
-	}
-	if len(cfg.ACUrls) == 1 && cfg.ACUrls[0].String() == DefaultAdvertiseClientURLs {
-		return defaultHostname, defaultHostStatus
-	}
-	return "", defaultHostStatus
+func (cfg Config) defaultPeerHost() bool {
+	return len(cfg.APUrls) == 1 && cfg.APUrls[0].String() == DefaultInitialAdvertisePeerURLs
 }
 
-// UpdateDefaultClusterFromName updates cluster advertise URLs with default host.
+func (cfg Config) defaultClientHost() bool {
+	return len(cfg.ACUrls) == 1 && cfg.ACUrls[0].String() == DefaultAdvertiseClientURLs
+}
+
+// UpdateDefaultClusterFromName updates cluster advertise URLs with, if available, default host,
+// if advertise URLs are default values(localhost:2379,2380) AND if listen URL is 0.0.0.0.
+// e.g. advertise peer URL localhost:2380 or listen peer URL 0.0.0.0:2380
+// then the advertise peer host would be updated with machine's default host,
+// while keeping the listen URL's port.
+// User can work around this by explicitly setting URL with 127.0.0.1.
+// It returns the default hostname, if used, and the error, if any, from getting the machine's default host.
 // TODO: check whether fields are set instead of whether fields have default value
-func (cfg *Config) UpdateDefaultClusterFromName(defaultInitialCluster string) {
-	defaultHost, defaultHostErr := cfg.IsDefaultHost()
-	defaultHostOverride := defaultHost == "" || defaultHostErr == nil
-	if (defaultHostOverride || cfg.Name != DefaultName) && cfg.InitialCluster == defaultInitialCluster {
-		cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
-		ip, _, _ := net.SplitHostPort(cfg.LCUrls[0].Host)
-		// if client-listen-url is 0.0.0.0, just use detected default host
-		// otherwise, rewrite advertise-client-url with localhost
-		if ip != "0.0.0.0" {
-			_, acPort, _ := net.SplitHostPort(cfg.ACUrls[0].Host)
-			cfg.ACUrls[0] = url.URL{Scheme: cfg.ACUrls[0].Scheme, Host: fmt.Sprintf("localhost:%s", acPort)}
+func (cfg *Config) UpdateDefaultClusterFromName(defaultInitialCluster string) (string, error) {
+	if defaultHostname == "" || defaultHostStatus != nil {
+		// update 'initial-cluster' when only the name is specified (e.g. 'etcd --name=abc')
+		if cfg.Name != DefaultName && cfg.InitialCluster == defaultInitialCluster {
 			cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
 		}
+		return "", defaultHostStatus
 	}
+
+	used := false
+	pip, pport, _ := net.SplitHostPort(cfg.LPUrls[0].Host)
+	if cfg.defaultPeerHost() && pip == "0.0.0.0" {
+		cfg.APUrls[0] = url.URL{Scheme: cfg.APUrls[0].Scheme, Host: fmt.Sprintf("%s:%s", defaultHostname, pport)}
+		used = true
+	}
+	// update 'initial-cluster' when only the name is specified (e.g. 'etcd --name=abc')
+	if cfg.Name != DefaultName && cfg.InitialCluster == defaultInitialCluster {
+		cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
+	}
+
+	cip, cport, _ := net.SplitHostPort(cfg.LCUrls[0].Host)
+	if cfg.defaultClientHost() && cip == "0.0.0.0" {
+		cfg.ACUrls[0] = url.URL{Scheme: cfg.ACUrls[0].Scheme, Host: fmt.Sprintf("%s:%s", defaultHostname, cport)}
+		used = true
+	}
+	dhost := defaultHostname
+	if !used {
+		dhost = ""
+	}
+	return dhost, defaultHostStatus
 }
 
 // checkBindURLs returns an error if any URL uses a domain name.

embed/config_test.go

@@ -15,11 +15,15 @@
 package embed
 
 import (
+	"fmt"
 	"io/ioutil"
+	"net"
+	"net/url"
 	"os"
 	"testing"
 
 	"github.com/coreos/etcd/pkg/transport"
+
 	"github.com/ghodss/yaml"
 )
 
@@ -61,6 +65,70 @@ func TestConfigFileOtherFields(t *testing.T) {
 	}
 }
 
+// TestUpdateDefaultClusterFromName ensures that etcd can start with 'etcd --name=abc'.
+func TestUpdateDefaultClusterFromName(t *testing.T) {
+	cfg := NewConfig()
+	defaultInitialCluster := cfg.InitialCluster
+	oldscheme := cfg.APUrls[0].Scheme
+	origpeer := cfg.APUrls[0].String()
+	origadvc := cfg.ACUrls[0].String()
+
+	cfg.Name = "abc"
+	_, lpport, _ := net.SplitHostPort(cfg.LPUrls[0].Host)
+
+	// in case of 'etcd --name=abc'
+	exp := fmt.Sprintf("%s=%s://localhost:%s", cfg.Name, oldscheme, lpport)
+	cfg.UpdateDefaultClusterFromName(defaultInitialCluster)
+	if exp != cfg.InitialCluster {
+		t.Fatalf("initial-cluster expected %q, got %q", exp, cfg.InitialCluster)
+	}
+	// advertise peer URL should not be affected
+	if origpeer != cfg.APUrls[0].String() {
+		t.Fatalf("advertise peer url expected %q, got %q", origadvc, cfg.APUrls[0].String())
+	}
+	// advertise client URL should not be affected
+	if origadvc != cfg.ACUrls[0].String() {
+		t.Fatalf("advertise client url expected %q, got %q", origadvc, cfg.ACUrls[0].String())
+	}
+}
+
+// TestUpdateDefaultClusterFromNameOverwrite ensures that machine's default host is only used
+// if advertise URLs are default values(localhost:2379,2380) AND if listen URL is 0.0.0.0.
+func TestUpdateDefaultClusterFromNameOverwrite(t *testing.T) {
+	if defaultHostname == "" {
+		t.Skip("machine's default host not found")
+	}
+
+	cfg := NewConfig()
+	defaultInitialCluster := cfg.InitialCluster
+	oldscheme := cfg.APUrls[0].Scheme
+	origadvc := cfg.ACUrls[0].String()
+
+	cfg.Name = "abc"
+	_, lpport, _ := net.SplitHostPort(cfg.LPUrls[0].Host)
+	cfg.LPUrls[0] = url.URL{Scheme: cfg.LPUrls[0].Scheme, Host: fmt.Sprintf("0.0.0.0:%s", lpport)}
+	dhost, _ := cfg.UpdateDefaultClusterFromName(defaultInitialCluster)
+	if dhost != defaultHostname {
+		t.Fatalf("expected default host %q, got %q", defaultHostname, dhost)
+	}
+	aphost, apport, _ := net.SplitHostPort(cfg.APUrls[0].Host)
+	if apport != lpport {
+		t.Fatalf("advertise peer url got different port %s, expected %s", apport, lpport)
+	}
+	if aphost != defaultHostname {
+		t.Fatalf("advertise peer url expected machine default host %q, got %q", defaultHostname, aphost)
+	}
+	expected := fmt.Sprintf("%s=%s://%s:%s", cfg.Name, oldscheme, defaultHostname, lpport)
+	if expected != cfg.InitialCluster {
+		t.Fatalf("initial-cluster expected %q, got %q", expected, cfg.InitialCluster)
+	}
+
+	// advertise client URL should not be affected
+	if origadvc != cfg.ACUrls[0].String() {
+		t.Fatalf("advertise-client-url expected %q, got %q", origadvc, cfg.ACUrls[0].String())
+	}
+}
+
 func (s *securityConfig) equals(t *transport.TLSInfo) bool {
 	return s.CAFile == t.CAFile &&
 		s.CertFile == t.CertFile &&

etcdctl/ctlv3/command/make_mirror_command.go

@@ -125,18 +125,19 @@ func makeMirror(ctx context.Context, c *clientv3.Client, dc *clientv3.Client) er
 			return rpctypes.ErrCompacted
 		}
 
-		var rev int64
+		var lastRev int64
 		ops := []clientv3.Op{}
 
 		for _, ev := range wr.Events {
-			nrev := ev.Kv.ModRevision
-			if rev != 0 && nrev > rev {
+			nextRev := ev.Kv.ModRevision
+			if lastRev != 0 && nextRev > lastRev {
 				_, err := dc.Txn(ctx).Then(ops...).Commit()
 				if err != nil {
 					return err
 				}
 				ops = []clientv3.Op{}
 			}
+			lastRev = nextRev
 			switch ev.Type {
 			case mvccpb.PUT:
 				ops = append(ops, clientv3.OpPut(modifyPrefix(string(ev.Kv.Key)), string(ev.Kv.Value)))

etcdctl/ctlv3/command/member_command.go

@@ -107,7 +107,8 @@ func memberAddCommandFunc(cmd *cobra.Command, args []string) {
 
 	urls := strings.Split(memberPeerURLs, ",")
 	ctx, cancel := commandCtx(cmd)
-	resp, err := mustClientFromCmd(cmd).MemberAdd(ctx, urls)
+	cli := mustClientFromCmd(cmd)
+	resp, err := cli.MemberAdd(ctx, urls)
 	cancel()
 	if err != nil {
 		ExitWithError(ExitError, err)
@@ -118,12 +119,24 @@ func memberAddCommandFunc(cmd *cobra.Command, args []string) {
 
 	if _, ok := (display).(*simplePrinter); ok {
 		ctx, cancel = commandCtx(cmd)
-		listResp, err := mustClientFromCmd(cmd).MemberList(ctx)
-		cancel()
-
-		if err != nil {
-			ExitWithError(ExitError, err)
+		listResp, err := cli.MemberList(ctx)
+		// get latest member list; if there's failover new member might have outdated list
+		for {
+			if err != nil {
+				ExitWithError(ExitError, err)
+			}
+			if listResp.Header.MemberId == resp.Header.MemberId {
+				break
+			}
+			// quorum get to sync cluster list
+			gresp, gerr := cli.Get(ctx, "_")
+			if gerr != nil {
+				ExitWithError(ExitError, err)
+			}
+			resp.Header.MemberId = gresp.Header.MemberId
+			listResp, err = cli.MemberList(ctx)
 		}
+		cancel()
 
 		conf := []string{}
 		for _, memb := range listResp.Members {

etcdctl/ctlv3/ctl.go

@@ -45,7 +45,7 @@ var (
 func init() {
 	rootCmd.PersistentFlags().StringSliceVar(&globalFlags.Endpoints, "endpoints", []string{"127.0.0.1:2379"}, "gRPC endpoints")
 
-	rootCmd.PersistentFlags().StringVarP(&globalFlags.OutputFormat, "write-out", "w", "simple", "set the output format (fields, json, proto, simple, table)")
+	rootCmd.PersistentFlags().StringVarP(&globalFlags.OutputFormat, "write-out", "w", "simple", "set the output format (fields, json, protobuf, simple, table)")
 	rootCmd.PersistentFlags().BoolVar(&globalFlags.IsHex, "hex", false, "print byte strings as hex encoded strings")
 
 	rootCmd.PersistentFlags().DurationVar(&globalFlags.DialTimeout, "dial-timeout", defaultDialTimeout, "dial timeout for client connections")

etcdmain/etcd.go

@@ -39,8 +39,6 @@ import (
 	"github.com/coreos/etcd/pkg/types"
 	"github.com/coreos/etcd/proxy/httpproxy"
 	"github.com/coreos/etcd/version"
-	"github.com/coreos/go-systemd/daemon"
-	systemdutil "github.com/coreos/go-systemd/util"
 	"github.com/coreos/pkg/capnslog"
 	"github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/prometheus/client_golang/prometheus"
@@ -85,7 +83,13 @@ func startEtcdOrProxyV2() {
 	GoMaxProcs := runtime.GOMAXPROCS(0)
 	plog.Infof("setting maximum number of CPUs to %d, total number of available CPUs is %d", GoMaxProcs, runtime.NumCPU())
 
-	(&cfg.Config).UpdateDefaultClusterFromName(defaultInitialCluster)
+	defaultHost, dhErr := (&cfg.Config).UpdateDefaultClusterFromName(defaultInitialCluster)
+	if defaultHost != "" {
+		plog.Infof("advertising using detected default host %q", defaultHost)
+	}
+	if dhErr != nil {
+		plog.Noticef("failed to detect default host (%v)", dhErr)
+	}
 
 	if cfg.Dir == "" {
 		cfg.Dir = fmt.Sprintf("%v.etcd", cfg.Name)
@@ -157,20 +161,12 @@ func startEtcdOrProxyV2() {
 
 	osutil.HandleInterrupts()
 
-	if systemdutil.IsRunningSystemd() {
-		// At this point, the initialization of etcd is done.
-		// The listeners are listening on the TCP ports and ready
-		// for accepting connections. The etcd instance should be
-		// joined with the cluster and ready to serve incoming
-		// connections.
-		sent, err := daemon.SdNotify(false, "READY=1")
-		if err != nil {
-			plog.Errorf("failed to notify systemd for readiness: %v", err)
-		}
-		if !sent {
-			plog.Errorf("forgot to set Type=notify in systemd service file?")
-		}
-	}
+	// At this point, the initialization of etcd is done.
+	// The listeners are listening on the TCP ports and ready
+	// for accepting connections. The etcd instance should be
+	// joined with the cluster and ready to serve incoming
+	// connections.
+	notifySystemd()
 
 	select {
 	case lerr := <-errc:
@@ -184,15 +180,6 @@ func startEtcdOrProxyV2() {
 
 // startEtcd runs StartEtcd in addition to hooks needed for standalone etcd.
 func startEtcd(cfg *embed.Config) (<-chan struct{}, <-chan error, error) {
-	defaultHost, dhErr := cfg.IsDefaultHost()
-	if defaultHost != "" {
-		if dhErr == nil {
-			plog.Infof("advertising using detected default host %q", defaultHost)
-		} else {
-			plog.Noticef("failed to detect default host, advertise falling back to %q (%v)", defaultHost, dhErr)
-		}
-	}
-
 	if cfg.Metrics == "extensive" {
 		grpc_prometheus.EnableHandlingTimeHistogram()
 	}
@@ -202,7 +189,10 @@ func startEtcd(cfg *embed.Config) (<-chan struct{}, <-chan error, error) {
 		return nil, nil, err
 	}
 	osutil.RegisterInterruptHandler(e.Server.Stop)
-	<-e.Server.ReadyNotify() // wait for e.Server to join the cluster
+	select {
+	case <-e.Server.ReadyNotify(): // wait for e.Server to join the cluster
+	case <-e.Server.StopNotify(): // publish aborted from 'ErrStopped'
+	}
 	return e.Server.StopNotify(), e.Err(), nil
 }
 

etcdmain/gateway.go

@@ -17,12 +17,14 @@ package etcdmain
 import (
 	"fmt"
 	"net"
+	"net/url"
 	"os"
 	"time"
 
 	"github.com/coreos/etcd/client"
 	"github.com/coreos/etcd/pkg/transport"
 	"github.com/coreos/etcd/proxy/tcpproxy"
+
 	"github.com/spf13/cobra"
 )
 
@@ -77,6 +79,20 @@ func newGatewayStartCommand() *cobra.Command {
 	return &cmd
 }
 
+func stripSchema(eps []string) []string {
+	var endpoints []string
+
+	for _, ep := range eps {
+
+		if u, err := url.Parse(ep); err == nil && u.Host != "" {
+			ep = u.Host
+		}
+
+		endpoints = append(endpoints, ep)
+	}
+
+	return endpoints
+}
 func startGateway(cmd *cobra.Command, args []string) {
 	endpoints := gatewayEndpoints
 	if gatewayDNSCluster != "" {
@@ -101,6 +117,9 @@ func startGateway(cmd *cobra.Command, args []string) {
 		}
 	}
 
+	// Strip the schema from the endpoints because we start just a TCP proxy
+	endpoints = stripSchema(endpoints)
+
 	if len(endpoints) == 0 {
 		plog.Fatalf("no endpoints found")
 	}
@@ -117,5 +136,8 @@ func startGateway(cmd *cobra.Command, args []string) {
 		MonitorInterval: getewayRetryDelay,
 	}
 
+	// At this point, etcd gateway listener is initialized
+	notifySystemd()
+
 	tp.Run()
 }

etcdmain/grpc_proxy.go

@@ -144,6 +144,9 @@ func startGRPCProxy(cmd *cobra.Command, args []string) {
 
 	go func() { errc <- m.Serve() }()
 
+	// grpc-proxy is initialized, ready to serve
+	notifySystemd()
+
 	fmt.Fprintln(os.Stderr, <-errc)
 	os.Exit(1)
 }

etcdmain/main.go

@@ -17,6 +17,9 @@ package etcdmain
 import (
 	"fmt"
 	"os"
+
+	"github.com/coreos/go-systemd/daemon"
+	systemdutil "github.com/coreos/go-systemd/util"
 )
 
 func Main() {
@@ -35,3 +38,16 @@ func Main() {
 
 	startEtcdOrProxyV2()
 }
+
+func notifySystemd() {
+	if !systemdutil.IsRunningSystemd() {
+		return
+	}
+	sent, err := daemon.SdNotify(false, "READY=1")
+	if err != nil {
+		plog.Errorf("failed to notify systemd for readiness: %v", err)
+	}
+	if !sent {
+		plog.Errorf("forgot to set Type=notify in systemd service file?")
+	}
+}

etcdserver/server.go

@@ -594,6 +594,7 @@ func (s *EtcdServer) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
 type etcdProgress struct {
 	confState raftpb.ConfState
 	snapi     uint64
+	appliedt  uint64
 	appliedi  uint64
 }
 
@@ -666,6 +667,7 @@ func (s *EtcdServer) run() {
 	ep := etcdProgress{
 		confState: snap.Metadata.ConfState,
 		snapi:     snap.Metadata.Index,
+		appliedt:  snap.Metadata.Term,
 		appliedi:  snap.Metadata.Index,
 	}
 
@@ -765,7 +767,7 @@ func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
 	select {
 	// snapshot requested via send()
 	case m := <-s.r.msgSnapC:
-		merged := s.createMergedSnapshotMessage(m, ep.appliedi, ep.confState)
+		merged := s.createMergedSnapshotMessage(m, ep.appliedt, ep.appliedi, ep.confState)
 		s.sendMergedSnap(merged)
 	default:
 	}
@@ -867,6 +869,7 @@ func (s *EtcdServer) applySnapshot(ep *etcdProgress, apply *apply) {
 	}
 	plog.Info("finished adding peers from new cluster configuration into network...")
 
+	ep.appliedt = apply.snapshot.Metadata.Term
 	ep.appliedi = apply.snapshot.Metadata.Index
 	ep.snapi = ep.appliedi
 	ep.confState = apply.snapshot.Metadata.ConfState
@@ -888,7 +891,7 @@ func (s *EtcdServer) applyEntries(ep *etcdProgress, apply *apply) {
 		return
 	}
 	var shouldstop bool
-	if ep.appliedi, shouldstop = s.apply(ents, &ep.confState); shouldstop {
+	if ep.appliedt, ep.appliedi, shouldstop = s.apply(ents, &ep.confState); shouldstop {
 		go s.stopWithDelay(10*100*time.Millisecond, fmt.Errorf("the member has been permanently removed from the cluster"))
 	}
 }
@@ -1242,9 +1245,7 @@ func (s *EtcdServer) sendMergedSnap(merged snap.Message) {
 // apply takes entries received from Raft (after it has been committed) and
 // applies them to the current state of the EtcdServer.
 // The given entries should not be empty.
-func (s *EtcdServer) apply(es []raftpb.Entry, confState *raftpb.ConfState) (uint64, bool) {
-	var applied uint64
-	var shouldstop bool
+func (s *EtcdServer) apply(es []raftpb.Entry, confState *raftpb.ConfState) (appliedt uint64, appliedi uint64, shouldStop bool) {
 	for i := range es {
 		e := es[i]
 		switch e.Type {
@@ -1254,16 +1255,17 @@ func (s *EtcdServer) apply(es []raftpb.Entry, confState *raftpb.ConfState) (uint
 			var cc raftpb.ConfChange
 			pbutil.MustUnmarshal(&cc, e.Data)
 			removedSelf, err := s.applyConfChange(cc, confState)
-			shouldstop = shouldstop || removedSelf
+			shouldStop = shouldStop || removedSelf
 			s.w.Trigger(cc.ID, err)
 		default:
 			plog.Panicf("entry type should be either EntryNormal or EntryConfChange")
 		}
 		atomic.StoreUint64(&s.r.index, e.Index)
 		atomic.StoreUint64(&s.r.term, e.Term)
-		applied = e.Index
+		appliedt = e.Term
+		appliedi = e.Index
 	}
-	return applied, shouldstop
+	return appliedt, appliedi, shouldStop
 }
 
 // applyEntryNormal apples an EntryNormal type raftpb request to the EtcdServer

etcdserver/server_test.go

@@ -613,7 +613,7 @@ func TestApplyMultiConfChangeShouldStop(t *testing.T) {
 		ents = append(ents, ent)
 	}
 
-	_, shouldStop := srv.apply(ents, &raftpb.ConfState{})
+	_, _, shouldStop := srv.apply(ents, &raftpb.ConfState{})
 	if !shouldStop {
 		t.Errorf("shouldStop = %t, want %t", shouldStop, true)
 	}

etcdserver/snapshot_merge.go

@@ -16,7 +16,6 @@ package etcdserver
 
 import (
 	"io"
-	"log"
 
 	"github.com/coreos/etcd/mvcc/backend"
 	"github.com/coreos/etcd/raft/raftpb"
@@ -26,12 +25,7 @@ import (
 // createMergedSnapshotMessage creates a snapshot message that contains: raft status (term, conf),
 // a snapshot of v2 store inside raft.Snapshot as []byte, a snapshot of v3 KV in the top level message
 // as ReadCloser.
-func (s *EtcdServer) createMergedSnapshotMessage(m raftpb.Message, snapi uint64, confState raftpb.ConfState) snap.Message {
-	snapt, err := s.r.raftStorage.Term(snapi)
-	if err != nil {
-		log.Panicf("get term should never fail: %v", err)
-	}
-
+func (s *EtcdServer) createMergedSnapshotMessage(m raftpb.Message, snapt, snapi uint64, confState raftpb.ConfState) snap.Message {
 	// get a snapshot of v2 store as []byte
 	clone := s.store.Clone()
 	d, err := clone.SaveNoCopy()

integration/cluster.go

@@ -449,6 +449,8 @@ type member struct {
 	grpcServer *grpc.Server
 	grpcAddr   string
 	grpcBridge *bridge
+
+	keepDataDirTerminate bool
 }
 
 func (m *member) GRPCAddr() string { return m.grpcAddr }
@@ -746,8 +748,10 @@ func (m *member) Restart(t *testing.T) error {
 func (m *member) Terminate(t *testing.T) {
 	plog.Printf("terminating %s (%s)", m.Name, m.grpcAddr)
 	m.Close()
-	if err := os.RemoveAll(m.ServerConfig.DataDir); err != nil {
-		t.Fatal(err)
+	if !m.keepDataDirTerminate {
+		if err := os.RemoveAll(m.ServerConfig.DataDir); err != nil {
+			t.Fatal(err)
+		}
 	}
 	plog.Printf("terminated %s (%s)", m.Name, m.grpcAddr)
 }

integration/cluster_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/coreos/etcd/client"
 	"github.com/coreos/etcd/etcdserver"
 	"github.com/coreos/etcd/pkg/testutil"
+	"github.com/coreos/pkg/capnslog"
 
 	"golang.org/x/net/context"
 )
@@ -441,6 +442,51 @@ func TestRejectUnhealthyRemove(t *testing.T) {
 	}
 }
 
+// TestRestartRemoved ensures that restarting removed member must exit
+// if 'initial-cluster-state' is set 'new' and old data directory still exists
+// (see https://github.com/coreos/etcd/issues/7512 for more).
+func TestRestartRemoved(t *testing.T) {
+	defer testutil.AfterTest(t)
+	capnslog.SetGlobalLogLevel(capnslog.INFO)
+
+	// 1. start single-member cluster
+	c := NewCluster(t, 1)
+	for _, m := range c.Members {
+		m.ServerConfig.StrictReconfigCheck = true
+	}
+	c.Launch(t)
+	defer c.Terminate(t)
+
+	// 2. add a new member
+	c.AddMember(t)
+	c.WaitLeader(t)
+
+	oldm := c.Members[0]
+	oldm.keepDataDirTerminate = true
+
+	// 3. remove first member, shut down without deleting data
+	if err := c.removeMember(t, uint64(c.Members[0].s.ID())); err != nil {
+		t.Fatalf("expected to remove member, got error %v", err)
+	}
+	c.WaitLeader(t)
+
+	// 4. restart first member with 'initial-cluster-state=new'
+	// wrong config, expects exit within ReqTimeout
+	oldm.ServerConfig.NewCluster = false
+	if err := oldm.Restart(t); err != nil {
+		t.Fatalf("unexpected ForceRestart error: %v", err)
+	}
+	defer func() {
+		oldm.Close()
+		os.RemoveAll(oldm.ServerConfig.DataDir)
+	}()
+	select {
+	case <-oldm.s.StopNotify():
+	case <-time.After(time.Minute):
+		t.Fatalf("removed member didn't exit within %v", time.Minute)
+	}
+}
+
 // clusterMustProgress ensures that cluster can make progress. It creates
 // a random key first, and check the new key could be got from all client urls
 // of the cluster.

integration/v3_grpc_test.go

@@ -581,6 +581,37 @@ func TestV3Hash(t *testing.T) {
 	}
 }
 
+// TestV3HashRestart ensures that hash stays the same after restart.
+func TestV3HashRestart(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	cli := clus.RandClient()
+	resp, err := toGRPC(cli).Maintenance.Hash(context.Background(), &pb.HashRequest{})
+	if err != nil || resp.Hash == 0 {
+		t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
+	}
+	hash1 := resp.Hash
+
+	clus.Members[0].Stop(t)
+	clus.Members[0].Restart(t)
+	clus.waitLeader(t, clus.Members)
+	kvc := toGRPC(clus.Client(0)).KV
+	waitForRestart(t, kvc)
+
+	cli = clus.RandClient()
+	resp, err = toGRPC(cli).Maintenance.Hash(context.Background(), &pb.HashRequest{})
+	if err != nil || resp.Hash == 0 {
+		t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
+	}
+	hash2 := resp.Hash
+
+	if hash1 != hash2 {
+		t.Fatalf("hash expected %d, got %d", hash1, hash2)
+	}
+}
+
 // TestV3StorageQuotaAPI tests the V3 server respects quotas at the API layer
 func TestV3StorageQuotaAPI(t *testing.T) {
 	defer testutil.AfterTest(t)

lease/lessor.go

@@ -252,10 +252,7 @@ func (le *lessor) Revoke(id LeaseID) error {
 
 	// sort keys so deletes are in same order among all members,
 	// otherwise the backened hashes will be different
-	keys := make([]string, 0, len(l.itemSet))
-	for item := range l.itemSet {
-		keys = append(keys, item.Key)
-	}
+	keys := l.Keys()
 	sort.StringSlice(keys).Sort()
 	for _, key := range keys {
 		_, _, err := le.rd.TxnDeleteRange(tid, []byte(key), nil)
@@ -367,10 +364,12 @@ func (le *lessor) Attach(id LeaseID, items []LeaseItem) error {
 		return ErrLeaseNotFound
 	}
 
+	l.mu.Lock()
 	for _, it := range items {
 		l.itemSet[it] = struct{}{}
 		le.itemMap[it] = id
 	}
+	l.mu.Unlock()
 	return nil
 }
 
@@ -392,10 +391,12 @@ func (le *lessor) Detach(id LeaseID, items []LeaseItem) error {
 		return ErrLeaseNotFound
 	}
 
+	l.mu.Lock()
 	for _, it := range items {
 		delete(l.itemSet, it)
 		delete(le.itemMap, it)
 	}
+	l.mu.Unlock()
 	return nil
 }
 
@@ -506,6 +507,8 @@ type Lease struct {
 	// expiry is time when lease should expire; must be 64-bit aligned.
 	expiry monotime.Time
 
+	// mu protects concurrent accesses to itemSet
+	mu      sync.RWMutex
 	itemSet map[LeaseItem]struct{}
 	revokec chan struct{}
 }
@@ -544,10 +547,12 @@ func (l *Lease) forever() { atomic.StoreUint64((*uint64)(&l.expiry), uint64(fore
 
 // Keys returns all the keys attached to the lease.
 func (l *Lease) Keys() []string {
+	l.mu.RLock()
 	keys := make([]string, 0, len(l.itemSet))
 	for k := range l.itemSet {
 		keys = append(keys, k.Key)
 	}
+	l.mu.RUnlock()
 	return keys
 }
 

lease/lessor_test.go

@@ -15,11 +15,13 @@
 package lease
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path"
 	"reflect"
 	"sort"
+	"sync"
 	"testing"
 	"time"
 
@@ -76,6 +78,53 @@ func TestLessorGrant(t *testing.T) {
 	be.BatchTx().Unlock()
 }
 
+// TestLeaseConcurrentKeys ensures Lease.Keys method calls are guarded
+// from concurrent map writes on 'itemSet'.
+func TestLeaseConcurrentKeys(t *testing.T) {
+	dir, be := NewTestBackend(t)
+	defer os.RemoveAll(dir)
+	defer be.Close()
+
+	fd := &fakeDeleter{}
+
+	le := newLessor(be, minLeaseTTL)
+	le.SetRangeDeleter(fd)
+
+	// grant a lease with long term (100 seconds) to
+	// avoid early termination during the test.
+	l, err := le.Grant(1, 100)
+	if err != nil {
+		t.Fatalf("could not grant lease for 100s ttl (%v)", err)
+	}
+
+	itemn := 10
+	items := make([]LeaseItem, itemn)
+	for i := 0; i < itemn; i++ {
+		items[i] = LeaseItem{Key: fmt.Sprintf("foo%d", i)}
+	}
+	if err = le.Attach(l.ID, items); err != nil {
+		t.Fatalf("failed to attach items to the lease: %v", err)
+	}
+
+	donec := make(chan struct{})
+	go func() {
+		le.Detach(l.ID, items)
+		close(donec)
+	}()
+
+	var wg sync.WaitGroup
+	wg.Add(itemn)
+	for i := 0; i < itemn; i++ {
+		go func() {
+			defer wg.Done()
+			l.Keys()
+		}()
+	}
+
+	<-donec
+	wg.Wait()
+}
+
 // TestLessorRevoke ensures Lessor can revoke a lease.
 // The items in the revoked lease should be removed from
 // the backend.

mvcc/backend/backend.go

@@ -303,6 +303,7 @@ func defragdb(odb, tmpdb *bolt.DB, limit int) error {
 		}
 
 		tmpb, berr := tmptx.CreateBucketIfNotExists(next)
+		tmpb.FillPercent = 0.9 // for seq write in for each
 		if berr != nil {
 			return berr
 		}
@@ -319,6 +320,8 @@ func defragdb(odb, tmpdb *bolt.DB, limit int) error {
 					return err
 				}
 				tmpb = tmptx.Bucket(next)
+				tmpb.FillPercent = 0.9 // for seq write in for each
+
 				count = 0
 			}
 			return tmpb.Put(k, v)

pkg/cpuutil/doc.go

@@ -0,0 +1,16 @@
+// 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 cpuutil provides facilities for detecting cpu-specific features.
+package cpuutil

pkg/cpuutil/endian.go

@@ -0,0 +1,36 @@
+// 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 cpuutil
+
+import (
+	"encoding/binary"
+	"unsafe"
+)
+
+const intWidth int = int(unsafe.Sizeof(0))
+
+var byteOrder binary.ByteOrder
+
+// ByteOrder returns the byte order for the CPU's native endianness.
+func ByteOrder() binary.ByteOrder { return byteOrder }
+
+func init() {
+	var i int = 0x1
+	if v := (*[intWidth]byte)(unsafe.Pointer(&i)); v[0] == 0 {
+		byteOrder = binary.BigEndian
+	} else {
+		byteOrder = binary.LittleEndian
+	}
+}

pkg/netutil/isolate_linux.go

@@ -43,7 +43,7 @@ func RecoverPort(port int) error {
 
 // SetLatency adds latency in millisecond scale with random variations.
 func SetLatency(ms, rv int) error {
-	ifce, err := GetDefaultInterface()
+	ifces, err := GetDefaultInterfaces()
 	if err != nil {
 		return err
 	}
@@ -51,14 +51,16 @@ func SetLatency(ms, rv int) error {
 	if rv > ms {
 		rv = 1
 	}
-	cmdStr := fmt.Sprintf("sudo tc qdisc add dev %s root netem delay %dms %dms distribution normal", ifce, ms, rv)
-	_, err = exec.Command("/bin/sh", "-c", cmdStr).Output()
-	if err != nil {
-		// the rule has already been added. Overwrite it.
-		cmdStr = fmt.Sprintf("sudo tc qdisc change dev %s root netem delay %dms %dms distribution normal", ifce, ms, rv)
+	for ifce := range ifces {
+		cmdStr := fmt.Sprintf("sudo tc qdisc add dev %s root netem delay %dms %dms distribution normal", ifce, ms, rv)
 		_, err = exec.Command("/bin/sh", "-c", cmdStr).Output()
 		if err != nil {
-			return err
+			// the rule has already been added. Overwrite it.
+			cmdStr = fmt.Sprintf("sudo tc qdisc change dev %s root netem delay %dms %dms distribution normal", ifce, ms, rv)
+			_, err = exec.Command("/bin/sh", "-c", cmdStr).Output()
+			if err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -66,10 +68,15 @@ func SetLatency(ms, rv int) error {
 
 // RemoveLatency resets latency configurations.
 func RemoveLatency() error {
-	ifce, err := GetDefaultInterface()
+	ifces, err := GetDefaultInterfaces()
 	if err != nil {
 		return err
 	}
-	_, err = exec.Command("/bin/sh", "-c", fmt.Sprintf("sudo tc qdisc del dev %s root netem", ifce)).Output()
-	return err
+	for ifce := range ifces {
+		_, err = exec.Command("/bin/sh", "-c", fmt.Sprintf("sudo tc qdisc del dev %s root netem", ifce)).Output()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }

pkg/netutil/routes.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// +build !linux !386,!amd64
+// +build !linux
 
 package netutil
 
@@ -27,7 +27,7 @@ func GetDefaultHost() (string, error) {
 	return "", fmt.Errorf("default host not supported on %s_%s", runtime.GOOS, runtime.GOARCH)
 }
 
-// GetDefaultInterface fetches the device name of default routable interface.
-func GetDefaultInterface() (string, error) {
-	return "", fmt.Errorf("default host not supported on %s_%s", runtime.GOOS, runtime.GOARCH)
+// GetDefaultInterfaces fetches the device name of default routable interface.
+func GetDefaultInterfaces() (map[string]uint8, error) {
+	return nil, fmt.Errorf("default host not supported on %s_%s", runtime.GOOS, runtime.GOARCH)
 }

pkg/netutil/routes_linux.go

@@ -13,9 +13,6 @@
 // limitations under the License.
 
 // +build linux
-// +build 386 amd64
-
-// TODO support native endian but without using "unsafe"
 
 package netutil
 
@@ -24,27 +21,57 @@ import (
 	"encoding/binary"
 	"fmt"
 	"net"
+	"sort"
 	"syscall"
+
+	"github.com/coreos/etcd/pkg/cpuutil"
 )
 
 var errNoDefaultRoute = fmt.Errorf("could not find default route")
+var errNoDefaultHost = fmt.Errorf("could not find default host")
+var errNoDefaultInterface = fmt.Errorf("could not find default interface")
 
+// GetDefaultHost obtains the first IP address of machine from the routing table and returns the IP address as string.
+// An IPv4 address is preferred to an IPv6 address for backward compatibility.
 func GetDefaultHost() (string, error) {
-	rmsg, rerr := getDefaultRoute()
+	rmsgs, rerr := getDefaultRoutes()
 	if rerr != nil {
 		return "", rerr
 	}
 
-	host, oif, err := parsePREFSRC(rmsg)
-	if err != nil {
-		return "", err
+	// prioritize IPv4
+	if rmsg, ok := rmsgs[syscall.AF_INET]; ok {
+		if host, err := chooseHost(syscall.AF_INET, rmsg); host != "" || err != nil {
+			return host, err
+		}
+		delete(rmsgs, syscall.AF_INET)
 	}
-	if host != "" {
-		return host, nil
+
+	// sort so choice is deterministic
+	var families []int
+	for family := range rmsgs {
+		families = append(families, int(family))
+	}
+	sort.Ints(families)
+
+	for _, f := range families {
+		family := uint8(f)
+		if host, err := chooseHost(family, rmsgs[family]); host != "" || err != nil {
+			return host, err
+		}
+	}
+
+	return "", errNoDefaultHost
+}
+
+func chooseHost(family uint8, rmsg *syscall.NetlinkMessage) (string, error) {
+	host, oif, err := parsePREFSRC(rmsg)
+	if host != "" || err != nil {
+		return host, err
 	}
 
 	// prefsrc not detected, fall back to getting address from iface
-	ifmsg, ierr := getIface(oif)
+	ifmsg, ierr := getIfaceAddr(oif, family)
 	if ierr != nil {
 		return "", ierr
 	}
@@ -55,15 +82,16 @@ func GetDefaultHost() (string, error) {
 	}
 
 	for _, attr := range attrs {
-		if attr.Attr.Type == syscall.RTA_SRC {
+		// search for RTA_DST because ipv6 doesn't have RTA_SRC
+		if attr.Attr.Type == syscall.RTA_DST {
 			return net.IP(attr.Value).String(), nil
 		}
 	}
 
-	return "", errNoDefaultRoute
+	return "", nil
 }
 
-func getDefaultRoute() (*syscall.NetlinkMessage, error) {
+func getDefaultRoutes() (map[uint8]*syscall.NetlinkMessage, error) {
 	dat, err := syscall.NetlinkRIB(syscall.RTM_GETROUTE, syscall.AF_UNSPEC)
 	if err != nil {
 		return nil, err
@@ -74,26 +102,33 @@ func getDefaultRoute() (*syscall.NetlinkMessage, error) {
 		return nil, msgErr
 	}
 
+	routes := make(map[uint8]*syscall.NetlinkMessage)
 	rtmsg := syscall.RtMsg{}
 	for _, m := range msgs {
 		if m.Header.Type != syscall.RTM_NEWROUTE {
 			continue
 		}
 		buf := bytes.NewBuffer(m.Data[:syscall.SizeofRtMsg])
-		if rerr := binary.Read(buf, binary.LittleEndian, &rtmsg); rerr != nil {
+		if rerr := binary.Read(buf, cpuutil.ByteOrder(), &rtmsg); rerr != nil {
 			continue
 		}
-		if rtmsg.Dst_len == 0 {
+		if rtmsg.Dst_len == 0 && rtmsg.Table == syscall.RT_TABLE_MAIN {
 			// zero-length Dst_len implies default route
-			return &m, nil
+			msg := m
+			routes[rtmsg.Family] = &msg
 		}
 	}
 
+	if len(routes) > 0 {
+		return routes, nil
+	}
+
 	return nil, errNoDefaultRoute
 }
 
-func getIface(idx uint32) (*syscall.NetlinkMessage, error) {
-	dat, err := syscall.NetlinkRIB(syscall.RTM_GETADDR, syscall.AF_UNSPEC)
+// Used to get an address of interface.
+func getIfaceAddr(idx uint32, family uint8) (*syscall.NetlinkMessage, error) {
+	dat, err := syscall.NetlinkRIB(syscall.RTM_GETADDR, int(family))
 	if err != nil {
 		return nil, err
 	}
@@ -109,7 +144,7 @@ func getIface(idx uint32) (*syscall.NetlinkMessage, error) {
 			continue
 		}
 		buf := bytes.NewBuffer(m.Data[:syscall.SizeofIfAddrmsg])
-		if rerr := binary.Read(buf, binary.LittleEndian, &ifaddrmsg); rerr != nil {
+		if rerr := binary.Read(buf, cpuutil.ByteOrder(), &ifaddrmsg); rerr != nil {
 			continue
 		}
 		if ifaddrmsg.Index == idx {
@@ -117,38 +152,75 @@ func getIface(idx uint32) (*syscall.NetlinkMessage, error) {
 		}
 	}
 
-	return nil, errNoDefaultRoute
+	return nil, fmt.Errorf("could not find address for interface index %v", idx)
+
 }
 
-var errNoDefaultInterface = fmt.Errorf("could not find default interface")
-
-func GetDefaultInterface() (string, error) {
-	rmsg, rerr := getDefaultRoute()
-	if rerr != nil {
-		return "", rerr
-	}
-
-	_, oif, err := parsePREFSRC(rmsg)
+// Used to get a name of interface.
+func getIfaceLink(idx uint32) (*syscall.NetlinkMessage, error) {
+	dat, err := syscall.NetlinkRIB(syscall.RTM_GETLINK, syscall.AF_UNSPEC)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
-	ifmsg, ierr := getIface(oif)
-	if ierr != nil {
-		return "", ierr
+	msgs, msgErr := syscall.ParseNetlinkMessage(dat)
+	if msgErr != nil {
+		return nil, msgErr
 	}
 
-	attrs, aerr := syscall.ParseNetlinkRouteAttr(ifmsg)
-	if aerr != nil {
-		return "", aerr
-	}
-
-	for _, attr := range attrs {
-		if attr.Attr.Type == syscall.IFLA_IFNAME {
-			return string(attr.Value[:len(attr.Value)-1]), nil
+	ifinfomsg := syscall.IfInfomsg{}
+	for _, m := range msgs {
+		if m.Header.Type != syscall.RTM_NEWLINK {
+			continue
+		}
+		buf := bytes.NewBuffer(m.Data[:syscall.SizeofIfInfomsg])
+		if rerr := binary.Read(buf, cpuutil.ByteOrder(), &ifinfomsg); rerr != nil {
+			continue
+		}
+		if ifinfomsg.Index == int32(idx) {
+			return &m, nil
 		}
 	}
-	return "", errNoDefaultInterface
+
+	return nil, fmt.Errorf("could not find link for interface index %v", idx)
+}
+
+// GetDefaultInterfaces gets names of interfaces and returns a map[interface]families.
+func GetDefaultInterfaces() (map[string]uint8, error) {
+	interfaces := make(map[string]uint8)
+	rmsgs, rerr := getDefaultRoutes()
+	if rerr != nil {
+		return interfaces, rerr
+	}
+
+	for family, rmsg := range rmsgs {
+		_, oif, err := parsePREFSRC(rmsg)
+		if err != nil {
+			return interfaces, err
+		}
+
+		ifmsg, ierr := getIfaceLink(oif)
+		if ierr != nil {
+			return interfaces, ierr
+		}
+
+		attrs, aerr := syscall.ParseNetlinkRouteAttr(ifmsg)
+		if aerr != nil {
+			return interfaces, aerr
+		}
+
+		for _, attr := range attrs {
+			if attr.Attr.Type == syscall.IFLA_IFNAME {
+				// key is an interface name
+				// possible values: 2 - AF_INET, 10 - AF_INET6, 12 - dualstack
+				interfaces[string(attr.Value[:len(attr.Value)-1])] += family
+			}
+		}
+	}
+	if len(interfaces) > 0 {
+		return interfaces, nil
+	}
+	return interfaces, errNoDefaultInterface
 }
 
 // parsePREFSRC returns preferred source address and output interface index (RTA_OIF).
@@ -164,7 +236,7 @@ func parsePREFSRC(m *syscall.NetlinkMessage) (host string, oif uint32, err error
 			host = net.IP(attr.Value).String()
 		}
 		if attr.Attr.Type == syscall.RTA_OIF {
-			oif = binary.LittleEndian.Uint32(attr.Value)
+			oif = cpuutil.ByteOrder().Uint32(attr.Value)
 		}
 		if host != "" && oif != uint32(0) {
 			break

pkg/netutil/routes_linux_test.go

@@ -19,9 +19,17 @@ package netutil
 import "testing"
 
 func TestGetDefaultInterface(t *testing.T) {
-	ifc, err := GetDefaultInterface()
+	ifc, err := GetDefaultInterfaces()
 	if err != nil {
 		t.Fatal(err)
 	}
-	t.Logf("default network interface: %q\n", ifc)
+	t.Logf("default network interfaces: %+v\n", ifc)
+}
+
+func TestGetDefaultHost(t *testing.T) {
+	ip, err := GetDefaultHost()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Logf("default ip: %v", ip)
 }

proxy/tcpproxy/userspace.go

@@ -147,16 +147,17 @@ func (tp *TCPProxy) runMonitor() {
 		select {
 		case <-time.After(tp.MonitorInterval):
 			tp.mu.Lock()
-			for _, r := range tp.remotes {
-				if !r.isActive() {
-					go func() {
-						if err := r.tryReactivate(); err != nil {
-							plog.Warningf("failed to activate endpoint [%s] due to %v (stay inactive for another %v)", r.addr, err, tp.MonitorInterval)
-						} else {
-							plog.Printf("activated %s", r.addr)
-						}
-					}()
+			for _, rem := range tp.remotes {
+				if rem.isActive() {
+					continue
 				}
+				go func(r *remote) {
+					if err := r.tryReactivate(); err != nil {
+						plog.Warningf("failed to activate endpoint [%s] due to %v (stay inactive for another %v)", r.addr, err, tp.MonitorInterval)
+					} else {
+						plog.Printf("activated %s", r.addr)
+					}
+				}(rem)
 			}
 			tp.mu.Unlock()
 		case <-tp.donec:

raft/raft.go

@@ -823,6 +823,11 @@ func stepLeader(r *raft, m pb.Message) {
 		return
 	case pb.MsgReadIndex:
 		if r.quorum() > 1 {
+			if r.raftLog.zeroTermOnErrCompacted(r.raftLog.term(r.raftLog.committed)) != r.Term {
+				// Reject read only request when this leader has not committed any log entry at its term.
+				return
+			}
+
 			// thinking: use an interally defined context instead of the user given context.
 			// We can express this in terms of the term and index instead of a user-supplied value.
 			// This would allow multiple reads to piggyback on the same message.

raft/raft_test.go

@@ -1856,6 +1856,77 @@ func TestReadOnlyOptionLeaseWithoutCheckQuorum(t *testing.T) {
 	}
 }
 
+// TestReadOnlyForNewLeader ensures that a leader only accepts MsgReadIndex message
+// when it commits at least one log entry at it term.
+func TestReadOnlyForNewLeader(t *testing.T) {
+	cfg := newTestConfig(1, []uint64{1, 2, 3}, 10, 1,
+		&MemoryStorage{
+			ents:      []pb.Entry{{}, {Index: 1, Term: 1}, {Index: 2, Term: 1}},
+			hardState: pb.HardState{Commit: 1, Term: 1},
+		})
+	cfg.Applied = 1
+	a := newRaft(cfg)
+	cfg = newTestConfig(2, []uint64{1, 2, 3}, 10, 1,
+		&MemoryStorage{
+			ents:      []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}},
+			hardState: pb.HardState{Commit: 2, Term: 1},
+		})
+	cfg.Applied = 2
+	b := newRaft(cfg)
+	cfg = newTestConfig(2, []uint64{1, 2, 3}, 10, 1,
+		&MemoryStorage{
+			ents:      []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}},
+			hardState: pb.HardState{Commit: 2, Term: 1},
+		})
+	cfg.Applied = 2
+	c := newRaft(cfg)
+	nt := newNetwork(a, b, c)
+
+	// Drop MsgApp to forbid peer a to commit any log entry at its term after it becomes leader.
+	nt.ignore(pb.MsgApp)
+	// Force peer a to become leader.
+	nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
+	if a.state != StateLeader {
+		t.Fatalf("state = %s, want %s", a.state, StateLeader)
+	}
+
+	// Ensure peer a drops read only request.
+	var windex uint64 = 4
+	wctx := []byte("ctx")
+	nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: wctx}}})
+	if len(a.readStates) != 0 {
+		t.Fatalf("len(readStates) = %d, want zero", len(a.readStates))
+	}
+
+	nt.recover()
+
+	// Force peer a to commit a log entry at its term
+	for i := 0; i < a.heartbeatTimeout; i++ {
+		a.tick()
+	}
+	nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{}}})
+	if a.raftLog.committed != 4 {
+		t.Fatalf("committed = %d, want 4", a.raftLog.committed)
+	}
+	lastLogTerm := a.raftLog.zeroTermOnErrCompacted(a.raftLog.term(a.raftLog.committed))
+	if lastLogTerm != a.Term {
+		t.Fatalf("last log term = %d, want %d", lastLogTerm, a.Term)
+	}
+
+	// Ensure peer a accepts read only request after it commits a entry at its term.
+	nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: wctx}}})
+	if len(a.readStates) != 1 {
+		t.Fatalf("len(readStates) = %d, want 1", len(a.readStates))
+	}
+	rs := a.readStates[0]
+	if rs.Index != windex {
+		t.Fatalf("readIndex = %d, want %d", rs.Index, windex)
+	}
+	if !bytes.Equal(rs.RequestCtx, wctx) {
+		t.Fatalf("requestCtx = %v, want %v", rs.RequestCtx, wctx)
+	}
+}
+
 func TestLeaderAppResp(t *testing.T) {
 	// initial progress: match = 0; next = 3
 	tests := []struct {

version/version.go

@@ -26,7 +26,7 @@ import (
 var (
 	// MinClusterVersion is the min cluster version this etcd binary is compatible with.
 	MinClusterVersion = "3.0.0"
-	Version           = "3.1.1"
+	Version           = "3.1.4"
 	APIVersion        = "unknown"
 
 	// Git SHA Value will be set during build