Clone event context before setting the distributed tracing extension (#…

…351)

* Fixed #332 (#333)

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Clone event context before setting the distributed tracing extension

This prevents the client from mutating the caller's copy of the event
allowing for sharing across multiple clients.

Signed-off-by: Ian Milligan <ianmllgn@gmail.com>

Co-authored-by: Francesco Guardiani <francescoguard@gmail.com>

pkg/binding/transport_test.go

@@ -16,7 +16,7 @@ func TestTransportSend(t *testing.T) {
 	transport := binding.NewTransportAdapter(binding.ChanSender(messageChannel), binding.ChanReceiver(messageChannel), nil)
 	ev := test.MinEvent()
 
-	client, err := cloudevents.NewClient(transport)
+	client, err := cloudevents.NewClient(transport, cloudevents.WithoutTracePropagation())
 	require.NoError(t, err)
 
 	_, _, err = client.Send(context.Background(), ev)

pkg/cloudevents/client/client.go

@@ -121,6 +121,7 @@ func (c *ceClient) obsSend(ctx context.Context, event cloudevents.Event) (contex
 	// Set distributed tracing extension.
 	if !c.disableTracePropagation {
 		if span := trace.FromContext(ctx); span != nil {
+			event.Context = event.Context.Clone()
 			if err := extensions.FromSpanContext(span.SpanContext()).AddTracingAttributes(event.Context); err != nil {
 				return ctx, nil, fmt.Errorf("error setting distributed tracing extension: %w", err)
 			}

pkg/cloudevents/eventcontext_test.go

@@ -7,6 +7,7 @@ import (
 	ce "github.com/cloudevents/sdk-go/pkg/cloudevents"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
 )
 
 func TestContextAsV01(t *testing.T) {
@@ -313,3 +314,40 @@ func TestContextAsV1(t *testing.T) {
 		})
 	}
 }
+
+func TestEventContextClone(t *testing.T) {
+	tests := []struct {
+		name    string
+		context ce.EventContext
+	}{
+		{
+			name:    "v0.1",
+			context: FullEventContextV01(types.Timestamp{Time: time.Now()}),
+		},
+		{
+			name:    "v0.2",
+			context: FullEventContextV02(types.Timestamp{Time: time.Now()}),
+		},
+		{
+			name:    "v0.3",
+			context: FullEventContextV03(types.Timestamp{Time: time.Now()}),
+		},
+		{
+			name:    "v1.0",
+			context: FullEventContextV1(types.Timestamp{Time: time.Now()}),
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			initial := test.context
+			require.NoError(t, initial.SetExtension("aaa", "bbb"))
+
+			clone := initial.Clone()
+			require.NoError(t, clone.SetExtension("aaa", "ccc"))
+
+			val, err := initial.GetExtension("aaa")
+			require.NoError(t, err)
+			require.Equal(t, "bbb", val)
+		})
+	}
+}

pkg/cloudevents/eventcontext_v01.go

@@ -76,7 +76,21 @@ func (ec *EventContextV01) SetExtension(name string, value interface{}) error {
 
 // Clone implements EventContextConverter.Clone
 func (ec EventContextV01) Clone() EventContext {
-	return ec.AsV01()
+	ev01 := ec.AsV01()
+	ev01.Extensions = ev01.cloneExtensions()
+	return ev01
+}
+
+func (ec *EventContextV01) cloneExtensions() map[string]interface{} {
+	old := ec.Extensions
+	if old == nil {
+		return nil
+	}
+	new := make(map[string]interface{}, len(ec.Extensions))
+	for k, v := range old {
+		new[k] = v
+	}
+	return new
 }
 
 // AsV01 implements EventContextConverter.AsV01

pkg/cloudevents/eventcontext_v02.go

@@ -85,7 +85,21 @@ func (ec *EventContextV02) SetExtension(name string, value interface{}) error {
 
 // Clone implements EventContextConverter.Clone
 func (ec EventContextV02) Clone() EventContext {
-	return ec.AsV02()
+	ec02 := ec.AsV02()
+	ec02.Extensions = ec02.cloneExtensions()
+	return ec02
+}
+
+func (ec *EventContextV02) cloneExtensions() map[string]interface{} {
+	old := ec.Extensions
+	if old == nil {
+		return nil
+	}
+	new := make(map[string]interface{}, len(ec.Extensions))
+	for k, v := range old {
+		new[k] = v
+	}
+	return new
 }
 
 // AsV01 implements EventContextConverter.AsV01

pkg/cloudevents/eventcontext_v03.go

@@ -94,7 +94,21 @@ func (ec *EventContextV03) SetExtension(name string, value interface{}) error {
 
 // Clone implements EventContextConverter.Clone
 func (ec EventContextV03) Clone() EventContext {
-	return ec.AsV03()
+	ec03 := ec.AsV03()
+	ec03.Extensions = ec03.cloneExtensions()
+	return ec03
+}
+
+func (ec *EventContextV03) cloneExtensions() map[string]interface{} {
+	old := ec.Extensions
+	if old == nil {
+		return nil
+	}
+	new := make(map[string]interface{}, len(ec.Extensions))
+	for k, v := range old {
+		new[k] = v
+	}
+	return new
 }
 
 // AsV01 implements EventContextConverter.AsV01

pkg/cloudevents/eventcontext_v1.go

@@ -97,7 +97,21 @@ func (ec *EventContextV1) SetExtension(name string, value interface{}) error {
 
 // Clone implements EventContextConverter.Clone
 func (ec EventContextV1) Clone() EventContext {
-	return ec.AsV1()
+	ec1 := ec.AsV1()
+	ec1.Extensions = ec1.cloneExtensions()
+	return ec1
+}
+
+func (ec *EventContextV1) cloneExtensions() map[string]interface{} {
+	old := ec.Extensions
+	if old == nil {
+		return nil
+	}
+	new := make(map[string]interface{}, len(ec.Extensions))
+	for k, v := range old {
+		new[k] = v
+	}
+	return new
 }
 
 // AsV01 implements EventContextConverter.AsV01

test/http/direct_v1_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/cloudevents/types"
 )
 
 func TestSenderReceiver_binary_v01(t *testing.T) {
@@ -88,7 +89,7 @@ func TestSenderReceiver_structured_v01(t *testing.T) {
 				Header: map[string][]string{
 					"content-type": {"application/cloudevents+json"},
 				},
-				Body:          fmt.Sprintf(`{"data":{"hello":"unittest"},"id":"ABC-123","source":"/unit/test/client","specversion":"1.0","subject":"resource","time":%q,"type":"unit.test.client.sent"}`, now.UTC().Format(time.RFC3339Nano)),
+				Body:          fmt.Sprintf(`{"data":{"hello":"unittest"},"id":"ABC-123","source":"/unit/test/client","specversion":"1.0","subject":"resource","time":%q,"type":"unit.test.client.sent"}`, types.FormatTime(now.UTC())),
 				ContentLength: 182,
 			},
 		},