event trace propagation (#802)

Author: FoseFx

libs/hwes/event.go

@@ -16,19 +16,31 @@ import (
 )
 
 type Event struct {
-	EventID         uuid.UUID
-	EventType       string
-	AggregateID     uuid.UUID
-	AggregateType   AggregateType
-	Data            []byte
-	Timestamp       time.Time
-	Version         uint64
+	// identifier of this event
+	EventID uuid.UUID
+	// constant which can be used to route this event to its handler(s)
+	EventType string
+	// identifier of the aggregate that this event has an effect on
+	AggregateID uuid.UUID
+	// type of the aggregate that this event has an effect on
+	AggregateType AggregateType
+	// payload of the event, must be json
+	Data []byte
+	// time of event creation
+	Timestamp time.Time
+	// event's revision number (event's can, but should not be modified)
+	Version uint64
+	// user responsible for this event
 	CommitterUserID *uuid.UUID
+	// w3c trace context
+	TraceParent string
 }
 
 type metadata struct {
 	// CommitterUserID represents an optional UUID that identifies the user that is directly responsible for this event
 	CommitterUserID string `json:"committer_user_id"`
+	// w3c trace context
+	TraceParent string `json:"trace_parent"`
 	// The Timestamp represents the time when the event was created. Using the built-in eventstoreDB timestamp is discouraged.
 	Timestamp time.Time `json:"timestamp"`
 }
@@ -39,6 +51,7 @@ type EventOption func(*Event) error
 // WithContext applies SetCommitterFromCtx after construction
 func WithContext(ctx context.Context) EventOption {
 	return func(event *Event) error {
+		event.SetTracingContextFromCtx(ctx)
 		return event.SetCommitterFromCtx(ctx)
 	}
 }
@@ -129,6 +142,7 @@ func NewEventFromRecordedEvent(esdbEvent *esdb.RecordedEvent) (Event, error) {
 		Timestamp:       md.Timestamp,
 		Version:         esdbEvent.EventNumber,
 		CommitterUserID: nil,
+		TraceParent:     md.TraceParent,
 	}
 
 	eventCommitterUserID, err := uuid.Parse(md.CommitterUserID)
@@ -163,6 +177,7 @@ func (e *Event) GetVersion() uint64 {
 
 func (e *Event) ToEventData() (esdb.EventData, error) {
 	md := metadata{
+    TraceParent: e.TraceParent,
 		Timestamp: e.Timestamp,
 	}
 	if e.CommitterUserID != nil {
@@ -221,8 +236,9 @@ func (e *Event) SetCommitterFromCtx(ctx context.Context) error {
 
 	userID, err := common.GetUserID(ctx)
 	if err != nil {
-		return err
+		return nil // don't set a user, if no user is available
 	}
+
 	e.CommitterUserID = &userID
 
 	// Just to make sure we are actually dealing with a valid UUID
@@ -234,16 +250,25 @@ func (e *Event) SetCommitterFromCtx(ctx context.Context) error {
 	return nil
 }
 
+// SetTracingContextFromCtx propagates the currently active span in the context
+func (e *Event) SetTracingContextFromCtx(ctx context.Context) {
+	ctx, span, _ := telemetry.StartSpan(ctx, "SetTracingContextFromCtx")
+	e.TraceParent = telemetry.TraceParent(ctx)
+	span.End()
+}
+
 // GetZerologDict to enrich some logs
 //
 // Example:
 //
 // zerolog.Ctx(ctx).Debug().Dict("event", event.GetZerologDict()).Msg("process event")
 func (e *Event) GetZerologDict() *zerolog.Event {
+
 	dict := zerolog.Dict().
 		Str("eventId", e.EventID.String()).
 		Str("eventType", e.EventType).
-		Uint64("eventVersion", e.Version)
+		Uint64("eventVersion", e.Version).
+		Str("traceParent", e.TraceParent)
 
 	if e.CommitterUserID != nil {
 		dict.Str("committerUserID", e.CommitterUserID.String())

libs/hwes/eventstoredb/projections/custom/custom.go

@@ -165,7 +165,6 @@ func (p *CustomProjection) Subscribe(ctx context.Context) error {
 // and calls the according event handler based on the received event
 // This function blocks the thread until the passed context gets canceled
 func (p *CustomProjection) processReceivedEventFromStream(ctx context.Context, stream *esdb.PersistentSubscription, esdbEvent *esdb.PersistentSubscriptionEvent) error {
-	// TODO: Connect with source trace?
 	ctx, span, log := telemetry.StartSpan(ctx, "custom_projection.processReceivedEventFromStream")
 	defer span.End()
 
@@ -193,6 +192,27 @@ func (p *CustomProjection) processReceivedEventFromStream(ctx context.Context, s
 		return nil
 	}
 
+	//
+	// change tracing spans
+	//
+	if event.TraceParent != "" {
+		ctx = telemetry.FromTraceParent(ctx, event.TraceParent)
+
+		// end old span, start new one
+		span.End()
+		ctx, span, log = telemetry.StartSpan(ctx, "custom_projection.processReceivedEventFromStream")
+		defer span.End()
+
+		// set attributes
+		log = log.With().
+			Str("subscription_group_name", p.subscriptionGroupName).
+			Str("esdbEventID", event.EventID.String()).
+			Logger()
+
+		telemetry.SetSpanStr(ctx, "subscription_group_name", p.subscriptionGroupName)
+		telemetry.SetSpanStr(ctx, "esdbEventID", event.EventID.String())
+	}
+
 	log.Debug().Dict("event", event.GetZerologDict()).Msg("process event")
 
 	err, nackAction := p.HandleEvent(ctx, event)

libs/telemetry/otel.go

@@ -7,6 +7,7 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
 	"reflect"
 )
@@ -56,6 +57,21 @@ func SetSpanBool(ctx context.Context, key string, value bool) {
 	SetSpanAttributes(ctx, attribute.Bool(key, value))
 }
 
+// TraceParent returns the w3c Trace Context traceparent header for a SpanContext, and if the span context is valid
+func TraceParent(ctx context.Context) string {
+	propagator := propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
+	carrier := propagation.MapCarrier{}
+	propagator.Inject(ctx, carrier)
+	return carrier["traceparent"]
+}
+
+// FromTraceParent yields a new context using a propagated w3c traceparent header
+func FromTraceParent(ctx context.Context, traceparent string) context.Context {
+	propagator := propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
+	carrier := propagation.MapCarrier{"traceparent": traceparent}
+	return propagator.Extract(ctx, carrier)
+}
+
 // zerologTraceHook calls addSpanIdToLogEvent and TODO for log events
 func zerologTraceHook() zerolog.HookFunc {
 	return func(event *zerolog.Event, level zerolog.Level, message string) {

services/property-svc/internal/property-set/aggregate/actions.go

@@ -8,7 +8,7 @@ import (
 func (a *PropertySetAggregate) CreatePropertySet(ctx context.Context, name string) error {
 	id := a.GetID()
 
-	event, err := propertyEventsV1.NewPropertySetCreatedEvent(a, id, name)
+	event, err := propertyEventsV1.NewPropertySetCreatedEvent(ctx, a, id, name)
 	if err != nil {
 		return err
 	}

services/property-svc/internal/property-set/events/v1/events.go

@@ -1,6 +1,7 @@
 package v1
 
 import (
+	"context"
 	"github.com/google/uuid"
 	"hwes"
 )
@@ -14,10 +15,10 @@ type PropertySetCreatedEvent struct {
 	Name string `json:"name"`
 }
 
-func NewPropertySetCreatedEvent(a hwes.Aggregate, id uuid.UUID, name string) (hwes.Event, error) {
+func NewPropertySetCreatedEvent(ctx context.Context, a hwes.Aggregate, id uuid.UUID, name string) (hwes.Event, error) {
 	payload := PropertySetCreatedEvent{
 		ID:   id.String(),
 		Name: name,
 	}
-	return hwes.NewEvent(a, PropertySetCreated, hwes.WithData(payload))
+	return hwes.NewEvent(a, PropertySetCreated, hwes.WithData(payload), hwes.WithContext(ctx))
 }

services/property-svc/internal/property-value/aggregate/actions.go

@@ -9,15 +9,15 @@ import (
 func (a *PropertyValueAggregate) CreatePropertyValue(ctx context.Context, propertyID uuid.UUID, value interface{}, subjectID uuid.UUID) error {
 	id := a.GetID()
 
-	event, err := propertyEventsV1.NewPropertyValueCreatedEvent(a, id, propertyID, value, subjectID)
+	event, err := propertyEventsV1.NewPropertyValueCreatedEvent(ctx, a, id, propertyID, value, subjectID)
 	if err != nil {
 		return err
 	}
 	return a.Apply(event)
 }
 
 func (a *PropertyValueAggregate) UpdatePropertyValue(ctx context.Context, value interface{}) error {
-	event, err := propertyEventsV1.NewPropertyValueUpdatedEvent(a, value)
+	event, err := propertyEventsV1.NewPropertyValueUpdatedEvent(ctx, a, value)
 	if err != nil {
 		return err
 	}

services/property-svc/internal/property-value/events/v1/events.go

@@ -1,6 +1,7 @@
 package v1
 
 import (
+	"context"
 	"github.com/google/uuid"
 	"hwes"
 )
@@ -17,23 +18,23 @@ type PropertyValueCreatedEvent struct {
 	SubjectID  string      `json:"subject_id"`
 }
 
-func NewPropertyValueCreatedEvent(a hwes.Aggregate, id uuid.UUID, propertyID uuid.UUID, value interface{}, subjectID uuid.UUID) (hwes.Event, error) {
+func NewPropertyValueCreatedEvent(ctx context.Context, a hwes.Aggregate, id uuid.UUID, propertyID uuid.UUID, value interface{}, subjectID uuid.UUID) (hwes.Event, error) {
 	payload := PropertyValueCreatedEvent{
 		ID:         id.String(),
 		PropertyID: propertyID.String(),
 		Value:      value,
 		SubjectID:  subjectID.String(),
 	}
-	return hwes.NewEvent(a, PropertyValueCreated, hwes.WithData(payload))
+	return hwes.NewEvent(a, PropertyValueCreated, hwes.WithData(payload), hwes.WithContext(ctx))
 }
 
 type PropertyValueUpdatedEvent struct {
 	Value interface{} `json:"value"`
 }
 
-func NewPropertyValueUpdatedEvent(a hwes.Aggregate, value interface{}) (hwes.Event, error) {
+func NewPropertyValueUpdatedEvent(ctx context.Context, a hwes.Aggregate, value interface{}) (hwes.Event, error) {
 	payload := PropertyValueUpdatedEvent{
 		Value: value,
 	}
-	return hwes.NewEvent(a, PropertyValueUpdated, hwes.WithData(payload))
+	return hwes.NewEvent(a, PropertyValueUpdated, hwes.WithData(payload), hwes.WithContext(ctx))
 }