integration: use unix domain sockets for all connections

integration/bridge.go

@@ -18,8 +18,9 @@ import (
 	"fmt"
 	"io"
 	"net"
-	"os"
 	"sync"
+
+	"github.com/coreos/etcd/pkg/transport"
 )
 
 // bridge creates a unix socket bridge to another unix socket, making it possible
@@ -43,10 +44,7 @@ func newBridge(addr string) (*bridge, error) {
 		conns:   make(map[*bridgeConn]struct{}),
 		stopc:   make(chan struct{}, 1),
 	}
-	if err := os.RemoveAll(b.inaddr); err != nil {
+	l, err := transport.NewUnixListener(b.inaddr)
-		return nil, err
-	}
-	l, err := net.Listen("unix", b.inaddr)
 	if err != nil {
 		return nil, fmt.Errorf("listen failed on socket %s (%v)", addr, err)
 	}
@@ -79,7 +77,6 @@ func (b *bridge) Reset() {
 func (b *bridge) serveListen() {
 	defer func() {
 		b.l.Close()
-		os.RemoveAll(b.inaddr)
 		b.mu.Lock()
 		for bc := range b.conns {
 			bc.Close()

integration/cluster.go

@@ -25,7 +25,6 @@ import (
 	"os"
 	"reflect"
 	"sort"
-	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -53,14 +52,18 @@ const (
 	tickDuration   = 10 * time.Millisecond
 	clusterName    = "etcd"
 	requestTimeout = 20 * time.Second
+
+	basePort     = 21000
+	urlScheme    = "unix"
+	urlSchemeTLS = "unixs"
 )
 
 var (
 	electionTicks = 10
 
-	// integration test uses well-known ports to listen for each running member,
+	// integration test uses unique ports, counting up, to listen for each
-	// which ensures restarted member could listen on specific port again.
+	// member, ensuring restarted members can listen on the same port again.
-	nextListenPort int64 = 21000
+	localListenCount int64 = 0
 
 	testTLSInfo = transport.TLSInfo{
 		KeyFile:        "./fixtures/server.key.insecure",
@@ -91,6 +94,13 @@ func init() {
 	api.EnableCapability(api.V3rpcCapability)
 }
 
+func schemeFromTLSInfo(tls *transport.TLSInfo) string {
+	if tls == nil {
+		return urlScheme
+	}
+	return urlSchemeTLS
+}
+
 func (c *cluster) fillClusterForMembers() error {
 	if c.cfg.DiscoveryURL != "" {
 		// cluster will be discovered
@@ -99,10 +109,7 @@ func (c *cluster) fillClusterForMembers() error {
 
 	addrs := make([]string, 0)
 	for _, m := range c.Members {
-		scheme := "http"
+		scheme := schemeFromTLSInfo(m.PeerTLSInfo)
-		if m.PeerTLSInfo != nil {
-			scheme = "https"
-		}
 		for _, l := range m.PeerListeners {
 			addrs = append(addrs, fmt.Sprintf("%s=%s://%s", m.Name, scheme, l.Addr().String()))
 		}
@@ -186,13 +193,8 @@ func (c *cluster) URLs() []string {
 func (c *cluster) HTTPMembers() []client.Member {
 	ms := []client.Member{}
 	for _, m := range c.Members {
-		pScheme, cScheme := "http", "http"
+		pScheme := schemeFromTLSInfo(m.PeerTLSInfo)
-		if m.PeerTLSInfo != nil {
+		cScheme := schemeFromTLSInfo(m.ClientTLSInfo)
-			pScheme = "https"
-		}
-		if m.ClientTLSInfo != nil {
-			cScheme = "https"
-		}
 		cm := client.Member{Name: m.Name}
 		for _, ln := range m.PeerListeners {
 			cm.PeerURLs = append(cm.PeerURLs, pScheme+"://"+ln.Addr().String())
@@ -225,10 +227,7 @@ func (c *cluster) mustNewMember(t *testing.T) *member {
 func (c *cluster) addMember(t *testing.T) {
 	m := c.mustNewMember(t)
 
-	scheme := "http"
+	scheme := schemeFromTLSInfo(c.cfg.PeerTLS)
-	if c.cfg.PeerTLS != nil {
-		scheme = "https"
-	}
 
 	// send add request to the cluster
 	var err error
@@ -390,26 +389,13 @@ func isMembersEqual(membs []client.Member, wmembs []client.Member) bool {
 }
 
 func newLocalListener(t *testing.T) net.Listener {
-	port := atomic.AddInt64(&nextListenPort, 1)
+	c := atomic.AddInt64(&localListenCount, 1)
-	l, err := net.Listen("tcp", "127.0.0.1:"+strconv.FormatInt(port, 10))
+	addr := fmt.Sprintf("127.0.0.1:%d.%d.sock", c+basePort, os.Getpid())
-	if err != nil {
+	return newListenerWithAddr(t, addr)
-		t.Fatal(err)
-	}
-	return l
 }
 
 func newListenerWithAddr(t *testing.T, addr string) net.Listener {
-	var err error
+	l, err := transport.NewUnixListener(addr)
-	var l net.Listener
-	// TODO: we want to reuse a previous closed port immediately.
-	// a better way is to set SO_REUSExx instead of doing retry.
-	for i := 0; i < 5; i++ {
-		l, err = net.Listen("tcp", addr)
-		if err == nil {
-			break
-		}
-		time.Sleep(500 * time.Millisecond)
-	}
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -449,13 +435,8 @@ func mustNewMember(t *testing.T, mcfg memberConfig) *member {
 	var err error
 	m := &member{}
 
-	peerScheme, clientScheme := "http", "http"
+	peerScheme := schemeFromTLSInfo(mcfg.peerTLS)
-	if mcfg.peerTLS != nil {
+	clientScheme := schemeFromTLSInfo(mcfg.clientTLS)
-		peerScheme = "https"
-	}
-	if mcfg.clientTLS != nil {
-		clientScheme = "https"
-	}
 
 	pln := newLocalListener(t)
 	m.PeerListeners = []net.Listener{pln}
@@ -500,10 +481,7 @@ func mustNewMember(t *testing.T, mcfg memberConfig) *member {
 func (m *member) listenGRPC() error {
 	// prefix with localhost so cert has right domain
 	m.grpcAddr = "localhost:" + m.Name + ".sock"
-	if err := os.RemoveAll(m.grpcAddr); err != nil {
+	l, err := transport.NewUnixListener(m.grpcAddr)
-		return err
-	}
-	l, err := net.Listen("unix", m.grpcAddr)
 	if err != nil {
 		return fmt.Errorf("listen failed on grpc socket %s (%v)", m.grpcAddr, err)
 	}

integration/v2_http_kv_test.go

@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"net/url"
 	"reflect"
@@ -28,6 +27,7 @@ import (
 	"time"
 
 	"github.com/coreos/etcd/pkg/testutil"
+	"github.com/coreos/etcd/pkg/transport"
 	"github.com/coreos/pkg/capnslog"
 )
 
@@ -1038,10 +1038,8 @@ type testHttpClient struct {
 
 // Creates a new HTTP client with KeepAlive disabled.
 func NewTestClient() *testHttpClient {
-	tr := &http.Transport{
+	tr, _ := transport.NewTransport(transport.TLSInfo{}, time.Second)
-		Dial:              (&net.Dialer{Timeout: time.Second}).Dial,
+	tr.DisableKeepAlives = true
-		DisableKeepAlives: true,
-	}
 	return &testHttpClient{&http.Client{Transport: tr}}
 }