sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Pass context to event publishing.

Browse Source 
To allow tracing propagation via events, we need to pass the context
to the `Publish` function of reva events. This adds the context
everywhere where events are published. If there was no context to pass,
we started a new one with `context.Background()`.
This commit is contained in:
Daniël Franke
2023-08-11 12:01:19 +02:00
parent 6700214519
commit 952ebe7d7d
24 changed files with 86 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/github.com/cs3org/reva/v2/internal/grpc/interceptors/eventsmiddleware/events.go
Generated Vendored
+1 -1
View File
@@ -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)
}
}
vendor/github.com/cs3org/reva/v2/pkg/events/events.go
Generated Vendored
+40 -12
View File
@@ -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"]
}
vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/datatx.go
Generated Vendored
+1 -1
View File
@@ -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
vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go
Generated Vendored
+7 -10
View File
@@ -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)
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go
Generated Vendored
+2 -2
View File
@@ -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
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go
Generated Vendored
+1 -1
View File
@@ -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),