E2E Test for Broker Tracing

cmd/broker/ingress/main.go

@@ -34,17 +34,17 @@ import (
 	"go.opencensus.io/stats/view"
 	"knative.dev/eventing/pkg/broker/ingress"
 	"knative.dev/eventing/pkg/channel"
+	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/tracing"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/system"
 	pkgtracing "knative.dev/pkg/tracing"
-
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/injection"
-	"knative.dev/pkg/injection/sharedmain"
 )
 
 var (
@@ -124,7 +124,7 @@ func main() {
 		writer.WriteHeader(http.StatusOK)
 	})
 
-	ceClient, err := cloudevents.NewClient(httpTransport, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	ceClient, err := kncloudevents.NewDefaultClientGivenHttpTransport(httpTransport)
 	if err != nil {
 		logger.Fatal("Unable to create CE client", zap.Error(err))
 	}

pkg/broker/filter/filter_handler.go

@@ -27,13 +27,14 @@ import (
 
 	cloudevents "github.com/cloudevents/sdk-go"
 	cehttp "github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http"
+	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing/pkg/broker"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1alpha1"
+	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/logging"
 	"knative.dev/eventing/pkg/reconciler/trigger/path"
-	"knative.dev/eventing/pkg/tracing"
 	"knative.dev/eventing/pkg/utils"
 	pkgtracing "knative.dev/pkg/tracing"
 )
@@ -69,7 +70,7 @@ func NewHandler(logger *zap.Logger, triggerLister eventinglisters.TriggerNamespa
 		return nil, err
 	}
 
-	ceClient, err := cloudevents.NewClient(httpTransport, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	ceClient, err := kncloudevents.NewDefaultClientGivenHttpTransport(httpTransport)
 	if err != nil {
 		return nil, err
 	}
@@ -252,10 +253,9 @@ func (r *Handler) sendEvent(ctx context.Context, tctx cloudevents.HTTPTransportC
 	}
 
 	sendingCTX := utils.ContextFrom(tctx, subscriberURI)
-	sendingCTX, err = tracing.AddSpanFromTraceparentAttribute(sendingCTX, "name", *event)
-	if err != nil {
-		r.logger.Info("Unable to attach trace", zap.Error(err))
-	}
+	// Due to an issue in utils.ContextFrom, we don't retain the original trace context from ctx, so
+	// bring it in manually.
+	sendingCTX = trace.NewContext(sendingCTX, trace.FromContext(ctx))
 
 	start := time.Now()
 	rctx, replyEvent, err := r.ceClient.Send(sendingCTX, *event)

pkg/broker/ingress/ingress_handler.go

@@ -9,9 +9,9 @@ import (
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go"
+	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	"knative.dev/eventing/pkg/broker"
-	"knative.dev/eventing/pkg/tracing"
 	"knative.dev/eventing/pkg/utils"
 )
 
@@ -73,8 +73,6 @@ func (h *Handler) serveHTTP(ctx context.Context, event cloudevents.Event, resp *
 		return nil
 	}
 
-	tracing.AddTraceparentAttributeFromContext(ctx, event)
-
 	reporterArgs := &ReportArgs{
 		ns:          h.Namespace,
 		broker:      h.BrokerName,
@@ -91,6 +89,10 @@ func (h *Handler) serveHTTP(ctx context.Context, event cloudevents.Event, resp *
 
 	start := time.Now()
 	sendingCTX := utils.ContextFrom(tctx, h.ChannelURI)
+	// Due to an issue in utils.ContextFrom, we don't retain the original trace context from ctx, so
+	// bring it in manually.
+	sendingCTX = trace.NewContext(sendingCTX, trace.FromContext(ctx))
+
 	rctx, _, err := h.CeClient.Send(sendingCTX, event)
 	rtctx := cloudevents.HTTPTransportContextFrom(rctx)
 	// Record the dispatch time.

pkg/kncloudevents/good_client.go

@@ -25,6 +25,12 @@ func NewDefaultClient(target ...string) (cloudevents.Client, error) {
 	if err != nil {
 		return nil, err
 	}
+	return NewDefaultClientGivenHttpTransport(t)
+}
+
+// NewDefaultClientGivenHttpTransport creates a new CloudEvents client using the provided HTTP
+// transport. Note that it does modify the provided HTTP Transport by adding tracing to its Client.
+func NewDefaultClientGivenHttpTransport(t *cloudevents.HTTPTransport) (cloudevents.Client, error) {
 	// Add output tracing.
 	t.Client = &gohttp.Client{
 		Transport: &ochttp.Transport{

pkg/tracing/names.go

@@ -27,7 +27,7 @@ type BrokerIngressNameArgs struct {
 // BrokerIngressName creates the service name for Broker Ingresses to use when writing Zipkin
 // traces.
 func BrokerIngressName(args BrokerIngressNameArgs) string {
-	return fmt.Sprintf("%s-broker-ingress.%s", args.BrokerName, args.Namespace)
+	return fmt.Sprintf("%s-broker.%s", args.BrokerName, args.Namespace)
 }
 
 // BrokerFilterNameArgs are the arguments needed to generate the BrokerFilterName.

pkg/tracing/names_test.go

@@ -25,7 +25,7 @@ func TestBrokerIngressName(t *testing.T) {
 		Namespace:  testNS,
 		BrokerName: broker,
 	}
-	if got, want := BrokerIngressName(args), "my-broker-broker-ingress.test-namespace"; got != want {
+	if got, want := BrokerIngressName(args), "my-broker-broker.test-namespace"; got != want {
 		t.Errorf("BrokerIngressName = %q, want %q", got, want)
 	}
 }

test/common/creation.go

@@ -23,6 +23,7 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	messagingv1alpha1 "knative.dev/eventing/pkg/apis/messaging/v1alpha1"
 	sourcesv1alpha1 "knative.dev/eventing/pkg/apis/sources/v1alpha1"
 	"knative.dev/eventing/test/base"
@@ -96,7 +97,7 @@ func (client *Client) CreateSubscriptionsOrFail(
 }
 
 // CreateBrokerOrFail will create a Broker or fail the test if there is an error.
-func (client *Client) CreateBrokerOrFail(name string, channelTypeMeta *metav1.TypeMeta) {
+func (client *Client) CreateBrokerOrFail(name string, channelTypeMeta *metav1.TypeMeta) *v1alpha1.Broker {
 	namespace := client.Namespace
 	broker := resources.Broker(name, resources.WithChannelTemplateForBroker(*channelTypeMeta))
 
@@ -107,6 +108,7 @@ func (client *Client) CreateBrokerOrFail(name string, channelTypeMeta *metav1.Ty
 		client.T.Fatalf("Failed to create broker %q: %v", name, err)
 	}
 	client.Tracker.AddObj(broker)
+	return broker
 }
 
 // CreateBrokersOrFail will create a list of Brokers.
@@ -117,7 +119,7 @@ func (client *Client) CreateBrokersOrFail(names []string, channelTypeMeta *metav
 }
 
 // CreateTriggerOrFail will create a Trigger or fail the test if there is an error.
-func (client *Client) CreateTriggerOrFail(name string, options ...resources.TriggerOption) {
+func (client *Client) CreateTriggerOrFail(name string, options ...resources.TriggerOption) *v1alpha1.Trigger {
 	namespace := client.Namespace
 	trigger := resources.Trigger(name, options...)
 
@@ -128,6 +130,7 @@ func (client *Client) CreateTriggerOrFail(name string, options ...resources.Trig
 		client.T.Fatalf("Failed to create trigger %q: %v", name, err)
 	}
 	client.Tracker.AddObj(trigger)
+	return trigger
 }
 
 // CreateSequenceOrFail will create a Sequence or fail the test if there is an error.

test/common/operation.go

@@ -139,7 +139,9 @@ func (client *Client) WaitForResourcesReady(typemeta *metav1.TypeMeta) error {
 // WaitForAllTestResourcesReady waits until all test resources in the namespace are Ready.
 func (client *Client) WaitForAllTestResourcesReady() error {
 	// wait for all Knative resources created in this test to become ready.
-	client.Tracker.WaitForKResourcesReady()
+	if err := client.Tracker.WaitForKResourcesReady(); err != nil {
+		return err
+	}
 	// explicitly wait for all pods to become ready.
 	if err := pkgTest.WaitForAllPodsRunning(client.Kube, client.Namespace); err != nil {
 		return err

test/common/tracker.go

@@ -21,6 +21,7 @@ package common
 
 import (
 	"encoding/json"
+	"fmt"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -144,7 +145,7 @@ func (t *Tracker) WaitForKResourcesReady() error {
 	t.logf("Waiting for all KResources to become ready")
 	for _, metaResource := range t.resourcesToCheckStatus {
 		if err := base.WaitForResourceReady(t.dynamicClient, &metaResource); err != nil {
-			return err
+			return fmt.Errorf("failed waiting for %+v to become ready: %v", metaResource, err)
 		}
 	}
 	return nil

test/conformance/broker_tracing_test.go

@@ -0,0 +1,26 @@
+// +build e2e
+
+/*
+Copyright 2019 The Knative Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package conformance
+
+import (
+	"testing"
+
+	"knative.dev/eventing/test/conformance/helpers"
+)
+
+func TestBrokerTracing(t *testing.T) {
+	helpers.BrokerTracingTestHelper(t, channelTestRunner)
+}