[VOL-3197][VOL-3196] Enhanced Kafka RPC and gRPC interfaces to propagate Span context for log correlation
Also, made some corrections to helper method in log/utils based on testing
Change-Id: Ic0fec935dd8996b3c6c17116586c5bd307e7bebb
diff --git a/pkg/log/utils.go b/pkg/log/utils.go
index cdd0a4e..b22ca14 100644
--- a/pkg/log/utils.go
+++ b/pkg/log/utils.go
@@ -175,6 +175,15 @@
attrMap[k] = v
}
+
+ processBaggageItems := func(k, v string) bool {
+ if k != "rpc-span-name" {
+ attrMap[k] = v
+ }
+ return true
+ }
+
+ jspan.SpanContext().ForeachBaggageItem(processBaggageItems)
}
}
}
@@ -186,9 +195,23 @@
func EnrichSpan(ctx context.Context, keyAndValues ...Fields) {
span := opentracing.SpanFromContext(ctx)
if span != nil {
- for _, field := range keyAndValues {
- for k, v := range field {
- span.SetTag(k, v)
+ if jspan, ok := span.(*jtracing.Span); ok {
+ // Inject as a BaggageItem when the Span is the Root Span so that it propagates
+ // across the components along with Root Span (called as Trace)
+ // Else, inject as a Tag so that it is attached to the Child Task
+ isRootSpan := false
+ if jspan.SpanContext().TraceID().String() == jspan.SpanContext().SpanID().String() {
+ isRootSpan = true
+ }
+
+ for _, field := range keyAndValues {
+ for k, v := range field {
+ if isRootSpan {
+ span.SetBaggageItem(k, v.(string))
+ } else {
+ span.SetTag(k, v)
+ }
+ }
}
}
}