backup/showbridge-go
1
0
Fork 0
mirror of https://github.com/jwetzell/showbridge-go.git synced 2026-04-26 21:05:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

rework route and processing spans

Browse Source
This commit is contained in:
Joel Wetzell
2026-02-04 12:20:36 -06:00
parent 4dac1faae0
commit a920f34926
2 changed files with 24 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
internal/route/route.go
View File

@@ -6,6 +6,9 @@ import (
"github.com/jwetzell/showbridge-go/internal/config" "github.com/jwetzell/showbridge-go/internal/config"
"github.com/jwetzell/showbridge-go/internal/processor" "github.com/jwetzell/showbridge-go/internal/processor"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
) )
type routeContextKey string type routeContextKey string
@@ -73,17 +76,32 @@ func (r *ProcessorRoute) Output() string {
} }
func (r *ProcessorRoute) ProcessPayload(ctx context.Context, payload any) (any, error) { func (r *ProcessorRoute) ProcessPayload(ctx context.Context, payload any) (any, error) {
for _, processor := range r.processors { parentSpan := trace.SpanFromContext(ctx)
processedPayload, err := processor.Process(ctx, payload) tracer := parentSpan.TracerProvider().Tracer("route.ProcessPayload")
processCtx, processSpan := tracer.Start(ctx, "route.process")
defer processSpan.End()
for processorIndex, processor := range r.processors {
processorCtx, processorSpan := tracer.Start(processCtx, "route.processor", trace.WithAttributes(attribute.Int("processor.index", processorIndex), attribute.String("processor.type", processor.Type())))
processedPayload, err := processor.Process(processorCtx, payload)
if err != nil { if err != nil {
processorSpan.SetStatus(codes.Error, "route processor error")
processorSpan.RecordError(err)
processorSpan.End()
processSpan.SetStatus(codes.Error, "route processing error")
processSpan.RecordError(err)
return nil, err return nil, err
} }
processorSpan.SetStatus(codes.Ok, "processor successful")
//NOTE(jwetzell) nil payload will result in the route being "terminated" //NOTE(jwetzell) nil payload will result in the route being "terminated"
if processedPayload == nil { if processedPayload == nil {
processSpan.SetStatus(codes.Ok, "route processing terminated early due to nil payload")
processorSpan.End()
return nil, nil return nil, nil
} }
payload = processedPayload payload = processedPayload
processorSpan.End()
} }
processSpan.SetStatus(codes.Ok, "route processing successful")
return payload, nil return payload, nil
} }

23
router.go
View File

@@ -160,10 +160,8 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
routeFound = true routeFound = true
routeContext := context.WithValue(spanCtx, route.SourceContextKey, sourceId) routeContext := context.WithValue(spanCtx, route.SourceContextKey, sourceId)
routeSpanCtx, routeSpan := r.tracer.Start(routeContext, "route.input", trace.WithAttributes(attribute.Int("route.index", routeIndex))) routeCtx, routeSpan := r.tracer.Start(routeContext, "route", trace.WithAttributes(attribute.Int("route.index", routeIndex), attribute.String("route.input", routeInstance.Input()), attribute.String("route.output", routeInstance.Output())))
defer routeSpan.End() payload, err := routeInstance.ProcessPayload(routeCtx, payload)
routeProcessCtx, routeSpan := r.tracer.Start(routeSpanCtx, "route.process")
payload, err := routeInstance.ProcessPayload(routeProcessCtx, payload)
if err != nil { if err != nil {
if routeIOErrors == nil { if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{} routeIOErrors = []route.RouteIOError{}
@@ -173,13 +171,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
Index: routeIndex, Index: routeIndex,
ProcessError: err, ProcessError: err,
}) })
routeSpan.SetStatus(codes.Error, err.Error())
routeSpan.RecordError(err)
routeSpan.End()
return return
} else {
routeSpan.SetStatus(codes.Ok, "route processing successful")
routeSpan.End()
} }
if payload == nil { if payload == nil {
@@ -187,8 +179,7 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
return return
} }
routeOutputCtx, routeOutputSpan := r.tracer.Start(routeSpanCtx, "route.output", trace.WithAttributes(attribute.String("destination.id", routeInstance.Output()))) outputErrors := r.HandleOutput(routeCtx, routeInstance.Output(), payload)
outputErrors := r.HandleOutput(routeOutputCtx, routeInstance.Output(), payload)
if outputErrors != nil { if outputErrors != nil {
if routeIOErrors == nil { if routeIOErrors == nil {
routeIOErrors = []route.RouteIOError{} routeIOErrors = []route.RouteIOError{}
@@ -197,14 +188,8 @@ func (r *Router) HandleInput(ctx context.Context, sourceId string, payload any)
Index: routeIndex, Index: routeIndex,
OutputErrors: outputErrors, OutputErrors: outputErrors,
}) })
routeOutputSpan.SetStatus(codes.Error, "route output error")
for _, outputError := range outputErrors {
routeOutputSpan.RecordError(outputError)
}
} else {
routeOutputSpan.SetStatus(codes.Ok, "route output successful")
} }
routeOutputSpan.End() routeSpan.End()
}) })
} }
} }