Merge pull request #15940 from HubertZhang/backport-12469-3.5

[release-3.5]pkg/types: Support Unix sockets in NewURLS

client/pkg/types/urls.go

@@ -36,20 +36,25 @@ func NewURLs(strs []string) (URLs, error) {
 		if err != nil {
 			return nil, err
 		}
-		if u.Scheme != "http" && u.Scheme != "https" && u.Scheme != "unix" && u.Scheme != "unixs" {
+
+		switch u.Scheme {
+		case "http", "https":
+			if _, _, err := net.SplitHostPort(u.Host); err != nil {
+				return nil, fmt.Errorf(`URL address does not have the form "host:port": %s`, in)
+			}
+
+			if u.Path != "" {
+				return nil, fmt.Errorf("URL must not contain a path: %s", in)
+			}
+		case "unix", "unixs":
+			break
+		default:
 			return nil, fmt.Errorf("URL scheme must be http, https, unix, or unixs: %s", in)
 		}
-		if _, _, err := net.SplitHostPort(u.Host); err != nil {
-			return nil, fmt.Errorf(`URL address does not have the form "host:port": %s`, in)
-		}
-		if u.Path != "" {
-			return nil, fmt.Errorf("URL must not contain a path: %s", in)
-		}
 		all[i] = *u
 	}
 	us := URLs(all)
 	us.Sort()
-
 	return us, nil
 }
 

pkg/flags/urls_test.go

@@ -29,9 +29,6 @@ func TestValidateURLsValueBad(t *testing.T) {
 		// bad port specification
 		"127.0.0.1:foo",
 		"127.0.0.1:",
-		// unix sockets not supported
-		"unix://",
-		"unix://tmp/etcd.sock",
 		// bad strings
 		"somewhere",
 		"234#$",
@@ -56,6 +53,9 @@ func TestNewURLsValue(t *testing.T) {
 		{s: "http://10.1.1.1:80", exp: []url.URL{{Scheme: "http", Host: "10.1.1.1:80"}}},
 		{s: "http://localhost:80", exp: []url.URL{{Scheme: "http", Host: "localhost:80"}}},
 		{s: "http://:80", exp: []url.URL{{Scheme: "http", Host: ":80"}}},
+		{s: "unix://tmp/etcd.sock", exp: []url.URL{{Scheme: "unix", Host: "tmp", Path: "/etcd.sock"}}},
+		{s: "unix:///tmp/127.27.84.4:23432", exp: []url.URL{{Scheme: "unix", Path: "/tmp/127.27.84.4:23432"}}},
+		{s: "unix://127.0.0.5:1456", exp: []url.URL{{Scheme: "unix", Host: "127.0.0.5:1456"}}},
 		{
 			s: "http://localhost:1,https://localhost:2",
 			exp: []url.URL{

server/embed/etcd.go

@@ -622,10 +622,10 @@ func configureClientListeners(cfg *Config) (sctxs map[string]*serveCtx, err erro
 	for _, u := range append(cfg.ListenClientUrls, cfg.ListenClientHttpUrls...) {
 		if u.Scheme == "http" || u.Scheme == "unix" {
 			if !cfg.ClientTLSInfo.Empty() {
-				cfg.logger.Warn("scheme is HTTP while key and cert files are present; ignoring key and cert files", zap.String("client-url", u.String()))
+				cfg.logger.Warn("scheme is http or unix while key and cert files are present; ignoring key and cert files", zap.String("client-url", u.String()))
 			}
 			if cfg.ClientTLSInfo.ClientCertAuth {
-				cfg.logger.Warn("scheme is HTTP while --client-cert-auth is enabled; ignoring client cert auth for this URL", zap.String("client-url", u.String()))
+				cfg.logger.Warn("scheme is http or unix while --client-cert-auth is enabled; ignoring client cert auth for this URL", zap.String("client-url", u.String()))
 			}
 		}
 		if (u.Scheme == "https" || u.Scheme == "unixs") && cfg.ClientTLSInfo.Empty() {

tests/e2e/ctl_v3_grpc_test.go

@@ -33,6 +33,7 @@ import (
 func TestAuthority(t *testing.T) {
 	tcs := []struct {
 		name           string
+		useUnix        bool
 		useTLS         bool
 		useInsecureTLS bool
 		// Pattern used to generate endpoints for client. Fields filled
@@ -43,6 +44,33 @@ func TestAuthority(t *testing.T) {
 		// ${MEMBER_PORT} - will be filled with member grpc port
 		expectAuthorityPattern string
 	}{
+		{
+			name:                   "unix:path",
+			useUnix:                true,
+			clientURLPattern:       "unix:localhost:%d",
+			expectAuthorityPattern: "localhost:${MEMBER_PORT}",
+		},
+		{
+			name:                   "unix://absolute_path",
+			useUnix:                true,
+			clientURLPattern:       "unix://localhost:%d",
+			expectAuthorityPattern: "localhost:${MEMBER_PORT}",
+		},
+		// "unixs" is not standard schema supported by etcd
+		{
+			name:                   "unixs:absolute_path",
+			useUnix:                true,
+			useTLS:                 true,
+			clientURLPattern:       "unixs:localhost:%d",
+			expectAuthorityPattern: "localhost:${MEMBER_PORT}",
+		},
+		{
+			name:                   "unixs://absolute_path",
+			useUnix:                true,
+			useTLS:                 true,
+			clientURLPattern:       "unixs://localhost:%d",
+			expectAuthorityPattern: "localhost:${MEMBER_PORT}",
+		},
 		{
 			name:                   "http://domain[:port]",
 			clientURLPattern:       "http://localhost:%d",
@@ -93,6 +121,9 @@ func TestAuthority(t *testing.T) {
 				cfg.IsClientAutoTLS = tc.useInsecureTLS
 				// Enable debug mode to get logs with http2 headers (including authority)
 				cfg.EnvVars = map[string]string{"GODEBUG": "http2debug=2"}
+				if tc.useUnix {
+					cfg.BaseClientScheme = "unix"
+				}
 
 				epc, err := e2e.NewEtcdProcessCluster(t, cfg)
 				if err != nil {

tests/e2e/ctl_v3_test.go

@@ -58,7 +58,7 @@ func TestClusterVersion(t *testing.T) {
 			cfg := e2e.NewConfigNoTLS()
 			cfg.ExecPath = binary
 			cfg.SnapshotCount = 3
-			cfg.BaseScheme = "unix" // to avoid port conflict
+			cfg.BasePeerScheme = "unix" // to avoid port conflict
 			cfg.RollingStart = tt.rollingStart
 
 			epc, err := e2e.NewEtcdProcessCluster(t, cfg)

tests/e2e/etcd_release_upgrade_test.go

@@ -39,7 +39,7 @@ func TestReleaseUpgrade(t *testing.T) {
 	copiedCfg := e2e.NewConfigNoTLS()
 	copiedCfg.ExecPath = lastReleaseBinary
 	copiedCfg.SnapshotCount = 3
-	copiedCfg.BaseScheme = "unix" // to avoid port conflict
+	copiedCfg.BasePeerScheme = "unix" // to avoid port conflict
 
 	epc, err := e2e.NewEtcdProcessCluster(t, copiedCfg)
 	if err != nil {
@@ -125,7 +125,7 @@ func TestReleaseUpgradeWithRestart(t *testing.T) {
 	copiedCfg := e2e.NewConfigNoTLS()
 	copiedCfg.ExecPath = lastReleaseBinary
 	copiedCfg.SnapshotCount = 10
-	copiedCfg.BaseScheme = "unix"
+	copiedCfg.BasePeerScheme = "unix"
 
 	epc, err := e2e.NewEtcdProcessCluster(t, copiedCfg)
 	if err != nil {

tests/framework/e2e/cluster.go

@@ -148,8 +148,11 @@ type EtcdProcessClusterConfig struct {
 
 	ClusterSize int
 
-	BaseScheme string
-	BasePort   int
+	// BasePeerScheme specifies scheme of --listen-peer-urls and --initial-advertise-peer-urls
+	BasePeerScheme string
+	BasePort       int
+	// BaseClientScheme specifies scheme of --listen-client-urls, --listen-client-http-urls and --initial-advertise-client-urls
+	BaseClientScheme string
 
 	MetricsURLScheme string
 
@@ -243,21 +246,11 @@ func StartEtcdProcessCluster(t testing.TB, epc *EtcdProcessCluster, cfg *EtcdPro
 }
 
 func (cfg *EtcdProcessClusterConfig) ClientScheme() string {
-	if cfg.ClientTLS == ClientTLS {
-		return "https"
-	}
-	return "http"
+	return setupScheme(cfg.BaseClientScheme, cfg.ClientTLS == ClientTLS)
 }
 
 func (cfg *EtcdProcessClusterConfig) PeerScheme() string {
-	peerScheme := cfg.BaseScheme
-	if peerScheme == "" {
-		peerScheme = "http"
-	}
-	if cfg.IsPeerTLS {
-		peerScheme += "s"
-	}
-	return peerScheme
+	return setupScheme(cfg.BasePeerScheme, cfg.IsPeerTLS)
 }
 
 func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*EtcdServerProcessConfig {
@@ -285,10 +278,10 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 		clientHttpPort := port + 4
 
 		if cfg.ClientTLS == ClientTLSAndNonTLS {
-			curl = clientURL(clientPort, ClientNonTLS)
-			curls = []string{curl, clientURL(clientPort, ClientTLS)}
+			curl = clientURL(cfg.ClientScheme(), clientPort, ClientNonTLS)
+			curls = []string{curl, clientURL(cfg.ClientScheme(), clientPort, ClientTLS)}
 		} else {
-			curl = clientURL(clientPort, cfg.ClientTLS)
+			curl = clientURL(cfg.ClientScheme(), clientPort, cfg.ClientTLS)
 			curls = []string{curl}
 		}
 
@@ -326,7 +319,7 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 		}
 		var clientHttpUrl string
 		if cfg.ClientHttpSeparate {
-			clientHttpUrl = clientURL(clientHttpPort, cfg.ClientTLS)
+			clientHttpUrl = clientURL(cfg.ClientScheme(), clientHttpPort, cfg.ClientTLS)
 			args = append(args, "--listen-client-http-urls", clientHttpUrl)
 		}
 		args = AddV2Args(args)
@@ -429,13 +422,13 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 	return etcdCfgs
 }
 
-func clientURL(port int, connType ClientConnType) string {
+func clientURL(scheme string, port int, connType ClientConnType) string {
 	curlHost := fmt.Sprintf("localhost:%d", port)
 	switch connType {
 	case ClientNonTLS:
-		return (&url.URL{Scheme: "http", Host: curlHost}).String()
+		return (&url.URL{Scheme: scheme, Host: curlHost}).String()
 	case ClientTLS:
-		return (&url.URL{Scheme: "https", Host: curlHost}).String()
+		return (&url.URL{Scheme: ToTLS(scheme), Host: curlHost}).String()
 	default:
 		panic(fmt.Sprintf("Unsupported connection type %v", connType))
 	}

tests/framework/e2e/util.go

@@ -120,8 +120,24 @@ func CloseWithTimeout(p *expect.ExpectProcess, d time.Duration) error {
 	return fmt.Errorf("took longer than %v to Close process %+v", d, p)
 }
 
+func setupScheme(s string, isTLS bool) string {
+	if s == "" {
+		s = "http"
+	}
+	if isTLS {
+		s = ToTLS(s)
+	}
+	return s
+}
+
 func ToTLS(s string) string {
-	return strings.Replace(s, "http://", "https://", 1)
+	if strings.Contains(s, "http") && !strings.Contains(s, "https") {
+		return strings.Replace(s, "http", "https", 1)
+	}
+	if strings.Contains(s, "unix") && !strings.Contains(s, "unixs") {
+		return strings.Replace(s, "unix", "unixs", 1)
+	}
+	return s
 }
 
 func SkipInShortMode(t testing.TB) {

tests/integration/cluster.go

@@ -38,7 +38,7 @@ import (
 	"go.etcd.io/etcd/client/pkg/v3/transport"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 	"go.etcd.io/etcd/client/v2"
-	"go.etcd.io/etcd/client/v3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/pkg/v3/grpc_testing"
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/server/v3/config"
@@ -192,6 +192,7 @@ func schemeFromTLSInfo(tls *transport.TLSInfo) string {
 	return URLSchemeTLS
 }
 
+// fillClusterForMembers fills up Member.InitialPeerURLsMap from each member's [name, scheme and PeerListeners address]
 func (c *cluster) fillClusterForMembers() error {
 	if c.cfg.DiscoveryURL != "" {
 		// cluster will be discovered