Fix racy connect network tests that always fail in Docker due to listen races
This commit is contained in:
parent
5abd43a567
commit
93ff59a132
|
@ -25,6 +25,15 @@ type Listener struct {
|
||||||
stopFlag int32
|
stopFlag int32
|
||||||
stopChan chan struct{}
|
stopChan chan struct{}
|
||||||
|
|
||||||
|
// listeningChan is closed when listener is opened successfully. It's really
|
||||||
|
// only for use in tests where we need to coordinate wait for the Serve
|
||||||
|
// goroutine to be running before we proceed trying to connect. On my laptop
|
||||||
|
// this always works out anyway but on constrained VMs and especially docker
|
||||||
|
// containers (e.g. in CI) we often see the Dial routine win the race and get
|
||||||
|
// `connection refused`. Retry loops and sleeps are unpleasant workarounds and
|
||||||
|
// this is cheap and correct.
|
||||||
|
listeningChan chan struct{}
|
||||||
|
|
||||||
logger *log.Logger
|
logger *log.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -42,6 +51,7 @@ func NewPublicListener(svc *connect.Service, cfg PublicListenerConfig,
|
||||||
time.Duration(cfg.LocalConnectTimeoutMs)*time.Millisecond)
|
time.Duration(cfg.LocalConnectTimeoutMs)*time.Millisecond)
|
||||||
},
|
},
|
||||||
stopChan: make(chan struct{}),
|
stopChan: make(chan struct{}),
|
||||||
|
listeningChan: make(chan struct{}),
|
||||||
logger: logger,
|
logger: logger,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -65,16 +75,26 @@ func NewUpstreamListener(svc *connect.Service, cfg UpstreamConfig,
|
||||||
return svc.Dial(ctx, cfg.resolver)
|
return svc.Dial(ctx, cfg.resolver)
|
||||||
},
|
},
|
||||||
stopChan: make(chan struct{}),
|
stopChan: make(chan struct{}),
|
||||||
|
listeningChan: make(chan struct{}),
|
||||||
logger: logger,
|
logger: logger,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Serve runs the listener until it is stopped.
|
// Serve runs the listener until it is stopped. It is an error to call Serve
|
||||||
|
// more than once for any given Listener instance.
|
||||||
func (l *Listener) Serve() error {
|
func (l *Listener) Serve() error {
|
||||||
|
// Ensure we mark state closed if we fail before Close is called externally.
|
||||||
|
defer l.Close()
|
||||||
|
|
||||||
|
if atomic.LoadInt32(&l.stopFlag) != 0 {
|
||||||
|
return errors.New("serve called on a closed listener")
|
||||||
|
}
|
||||||
|
|
||||||
listen, err := l.listenFunc()
|
listen, err := l.listenFunc()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
close(l.listeningChan)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
conn, err := listen.Accept()
|
conn, err := listen.Accept()
|
||||||
|
@ -113,3 +133,8 @@ func (l *Listener) handleConn(src net.Conn) {
|
||||||
func (l *Listener) Close() error {
|
func (l *Listener) Close() error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Wait for the listener to be ready to accept connections.
|
||||||
|
func (l *Listener) Wait() {
|
||||||
|
<-l.listeningChan
|
||||||
|
}
|
||||||
|
|
|
@ -24,7 +24,7 @@ func TestPublicListener(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
testApp, err := NewTestTCPServer(t, cfg.LocalServiceAddress)
|
testApp, err := NewTestTCPServer(t, cfg.LocalServiceAddress)
|
||||||
require.Nil(t, err)
|
require.NoError(t, err)
|
||||||
defer testApp.Close()
|
defer testApp.Close()
|
||||||
|
|
||||||
svc := connect.TestService(t, "db", ca)
|
svc := connect.TestService(t, "db", ca)
|
||||||
|
@ -34,9 +34,10 @@ func TestPublicListener(t *testing.T) {
|
||||||
// Run proxy
|
// Run proxy
|
||||||
go func() {
|
go func() {
|
||||||
err := l.Serve()
|
err := l.Serve()
|
||||||
require.Nil(t, err)
|
require.NoError(t, err)
|
||||||
}()
|
}()
|
||||||
defer l.Close()
|
defer l.Close()
|
||||||
|
l.Wait()
|
||||||
|
|
||||||
// Proxy and backend are running, play the part of a TLS client using same
|
// Proxy and backend are running, play the part of a TLS client using same
|
||||||
// cert for now.
|
// cert for now.
|
||||||
|
@ -44,7 +45,7 @@ func TestPublicListener(t *testing.T) {
|
||||||
Addr: addrs[0],
|
Addr: addrs[0],
|
||||||
CertURI: agConnect.TestSpiffeIDService(t, "db"),
|
CertURI: agConnect.TestSpiffeIDService(t, "db"),
|
||||||
})
|
})
|
||||||
require.Nilf(t, err, "unexpected err: %s", err)
|
require.NoError(t, err)
|
||||||
TestEchoConn(t, conn, "")
|
TestEchoConn(t, conn, "")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -56,9 +57,10 @@ func TestUpstreamListener(t *testing.T) {
|
||||||
testSvr := connect.NewTestServer(t, "db", ca)
|
testSvr := connect.NewTestServer(t, "db", ca)
|
||||||
go func() {
|
go func() {
|
||||||
err := testSvr.Serve()
|
err := testSvr.Serve()
|
||||||
require.Nil(t, err)
|
require.NoError(t, err)
|
||||||
}()
|
}()
|
||||||
defer testSvr.Close()
|
defer testSvr.Close()
|
||||||
|
<-testSvr.Listening
|
||||||
|
|
||||||
cfg := UpstreamConfig{
|
cfg := UpstreamConfig{
|
||||||
DestinationType: "service",
|
DestinationType: "service",
|
||||||
|
@ -79,13 +81,14 @@ func TestUpstreamListener(t *testing.T) {
|
||||||
// Run proxy
|
// Run proxy
|
||||||
go func() {
|
go func() {
|
||||||
err := l.Serve()
|
err := l.Serve()
|
||||||
require.Nil(t, err)
|
require.NoError(t, err)
|
||||||
}()
|
}()
|
||||||
defer l.Close()
|
defer l.Close()
|
||||||
|
l.Wait()
|
||||||
|
|
||||||
// Proxy and fake remote service are running, play the part of the app
|
// Proxy and fake remote service are running, play the part of the app
|
||||||
// connecting to a remote connect service over TCP.
|
// connecting to a remote connect service over TCP.
|
||||||
conn, err := net.Dial("tcp", cfg.LocalBindAddress)
|
conn, err := net.Dial("tcp", cfg.LocalBindAddress)
|
||||||
require.Nilf(t, err, "unexpected err: %s", err)
|
require.NoError(t, err)
|
||||||
TestEchoConn(t, conn, "")
|
TestEchoConn(t, conn, "")
|
||||||
}
|
}
|
||||||
|
|
|
@ -73,9 +73,10 @@ func TestService_Dial(t *testing.T) {
|
||||||
if tt.accept {
|
if tt.accept {
|
||||||
go func() {
|
go func() {
|
||||||
err := testSvr.Serve()
|
err := testSvr.Serve()
|
||||||
require.Nil(err)
|
require.NoError(err)
|
||||||
}()
|
}()
|
||||||
defer testSvr.Close()
|
defer testSvr.Close()
|
||||||
|
<-testSvr.Listening
|
||||||
}
|
}
|
||||||
|
|
||||||
// Always expect to be connecting to a "DB"
|
// Always expect to be connecting to a "DB"
|
||||||
|
@ -95,10 +96,10 @@ func TestService_Dial(t *testing.T) {
|
||||||
testTimer.Stop()
|
testTimer.Stop()
|
||||||
|
|
||||||
if tt.wantErr == "" {
|
if tt.wantErr == "" {
|
||||||
require.Nil(err)
|
require.NoError(err)
|
||||||
require.IsType(&tls.Conn{}, conn)
|
require.IsType(&tls.Conn{}, conn)
|
||||||
} else {
|
} else {
|
||||||
require.NotNil(err)
|
require.Error(err)
|
||||||
require.Contains(err.Error(), tt.wantErr)
|
require.Contains(err.Error(), tt.wantErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -117,7 +118,6 @@ func TestService_ServerTLSConfig(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestService_HTTPClient(t *testing.T) {
|
func TestService_HTTPClient(t *testing.T) {
|
||||||
require := require.New(t)
|
|
||||||
ca := connect.TestCA(t, nil)
|
ca := connect.TestCA(t, nil)
|
||||||
|
|
||||||
s := TestService(t, "web", ca)
|
s := TestService(t, "web", ca)
|
||||||
|
@ -129,8 +129,9 @@ func TestService_HTTPClient(t *testing.T) {
|
||||||
err := testSvr.ServeHTTPS(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
err := testSvr.ServeHTTPS(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
w.Write([]byte("Hello, I am Backend"))
|
w.Write([]byte("Hello, I am Backend"))
|
||||||
}))
|
}))
|
||||||
require.Nil(t, err)
|
require.NoError(t, err)
|
||||||
}()
|
}()
|
||||||
|
<-testSvr.Listening
|
||||||
|
|
||||||
// TODO(banks): this will talk http2 on both client and server. I hit some
|
// TODO(banks): this will talk http2 on both client and server. I hit some
|
||||||
// compatibility issues when testing though need to make sure that the http
|
// compatibility issues when testing though need to make sure that the http
|
||||||
|
|
|
@ -105,6 +105,8 @@ type TestServer struct {
|
||||||
// Addr is the listen address. It is set to a random free port on `localhost`
|
// Addr is the listen address. It is set to a random free port on `localhost`
|
||||||
// by default.
|
// by default.
|
||||||
Addr string
|
Addr string
|
||||||
|
// Listening is closed when the listener is run.
|
||||||
|
Listening chan struct{}
|
||||||
|
|
||||||
l net.Listener
|
l net.Listener
|
||||||
stopFlag int32
|
stopFlag int32
|
||||||
|
@ -120,7 +122,8 @@ func NewTestServer(t testing.T, service string, ca *structs.CARoot) *TestServer
|
||||||
CA: ca,
|
CA: ca,
|
||||||
stopChan: make(chan struct{}),
|
stopChan: make(chan struct{}),
|
||||||
TLSCfg: TestTLSConfig(t, service, ca),
|
TLSCfg: TestTLSConfig(t, service, ca),
|
||||||
Addr: fmt.Sprintf("localhost:%d", ports[0]),
|
Addr: fmt.Sprintf("127.0.0.1:%d", ports[0]),
|
||||||
|
Listening: make(chan struct{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -132,6 +135,7 @@ func (s *TestServer) Serve() error {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
close(s.Listening)
|
||||||
s.l = l
|
s.l = l
|
||||||
log.Printf("test connect service listening on %s", s.Addr)
|
log.Printf("test connect service listening on %s", s.Addr)
|
||||||
|
|
||||||
|
@ -172,7 +176,21 @@ func (s *TestServer) ServeHTTPS(h http.Handler) error {
|
||||||
Handler: h,
|
Handler: h,
|
||||||
}
|
}
|
||||||
log.Printf("starting test connect HTTPS server on %s", s.Addr)
|
log.Printf("starting test connect HTTPS server on %s", s.Addr)
|
||||||
return srv.ListenAndServeTLS("", "")
|
|
||||||
|
// Use our own listener so we can signal when it's ready.
|
||||||
|
l, err := net.Listen("tcp", s.Addr)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
close(s.Listening)
|
||||||
|
s.l = l
|
||||||
|
log.Printf("test connect service listening on %s", s.Addr)
|
||||||
|
|
||||||
|
err = srv.ServeTLS(l, "", "")
|
||||||
|
if atomic.LoadInt32(&s.stopFlag) == 1 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Close stops a TestServer
|
// Close stops a TestServer
|
||||||
|
|
Loading…
Reference in New Issue