Monitoring platform events with zipkin-go and opentracing

observer.go

@@ -0,0 +1,89 @@
+package zipkintracer
+
+import (
+	opentracing "github.com/opentracing/opentracing-go"
+)
+
+// Observer can be registered with the zipkin to recieve notifications
+// about new Spans.
+// The actual registration depends on the implementation, which might look
+// like the below e.g :
+// observer := myobserver.NewObserver()
+// tracer := zipkin.NewTracer(..., zipkin.WithObserver(observer))
+//
+type Observer interface {
+	// Create and return a span observer. Called when a span starts.
+	// E.g :
+	//     func StartSpan(opName string, opts ...opentracing.StartSpanOption) {
+	//     var sp opentracing.Span
+	//     sso := opentracing.StartSpanOptions{}
+	//     var spObs opentracing.SpanObserver = observer.OnStartSpan(span, opName, sso)
+	//     ...
+	// }
+	// OnStartSpan function needs to be defined for a package exporting
+	// metrics as well.
+	OnStartSpan(sp opentracing.Span, operationName string, options opentracing.StartSpanOptions) SpanObserver
+}
+
+// SpanObserver is created by the Observer and receives notifications about
+// other Span events.
+// zipkin should define these functions for each of the span operations
+// which should call the registered (observer) callbacks.
+type SpanObserver interface {
+	// Callback called from opentracing.Span.SetOperationName()
+	OnSetOperationName(operationName string)
+	// Callback called from opentracing.Span.SetTag()
+	OnSetTag(key string, value interface{})
+	// Callback called from opentracing.Span.Finish()
+	OnFinish(options opentracing.FinishOptions)
+}
+
+// observer is a dispatcher to other observers
+type observer struct {
+	observers []Observer
+}
+
+// spanObserver is a dispatcher to other span observers
+type spanObserver struct {
+	observers []SpanObserver
+}
+
+// noopSpanObserver is used when there are no observers registered on the
+// Tracer or none of them returns span observers
+var noopSpanObserver = spanObserver{}
+
+func (o observer) OnStartSpan(sp opentracing.Span, operationName string, options opentracing.StartSpanOptions) SpanObserver {
+	var spanObservers []SpanObserver
+	for _, obs := range o.observers {
+		spanObs := obs.OnStartSpan(sp, operationName, options)
+		if spanObs != nil {
+			if spanObservers == nil {
+				spanObservers = make([]SpanObserver, 0, len(o.observers))
+			}
+			spanObservers = append(spanObservers, spanObs)
+		}
+	}
+	if len(spanObservers) == 0 {
+		return noopSpanObserver
+	}
+
+	return spanObserver{observers: spanObservers}
+}
+
+func (o spanObserver) OnSetOperationName(operationName string) {
+	for _, obs := range o.observers {
+		obs.OnSetOperationName(operationName)
+	}
+}
+
+func (o spanObserver) OnSetTag(key string, value interface{}) {
+	for _, obs := range o.observers {
+		obs.OnSetTag(key, value)
+	}
+}
+
+func (o spanObserver) OnFinish(options opentracing.FinishOptions) {
+	for _, obs := range o.observers {
+		obs.OnFinish(options)
+	}
+}

span.go

@@ -29,6 +29,7 @@ type Span interface {
 type spanImpl struct {
 	tracer     *tracerImpl
 	event      func(SpanEvent)
+	Observer   SpanObserver
 	sync.Mutex // protects the fields below
 	raw        RawSpan
 	// The number of logs dropped because of MaxLogsPerSpan.
@@ -63,6 +64,9 @@ func (s *spanImpl) reset() {
 }
 
 func (s *spanImpl) SetOperationName(operationName string) opentracing.Span {
+	if s.Observer != nil {
+		s.Observer.OnSetOperationName(operationName)
+	}
 	s.Lock()
 	defer s.Unlock()
 	s.raw.Operation = operationName
@@ -75,6 +79,10 @@ func (s *spanImpl) trim() bool {
 
 func (s *spanImpl) SetTag(key string, value interface{}) opentracing.Span {
 	defer s.onTag(key, value)
+	if s.Observer != nil {
+		s.Observer.OnSetTag(key, value)
+	}
+
 	s.Lock()
 	defer s.Unlock()
 	if key == string(ext.SamplingPriority) {
@@ -190,6 +198,10 @@ func (s *spanImpl) FinishWithOptions(opts opentracing.FinishOptions) {
 	}
 	duration := finishTime.Sub(s.raw.Start)
 
+	if s.Observer != nil {
+		s.Observer.OnFinish(opts)
+	}
+
 	s.Lock()
 	defer s.Unlock()
 

tracer.go

@@ -107,6 +107,8 @@ type TracerOptions struct {
 	// Regardless of this setting, the library will propagate and support both
 	// 64 and 128 bit incoming traces from upstream sources.
 	traceID128Bit bool
+
+	observer Observer
 }
 
 // TracerOption allows for functional options.
@@ -228,6 +230,7 @@ func NewTracer(recorder SpanRecorder, options ...TracerOption) (opentracing.Trac
 		debugMode:                  false,
 		traceID128Bit:              false,
 		maxLogsPerSpan:             10000,
+		observer:                   nil,
 	}
 	for _, o := range options {
 		err := o(opts)
@@ -286,6 +289,11 @@ func (t *tracerImpl) startSpanWithOptions(
 	// Build the new span. This is the only allocation: We'll return this as
 	// an opentracing.Span.
 	sp := t.getSpan()
+
+	if t.options.observer != nil {
+		sp.Observer = t.options.observer.OnStartSpan(sp, operationName, opts)
+	}
+
 	// Look for a parent in the list of References.
 	//
 	// TODO: would be nice if basictracer did something with all
@@ -376,6 +384,7 @@ func (t *tracerImpl) startSpanInternal(
 	sp.raw.Start = startTime
 	sp.raw.Duration = -1
 	sp.raw.Tags = tags
+
 	if t.options.debugAssertSingleGoroutine {
 		sp.SetTag(debugGoroutineIDTag, curGoroutineID())
 	}
@@ -417,3 +426,11 @@ func (t *tracerImpl) Extract(format interface{}, carrier interface{}) (opentraci
 func (t *tracerImpl) Options() TracerOptions {
 	return t.options
 }
+
+// WithObserver assigns an initialized observer to opts.observer
+func WithObserver(observer Observer) TracerOption {
+	return func(opts *TracerOptions) error {
+		opts.observer = observer
+		return nil
+	}
+}