Export traces for eventshub in failed tests

README.md

@@ -396,6 +396,62 @@ for running later via `env.Test` or `env.TestSet`. With state, we are attempting
 to make it less difficult to communicate between `Setup` and `Assert` phases of
 testing.
 
+### Inspecting Zipkin traces for failed tests
+
+When the [eventshub](./pkg/eventshub) component is used for sending events then Zipkin traces
+can be collected on test exit. Traces for each test namespace are stored in a separate
+file under `$ARTIFACTS/traces/<namespace>.json` and will be collected only for
+failed tests.
+
+To enable collecting traces from Zipkin, set up the test environment as follows:
+```go
+
+import (
+    "knative.dev/reconciler-test/pkg/environment"
+    "knative.dev/reconciler-test/pkg/knative"
+    "knative.dev/reconciler-test/pkg/tracing"
+)
+
+ctx, env := global.Environment(
+    // Will call env.Finish() when the test exits.
+    environment.Managed(t),
+    // Set the knative namespace which holds the tracing config map.
+    knative.WithKnativeNamespace(system.Namespace()),
+    // Configure tracing for the eventshub component.
+    knative.WithTracingConfig,
+    // Configure logging for the eventshub component.
+    knative.WithLoggingConfig,
+    // Register the tracing listener which will gather event traces on env.Finish().
+    tracing.WithGatherer(t),
+)
+```
+
+The TestMain function should include the cleanup code for Zipkin:
+```go
+func TestMain(m *testing.M) {
+    os.Exit(func() int {
+        // Any tests may set up Zipkin tracing via tracing.WithGatherer, it will only actually be done once.
+        // This should be the ONLY place that cleans it up. If an individual test calls this instead, then
+        // it will break other tests that need the tracing in place.
+        defer tracing.Cleanup()
+        return m.Run()
+    }())
+}
+```
+
+Traces can be viewed as follows:
+- Start a Zipkin container on localhost:
+   ```
+   $ docker run -d -p 9411:9411 ghcr.io/openzipkin/zipkin:2
+   ```
+- Send traces to the Zipkin endpoint:
+   ```
+   $ curl -v localhost:9411/api/v2/spans \
+     -H 'Content-Type: application/json' \
+     -d @$ARTIFACTS/traces/<namespace>.json
+   ```
+- View traces in Zipkin UI at `http://localhost:9411/zipkin`
+
 ### Running Tests
 
 Running tests is nothing more than using `go test`.

cmd/eventshub/main.go

@@ -40,10 +40,10 @@ func main() {
 		},
 		map[string]eventshub.EventGeneratorStarter{
 			eventshub.ReceiverEventGenerator: func(ctx context.Context, logs *eventshub.EventLogs) error {
-				return receiver.NewFromEnv(ctx, logs).Start(ctx, eventshub.WithTracing)
+				return receiver.NewFromEnv(ctx, logs).Start(ctx, eventshub.WithServerTracing)
 			},
 			eventshub.SenderEventGenerator: func(ctx context.Context, logs *eventshub.EventLogs) error {
-				return sender.Start(ctx, logs)
+				return sender.Start(ctx, logs, eventshub.WithClientTracing)
 			},
 		},
 	)

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/kelseyhightower/envconfig v1.4.0
+	github.com/openzipkin/zipkin-go v0.3.0
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.7.0
 	go.opencensus.io v0.23.0
@@ -25,6 +26,7 @@ require (
 
 require (
 	cloud.google.com/go v0.98.0 // indirect
+	cloud.google.com/go/storage v1.18.2 // indirect
 	contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d // indirect
 	contrib.go.opencensus.io/exporter/prometheus v0.4.0 // indirect
 	contrib.go.opencensus.io/exporter/zipkin v0.1.2 // indirect
@@ -34,8 +36,12 @@ require (
 	github.com/blendle/zapdriver v1.3.1 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 // indirect
+	github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful v2.9.5+incompatible // indirect
+	github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021 // indirect
+	github.com/envoyproxy/protoc-gen-validate v0.1.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.6.0 // indirect
 	github.com/go-kit/log v0.1.0 // indirect
 	github.com/go-logfmt/logfmt v0.5.0 // indirect
@@ -47,6 +53,7 @@ require (
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/googleapis/gax-go/v2 v2.1.1 // indirect
 	github.com/googleapis/gnostic v0.5.5 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
 	github.com/imdario/mergo v0.3.9 // indirect
@@ -56,7 +63,6 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/openzipkin/zipkin-go v0.3.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.11.1 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect

go.sum

@@ -45,6 +45,7 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
+cloud.google.com/go/storage v1.18.2 h1:5NQw6tOn3eMm0oE8vTkfjau18kjL79FlMjy/CHTpmoY=
 cloud.google.com/go/storage v1.18.2/go.mod h1:AiIj7BWXyhO5gGVmYJ+S8tbkCx3yb0IMjua8Aw4naVM=
 contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d h1:LblfooH1lKOpp1hIhukktmSAxFkqMPFk9KR6iZ0MJNI=
 contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d/go.mod h1:IshRmMJBhDfFj5Y67nVhMYTTIze91RUeT73ipWKs/GY=
@@ -124,10 +125,12 @@ github.com/cloudevents/sdk-go/v2 v2.4.1/go.mod h1:MZiMwmAh5tGj+fPFvtHv9hKurKqXtd
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
+github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 h1:hzAQntlaYRkVSFEfj9OTWlVV1H155FMD8BTKktLv0QI=
 github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 h1:zH8ljVhhq7yC0MIeUL/IviMtY8hx2mK8cN9wEYb8ggw=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cockroachdb/datadriven v0.0.0-20200714090401-bf6692d28da5/go.mod h1:h6jFvWxBdQXxjopDMZyH2UVceIRfR84bdzbkoKrsWNo=
 github.com/cockroachdb/errors v1.2.4/go.mod h1:rQD95gz6FARkaKkQXUksEje/d9a6wBJoCr5oaCLELYA=
@@ -173,7 +176,9 @@ github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5y
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
+github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021 h1:fP+fF0up6oPY49OrjPrhIJ8yQfdIM85NXMLkMg1EXVs=
 github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
+github.com/envoyproxy/protoc-gen-validate v0.1.0 h1:EQciDnbrYxy13PgWoY8AqoxGiPrpgBZ1R8UNe3ddc+A=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
 github.com/evanphx/json-patch v4.12.0+incompatible h1:4onqiflcdA9EOZ4RxV643DvftH5pOlLGNtQ5lPWQu84=
@@ -303,9 +308,11 @@ github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/mako v0.0.0-20190821191249-122f8dcef9e3/go.mod h1:YzLcVlL+NqWnmUEPuhS1LxDDwGO9WNbVlEXaF4IH35g=
+github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
 github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
+github.com/google/martian/v3 v3.2.1 h1:d8MncMlErDFTwQGBK1xhv026j9kqhvw1Qv9IbWT1VLQ=
 github.com/google/martian/v3 v3.2.1/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -329,6 +336,7 @@ github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gax-go/v2 v2.1.0/go.mod h1:Q3nei7sK6ybPYH7twZdmQpAd1MKb7pfu6SK+H1/DsU0=
+github.com/googleapis/gax-go/v2 v2.1.1 h1:dp3bWCh+PPO1zjRRiCSczJav13sBvG4UhNyVTa1KqdU=
 github.com/googleapis/gax-go/v2 v2.1.1/go.mod h1:hddJymUZASv3XPyGkUpKj8pPO47Rmb0eJc8R6ouapiM=
 github.com/googleapis/gnostic v0.5.1/go.mod h1:6U4PtQXGIEt/Z3h5MAT7FNofLnw9vXk2cUuW7uA/OeU=
 github.com/googleapis/gnostic v0.5.5 h1:9fHAtK0uDfpveeqqo1hkEZJcFvYXAiCN3UutL8F9xHw=

pkg/environment/magic.go

@@ -28,7 +28,6 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"knative.dev/pkg/logging"
-
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/milestone"
 	"knative.dev/reconciler-test/pkg/state"
@@ -136,6 +135,15 @@ func Cleanup(t feature.T) EnvOpts {
 	}
 }
 
+func WithEmitter(emitter milestone.Emitter) EnvOpts {
+	return func(ctx context.Context, env Environment) (context.Context, error) {
+		if e, ok := env.(*MagicEnvironment); ok {
+			e.milestones = emitter
+		}
+		return ctx, nil
+	}
+}
+
 func (mr *MagicGlobalEnvironment) Environment(opts ...EnvOpts) (context.Context, Environment) {
 	namespace := feature.MakeK8sNamePrefix(feature.AppendRandomString("test"))
 
@@ -168,16 +176,18 @@ func (mr *MagicGlobalEnvironment) Environment(opts ...EnvOpts) (context.Context,
 		}
 	})
 
-	// It is possible to have milestones set in the options, check for nil in
-	// env first before attempting to pull one from the os environment.
+	eventEmitter, err := milestone.NewMilestoneEmitterFromEnv(mr.instanceID, namespace)
+	if err != nil {
+		// This is just an FYI error, don't block the test run.
+		logging.FromContext(ctx).Error("failed to create the milestone event sender", zap.Error(err))
+	}
+	logEmitter := milestone.NewLogEmitter(ctx, namespace)
+
 	if env.milestones == nil {
-		eventEmitter, err := milestone.NewMilestoneEmitterFromEnv(mr.instanceID, namespace)
-		if err != nil {
-			// This is just an FYI error, don't block the test run.
-			logging.FromContext(ctx).Error("failed to create the milestone event sender", zap.Error(err))
-		}
-		logEmitter := milestone.NewLogEmitter(ctx, namespace)
 		env.milestones = milestone.Compose(eventEmitter, logEmitter)
+	} else {
+		// Compose the emitters with those passed through opts.
+		env.milestones = milestone.Compose(env.milestones, eventEmitter, logEmitter)
 	}
 
 	if err := env.CreateNamespaceIfNeeded(); err != nil {

pkg/eventshub/options.go

@@ -206,6 +206,7 @@ func InputMethod(method string) EventsHubOption {
 }
 
 // AddTracing adds tracing headers when sending events.
+// Deprecated: Exporting traces from the client/sender is enabled by default.
 var AddTracing = envOption("ADD_TRACING", "true")
 
 // AddSequence adds an extension named 'sequence' which contains the incremental number of sent events

pkg/eventshub/sender/sender.go

@@ -35,11 +35,9 @@ import (
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"github.com/cloudevents/sdk-go/v2/types"
 	"github.com/kelseyhightower/envconfig"
-	"go.opencensus.io/plugin/ochttp"
+	"go.opencensus.io/trace"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/logging"
-	"knative.dev/pkg/tracing/propagation/tracecontextb3"
-
 	"knative.dev/reconciler-test/pkg/eventshub"
 )
 
@@ -77,9 +75,6 @@ type generator struct {
 	// InputMethod to use when sending the http request
 	InputMethod string `envconfig:"INPUT_METHOD" default:"POST" required:"false"`
 
-	// Should tracing be added to events sent.
-	AddTracing bool `envconfig:"ADD_TRACING" default:"false" required:"false"`
-
 	// Should add extension 'sequence' identifying the sequence number.
 	AddSequence bool `envconfig:"ADD_SEQUENCE" default:"false" required:"false"`
 
@@ -95,6 +90,9 @@ type generator struct {
 	// The number of messages to attempt to send. -1 for inferred, 0 for unlimited.
 	MaxMessages int `envconfig:"MAX_MESSAGES" default:"-1" required:"false"`
 
+	// The current namespace.
+	SystemNamespace string `envconfig:"SYSTEM_NAMESPACE" required:"true"`
+
 	// --- Processed State ---
 
 	// baseEvent is parsed from InputEvent.
@@ -107,7 +105,9 @@ type generator struct {
 	eventQueue []conformanceevent.Event
 }
 
-func Start(ctx context.Context, logs *eventshub.EventLogs) error {
+type Option func(*nethttp.Client) error
+
+func Start(ctx context.Context, logs *eventshub.EventLogs, clientOpts ...Option) error {
 	var env generator
 	if err := envconfig.Process("", &env); err != nil {
 		return fmt.Errorf("failed to process env var. %w", err)
@@ -153,10 +153,10 @@ func Start(ctx context.Context, logs *eventshub.EventLogs) error {
 	}
 
 	httpClient := &nethttp.Client{}
-	if env.AddTracing {
-		httpClient.Transport = &ochttp.Transport{
-			Base:        nethttp.DefaultTransport,
-			Propagation: tracecontextb3.TraceContextEgress,
+
+	for _, opt := range clientOpts {
+		if err := opt(httpClient); err != nil {
+			return fmt.Errorf("unable to apply option: %w", err)
 		}
 	}
 
@@ -172,12 +172,22 @@ func Start(ctx context.Context, logs *eventshub.EventLogs) error {
 	ticker := time.NewTicker(period)
 	for {
 
+		ctx, span := trace.StartSpan(ctx, "eventshub-sender")
+
 		req, event, err := env.next(ctx)
 		if err != nil {
 			return err
 		}
 
+		span.AddAttributes(
+			trace.StringAttribute("namespace", env.SystemNamespace),
+			trace.StringAttribute("event", event.String()),
+		)
+
 		res, err := httpClient.Do(req)
+
+		span.End()
+
 		// Publish sent event info
 		if err := logs.Vent(env.sentInfo(event, req, err)); err != nil {
 			return fmt.Errorf("cannot forward event info: %w", err)
@@ -373,6 +383,7 @@ func (g *generator) nextQueued(ctx context.Context) (*nethttp.Request, *cloudeve
 	if err != nil {
 		return nil, nil, err
 	}
+	req = req.WithContext(ctx)
 
 	event := eventToEvent(next)
 
@@ -406,7 +417,7 @@ func eventToEvent(conf conformanceevent.Event) *cloudevents.Event {
 }
 
 func (g *generator) nextGenerated(ctx context.Context) (*nethttp.Request, *cloudevents.Event, error) {
-	req, err := nethttp.NewRequest(g.InputMethod, g.Sink, nil)
+	req, err := nethttp.NewRequestWithContext(ctx, g.InputMethod, g.Sink, nil)
 	if err != nil {
 		logging.FromContext(ctx).Error("Cannot create the request: ", err)
 		return nil, nil, err

pkg/eventshub/utils.go

@@ -86,10 +86,19 @@ func ConfigureLogging(ctx context.Context, name string) context.Context {
 	return logging.WithLogger(ctx, l)
 }
 
-// WithTracing wraps the provided handler in a tracing handler
-func WithTracing(handler http.Handler) http.Handler {
+// WithServerTracing wraps the provided handler in a tracing handler
+func WithServerTracing(handler http.Handler) http.Handler {
 	return &ochttp.Handler{
 		Propagation: tracecontextb3.TraceContextEgress,
 		Handler:     handler,
 	}
 }
+
+// WithClientTracing enables exporting traces by the client's transport.
+func WithClientTracing(client *http.Client) error {
+	client.Transport = &ochttp.Transport{
+		Base:        http.DefaultTransport,
+		Propagation: tracecontextb3.TraceContextEgress,
+	}
+	return nil
+}