internal: test reliability improvements (projectcontour#2780)

* Switch from shutting down gRPC server with `Stop()` to `GracefulStop()`.
  which waits for current requests to drain before returning. This
  means that tests which log to the test runner won't ever attempt to
  log after the test has completed.

* Hoist test loggers into the fixtures package and use them from there.

* Add a workgroup context cancellation test.

This updates projectcontour#2775.

Signed-off-by: James Peach <jpeach@vmware.com>

cmd/contour/serve.go

@@ -434,7 +434,7 @@ func doServe(log logrus.FieldLogger, ctx *serveContext) error {
 
 		go func() {
 			<-stop
-			s.Stop()
+			s.GracefulStop()
 		}()
 
 		return s.Serve(l)

cmd/contour/servecontext_test.go

@@ -352,7 +352,7 @@ func TestServeContextCertificateHandling(t *testing.T) {
 		err := g.Serve(l)
 		checkFatalErr(t, err)
 	}()
-	defer g.Stop()
+	defer g.GracefulStop()
 
 	for name, tc := range tests {
 		t.Run(name, func(t *testing.T) {

internal/contour/server_test.go

@@ -216,7 +216,7 @@ func TestGRPC(t *testing.T) {
 				done <- srv.Serve(l)
 			}()
 			defer func() {
-				srv.Stop()
+				srv.GracefulStop()
 				close(stop)
 				<-done
 			}()

internal/e2e/e2e.go

@@ -29,12 +29,12 @@ import (
 	"github.com/projectcontour/contour/internal/assert"
 	"github.com/projectcontour/contour/internal/contour"
 	"github.com/projectcontour/contour/internal/dag"
+	"github.com/projectcontour/contour/internal/fixture"
 	cgrpc "github.com/projectcontour/contour/internal/grpc"
 	"github.com/projectcontour/contour/internal/k8s"
 	"github.com/projectcontour/contour/internal/protobuf"
 	"github.com/projectcontour/contour/internal/workgroup"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
@@ -47,27 +47,10 @@ const (
 	statsPort    = 8002
 )
 
-type testWriter struct {
-	*testing.T
-}
-
-func (t *testWriter) Write(buf []byte) (int, error) {
-	t.Logf("%s", buf)
-	return len(buf), nil
-}
-
-type discardWriter struct {
-}
-
-func (d *discardWriter) Write(buf []byte) (int, error) {
-	return len(buf), nil
-}
-
 func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *grpc.ClientConn, func()) {
 	t.Parallel()
 
-	log := logrus.New()
-	log.Out = &testWriter{t}
+	log := fixture.NewTestLogger(t)
 
 	et := &contour.EndpointsTranslator{
 		FieldLogger: log,
@@ -113,21 +96,18 @@ func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *grpc
 
 	l, err := net.Listen("tcp", "127.0.0.1:0")
 	check(t, err)
-	discard := logrus.New()
-	discard.Out = new(discardWriter)
 	// Resource types in xDS v2.
-	srv := cgrpc.NewAPI(discard, append(contour.ResourcesOf(resources), et), r)
+	srv := cgrpc.NewAPI(log, append(contour.ResourcesOf(resources), et), r)
 
 	var g workgroup.Group
 
 	g.Add(func(stop <-chan struct{}) error {
-		done := make(chan error)
 		go func() {
-			done <- srv.Serve(l) // srv now owns l and will close l before returning
+			<-stop
+			srv.GracefulStop()
 		}()
-		<-stop
-		srv.Stop()
-		return <-done
+
+		return srv.Serve(l) // srv now owns l and will close l before returning
 	})
 	g.Add(eh.Start())
 

internal/featuretests/featuretests.go

@@ -31,14 +31,14 @@ import (
 	"github.com/projectcontour/contour/internal/assert"
 	"github.com/projectcontour/contour/internal/contour"
 	"github.com/projectcontour/contour/internal/dag"
+	"github.com/projectcontour/contour/internal/fixture"
 	cgrpc "github.com/projectcontour/contour/internal/grpc"
 	"github.com/projectcontour/contour/internal/k8s"
 	"github.com/projectcontour/contour/internal/metrics"
 	"github.com/projectcontour/contour/internal/protobuf"
 	"github.com/projectcontour/contour/internal/sorter"
 	"github.com/projectcontour/contour/internal/workgroup"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
@@ -54,17 +54,10 @@ const (
 	statsPort    = 8002
 )
 
-type discardWriter struct{}
-
-func (d *discardWriter) Write(buf []byte) (int, error) {
-	return len(buf), nil
-}
-
 func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *Contour, func()) {
 	t.Parallel()
 
-	log := logrus.New()
-	log.Out = new(discardWriter)
+	log := fixture.NewTestLogger(t)
 
 	et := &contour.EndpointsTranslator{
 		FieldLogger: log,
@@ -118,21 +111,17 @@ func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *Cont
 
 	l, err := net.Listen("tcp", "127.0.0.1:0")
 	check(t, err)
-	discard := logrus.New()
-	discard.Out = new(discardWriter)
 	// Resource types in xDS v2.
-	srv := cgrpc.NewAPI(discard, append(contour.ResourcesOf(resources), et), r)
+	srv := cgrpc.NewAPI(log, append(contour.ResourcesOf(resources), et), r)
 
 	var g workgroup.Group
 
 	g.Add(func(stop <-chan struct{}) error {
-		done := make(chan error)
 		go func() {
-			done <- srv.Serve(l) // srv now owns l and will close l before returning
+			<-stop
+			srv.GracefulStop()
 		}()
-		<-stop
-		srv.Stop()
-		return <-done
+		return srv.Serve(l) // srv now owns l and will close l before returning
 	})
 	g.Add(eh.Start())
 

internal/fixture/log.go

@@ -0,0 +1,36 @@
+package fixture
+
+import (
+	"testing"
+
+	"github.com/sirupsen/logrus"
+)
+
+type testWriter struct {
+	*testing.T
+}
+
+func (t *testWriter) Write(buf []byte) (int, error) {
+	t.Logf("%s", buf)
+	return len(buf), nil
+}
+
+// NewTestLogger returns logrus.Logger that writes messages using (*testing.T)Logf.
+func NewTestLogger(t *testing.T) *logrus.Logger {
+	log := logrus.New()
+	log.Out = &testWriter{t}
+	return log
+}
+
+type discardWriter struct{}
+
+func (d *discardWriter) Write(buf []byte) (int, error) {
+	return len(buf), nil
+}
+
+// NewDiscardLogger returns logrus.Logger that discards log messages.
+func NewDiscardLogger() *logrus.Logger {
+	log := logrus.New()
+	log.Out = &discardWriter{}
+	return log
+}

internal/workgroup/group_test.go

@@ -17,7 +17,9 @@ import (
 	"context"
 	"errors"
 	"io"
+	"sync/atomic"
 	"testing"
+	"time"
 )
 
 func TestGroupRunWithNoRegisteredFunctions(t *testing.T) {
@@ -67,6 +69,35 @@ func TestGroupAddContext(t *testing.T) {
 	assert(t, io.EOF, <-result)
 }
 
+func TestGroupCancellation(t *testing.T) {
+	var g Group
+	ctx, cancel := context.WithCancel(context.Background())
+
+	const tasks = 100
+	var count int32
+
+	for i := 0; i < tasks; i++ {
+		g.Add(func(stop <-chan struct{}) error {
+			defer atomic.AddInt32(&count, 1)
+			defer time.Sleep(time.Millisecond * time.Duration(i))
+			<-stop
+			return nil
+		})
+	}
+
+	done := make(chan error)
+	go func() {
+		done <- g.Run(ctx)
+	}()
+
+	cancel()
+	<-done
+
+	if got := atomic.LoadInt32(&count); got != tasks {
+		t.Errorf("expected: %d, got: %d", tasks, got)
+	}
+}
+
 func assert(t *testing.T, want, got error) {
 	t.Helper()
 	if want != got {