Add tracing propagation to events

changelog/unreleased/tracing-events-propagator.md

@@ -0,0 +1,5 @@
+Enhancement: tracing events propgation
+
+tracing information will now be propagated via events
+
+https://github.com/cs3org/reva/pull/4110

internal/grpc/interceptors/eventsmiddleware/events.go

@@ -194,7 +194,7 @@ func NewUnary(m map[string]interface{}) (grpc.UnaryServerInterceptor, int, error
 		}
 
 		if ev != nil {
-			if err := events.Publish(publisher, ev); err != nil {
+			if err := events.Publish(ctx, publisher, ev); err != nil {
 				log.Error(err)
 			}
 		}

pkg/events/events.go

@@ -19,11 +19,13 @@
 package events
 
 import (
+	"context"
 	"log"
 	"reflect"
 
 	"github.com/google/uuid"
 	"go-micro.dev/v4/events"
+	"go.opentelemetry.io/otel/propagation"
 )
 
 var (
@@ -38,6 +40,9 @@ var (
 
 	// MetadatakeyEventID is the key used for the eventID in the metadata map of the event
 	MetadatakeyEventID = "eventid"
+
+	// MetadatakeyTraceParent is the key used for the traceparent in the metadata map of the event
+	MetadatakeyTraceParent = "traceparent"
 )
 
 type (
@@ -64,9 +69,10 @@ type (
 
 	// Event is the envelope for events
 	Event struct {
-		Type  string
-		ID    string
-		Event interface{}
+		Type        string
+		ID          string
+		TraceParent string
+		Event       interface{}
 	}
 )
 
@@ -102,9 +108,10 @@ func Consume(s Consumer, group string, evs ...Unmarshaller) (<-chan Event, error
 			}
 
 			outchan <- Event{
-				Type:  et,
-				ID:    e.Metadata[MetadatakeyEventID],
-				Event: event,
+				Type:        et,
+				ID:          e.Metadata[MetadatakeyEventID],
+				TraceParent: e.Metadata[MetadatakeyTraceParent],
+				Event:       event,
 			}
 		}
 	}()
@@ -123,9 +130,10 @@ func ConsumeAll(s Consumer, group string) (<-chan Event, error) {
 		for {
 			e := <-c
 			outchan <- Event{
-				Type:  e.Metadata[MetadatakeyEventType],
-				ID:    e.Metadata[MetadatakeyEventID],
-				Event: e.Payload,
+				Type:        e.Metadata[MetadatakeyEventType],
+				ID:          e.Metadata[MetadatakeyEventID],
+				TraceParent: e.Metadata[MetadatakeyTraceParent],
+				Event:       e.Payload,
 			}
 		}
 	}()
@@ -134,10 +142,30 @@ func ConsumeAll(s Consumer, group string) (<-chan Event, error) {
 
 // Publish publishes the ev to the MainQueue from where it is distributed to all subscribers
 // NOTE: needs to use reflect on runtime
-func Publish(s Publisher, ev interface{}) error {
+func Publish(ctx context.Context, s Publisher, ev interface{}) error {
 	evName := reflect.TypeOf(ev).String()
+	traceParent := getTraceParentFromCtx(ctx)
 	return s.Publish(MainQueueName, ev, events.WithMetadata(map[string]string{
-		MetadatakeyEventType: evName,
-		MetadatakeyEventID:   uuid.New().String(),
+		MetadatakeyEventType:   evName,
+		MetadatakeyEventID:     uuid.New().String(),
+		MetadatakeyTraceParent: traceParent,
 	}))
 }
+
+// GetTraceContext extracts the trace context from the event and injects it into the given
+// context.
+func (e *Event) GetTraceContext(ctx context.Context) context.Context {
+	return propagation.TraceContext{}.Extract(ctx, propagation.MapCarrier{
+		"traceparent": e.TraceParent,
+	})
+}
+
+// getTraceParentFromCtx will return a traceparent from the context if it exists.
+// it will be a string as specificied here: https://www.w3.org/TR/trace-context/
+// If no trace info in the context, the return will be an empty string
+func getTraceParentFromCtx(ctx context.Context) string {
+	mc := propagation.MapCarrier{}
+	tc := propagation.TraceContext{}
+	tc.Inject(ctx, &mc)
+	return mc["traceparent"]
+}

pkg/events/example/publisher/publisher.go

@@ -20,6 +20,7 @@
 package publisher
 
 import (
+	"context"
 	"log"
 
 	user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
@@ -29,7 +30,7 @@ import (
 // Example publishes events to the queue
 func Example(p events.Publisher) {
 	// nothing to do - just publish!
-	err := events.Publish(p, events.ShareCreated{
+	err := events.Publish(context.Background(), p, events.ShareCreated{
 		Sharer: &user.UserId{
 			OpaqueId: "123",
 		},
@@ -40,5 +41,4 @@ func Example(p events.Publisher) {
 	if err != nil {
 		log.Fatal(err)
 	}
-
 }

pkg/rhttp/datatx/datatx.go

@@ -51,7 +51,7 @@ func EmitFileUploadedEvent(spaceOwnerOrManager, executant *userv1beta1.UserId, r
 		Timestamp:  utils.TSNow(),
 	}
 
-	return events.Publish(publisher, uploadedEv)
+	return events.Publish(context.Background(), publisher, uploadedEv)
 }
 
 // InvalidateCache is a helper function which invalidates the stat cache

pkg/share/manager/jsoncs3/jsoncs3.go

@@ -304,7 +304,7 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
 
 	// check if share already exists.
 	key := &collaboration.ShareKey{
-		//Owner:      md.Owner, owner no longer matters as it belongs to the space
+		// Owner:      md.Owner, owner no longer matters as it belongs to the space
 		ResourceId: md.Id,
 		Grantee:    g.Grantee,
 	}
@@ -337,7 +337,6 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
 
 	eg.Go(func() error {
 		err := m.Cache.Add(ctx, md.Id.StorageId, md.Id.SpaceId, shareID, s)
-
 		if err != nil {
 			span.RecordError(err)
 			span.SetStatus(codes.Error, err.Error())
@@ -348,7 +347,6 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
 
 	eg.Go(func() error {
 		err := m.CreatedCache.Add(ctx, s.GetCreator().GetOpaqueId(), shareID)
-
 		if err != nil {
 			span.RecordError(err)
 			span.SetStatus(codes.Error, err.Error())
@@ -369,7 +367,6 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
 				State: collaboration.ShareState_SHARE_STATE_PENDING,
 			}
 			err := m.UserReceivedStates.Add(ctx, userid, spaceID, rs)
-
 			if err != nil {
 				span.RecordError(err)
 				span.SetStatus(codes.Error, err.Error())
@@ -381,7 +378,6 @@ func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *colla
 		eg.Go(func() error {
 			groupid := g.Grantee.GetGroupId().GetOpaqueId()
 			err := m.GroupReceivedCache.Add(ctx, groupid, shareID)
-
 			if err != nil {
 				span.RecordError(err)
 				span.SetStatus(codes.Error, err.Error())
@@ -458,7 +454,7 @@ func (m *Manager) GetShare(ctx context.Context, ref *collaboration.ShareReferenc
 			log.Error().Err(err).
 				Msg("failed to unshare expired share")
 		}
-		if err := events.Publish(m.eventStream, events.ShareExpired{
+		if err := events.Publish(ctx, m.eventStream, events.ShareExpired{
 			ShareID:        s.GetId(),
 			ShareOwner:     s.GetOwner(),
 			ItemID:         s.GetResourceId(),
@@ -643,7 +639,7 @@ func (m *Manager) listSharesByIDs(ctx context.Context, user *userv1beta1.User, f
 						log.Error().Err(err).
 							Msg("failed to unshare expired share")
 					}
-					if err := events.Publish(m.eventStream, events.ShareExpired{
+					if err := events.Publish(ctx, m.eventStream, events.ShareExpired{
 						ShareOwner:     s.GetOwner(),
 						ItemID:         s.GetResourceId(),
 						ExpiredAt:      time.Unix(int64(s.GetExpiration().GetSeconds()), int64(s.GetExpiration().GetNanos())),
@@ -711,7 +707,7 @@ func (m *Manager) listCreatedShares(ctx context.Context, user *userv1beta1.User,
 					log.Error().Err(err).
 						Msg("failed to unshare expired share")
 				}
-				if err := events.Publish(m.eventStream, events.ShareExpired{
+				if err := events.Publish(ctx, m.eventStream, events.ShareExpired{
 					ShareOwner:     s.GetOwner(),
 					ItemID:         s.GetResourceId(),
 					ExpiredAt:      time.Unix(int64(s.GetExpiration().GetSeconds()), int64(s.GetExpiration().GetNanos())),
@@ -834,7 +830,7 @@ func (m *Manager) ListReceivedShares(ctx context.Context, filters []*collaborati
 							log.Error().Err(err).
 								Msg("failed to unshare expired share")
 						}
-						if err := events.Publish(m.eventStream, events.ShareExpired{
+						if err := events.Publish(ctx, m.eventStream, events.ShareExpired{
 							ShareOwner:     s.GetOwner(),
 							ItemID:         s.GetResourceId(),
 							ExpiredAt:      time.Unix(int64(s.GetExpiration().GetSeconds()), int64(s.GetExpiration().GetNanos())),
@@ -934,7 +930,7 @@ func (m *Manager) getReceived(ctx context.Context, ref *collaboration.ShareRefer
 			log.Error().Err(err).
 				Msg("failed to unshare expired share")
 		}
-		if err := events.Publish(m.eventStream, events.ShareExpired{
+		if err := events.Publish(ctx, m.eventStream, events.ShareExpired{
 			ShareOwner:     s.GetOwner(),
 			ItemID:         s.GetResourceId(),
 			ExpiredAt:      time.Unix(int64(s.GetExpiration().GetSeconds()), int64(s.GetExpiration().GetNanos())),
@@ -988,6 +984,7 @@ func (m *Manager) UpdateReceivedShare(ctx context.Context, receivedShare *collab
 func shareIsRoutable(share *collaboration.Share) bool {
 	return strings.Contains(share.Id.OpaqueId, shareid.IDDelimiter)
 }
+
 func updateShareID(share *collaboration.Share) {
 	share.Id.OpaqueId = shareid.Encode(share.ResourceId.StorageId, share.ResourceId.SpaceId, share.Id.OpaqueId)
 }

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -305,6 +305,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 			fs.cache.RemoveStatContext(ctx, ev.ExecutingUser.GetId(), &provider.ResourceId{SpaceId: n.SpaceID, OpaqueId: n.ID})
 
 			if err := events.Publish(
+				ctx,
 				fs.stream,
 				events.UploadReady{
 					UploadID:      ev.UploadID,
@@ -342,7 +343,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 				continue
 			}
 			// restart postprocessing
-			if err := events.Publish(fs.stream, events.BytesReceived{
+			if err := events.Publish(ctx, fs.stream, events.BytesReceived{
 				UploadID:      up.Info.ID,
 				URL:           s,
 				SpaceOwner:    n.SpaceOwnerOrManager(up.Ctx),
@@ -474,7 +475,6 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 			log.Error().Interface("event", ev).Msg("Unknown event")
 		}
 	}
-
 }
 
 // Shutdown shuts down the storage

pkg/storage/utils/decomposedfs/upload/upload.go

@@ -274,7 +274,7 @@ func (upload *Upload) FinishUpload(_ context.Context) error {
 			return err
 		}
 
-		if err := events.Publish(upload.pub, events.BytesReceived{
+		if err := events.Publish(ctx, upload.pub, events.BytesReceived{
 			UploadID:      upload.Info.ID,
 			URL:           s,
 			SpaceOwner:    n.SpaceOwnerOrManager(upload.Ctx),