add trace propagation

opencensus/opencensus.go

@@ -4,12 +4,17 @@ import (
 	"context"
 	"github.com/Azure/azure-amqp-common-go/trace"
 	oct "go.opencensus.io/trace"
+	"go.opencensus.io/trace/propagation"
 )
 
 func init() {
 	trace.Register(new(Trace))
 }
 
+const (
+	propagationKey = "_oc_prop"
+)
+
 type (
 	// Trace is the implementation of the OpenCensus trace abstraction
 	Trace struct{}
@@ -37,11 +42,15 @@ func (t *Trace) StartSpan(ctx context.Context, operationName string, opts ...int
 //
 // Returned context contains the newly created span. You can use it to
 // propagate the returned span in process.
-func (t *Trace) StartSpanWithRemoteParent(ctx context.Context, operationName string, reference interface{}, opts ...interface{}) (context.Context, trace.Spanner) {
-	if sp, ok := reference.(oct.SpanContext); ok {
-		ctx, span := oct.StartSpanWithRemoteParent(ctx, operationName, sp, toOCOption(opts...)...)
-		return ctx, &Span{span: span}
+func (t *Trace) StartSpanWithRemoteParent(ctx context.Context, operationName string, carrier trace.Carrier, opts ...interface{}) (context.Context, trace.Spanner) {
+	keysValues := carrier.GetKeyValues()
+	if val, ok := keysValues[propagationKey]; ok {
+		if sc, ok := propagation.FromBinary(val.([]byte)); ok {
+			ctx, span := oct.StartSpanWithRemoteParent(ctx, operationName, sc)
+			return ctx, &Span{span: span}
+		}
 	}
+
 	return t.StartSpan(ctx, operationName)
 }
 
@@ -68,6 +77,12 @@ func (s *Span) Logger() trace.Logger {
 	return &trace.SpanLogger{Span: s}
 }
 
+// Inject propagation key onto the carrier
+func (s *Span) Inject(carrier trace.Carrier) error {
+	carrier.Set(propagationKey, propagation.Binary(s.span.SpanContext()))
+	return nil
+}
+
 func toOCOption(opts ...interface{}) []oct.StartOption {
 	var ocStartOptions []oct.StartOption
 	for _, opt := range opts {

opentracing/opentracing.go

@@ -18,6 +18,10 @@ type (
 	Span struct {
 		span opentracing.Span
 	}
+
+	carrierAdapter struct {
+		carrier trace.Carrier
+	}
 )
 
 // StartSpan starts and returns a Span with `operationName`, using
@@ -30,13 +34,15 @@ func (t *Trace) StartSpan(ctx context.Context, operationName string, opts ...int
 
 // StartSpanWithRemoteParent starts and returns a Span with `operationName`, using
 // reference span as FollowsFrom
-func (t *Trace) StartSpanWithRemoteParent(ctx context.Context, operationName string, reference interface{}, opts ...interface{}) (context.Context, trace.Spanner) {
-	if sp, ok := reference.(opentracing.SpanContext); ok {
-		span := opentracing.StartSpan(operationName, append(toOTOption(opts...), opentracing.FollowsFrom(sp))...)
-		ctx = opentracing.ContextWithSpan(ctx, span)
-		return ctx, &Span{span: span}
+func (t *Trace) StartSpanWithRemoteParent(ctx context.Context, operationName string, carrier trace.Carrier, opts ...interface{}) (context.Context, trace.Spanner) {
+	sc, err := opentracing.GlobalTracer().Extract(opentracing.TextMap, carrierAdapter{carrier: carrier})
+	if err != nil {
+		return t.StartSpan(ctx, operationName)
 	}
-	return t.StartSpan(ctx, operationName)
+
+	span := opentracing.StartSpan(operationName, append(toOTOption(opts...), opentracing.FollowsFrom(sc))...)
+	ctx = opentracing.ContextWithSpan(ctx, span)
+	return ctx, &Span{span: span}
 }
 
 // FromContext returns the `Span` previously associated with `ctx`, or
@@ -69,6 +75,28 @@ func (s *Span) Logger() trace.Logger {
 	return &trace.SpanLogger{Span: s}
 }
 
+// Inject span context into carrier
+func (s *Span) Inject(carrier trace.Carrier) error {
+	return opentracing.GlobalTracer().Inject(s.span.Context(), opentracing.TextMap, carrierAdapter{carrier: carrier})
+}
+
+// Set a key and value on the carrier
+func (ca *carrierAdapter) Set(key, value string) {
+	ca.carrier.Set(key, value)
+}
+
+// ForeachKey runs the handler across the map of carrier key / values
+func (ca *carrierAdapter) ForeachKey(handler func(key, val string) error) error {
+	for k, v := range ca.carrier.GetKeyValues() {
+		if vStr, ok := v.(string); ok {
+			if err := handler(k, vStr); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func toOTOption(opts ...interface{}) []opentracing.StartSpanOption {
 	var ocStartOptions []opentracing.StartSpanOption
 	for _, opt := range opts {

trace/trace.go

@@ -29,23 +29,23 @@ func Int64Attribute(key string, value int64) Attribute {
 // StartSpan starts a new child span
 func StartSpan(ctx context.Context, operationName string, opts ...interface{}) (context.Context, Spanner) {
 	if tracer == nil {
-		return ctx, new(nopSpanner)
+		return ctx, new(noOpSpanner)
 	}
 	return tracer.StartSpan(ctx, operationName, opts)
 }
 
 // StartSpanWithRemoteParent starts a new child span of the span from the given parent.
-func StartSpanWithRemoteParent(ctx context.Context, operationName string, reference interface{}, opts ...interface{}) (context.Context, Spanner) {
+func StartSpanWithRemoteParent(ctx context.Context, operationName string, carrier Carrier, opts ...interface{}) (context.Context, Spanner) {
 	if tracer == nil {
-		return ctx, new(nopSpanner)
+		return ctx, new(noOpSpanner)
 	}
-	return tracer.StartSpanWithRemoteParent(ctx, operationName, reference, opts)
+	return tracer.StartSpanWithRemoteParent(ctx, operationName, carrier, opts)
 }
 
 // FromContext returns the Span stored in a context, or nil if there isn't one.
 func FromContext(ctx context.Context) Spanner {
 	if tracer == nil {
-		return new(nopSpanner)
+		return new(noOpSpanner)
 	}
 	return tracer.FromContext(ctx)
 }
@@ -57,17 +57,24 @@ type (
 		Value interface{}
 	}
 
+	// Carrier is an abstraction over OpenTracing and OpenCensus propagation carrier
+	Carrier interface {
+		Set(key string, value interface{})
+		GetKeyValues() map[string]interface{}
+	}
+
 	// Spanner is an abstraction over OpenTracing and OpenCensus Spans
 	Spanner interface {
 		AddAttributes(attributes ...Attribute)
 		End()
 		Logger() Logger
+		Inject(carrier Carrier) error
 	}
 
 	// Tracer is an abstraction over OpenTracing and OpenCensus trace implementations
 	Tracer interface {
 		StartSpan(ctx context.Context, operationName string, opts ...interface{}) (context.Context, Spanner)
-		StartSpanWithRemoteParent(ctx context.Context, operationName string, reference interface{}, opts ...interface{}) (context.Context, Spanner)
+		StartSpanWithRemoteParent(ctx context.Context, operationName string, carrier Carrier, opts ...interface{}) (context.Context, Spanner)
 		FromContext(ctx context.Context) Spanner
 	}
 
@@ -84,20 +91,25 @@ type (
 		Span Spanner
 	}
 
-	nopLogger struct{}
+	noOpLogger struct{}
 
-	nopSpanner struct{}
+	noOpSpanner struct{}
 )
 
 // AddAttributes is a nop
-func (ns *nopSpanner) AddAttributes(attributes ...Attribute) {}
+func (ns *noOpSpanner) AddAttributes(attributes ...Attribute) {}
 
 // End is a nop
-func (ns *nopSpanner) End() {}
+func (ns *noOpSpanner) End() {}
 
 // Logger returns a nopLogger
-func (ns *nopSpanner) Logger() Logger {
-	return nopLogger{}
+func (ns *noOpSpanner) Logger() Logger {
+	return noOpLogger{}
+}
+
+// Inject is a nop
+func (ns *noOpSpanner) Inject(carrier Carrier) error {
+	return nil
 }
 
 // For will return a logger for a given context
@@ -105,7 +117,7 @@ func For(ctx context.Context) Logger {
 	if span := tracer.FromContext(ctx); span != nil {
 		return span.Logger()
 	}
-	return new(nopLogger)
+	return new(noOpLogger)
 }
 
 // Info logs an info tag with message to a span
@@ -136,13 +148,13 @@ func (sl SpanLogger) logToSpan(level string, msg string, attributes ...Attribute
 }
 
 // Info nops log entry
-func (sl nopLogger) Info(msg string, attributes ...Attribute) {}
+func (sl noOpLogger) Info(msg string, attributes ...Attribute) {}
 
 // Error nops log entry
-func (sl nopLogger) Error(err error, attributes ...Attribute) {}
+func (sl noOpLogger) Error(err error, attributes ...Attribute) {}
 
 // Fatal nops log entry
-func (sl nopLogger) Fatal(msg string, attributes ...Attribute) {}
+func (sl noOpLogger) Fatal(msg string, attributes ...Attribute) {}
 
 // Debug nops log entry
-func (sl nopLogger) Debug(msg string, attributes ...Attribute) {}
+func (sl noOpLogger) Debug(msg string, attributes ...Attribute) {}