Fix watch stream starvation

pkg/stringutil/rand.go

@@ -24,7 +24,7 @@ func UniqueStrings(slen uint, n int) (ss []string) {
 	exist := make(map[string]struct{})
 	ss = make([]string, 0, n)
 	for len(ss) < n {
-		s := randString(slen)
+		s := RandString(slen)
 		if _, ok := exist[s]; !ok {
 			ss = append(ss, s)
 			exist[s] = struct{}{}
@@ -37,14 +37,14 @@ func UniqueStrings(slen uint, n int) (ss []string) {
 func RandomStrings(slen uint, n int) (ss []string) {
 	ss = make([]string, 0, n)
 	for i := 0; i < n; i++ {
-		ss = append(ss, randString(slen))
+		ss = append(ss, RandString(slen))
 	}
 	return ss
 }
 
 const chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 
-func randString(l uint) string {
+func RandString(l uint) string {
 	rand.Seed(time.Now().UnixNano())
 	s := make([]byte, l)
 	for i := 0; i < int(l); i++ {

server/embed/etcd.go

@@ -455,6 +455,7 @@ func (e *Etcd) Close() {
 }
 
 func stopServers(ctx context.Context, ss *servers) {
+	ss.cmux.Close()
 	// first, close the http.Server
 	ss.http.Shutdown(ctx)
 	// do not grpc.Server.GracefulStop with TLS enabled etcd server

server/embed/serve.go

@@ -16,6 +16,7 @@ package embed
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"io"
 	defaultLog "log"
@@ -27,7 +28,7 @@ import (
 
 	etcdservergw "go.etcd.io/etcd/api/v3/etcdserverpb/gw"
 	"go.etcd.io/etcd/client/pkg/v3/transport"
-	"go.etcd.io/etcd/client/v3/credentials"
+	clientcreds "go.etcd.io/etcd/client/v3/credentials"
 	"go.etcd.io/etcd/pkg/v3/debugutil"
 	"go.etcd.io/etcd/pkg/v3/httputil"
 	"go.etcd.io/etcd/server/v3/config"
@@ -48,6 +49,7 @@ import (
 	"golang.org/x/net/http2"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
@@ -71,6 +73,7 @@ type servers struct {
 	secure bool
 	grpc   *grpc.Server
 	http   *http.Server
+	cmux   cmux.CMux
 }
 
 func newServeCtx(lg *zap.Logger) *serveCtx {
@@ -109,113 +112,102 @@ func (sctx *serveCtx) serve(
 
 	sctx.lg.Info("ready to serve client requests")
 
-	m := cmux.New(sctx.l)
 	v3c := v3client.New(s)
 	servElection := v3election.NewElectionServer(v3c)
 	servLock := v3lock.NewLockServer(v3c)
 
-	var gs *grpc.Server
+	var tlscfg *tls.Config
+
+	// Make sure serversC is closed even if we prematurely exit the function.
+	var closed bool
 	defer func() {
-		if err != nil && gs != nil {
-			sctx.lg.Warn("stopping grpc server due to error", zap.Error(err))
-			gs.Stop()
-			sctx.lg.Warn("stopped grpc server due to error", zap.Error(err))
+		if !closed {
+			close(sctx.serversC)
 		}
 	}()
-
-	// Make sure serversC is closed even if we prematurely exit the function.
-	defer close(sctx.serversC)
-
-	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)
+	if sctx.secure {
+		tlscfg, err = tlsinfo.ServerConfig()
+		if err != nil {
+			return err
 		}
-		grpcl := m.Match(cmux.HTTP2())
-		go func() { errHandler(gs.Serve(grpcl)) }()
-
-		var gwmux *gw.ServeMux
-		if s.Cfg.EnableGRPCGateway {
-			gwmux, err = sctx.registerGateway([]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())})
-			if err != nil {
-				sctx.lg.Error("registerGateway failed", zap.Error(err))
-				return err
+	}
+	gs := v3rpc.Server(s, nil, nil, gopts...)
+	defer func() {
+		sctx.lg.Warn("stopping grpc server due to error", zap.Error(err))
+		gs.Stop()
+		sctx.lg.Warn("stopped grpc server due to error", zap.Error(err))
+	}()
+	v3electionpb.RegisterElectionServer(gs, servElection)
+	v3lockpb.RegisterLockServer(gs, servLock)
+	if sctx.serviceRegister != nil {
+		sctx.serviceRegister(gs)
+	}
+	var srv *http.Server
+	var gwmux *gw.ServeMux
+	if s.Cfg.EnableGRPCGateway {
+		var creds credentials.TransportCredentials
+		if sctx.insecure {
+			creds = insecure.NewCredentials()
+		}
+		if sctx.secure {
+			if s.Cfg.EnableGRPCGateway {
+				dtls := tlscfg.Clone()
+				// trust local server
+				dtls.InsecureSkipVerify = true
+				bundle := clientcreds.NewBundle(clientcreds.Config{TLSConfig: dtls})
+				creds = bundle.TransportCredentials()
 			}
 		}
-
-		httpmux := sctx.createMux(gwmux, handler)
-
-		srvhttp := &http.Server{
-			Handler:  createAccessController(sctx.lg, s, httpmux),
-			ErrorLog: logger, // do not log user error
+		gwmux, err = sctx.registerGateway([]grpc.DialOption{grpc.WithTransportCredentials(creds)})
+		if err != nil {
+			sctx.lg.Error("registerGateway failed", zap.Error(err))
+			return err
 		}
-		if err := configureHttpServer(srvhttp, s.Cfg); err != nil {
-			sctx.lg.Error("Configure http server failed", zap.Error(err))
+	}
+	listener := sctx.l
+	if sctx.secure {
+		listener, err = transport.NewTLSListener(listener, tlsinfo)
+		if err != nil {
 			return err
 		}
-		httpl := m.Match(cmux.HTTP1())
-		go func() { errHandler(srvhttp.Serve(httpl)) }()
+	}
+	m := cmux.New(listener)
+	// TODO: add debug flag; enable logging when debug flag is set
+	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
+	}
+	if sctx.insecure {
+		grpcl := m.Match(cmux.HTTP2())
+		go func() { errHandler(gs.Serve(grpcl)) }()
 
-		sctx.serversC <- &servers{grpc: gs, http: srvhttp}
-		sctx.lg.Info(
-			"serving client traffic insecurely; this is strongly discouraged!",
-			zap.String("address", sctx.l.Addr().String()),
-		)
+		httpl := m.Match(cmux.HTTP1())
+		go func() { errHandler(srv.Serve(httpl)) }()
 	}
 
 	if sctx.secure {
-		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)
-		}
-		handler = grpcHandlerFunc(gs, handler)
-
-		var gwmux *gw.ServeMux
-		if s.Cfg.EnableGRPCGateway {
-			dtls := tlscfg.Clone()
-			// trust local server
-			dtls.InsecureSkipVerify = true
-			bundle := credentials.NewBundle(credentials.Config{TLSConfig: dtls})
-			opts := []grpc.DialOption{grpc.WithTransportCredentials(bundle.TransportCredentials())}
-			gwmux, err = sctx.registerGateway(opts)
-			if err != nil {
-				return err
-			}
-		}
+		grpcl := m.MatchWithWriters(cmux.HTTP2MatchHeaderFieldSendSettings("content-type", "application/grpc"))
+		go func() { errHandler(gs.Serve(grpcl)) }()
 
-		var tlsl net.Listener
-		tlsl, err = transport.NewTLSListener(m.Match(cmux.Any()), tlsinfo)
-		if err != nil {
-			return err
-		}
-		// TODO: add debug flag; enable logging when debug flag is set
-		httpmux := sctx.createMux(gwmux, handler)
+		httpl := m.Match(cmux.Any())
+		go func() { errHandler(srv.Serve(httpl)) }()
+	}
 
-		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() { errHandler(srv.Serve(tlsl)) }()
+	sctx.serversC <- &servers{secure: sctx.secure, grpc: gs, http: srv, cmux: m}
+	close(sctx.serversC)
+	closed = true
 
-		sctx.serversC <- &servers{secure: true, grpc: gs, http: srv}
-		sctx.lg.Info(
-			"serving client traffic securely",
-			zap.String("address", sctx.l.Addr().String()),
-		)
+	msg := "serving client traffic securely"
+	if sctx.insecure {
+		msg = "serving client traffic insecurely; this is strongly discouraged!"
 	}
+	sctx.lg.Info(msg, zap.String("address", sctx.l.Addr().String()))
 
 	return m.Serve()
 }
@@ -227,23 +219,6 @@ func configureHttpServer(srv *http.Server, cfg config.ServerConfig) error {
 	})
 }
 
-// grpcHandlerFunc returns an http.Handler that delegates to grpcServer on incoming gRPC
-// connections or otherHandler otherwise. Given in gRPC docs.
-func grpcHandlerFunc(grpcServer *grpc.Server, otherHandler http.Handler) http.Handler {
-	if otherHandler == nil {
-		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			grpcServer.ServeHTTP(w, r)
-		})
-	}
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.ProtoMajor == 2 && strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
-			grpcServer.ServeHTTP(w, r)
-		} else {
-			otherHandler.ServeHTTP(w, r)
-		}
-	})
-}
-
 type registerHandlerFunc func(context.Context, *gw.ServeMux, *grpc.ClientConn) error
 
 func (sctx *serveCtx) registerGateway(opts []grpc.DialOption) (*gw.ServeMux, error) {