Add Open Telemetry Spec attributes to IMC Dispatcher span

docs/spec/channel.md

@@ -327,13 +327,6 @@ discretion (e.g. expose a gRPC endpoint to accept events).
 If a Channel receives an event queueing request and is unable to parse a valid
 CloudEvent, then it MUST reject the request.
 
-The Channel MUST recognize and pass through all tracing information from sender
-to subscribers using [W3C Tracecontext](https://w3c.github.io/trace-context/),
-although internally it MAY use another mechanism(s) to propagate the tracing
-information. The Channel SHOULD sample and write traces to the location
-specified in
-[`config-tracing`](https://github.com/knative/eventing/blob/master/config/config-tracing.yaml).
-
 ##### HTTP
 
 Channels MUST reject all HTTP event queueing requests with a method other than
@@ -388,7 +381,7 @@ not limited to:
 - the time in-between retries
 - the backoff rate
 
-#### Metrics
+#### Observability
 
 Channels SHOULD expose a variety of metrics, including, but not limited to:
 
@@ -401,6 +394,23 @@ Channels SHOULD expose a variety of metrics, including, but not limited to:
 Metrics SHOULD be enabled by default, with a configuration parameter included to
 disable them if desired.
 
+The Channel MUST recognize and pass through all tracing information from sender
+to subscribers using [W3C Tracecontext](https://w3c.github.io/trace-context/),
+although internally it MAY use another mechanism(s) to propagate the tracing
+information. The Channel SHOULD sample and write traces to the location
+specified in
+[`config-tracing`](https://github.com/knative/eventing/blob/master/config/config-tracing.yaml).
+
+Spans emitted by the Channel SHOULD follow the
+[OpenTelemetry Semantic Conventions for Messaging Systems](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md)
+whenever possible. In particular, spans emitted by the Channel SHOULD set the
+following attributes:
+
+- messaging.system: "knative"
+- messaging.destination: url to which the event is being routed
+- messaging.protocol: the name of the underlying transport protocol
+- messaging.message_id: the event ID
+
 ## Changelog
 
 - `0.11.x release`: CloudEvents in 0.3 and 1.0 are supported.

pkg/channel/message_dispatcher.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	"knative.dev/eventing/pkg/kncloudevents"
+	"knative.dev/eventing/pkg/tracing"
 	"knative.dev/eventing/pkg/utils"
 )
 
@@ -190,7 +191,7 @@ func (d *MessageDispatcherImpl) executeRequest(ctx context.Context, url *url.URL
 	}
 
 	if span.IsRecordingEvents() {
-		err = kncloudevents.WriteHTTPRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders, kncloudevents.PopulateSpan(span))
+		err = kncloudevents.WriteHTTPRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders, tracing.PopulateSpan(span, url.String()))
 	} else {
 		err = kncloudevents.WriteHTTPRequestWithAdditionalHeaders(ctx, message, req, additionalHeaders)
 	}

pkg/kncloudevents/populate_span_transformer.go → pkg/tracing/populate_span_transformer.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kncloudevents
+package tracing
 
 import (
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -23,26 +23,32 @@ import (
 	"go.opencensus.io/trace"
 )
 
-func PopulateSpan(span *trace.Span) binding.TransformerFunc {
+func PopulateSpan(span *trace.Span, destination string) binding.TransformerFunc {
 	return func(reader binding.MessageMetadataReader, writer binding.MessageMetadataWriter) error {
-		_, specVersion := reader.GetAttribute(spec.SpecVersion)
-		if specVersion != nil {
-			specVersionParsed, err := types.Format(specVersion)
-			if err != nil {
-				return err
-			}
-			span.AddAttributes(trace.StringAttribute("cloudevents.specversion", specVersionParsed))
-		}
+
+		span.AddAttributes(MessagingProtocolHTTP)
+		span.AddAttributes(MessagingSystemAttribute)
+		span.AddAttributes(trace.StringAttribute(MessagingDestinationAttributeName, destination))
 
 		_, id := reader.GetAttribute(spec.ID)
 		if id != nil {
 			idParsed, err := types.Format(id)
 			if err != nil {
 				return err
 			}
+			span.AddAttributes(trace.StringAttribute(MessagingMessageIDAttributeName, idParsed))
 			span.AddAttributes(trace.StringAttribute("cloudevents.id", idParsed))
 		}
 
+		_, specVersion := reader.GetAttribute(spec.SpecVersion)
+		if specVersion != nil {
+			specVersionParsed, err := types.Format(specVersion)
+			if err != nil {
+				return err
+			}
+			span.AddAttributes(trace.StringAttribute("cloudevents.specversion", specVersionParsed))
+		}
+
 		_, ty := reader.GetAttribute(spec.Type)
 		if ty != nil {
 			tyParsed, err := types.Format(ty)

pkg/kncloudevents/populate_span_transformer_test.go → pkg/tracing/populate_span_transformer_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kncloudevents
+package tracing
 
 import (
 	"context"
@@ -46,7 +46,13 @@ func TestPopulateSpan(t *testing.T) {
 	wantEvent.SetType("hello.world")
 	wantEvent.SetSource("example.com")
 
+	destination := "some-url"
+
 	expectedAttributes := map[string]interface{}{
+		"messaging.system":        "knative",
+		"messaging.protocol":      "HTTP",
+		"messaging.message_id":    "aaa",
+		"messaging.destination":   "some-url",
 		"cloudevents.id":          "aaa",
 		"cloudevents.type":        "hello.world",
 		"cloudevents.source":      "example.com",
@@ -63,7 +69,7 @@ func TestPopulateSpan(t *testing.T) {
 				spanData := <-mockExp
 				require.Equal(t, expectedAttributes, spanData.Attributes)
 			},
-			Transformers: binding.Transformers{PopulateSpan(testSpanBinary)},
+			Transformers: binding.Transformers{PopulateSpan(testSpanBinary, destination)},
 		},
 		{
 			Name:       "Populate span for event messages",
@@ -74,7 +80,7 @@ func TestPopulateSpan(t *testing.T) {
 				spanData := <-mockExp
 				require.Equal(t, expectedAttributes, spanData.Attributes)
 			},
-			Transformers: binding.Transformers{PopulateSpan(testSpanEvent)},
+			Transformers: binding.Transformers{PopulateSpan(testSpanEvent, destination)},
 		},
 	})
 }

test/conformance/helpers/channel_tracing_test_helper.go

@@ -19,6 +19,7 @@ package helpers
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"testing"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -73,10 +74,10 @@ func setupChannelTracingWithReply(
 	recordEventsPod := recordevents.DeployEventRecordOrFail(ctx, client, recordEventsPodName)
 
 	// Create the subscriber, a Pod that mutates the event.
-	transformerPod := recordevents.DeployEventRecordOrFail(
+	mutatingPod := recordevents.DeployEventRecordOrFail(
 		ctx,
 		client,
-		"transformer",
+		"mutator",
 		recordevents.ReplyWithTransformedEvent(
 			"mutated",
 			eventSource,
@@ -89,7 +90,7 @@ func setupChannelTracingWithReply(
 		"sub",
 		channelName,
 		channel,
-		resources.WithSubscriberForSubscription(transformerPod.Name),
+		resources.WithSubscriberForSubscription(mutatingPod.Name),
 		resources.WithReplyForSubscription(replyChannelName, channel))
 
 	// Create the Subscription linking the reply Channel to the LogEvents K8s Service.
@@ -125,12 +126,15 @@ func setupChannelTracingWithReply(
 	// We expect the following spans:
 	// 1. Sending pod sends event to Channel (only if the sending pod generates a span).
 	// 2. Channel receives event from sending pod.
-	// 3. Channel sends event to transformer pod.
-	// 4. Transformer Pod receives event from Channel.
-	// 5. Channel sends reply from Transformer Pod to the reply Channel.
-	// 6. Reply Channel receives event from the original Channel's reply.
-	// 7. Reply Channel sends event to the logging Pod.
-	// 8. Logging pod receives event from Channel.
+	// 3. Channel Dispatcher span
+	// 4. Channel sends event to Mutator pod.
+	// 5. Mutator Pod receives event from Channel.
+	// 6. Channel Dispatcher span
+	// 7. Channel sends reply from Mutator Pod to the reply Channel.
+	// 8. Reply Channel receives event from the original Channel's reply.
+	// 9. Channel Dispatcher span
+	// 10. Reply Channel sends event to the logging Pod.
+	// 11. Logging pod receives event from Channel.
 	expected := tracinghelper.TestSpanTree{
 		// 1 is added below if it is needed.
 		// 2. Channel receives event from sending pod.
@@ -143,68 +147,86 @@ func setupChannelTracingWithReply(
 		),
 		Children: []tracinghelper.TestSpanTree{
 			{
-				// 3. Channel sends event to transformer pod.
-				Span: tracinghelper.MatchHTTPSpanWithReply(
-					model.Client,
-					tracinghelper.WithHTTPHostAndPath(
-						fmt.Sprintf("%s.%s.svc", transformerPod.Name, client.Namespace),
-						"/",
-					),
-				),
+				// 3. Channel Dispatcher span
+				Span: channelSpan(eventID, fmt.Sprintf("%s.%s.svc", mutatingPod.Name, client.Namespace), "/"),
 				Children: []tracinghelper.TestSpanTree{
 					{
-						// 4. Transformer Pod receives event from Channel.
+						// 4. Channel sends event to Mutator pod.
 						Span: tracinghelper.MatchHTTPSpanWithReply(
-							model.Server,
+							model.Client,
 							tracinghelper.WithHTTPHostAndPath(
-								fmt.Sprintf("%s.%s.svc", transformerPod.Name, client.Namespace),
+								fmt.Sprintf("%s.%s.svc", mutatingPod.Name, client.Namespace),
 								"/",
 							),
-							tracinghelper.WithLocalEndpointServiceName(transformerPod.Name),
 						),
+						Children: []tracinghelper.TestSpanTree{
+							{
+								// 5. Mutator Pod receives event from Channel.
+								Span: tracinghelper.MatchHTTPSpanWithReply(
+									model.Server,
+									tracinghelper.WithHTTPHostAndPath(
+										fmt.Sprintf("%s.%s.svc", mutatingPod.Name, client.Namespace),
+										"/",
+									),
+									tracinghelper.WithLocalEndpointServiceName(mutatingPod.Name),
+								),
+							},
+						},
 					},
-				},
-			},
-			{
-				// 5. Channel sends reply from Transformer Pod to the reply Channel.
-				Span: tracinghelper.MatchHTTPSpanNoReply(
-					model.Client,
-					tracinghelper.WithHTTPHostAndPath(
-						fmt.Sprintf("%s-kn-channel.%s.svc", replyChannelName, client.Namespace),
-						"",
-					),
-				),
-				Children: []tracinghelper.TestSpanTree{
-					// 6. Reply Channel receives event from the original Channel's reply.
 					{
-						Span: tracinghelper.MatchHTTPSpanNoReply(
-							model.Server,
-							tracinghelper.WithHTTPHostAndPath(
-								fmt.Sprintf("%s-kn-channel.%s.svc", replyChannelName, client.Namespace),
-								"/",
-							),
-						),
+						// 6. Channel Dispatcher span
+						Span: channelSpan(eventID, fmt.Sprintf("%s-kn-channel.%s.svc", replyChannelName, client.Namespace), ""),
 						Children: []tracinghelper.TestSpanTree{
 							{
-								// 7. Reply Channel sends event to the logging Pod.
+								// 7. Channel sends reply from Mutator Pod to the reply Channel.
 								Span: tracinghelper.MatchHTTPSpanNoReply(
 									model.Client,
 									tracinghelper.WithHTTPHostAndPath(
-										fmt.Sprintf("%s.%s.svc", recordEventsPod.Name, client.Namespace),
-										"/",
+										fmt.Sprintf("%s-kn-channel.%s.svc", replyChannelName, client.Namespace),
+										"",
 									),
 								),
 								Children: []tracinghelper.TestSpanTree{
 									{
-										// 8. Logging pod receives event from Channel.
+										// 8. Reply Channel receives event from the original Channel's reply.
 										Span: tracinghelper.MatchHTTPSpanNoReply(
 											model.Server,
 											tracinghelper.WithHTTPHostAndPath(
-												fmt.Sprintf("%s.%s.svc", recordEventsPod.Name, client.Namespace),
+												fmt.Sprintf("%s-kn-channel.%s.svc", replyChannelName, client.Namespace),
 												"/",
 											),
-											tracinghelper.WithLocalEndpointServiceName(recordEventsPod.Name),
 										),
+										Children: []tracinghelper.TestSpanTree{
+											{
+												// 9. Channel Dispatcher span
+												Span: channelSpan(eventID, fmt.Sprintf("%s.%s.svc", recordEventsPod.Name, client.Namespace), "/"),
+												Children: []tracinghelper.TestSpanTree{
+													{
+														// 10. Reply Channel sends event to the logging Pod.
+														Span: tracinghelper.MatchHTTPSpanNoReply(
+															model.Client,
+															tracinghelper.WithHTTPHostAndPath(
+																fmt.Sprintf("%s.%s.svc", recordEventsPod.Name, client.Namespace),
+																"/",
+															),
+														),
+														Children: []tracinghelper.TestSpanTree{
+															{
+																// 11. Logging pod receives event from Channel.
+																Span: tracinghelper.MatchHTTPSpanNoReply(
+																	model.Server,
+																	tracinghelper.WithHTTPHostAndPath(
+																		fmt.Sprintf("%s.%s.svc", recordEventsPod.Name, client.Namespace),
+																		"/",
+																	),
+																	tracinghelper.WithLocalEndpointServiceName(recordEventsPod.Name),
+																),
+															},
+														},
+													},
+												},
+											},
+										},
 									},
 								},
 							},
@@ -236,3 +258,15 @@ func setupChannelTracingWithReply(
 		cetest.DataContains(body),
 	)
 }
+
+func channelSpan(eventID, host, path string) *tracinghelper.SpanMatcher {
+	k := model.Client
+	return &tracinghelper.SpanMatcher{
+		Kind: &k,
+		Tags: map[string]*regexp.Regexp{
+			"messaging.system":      regexp.MustCompile("^knative$"),
+			"messaging.destination": regexp.MustCompile("^http://" + host + tracinghelper.HostSuffix + path + "$"),
+			"messaging.message_id":  regexp.MustCompile("^" + eventID + "$"),
+		},
+	}
+}