server: Add --listen-client-http-urls flag to allow running grpc server separate from http server

Difference in load configuration for watch delay tests show how huge the
impact is. Even with random write scheduler grpc under http
server can only handle 500 KB with 2 seconds delay. On the other hand,
separate grpc server easily hits 10, 100 or even 1000 MB within 100 miliseconds.

Priority write scheduler that was used in most previous releases
is far worse than random one.

Tests configured to only 5 MB to avoid flakes and taking too long to fill
etcd.

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>

server/embed/config.go

@@ -205,12 +205,12 @@ type Config struct {
 	// streams that each client can open at a time.
 	MaxConcurrentStreams uint32 `json:"max-concurrent-streams"`
 
-	ListenPeerUrls, ListenClientUrls       []url.URL
-	AdvertisePeerUrls, AdvertiseClientUrls []url.URL
-	ClientTLSInfo                          transport.TLSInfo
-	ClientAutoTLS                          bool
-	PeerTLSInfo                            transport.TLSInfo
-	PeerAutoTLS                            bool
+	ListenPeerUrls, ListenClientUrls, ListenClientHttpUrls []url.URL
+	AdvertisePeerUrls, AdvertiseClientUrls                 []url.URL
+	ClientTLSInfo                                          transport.TLSInfo
+	ClientAutoTLS                                          bool
+	PeerTLSInfo                                            transport.TLSInfo
+	PeerAutoTLS                                            bool
 	// SelfSignedCertValidity specifies the validity period of the client and peer certificates
 	// that are automatically generated by etcd when you specify ClientAutoTLS and PeerAutoTLS,
 	// the unit is year, and the default is 1
@@ -423,10 +423,11 @@ type configYAML struct {
 
 // configJSON has file options that are translated into Config options
 type configJSON struct {
-	ListenPeerUrls      string `json:"listen-peer-urls"`
-	ListenClientUrls    string `json:"listen-client-urls"`
-	AdvertisePeerUrls   string `json:"initial-advertise-peer-urls"`
-	AdvertiseClientUrls string `json:"advertise-client-urls"`
+	ListenPeerUrls       string `json:"listen-peer-urls"`
+	ListenClientUrls     string `json:"listen-client-urls"`
+	ListenClientHttpUrls string `json:"listen-client-http-urls"`
+	AdvertisePeerUrls    string `json:"initial-advertise-peer-urls"`
+	AdvertiseClientUrls  string `json:"advertise-client-urls"`
 
 	CORSJSON          string `json:"cors"`
 	HostWhitelistJSON string `json:"host-whitelist"`
@@ -557,6 +558,15 @@ func (cfg *configYAML) configFromFile(path string) error {
 		cfg.Config.ListenClientUrls = u
 	}
 
+	if cfg.configJSON.ListenClientHttpUrls != "" {
+		u, err := types.NewURLs(strings.Split(cfg.configJSON.ListenClientHttpUrls, ","))
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "unexpected error setting up listen-client-http-urls: %v\n", err)
+			os.Exit(1)
+		}
+		cfg.Config.ListenClientHttpUrls = u
+	}
+
 	if cfg.configJSON.AdvertisePeerUrls != "" {
 		u, err := types.NewURLs(strings.Split(cfg.configJSON.AdvertisePeerUrls, ","))
 		if err != nil {
@@ -656,6 +666,12 @@ func (cfg *Config) Validate() error {
 	if err := checkBindURLs(cfg.ListenClientUrls); err != nil {
 		return err
 	}
+	if err := checkBindURLs(cfg.ListenClientHttpUrls); err != nil {
+		return err
+	}
+	if len(cfg.ListenClientHttpUrls) == 0 {
+		cfg.logger.Warn("Running http and grpc server on single port. This is not recommended for production.")
+	}
 	if err := checkBindURLs(cfg.ListenMetricsUrls); err != nil {
 		return err
 	}
@@ -877,9 +893,12 @@ func (cfg *Config) ClientSelfCert() (err error) {
 		cfg.logger.Warn("ignoring client auto TLS since certs given")
 		return nil
 	}
-	chosts := make([]string, len(cfg.ListenClientUrls))
-	for i, u := range cfg.ListenClientUrls {
-		chosts[i] = u.Host
+	chosts := make([]string, 0, len(cfg.ListenClientUrls)+len(cfg.ListenClientHttpUrls))
+	for _, u := range cfg.ListenClientUrls {
+		chosts = append(chosts, u.Host)
+	}
+	for _, u := range cfg.ListenClientHttpUrls {
+		chosts = append(chosts, u.Host)
 	}
 	cfg.ClientTLSInfo, err = transport.SelfCert(cfg.logger, filepath.Join(cfg.Dir, "fixtures", "client"), chosts, cfg.SelfSignedCertValidity)
 	if err != nil {
@@ -1014,6 +1033,14 @@ func (cfg *Config) getListenClientUrls() (ss []string) {
 	return ss
 }
 
+func (cfg *Config) getListenClientHttpUrls() (ss []string) {
+	ss = make([]string, len(cfg.ListenClientHttpUrls))
+	for i := range cfg.ListenClientHttpUrls {
+		ss[i] = cfg.ListenClientHttpUrls[i].String()
+	}
+	return ss
+}
+
 func (cfg *Config) getMetricsURLs() (ss []string) {
 	ss = make([]string, len(cfg.ListenMetricsUrls))
 	for i := range cfg.ListenMetricsUrls {

server/embed/etcd.go

@@ -440,11 +440,16 @@ func (e *Etcd) Close() {
 
 func stopServers(ctx context.Context, ss *servers) {
 	// first, close the http.Server
-	ss.http.Shutdown(ctx)
-	// do not grpc.Server.GracefulStop with TLS enabled etcd server
+	if ss.http != nil {
+		ss.http.Shutdown(ctx)
+	}
+	if ss.grpc == nil {
+		return
+	}
+	// do not grpc.Server.GracefulStop when grpc runs under http server
 	// See https://github.com/grpc/grpc-go/issues/1384#issuecomment-317124531
 	// and https://github.com/etcd-io/etcd/issues/8916
-	if ss.secure {
+	if ss.secure && ss.http != nil {
 		ss.grpc.Stop()
 		return
 	}
@@ -614,7 +619,7 @@ func configureClientListeners(cfg *Config) (sctxs map[string]*serveCtx, err erro
 	}
 
 	sctxs = make(map[string]*serveCtx)
-	for _, u := range cfg.ListenClientUrls {
+	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()))
@@ -641,6 +646,24 @@ func configureClientListeners(cfg *Config) (sctxs map[string]*serveCtx, err erro
 		sctx.addr = addr
 		sctx.network = network
 	}
+	for _, u := range cfg.ListenClientHttpUrls {
+		addr, secure, network := resolveUrl(u)
+
+		sctx := sctxs[addr]
+		if sctx == nil {
+			sctx = newServeCtx(cfg.logger)
+			sctxs[addr] = sctx
+		} else if !sctx.httpOnly {
+			return nil, fmt.Errorf("cannot bind both --client-listen-urls and --client-listen-http-urls on the same url %s", u.String())
+		}
+		sctx.secure = sctx.secure || secure
+		sctx.insecure = sctx.insecure || !secure
+		sctx.scheme = u.Scheme
+		sctx.addr = addr
+		sctx.network = network
+		sctx.httpOnly = true
+	}
+
 	for _, sctx := range sctxs {
 		if sctx.l, err = transport.NewListenerWithOpts(sctx.addr, sctx.scheme,
 			transport.WithSocketOpts(&cfg.SocketOpts),
@@ -663,7 +686,7 @@ func configureClientListeners(cfg *Config) (sctxs map[string]*serveCtx, err erro
 		}
 
 		defer func(addr string) {
-			if err == nil {
+			if err == nil || sctx.l == nil {
 				return
 			}
 			sctx.l.Close()
@@ -743,20 +766,27 @@ func (e *Etcd) serveClients() (err error) {
 		}))
 	}
 
+	splitHttp := false
+	for _, sctx := range e.sctxs {
+		if sctx.httpOnly {
+			splitHttp = true
+		}
+	}
+
 	// start client servers in each goroutine
 	for _, sctx := range e.sctxs {
 		go func(s *serveCtx) {
-			e.errHandler(s.serve(e.Server, &e.cfg.ClientTLSInfo, h, e.errHandler, e.grpcGatewayDial(), gopts...))
+			e.errHandler(s.serve(e.Server, &e.cfg.ClientTLSInfo, h, e.errHandler, e.grpcGatewayDial(splitHttp), splitHttp, gopts...))
 		}(sctx)
 	}
 	return nil
 }
 
-func (e *Etcd) grpcGatewayDial() (grpcDial func(ctx context.Context) (*grpc.ClientConn, error)) {
+func (e *Etcd) grpcGatewayDial(splitHttp bool) (grpcDial func(ctx context.Context) (*grpc.ClientConn, error)) {
 	if !e.cfg.EnableGRPCGateway {
 		return nil
 	}
-	sctx := e.pickGrpcGatewayServeContext()
+	sctx := e.pickGrpcGatewayServeContext(splitHttp)
 	addr := sctx.addr
 	if network := sctx.network; network == "unix" {
 		// explicitly define unix network for gRPC socket support
@@ -790,9 +820,11 @@ func (e *Etcd) grpcGatewayDial() (grpcDial func(ctx context.Context) (*grpc.Clie
 	}
 }
 
-func (e *Etcd) pickGrpcGatewayServeContext() *serveCtx {
+func (e *Etcd) pickGrpcGatewayServeContext(splitHttp bool) *serveCtx {
 	for _, sctx := range e.sctxs {
-		return sctx
+		if !splitHttp || !sctx.httpOnly {
+			return sctx
+		}
 	}
 	panic("Expect at least one context able to serve grpc")
 }

server/embed/serve.go

@@ -56,6 +56,7 @@ type serveCtx struct {
 	network  string
 	secure   bool
 	insecure bool
+	httpOnly bool
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -94,6 +95,7 @@ func (sctx *serveCtx) serve(
 	handler http.Handler,
 	errHandler func(error),
 	grpcDialForRestGatewayBackends func(ctx context.Context) (*grpc.ClientConn, error),
+	splitHttp bool,
 	gopts ...grpc.ServerOption) (err error) {
 	logger := defaultLog.New(ioutil.Discard, "etcdhttp", 0)
 	<-s.ReadyNotify()
@@ -101,6 +103,12 @@ func (sctx *serveCtx) serve(
 	sctx.lg.Info("ready to serve client requests")
 
 	m := cmux.New(sctx.l)
+	var server func() error
+	onlyGRPC := splitHttp && !sctx.httpOnly
+	onlyHttp := splitHttp && sctx.httpOnly
+	grpcEnabled := !onlyHttp
+	httpEnabled := !onlyGRPC
+
 	v3c := v3client.New(s)
 	servElection := v3election.NewElectionServer(v3c)
 	servLock := v3lock.NewLockServer(v3c)
@@ -116,104 +124,137 @@ func (sctx *serveCtx) serve(
 			return err
 		}
 	}
+	var traffic string
+	switch {
+	case onlyGRPC:
+		traffic = "grpc"
+	case onlyHttp:
+		traffic = "http"
+	default:
+		traffic = "grpc+http"
+	}
 
 	if sctx.insecure {
-		gs := v3rpc.Server(s, nil, nil, gopts...)
-		v3electionpb.RegisterElectionServer(gs, servElection)
-		v3lockpb.RegisterLockServer(gs, servLock)
-		if sctx.serviceRegister != nil {
-			sctx.serviceRegister(gs)
-		}
-
-		defer func(gs *grpc.Server) {
-			if err != nil {
-				sctx.lg.Warn("stopping insecure grpc server due to error", zap.Error(err))
-				gs.Stop()
-				sctx.lg.Warn("stopped insecure grpc server due to error", zap.Error(err))
+		var gs *grpc.Server
+		var srv *http.Server
+		if httpEnabled {
+			httpmux := sctx.createMux(gwmux, handler)
+			srv = &http.Server{
+				Handler:  createAccessController(sctx.lg, s, httpmux),
+				ErrorLog: logger, // do not log user error
+			}
+			if err := configureHttpServer(srv, s.Cfg); err != nil {
+				sctx.lg.Error("Configure http server failed", zap.Error(err))
+				return err
 			}
-		}(gs)
-
-		grpcl := m.Match(cmux.HTTP2())
-		go func(gs *grpc.Server, grpcLis net.Listener) {
-			errHandler(gs.Serve(grpcLis))
-		}(gs, grpcl)
-
-		httpmux := sctx.createMux(gwmux, handler)
-
-		srvhttp := &http.Server{
-			Handler:  createAccessController(sctx.lg, s, httpmux),
-			ErrorLog: logger, // do not log user error
 		}
-		if err := configureHttpServer(srvhttp, s.Cfg); err != nil {
-			sctx.lg.Error("Configure http server failed", zap.Error(err))
-			return err
+		if grpcEnabled {
+			gs = v3rpc.Server(s, nil, nil, gopts...)
+			v3electionpb.RegisterElectionServer(gs, servElection)
+			v3lockpb.RegisterLockServer(gs, servLock)
+			if sctx.serviceRegister != nil {
+				sctx.serviceRegister(gs)
+			}
+			defer func(gs *grpc.Server) {
+				if err != nil {
+					sctx.lg.Warn("stopping insecure grpc server due to error", zap.Error(err))
+					gs.Stop()
+					sctx.lg.Warn("stopped insecure grpc server due to error", zap.Error(err))
+				}
+			}(gs)
 		}
-		httpl := m.Match(cmux.HTTP1())
+		if onlyGRPC {
+			server = func() error {
+				return gs.Serve(sctx.l)
+			}
+		} else {
+			server = m.Serve
 
-		go func(srvhttp *http.Server, httpLis net.Listener) {
-			errHandler(srvhttp.Serve(httpLis))
-		}(srvhttp, httpl)
+			httpl := m.Match(cmux.HTTP1())
+			go func(srvhttp *http.Server, tlsLis net.Listener) {
+				errHandler(srvhttp.Serve(tlsLis))
+			}(srv, httpl)
 
-		sctx.serversC <- &servers{grpc: gs, http: srvhttp}
+			if grpcEnabled {
+				grpcl := m.Match(cmux.HTTP2())
+				go func(gs *grpc.Server, l net.Listener) {
+					errHandler(gs.Serve(l))
+				}(gs, grpcl)
+			}
+		}
+
+		sctx.serversC <- &servers{grpc: gs, http: srv}
 		sctx.lg.Info(
 			"serving client traffic insecurely; this is strongly discouraged!",
+			zap.String("traffic", traffic),
 			zap.String("address", sctx.l.Addr().String()),
 		)
 	}
 
 	if sctx.secure {
+		var gs *grpc.Server
+		var srv *http.Server
+
 		tlscfg, tlsErr := tlsinfo.ServerConfig()
 		if tlsErr != nil {
 			return tlsErr
 		}
 
-		gs := v3rpc.Server(s, tlscfg, nil, gopts...)
-		v3electionpb.RegisterElectionServer(gs, servElection)
-		v3lockpb.RegisterLockServer(gs, servLock)
-		if sctx.serviceRegister != nil {
-			sctx.serviceRegister(gs)
-		}
-
-		defer func(gs *grpc.Server) {
-			if err != nil {
-				sctx.lg.Warn("stopping secure grpc server due to error", zap.Error(err))
-				gs.Stop()
-				sctx.lg.Warn("stopped secure grpc server due to error", zap.Error(err))
+		if grpcEnabled {
+			gs = v3rpc.Server(s, tlscfg, nil, gopts...)
+			v3electionpb.RegisterElectionServer(gs, servElection)
+			v3lockpb.RegisterLockServer(gs, servLock)
+			if sctx.serviceRegister != nil {
+				sctx.serviceRegister(gs)
 			}
-		}(gs)
-
-		handler = grpcHandlerFunc(gs, handler)
-		var tlsl net.Listener
-		tlsl, err = transport.NewTLSListener(m.Match(cmux.Any()), tlsinfo)
-		if err != nil {
-			return err
+			defer func(gs *grpc.Server) {
+				if err != nil {
+					sctx.lg.Warn("stopping secure grpc server due to error", zap.Error(err))
+					gs.Stop()
+					sctx.lg.Warn("stopped secure grpc server due to error", zap.Error(err))
+				}
+			}(gs)
 		}
-		// TODO: add debug flag; enable logging when debug flag is set
-		httpmux := sctx.createMux(gwmux, handler)
+		if httpEnabled {
+			if grpcEnabled {
+				handler = grpcHandlerFunc(gs, handler)
+			}
+			httpmux := sctx.createMux(gwmux, handler)
 
-		srv := &http.Server{
-			Handler:   createAccessController(sctx.lg, s, httpmux),
-			TLSConfig: tlscfg,
-			ErrorLog:  logger, // do not log user error
-		}
-		if err := configureHttpServer(srv, s.Cfg); err != nil {
-			sctx.lg.Error("Configure https server failed", zap.Error(err))
-			return err
+			srv = &http.Server{
+				Handler:   createAccessController(sctx.lg, s, httpmux),
+				TLSConfig: tlscfg,
+				ErrorLog:  logger, // do not log user error
+			}
+			if err := configureHttpServer(srv, s.Cfg); err != nil {
+				sctx.lg.Error("Configure https server failed", zap.Error(err))
+				return err
+			}
 		}
 
-		go func(srvhttp *http.Server, tlsLis net.Listener) {
-			errHandler(srvhttp.Serve(tlsLis))
-		}(srv, tlsl)
+		if onlyGRPC {
+			server = func() error { return gs.Serve(sctx.l) }
+		} else {
+			server = m.Serve
+
+			tlsl, err := transport.NewTLSListener(m.Match(cmux.Any()), tlsinfo)
+			if err != nil {
+				return err
+			}
+			go func(srvhttp *http.Server, tlsl net.Listener) {
+				errHandler(srvhttp.Serve(tlsl))
+			}(srv, tlsl)
+		}
 
 		sctx.serversC <- &servers{secure: true, grpc: gs, http: srv}
 		sctx.lg.Info(
 			"serving client traffic securely",
+			zap.String("traffic", traffic),
 			zap.String("address", sctx.l.Addr().String()),
 		)
 	}
 
-	close(sctx.serversC)
-	return m.Serve()
+	return server()
 }
 
 func configureHttpServer(srv *http.Server, cfg config.ServerConfig) error {