Add --trace option to print full trace of HTTP requests and respons…

…es for failed test cases (#761)

This is a bit of a bear -- most of the complexity is in the minimal
amount of parsing that the middleware must do on the wire format, to
report relevant information about it in the trace and then to also
capture the end-of-stream message in the Connect streaming and gRPC-web
protocols.

So most of the code is in the new `tracer` internal package, and that's
where all of the gnarly bits are, too. The integrations into the
command-line and into the reference client and server are pretty
straight-forward.

This does not instrument the grpc-go reference client and server
implementations. We could add that later. It's relatively
straight-forward with the server, since it _can_ be configured to use
`net/http` (though it is not currently configured this way). But the
grpc-go client cannot be configured that way. So to collect traces, we
must point the grpc-go client at a local (in-process) HTTP proxy, that
can record the traces and then forward the requests/responses to the
real backend. This seems like enough of a headache that we can punt on
it and maybe revisit if people really need traces against the grpc-go
implementations.

cmd/connectconformance/main.go

@@ -47,6 +47,7 @@ const (
 	tlsKeyFlagName        = "key"
 	portFlagName          = "port"
 	bindFlagName          = "bind"
+	traceFlagName         = "trace"
 )
 
 type flags struct {
@@ -65,6 +66,7 @@ type flags struct {
 	tlsKeyFile           string
 	port                 uint
 	bind                 string
+	trace                bool
 }
 
 func main() {
@@ -157,6 +159,8 @@ func bind(cmd *cobra.Command, flags *flags) {
 		"in client mode, the port number on which the reference server should listen (implies --max-servers=1)")
 	cmd.Flags().StringVar(&flags.bind, bindFlagName, internal.DefaultHost,
 		"in client mode, the bind address on which the reference server should listen (0.0.0.0 means listen on all interfaces)")
+	cmd.Flags().BoolVar(&flags.trace, traceFlagName, false,
+		"if true, full HTTP traces will be captured and shown alongside failing test cases")
 }
 
 func run(flags *flags, cobraFlags *pflag.FlagSet, command []string) { //nolint:gocyclo
@@ -298,6 +302,7 @@ func run(flags *flags, cobraFlags *pflag.FlagSet, command []string) { //nolint:g
 			TLSKeyFile:           flags.tlsKeyFile,
 			ServerPort:           flags.port,
 			ServerBind:           flags.bind,
+			HTTPTrace:            flags.trace,
 		},
 		internal.NewPrinter(os.Stdout),
 		internal.NewPrinter(os.Stderr),

cmd/referenceclient/main.go

@@ -23,7 +23,7 @@ import (
 )
 
 func main() {
-	err := referenceclient.Run(context.Background(), os.Args, os.Stdin, os.Stdout, os.Stderr)
+	err := referenceclient.Run(context.Background(), os.Args, os.Stdin, os.Stdout, os.Stderr, nil)
 	if err != nil {
 		log.Fatalf("an error occurred running the reference client: %s", err.Error())
 	}

internal/app/connectconformance/connectconformance.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"sort"
@@ -32,6 +33,7 @@ import (
 	"connectrpc.com/conformance/internal/app/referenceclient"
 	"connectrpc.com/conformance/internal/app/referenceserver"
 	conformancev1 "connectrpc.com/conformance/internal/gen/proto/go/connectrpc/conformance/v1"
+	"connectrpc.com/conformance/internal/tracer"
 	"golang.org/x/sync/semaphore"
 	"google.golang.org/protobuf/proto"
 )
@@ -54,6 +56,7 @@ type Flags struct {
 	TLSKeyFile           string
 	ServerPort           uint
 	ServerBind           string
+	HTTPTrace            bool
 }
 
 func Run(flags *Flags, logPrinter internal.Printer, errPrinter internal.Printer) (bool, error) {
@@ -221,6 +224,11 @@ func run( //nolint:gocyclo
 		}
 	}
 
+	var trace *tracer.Tracer
+	if flags.HTTPTrace {
+		trace = &tracer.Tracer{}
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -232,7 +240,9 @@ func run( //nolint:gocyclo
 				start: runInProcess([]string{
 					"reference-client",
 					"-p", strconv.Itoa(int(flags.Parallelism)),
-				}, referenceclient.Run),
+				}, func(ctx context.Context, args []string, inReader io.ReadCloser, outWriter, errWriter io.WriteCloser) error {
+					return referenceclient.Run(ctx, args, inReader, outWriter, errWriter, trace)
+				}),
 				isReferenceImpl: true,
 			},
 			{
@@ -252,7 +262,7 @@ func run( //nolint:gocyclo
 		}
 	}
 
-	results := newResults(knownFailing, knownFlaky)
+	results := newResults(knownFailing, knownFlaky, trace)
 
 	for _, clientInfo := range clients {
 		clientProcess, err := runClient(ctx, clientInfo.start)
@@ -272,7 +282,9 @@ func run( //nolint:gocyclo
 						"-bind", flags.ServerBind,
 						"-cert", flags.TLSCertFile,
 						"-key", flags.TLSKeyFile,
-					}, referenceserver.RunInReferenceMode),
+					}, func(ctx context.Context, args []string, inReader io.ReadCloser, outWriter, errWriter io.WriteCloser) error {
+						return referenceserver.RunInReferenceMode(ctx, args, inReader, outWriter, errWriter, trace)
+					}),
 					isReferenceImpl: true,
 				},
 				{
@@ -349,6 +361,7 @@ func run( //nolint:gocyclo
 							errPrinter,
 							results,
 							clientProcess,
+							trace,
 						)
 					}(ctx, clientInfo, serverInfo, svrInstance)
 				}

internal/app/connectconformance/results.go

@@ -16,16 +16,19 @@ package connectconformance
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"reflect"
 	"sort"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"connectrpc.com/conformance/internal"
 	conformancev1 "connectrpc.com/conformance/internal/gen/proto/go/connectrpc/conformance/v1"
+	"connectrpc.com/conformance/internal/tracer"
 	"connectrpc.com/connect"
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/protobuf/proto"
@@ -42,16 +45,21 @@ const timeoutCheckGracePeriodMillis = 500
 type testResults struct {
 	knownFailing *testTrie
 	knownFlaky   *testTrie
+	tracer       *tracer.Tracer
+
+	traceWaitGroup sync.WaitGroup
 
 	mu             sync.Mutex
 	outcomes       map[string]testOutcome
+	traces         map[string]*tracer.Trace
 	serverSideband map[string]string
 }
 
-func newResults(knownFailing, knownFlaky *testTrie) *testResults {
+func newResults(knownFailing, knownFlaky *testTrie, tracer *tracer.Tracer) *testResults {
 	return &testResults{
 		knownFailing:   knownFailing,
 		knownFlaky:     knownFlaky,
+		tracer:         tracer,
 		outcomes:       map[string]testOutcome{},
 		serverSideband: map[string]string{},
 	}
@@ -74,6 +82,36 @@ func (r *testResults) setOutcomeLocked(testCase string, setupError bool, err err
 		knownFailing:  r.knownFailing.match(strings.Split(testCase, "/")),
 		knownFlaky:    r.knownFlaky.match(strings.Split(testCase, "/")),
 	}
+	r.fetchTrace(testCase)
+}
+
+//nolint:contextcheck,nolintlint // intentionally using context.Background; nolintlint incorrectly complains about this
+func (r *testResults) fetchTrace(testCase string) {
+	if r.tracer == nil {
+		return
+	}
+	r.traceWaitGroup.Add(1)
+	go func() {
+		defer r.traceWaitGroup.Done()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		trace, err := r.tracer.Await(ctx, testCase)
+		r.tracer.Clear(testCase)
+		if err != nil {
+			return
+		}
+
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		outcome := r.outcomes[testCase]
+		if outcome.actualFailure == nil || outcome.setupError || outcome.knownFlaky || outcome.knownFailing {
+			return
+		}
+		if r.traces == nil {
+			r.traces = map[string]*tracer.Trace{}
+		}
+		r.traces[testCase] = trace
+	}()
 }
 
 // failedToStart marks all the given test cases with the given setup error.
@@ -186,6 +224,7 @@ func (r *testResults) processSidebandInfoLocked() {
 }
 
 func (r *testResults) report(printer internal.Printer) bool {
+	r.traceWaitGroup.Wait() // make sure all traces have been received
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	if len(r.serverSideband) > 0 {
@@ -208,6 +247,12 @@ func (r *testResults) report(printer internal.Printer) bool {
 		switch {
 		case !expectError && outcome.actualFailure != nil:
 			printer.Printf("FAILED: %s:\n%s", name, indent(outcome.actualFailure.Error()))
+			trace := r.traces[name]
+			if trace != nil {
+				printer.Printf("---- HTTP Trace ----")
+				trace.Print(printer)
+				printer.Printf("--------------------")
+			}
 			failed++
 		case expectError && outcome.actualFailure == nil:
 			printer.Printf("FAILED: %s was expected to fail but did not", name)

internal/app/connectconformance/results_test.go

@@ -30,7 +30,7 @@ import (
 
 func TestResults_SetOutcome(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", false, nil)
 	results.setOutcome("foo/bar/2", true, errors.New("fail"))
 	results.setOutcome("foo/bar/3", false, errors.New("fail"))
@@ -58,7 +58,7 @@ func TestResults_SetOutcome(t *testing.T) {
 
 func TestResults_FailedToStart(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.failedToStart([]*conformancev1.TestCase{
 		{Request: &conformancev1.ClientCompatRequest{TestName: "foo/bar/1"}},
 		{Request: &conformancev1.ClientCompatRequest{TestName: "known-to-fail/1"}},
@@ -76,7 +76,7 @@ func TestResults_FailedToStart(t *testing.T) {
 
 func TestResults_FailRemaining(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", false, nil)
 	results.setOutcome("known-to-fail/1", false, errors.New("fail"))
 	results.failRemaining([]*conformancev1.TestCase{
@@ -101,7 +101,7 @@ func TestResults_FailRemaining(t *testing.T) {
 
 func TestResults_Failed(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.failed("foo/bar/1", &conformancev1.ClientErrorResult{Message: "fail"})
 	results.failed("known-to-fail/1", &conformancev1.ClientErrorResult{Message: "fail"})
 
@@ -116,7 +116,7 @@ func TestResults_Failed(t *testing.T) {
 
 func TestResults_Assert(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	payload1 := &conformancev1.ClientResponseResult{
 		Payloads: []*conformancev1.ConformancePayload{
 			{Data: []byte{0, 1, 2, 3, 4}},
@@ -688,7 +688,7 @@ func TestResults_Assert_ReportsAllErrors(t *testing.T) {
 		testCase := testCase
 		t.Run(testCase.name, func(t *testing.T) {
 			t.Parallel()
-			results := newResults(&testTrie{}, &testTrie{})
+			results := newResults(&testTrie{}, &testTrie{}, nil)
 
 			expected := &conformancev1.ClientResponseResult{}
 			err := protojson.Unmarshal(([]byte)(testCase.expected), expected)
@@ -722,7 +722,7 @@ func TestResults_Assert_ReportsAllErrors(t *testing.T) {
 
 func TestResults_ServerSideband(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", false, nil)
 	results.setOutcome("foo/bar/2", false, errors.New("fail"))
 	results.setOutcome("foo/bar/3", false, nil)
@@ -745,50 +745,50 @@ func TestResults_ServerSideband(t *testing.T) {
 
 func TestResults_Report(t *testing.T) {
 	t.Parallel()
-	results := newResults(makeKnownFailing(), makeKnownFlaky())
+	results := newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	logger := &linePrinter{}
 
 	// No test cases? Report success.
 	success := results.report(logger)
 	require.True(t, success)
 
 	// Only successful outcomes? Report success.
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", false, nil)
 	success = results.report(logger)
 	require.True(t, success)
 
 	// Unexpected failure? Report failure.
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", false, errors.New("ruh roh"))
 	success = results.report(logger)
 	require.False(t, success)
 
 	// Unexpected failure during setup? Report failure.
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("foo/bar/1", true, errors.New("ruh roh"))
 	success = results.report(logger)
 	require.False(t, success)
 
 	// Expected failure? Report success.
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("known-to-fail/1", false, errors.New("ruh roh"))
 	success = results.report(logger)
 	require.True(t, success)
 
 	// Setup error from expected failure? Report failure (setup errors never acceptable).
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("known-to-fail/1", true, errors.New("ruh roh"))
 	success = results.report(logger)
 	require.False(t, success)
 
 	// Flaky? Report success whether it passes or fails
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("known-to-flake/1", false, nil) // succeeds
 	success = results.report(logger)
 	require.True(t, success)
 
-	results = newResults(makeKnownFailing(), makeKnownFlaky())
+	results = newResults(makeKnownFailing(), makeKnownFlaky(), nil)
 	results.setOutcome("known-to-flake/1", false, errors.New("ruh roh"))
 	success = results.report(logger)
 	require.True(t, success)

internal/app/connectconformance/server_runner.go

@@ -26,6 +26,7 @@ import (
 
 	"connectrpc.com/conformance/internal"
 	conformancev1 "connectrpc.com/conformance/internal/gen/proto/go/connectrpc/conformance/v1"
+	"connectrpc.com/conformance/internal/tracer"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -51,6 +52,7 @@ func runTestCasesForServer(
 	errPrinter internal.Printer,
 	results *testResults,
 	client clientRunner,
+	tracer *tracer.Tracer,
 ) {
 	expectations := make(map[string]*conformancev1.ClientResponseResult, len(testCases))
 	for _, testCase := range testCases {
@@ -181,6 +183,7 @@ func runTestCasesForServer(
 			}
 		}
 
+		tracer.Init(req.TestName)
 		wg.Add(1)
 		err := client.sendRequest(req, func(name string, resp *conformancev1.ClientCompatResponse, err error) {
 			defer wg.Done()

internal/app/connectconformance/server_runner_test.go

@@ -194,7 +194,7 @@ func TestRunTestCasesForServer(t *testing.T) {
 		testCase := testCase
 		t.Run(testCase.name, func(t *testing.T) {
 			t.Parallel()
-			results := newResults(&testTrie{}, &testTrie{})
+			results := newResults(&testTrie{}, &testTrie{}, nil)
 
 			var procAddr atomic.Pointer[process] // populated when server process created
 			var actualSvrRequest bytes.Buffer
@@ -271,6 +271,7 @@ func TestRunTestCasesForServer(t *testing.T) {
 				discardPrinter{},
 				results,
 				&client,
+				nil,
 			)
 
 			if testCase.svrFailsToStart {