[FAB-9104] Improve core/comm tests

There were a number of flaky tests in
the core/comm package after upgrading the
gRPC package.  Many seemed to be due to
running tests in parallel while others
were just due to changes in library
behavior.

Fixes flaky TestNewConnection test as well as
eliminating race conditions for the tests

Change-Id: If2465ab3332408d7c7d7b87abe14ec4ac4393bc8
Signed-off-by: Gari Singh <gari.r.singh@gmail.com>

Author: mastersingh24

core/comm/client.go

@@ -60,8 +60,8 @@ func NewGRPCClient(config ClientConfig) (*GRPCClient, error) {
 		grpc.WithBlock())
 	client.timeout = config.Timeout
 	// set send/recv message size to package defaults
-	client.maxRecvMsgSize = maxRecvMsgSize
-	client.maxSendMsgSize = maxSendMsgSize
+	client.maxRecvMsgSize = MaxRecvMsgSize
+	client.maxSendMsgSize = MaxSendMsgSize
 
 	return client, nil
 }

core/comm/client_test.go

@@ -21,23 +21,23 @@ import (
 	testpb "github.com/hyperledger/fabric/core/comm/testdata/grpc"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
-	netctx "golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 )
 
-var caPEM, certPEM, keyPEM, serverKey, serverPEM []byte
 var testClientCert, testServerCert tls.Certificate
 var testTimeout = 1 * time.Second // conservative
 
 type echoServer struct{}
 
-func (es *echoServer) EchoCall(ctx netctx.Context,
+func (es *echoServer) EchoCall(ctx context.Context,
 	echo *testpb.Echo) (*testpb.Echo, error) {
 	return echo, nil
 }
 
 func TestNewGRPCClient_GoodConfig(t *testing.T) {
 	t.Parallel()
-	loadCerts(t)
+	caPEM, certPEM, keyPEM, _, _ := loadCerts(t)
 
 	config := comm.ClientConfig{}
 	client, err := comm.NewGRPCClient(config)
@@ -82,7 +82,7 @@ func TestNewGRPCClient_GoodConfig(t *testing.T) {
 
 func TestNewGRPCClient_BadConfig(t *testing.T) {
 	t.Parallel()
-	loadCerts(t)
+	_, certPEM, keyPEM, _, _ := loadCerts(t)
 	// bad root cert
 	config := comm.ClientConfig{
 		SecOpts: &comm.SecureOptions{
@@ -143,7 +143,7 @@ func TestNewConnection_Timeout(t *testing.T) {
 
 func TestNewConnection(t *testing.T) {
 	t.Parallel()
-	loadCerts(t)
+	caPEM, certPEM, keyPEM, _, _ := loadCerts(t)
 	certPool := x509.NewCertPool()
 	ok := certPool.AppendCertsFromPEM(caPEM)
 	if !ok {
@@ -269,24 +269,15 @@ func TestNewConnection(t *testing.T) {
 				t.Fatalf("error creating server for test: %v", err)
 			}
 			defer lis.Close()
+			serverOpts := []grpc.ServerOption{}
 			if test.serverTLS != nil {
-				t.Log("starting server with TLS")
-				tlsLis := tls.NewListener(lis, test.serverTLS)
-				defer tlsLis.Close()
-				go func() {
-					rawConn, err := tlsLis.Accept()
-					defer rawConn.Close()
-					if err != nil {
-						t.Fatalf("error accepting TLS connection: %v", err)
-					}
-					t.Logf("server: accepted from %s", rawConn.RemoteAddr())
-					sconn := tls.Server(rawConn, test.serverTLS)
-					err = sconn.Handshake()
-					if err != nil {
-						t.Logf("tls handshake error: %v", err)
-					}
-				}()
+				serverOpts = append(
+					serverOpts,
+					grpc.Creds(credentials.NewTLS(test.serverTLS)))
 			}
+			srv := grpc.NewServer(serverOpts...)
+			defer srv.Stop()
+			go srv.Serve(lis)
 			client, err := comm.NewGRPCClient(test.config)
 			if err != nil {
 				t.Fatalf("error creating client for test: %v", err)
@@ -306,7 +297,7 @@ func TestNewConnection(t *testing.T) {
 
 func TestSetServerRootCAs(t *testing.T) {
 	t.Parallel()
-	loadCerts(t)
+	caPEM, _, _, _, _ := loadCerts(t)
 
 	config := comm.ClientConfig{
 		SecOpts: &comm.SecureOptions{
@@ -325,26 +316,10 @@ func TestSetServerRootCAs(t *testing.T) {
 		t.Fatalf("failed to create listener for test server: %v", err)
 	}
 	defer lis.Close()
-	tlsConf := &tls.Config{
-		Certificates: []tls.Certificate{testServerCert}}
-	tlsLis := tls.NewListener(lis, tlsConf)
-	defer tlsLis.Close()
-	go func() {
-		// 3 connection tests
-		for i := 0; i < 3; i++ {
-			rawConn, err := tlsLis.Accept()
-			defer rawConn.Close()
-			if err != nil {
-				t.Fatalf("error accepting TLS connection: %v", err)
-			}
-			t.Logf("server: accepted from %s", rawConn.RemoteAddr())
-			sconn := tls.Server(rawConn, tlsConf)
-			err = sconn.Handshake()
-			if err != nil {
-				t.Logf("tls handshake error: %v", err)
-			}
-		}
-	}()
+	srv := grpc.NewServer(grpc.Creds(credentials.NewTLS(
+		&tls.Config{Certificates: []tls.Certificate{testServerCert}})))
+	defer srv.Stop()
+	go srv.Serve(lis)
 
 	// initial config should work
 	t.Log("running initial good config")
@@ -371,7 +346,9 @@ func TestSetServerRootCAs(t *testing.T) {
 	conn, err = client.NewConnection(address, "")
 	assert.NoError(t, err)
 	assert.NotNil(t, conn)
-	conn.Close()
+	if conn != nil {
+		conn.Close()
+	}
 
 	// bad root cert
 	t.Log("running bad root cert")
@@ -472,7 +449,13 @@ func TestSetMessageSize(t *testing.T) {
 	}
 }
 
-func loadCerts(t *testing.T) {
+func loadCerts(t *testing.T) (
+	caPEM,
+	certPEM,
+	keyPEM,
+	serverKey,
+	serverPEM []byte) {
+
 	t.Helper()
 	var err error
 	caPEM, err = ioutil.ReadFile(filepath.Join("testdata", "certs",
@@ -497,4 +480,6 @@ func loadCerts(t *testing.T) {
 	testServerCert, err = tls.LoadX509KeyPair(filepath.Join("testdata", "certs",
 		"Org1-server1-cert.pem"), filepath.Join("testdata", "certs",
 		"Org1-server1-key.pem"))
+
+	return caPEM, certPEM, keyPEM, serverKey, serverPEM
 }

core/comm/config.go

@@ -21,8 +21,8 @@ var (
 	// Is TLS enabled
 	tlsEnabled bool
 	// Max send and receive bytes for grpc clients and servers
-	maxRecvMsgSize = 100 * 1024 * 1024
-	maxSendMsgSize = 100 * 1024 * 1024
+	MaxRecvMsgSize = 100 * 1024 * 1024
+	MaxSendMsgSize = 100 * 1024 * 1024
 	// Default peer keepalive options
 	keepaliveOptions = &KeepaliveOptions{
 		ClientInterval:    time.Duration(1) * time.Minute,  // 1 min
@@ -123,30 +123,6 @@ func TLSEnabled() bool {
 	return tlsEnabled
 }
 
-// MaxRecvMsgSize returns the maximum message size in bytes that gRPC clients
-// and servers can receive
-func MaxRecvMsgSize() int {
-	return maxRecvMsgSize
-}
-
-// SetMaxRecvMsgSize sets the maximum message size in bytes that gRPC clients
-// and servers can receive
-func SetMaxRecvMsgSize(size int) {
-	maxRecvMsgSize = size
-}
-
-// MaxSendMsgSize returns the maximum message size in bytes that gRPC clients
-// and servers can send
-func MaxSendMsgSize() int {
-	return maxSendMsgSize
-}
-
-// SetMaxSendMsgSize sets the maximum message size in bytes that gRPC clients
-// and servers can send
-func SetMaxSendMsgSize(size int) {
-	maxSendMsgSize = size
-}
-
 // DefaultKeepaliveOptions returns sane default keepalive settings for gRPC
 // servers and clients
 func DefaultKeepaliveOptions() *KeepaliveOptions {

core/comm/config_test.go

@@ -16,19 +16,10 @@ import (
 func TestConfig(t *testing.T) {
 	t.Parallel()
 	// check the defaults
-	assert.EqualValues(t, maxRecvMsgSize, MaxRecvMsgSize())
-	assert.EqualValues(t, maxSendMsgSize, MaxSendMsgSize())
 	assert.EqualValues(t, keepaliveOptions, DefaultKeepaliveOptions())
 	assert.EqualValues(t, false, TLSEnabled())
 	assert.EqualValues(t, true, configurationCached)
 
-	// set send/recv msg sizes
-	size := 10 * 1024 * 1024
-	SetMaxRecvMsgSize(size)
-	SetMaxSendMsgSize(size)
-	assert.EqualValues(t, size, MaxRecvMsgSize())
-	assert.EqualValues(t, size, MaxSendMsgSize())
-
 	// reset cache
 	configurationCached = false
 	viper.Set("peer.tls.enabled", true)

core/comm/connection.go

@@ -205,8 +205,8 @@ func NewClientConnectionWithAddress(peerAddress string, block bool, tslEnabled b
 	if block {
 		opts = append(opts, grpc.WithBlock())
 	}
-	opts = append(opts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(MaxRecvMsgSize()),
-		grpc.MaxCallSendMsgSize(MaxSendMsgSize())))
+	opts = append(opts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(MaxRecvMsgSize),
+		grpc.MaxCallSendMsgSize(MaxSendMsgSize)))
 	ctx := context.Background()
 	ctx, _ = context.WithTimeout(ctx, defaultTimeout)
 	conn, err := grpc.DialContext(ctx, peerAddress, opts...)

core/comm/server.go

@@ -118,8 +118,8 @@ func NewGRPCServerFromListener(listener net.Listener, serverConfig ServerConfig)
 		}
 	}
 	// set max send and recv msg sizes
-	serverOpts = append(serverOpts, grpc.MaxSendMsgSize(MaxSendMsgSize()))
-	serverOpts = append(serverOpts, grpc.MaxRecvMsgSize(MaxRecvMsgSize()))
+	serverOpts = append(serverOpts, grpc.MaxSendMsgSize(MaxSendMsgSize))
+	serverOpts = append(serverOpts, grpc.MaxRecvMsgSize(MaxRecvMsgSize))
 	// set the keepalive options
 	serverOpts = append(serverOpts, ServerKeepaliveOptions(serverConfig.KaOpts)...)
 	// set connection timeout

core/comm/server_test.go

@@ -637,7 +637,7 @@ func TestNewSecureGRPCServer(t *testing.T) {
 		tlsVersion := tlsVersion
 		t.Run(tlsVersions[counter], func(t *testing.T) {
 			t.Parallel()
-			_, err = invokeEmptyCall(testAddress,
+			_, err := invokeEmptyCall(testAddress,
 				[]grpc.DialOption{grpc.WithTransportCredentials(
 					credentials.NewTLS(&tls.Config{
 						RootCAs:    certPool,

core/deliverservice/deliveryclient.go

@@ -225,8 +225,8 @@ func DefaultConnectionFactory(channelID string) func(endpoint string) (*grpc.Cli
 	return func(endpoint string) (*grpc.ClientConn, error) {
 		dialOpts := []grpc.DialOption{grpc.WithBlock()}
 		// set max send/recv msg sizes
-		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(comm.MaxRecvMsgSize()),
-			grpc.MaxCallSendMsgSize(comm.MaxSendMsgSize())))
+		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(comm.MaxRecvMsgSize),
+			grpc.MaxCallSendMsgSize(comm.MaxSendMsgSize)))
 		// set the keepalive options
 		kaOpts := comm.DefaultKeepaliveOptions()
 		if viper.IsSet("peer.keepalive.deliveryClient.interval") {

peer/node/start.go

@@ -249,8 +249,11 @@ func serve(args []string) error {
 	secureDialOpts := func() []grpc.DialOption {
 		var dialOpts []grpc.DialOption
 		// set max send/recv msg sizes
-		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(comm.MaxRecvMsgSize()),
-			grpc.MaxCallSendMsgSize(comm.MaxSendMsgSize())))
+		dialOpts = append(
+			dialOpts,
+			grpc.WithDefaultCallOptions(
+				grpc.MaxCallRecvMsgSize(comm.MaxRecvMsgSize),
+				grpc.MaxCallSendMsgSize(comm.MaxSendMsgSize)))
 		// set the keepalive options
 		kaOpts := comm.DefaultKeepaliveOptions()
 		if viper.IsSet("peer.keepalive.client.interval") {