- finish spans

- updated readme
- _getSpanFromTChannelRequest -> _extractSpan
- _saveSpanStateToCarrier -> _injectSpan
- span -> openTracingSpan
- Remove $tracing$ prefixed headers
- change default arguments
- Fixed patching for different ways of calling send
- add tags to tchannel bridge
- add error tagging on response, and handler callbacks
- move context to request options

README.md

@@ -14,10 +14,12 @@ Please see [CONTRIBUTING.md](./CONTRIBUTING.md).
 
 ### TChannel Span Bridging
 
-Because [tchannel-node](https://github.com/uber/tchannel-node) does not have instrumentation for opentracing Jaeger-Client exposes methods wrapping tchannel handlers, and encoded channels.
+Because [tchannel-node](https://github.com/uber/tchannel-node) does not have instrumentation for OpenTracing Jaeger-Client exposes methods wrapping tchannel handlers, and encoded channels.
 An encoded channel is a channel wrapped in either a thrift encoder `TChannelAsThrift`, or json encoder `TChannelAsJson`.  To wrap a server handler for thrift one can initialize a tchannel bridge, and wrap there encoded handler function with `tracedHandler`.
 
 ```javascript
+    import { TChannelBridge } from 'jaeger-client';
+
     let bridge = new TChannelBridge(tracer);
     let server = new TChannel({ serviceName: 'server' });
     server.listen(4040, '127.0.0.1');
@@ -28,7 +30,7 @@ An encoded channel is a channel wrapped in either a thrift encoder `TChannelAsTh
 
     serverThriftChannel.register(server, 'Echo::echo', context, bridge.tracedHandler(
         (context, req, head, body, callback) => {
-            //context will be populated witha span field represents the server side span
+            // context will contain an 'openTracingSpan' field that stores the tracing span for the inbound request.
         }
     ));
 ```
@@ -37,6 +39,8 @@ An encoded channel is a channel wrapped in either a thrift encoder `TChannelAsTh
 In the case of making an outgoing request you can wrap an encoded channel in a call to `tracedChannel`.
 
 ```javascript
+    import { TChannelBridge } from 'jaeger-client';
+
     let bridge = new TChannelBridge(tracer);
     // Create the toplevel client channel.
     let client = new TChannel();
@@ -52,15 +56,18 @@ In the case of making an outgoing request you can wrap an encoded channel in a c
         entryPoint: path.join(__dirname, 'thrift', 'echo.thrift') // file path to a thrift file
     });
 
-    let tracedChannel = bridge.tracedChannel(encodedChannel, contextForOutgoingCall);
+    // The 'context' object must be passed through from the request method with the field name 'openTracingContext' to ensure an uninterrupted trace.
+    // If the context is empty, a new trace will be started for the outbound call.
+    // The 'openTracingContext' object must also have an 'openTracingSpan' field that represents the current span.
+    let tracedChannel = bridge.tracedChannel(encodedChannel);
     let req = tracedChannel.request({
         serviceName: 'server',
+        openTracingContext: { openTracingSpan: span }, // where span is the current context's span
         headers: { cn: 'echo' }
     });
 
-    // Your app should have a context that holds the incoming span, or a new span will be created.
-    let context = {};
-    req.send('Echo::echo', context, { value: 'some-string' });
+    // headers should contain your outgoing tchannel headers if any.
+    req.send('Echo::echo', headers, { value: 'some-string' });
 ```
 
 ### Debug Traces (Forced Sampling)
@@ -86,7 +93,7 @@ curl -H "jaeger-debug-id: some-correlation-id" http://myhost.com
 When Jaeger sees this header in the request that otherwise has no
 tracing context, it ensures that the new trace started for this
 request will be sampled in the "debug" mode (meaning it should survive
-all downsampling that might happen in the collection pipeline), and the 
+all downsampling that might happen in the collection pipeline), and the
 root span will have a tag as if this statement was executed:
 
 ```javascript

entrypoint.js

@@ -10,6 +10,7 @@ var NoopReporter = require('./dist/src/reporters/noop_reporter.js').default;
 var RemoteReporter = require('./dist/src/reporters/remote_reporter.js').default;
 var SpanContext = require('./dist/src/span_context.js').default;
 var TestUtils = require('./dist/src/test_util.js').default
+var TChannelBridge = require('./dist/src/tchannel_bridge.js').default;
 
 module.exports = {
     initTracer: Configuration.initTracer,
@@ -23,5 +24,6 @@ module.exports = {
     NoopReporter: NoopReporter,
     RemoteReporter: RemoteReporter,
     TestUtils: TestUtils,
-    SpanContext: SpanContext
+    SpanContext: SpanContext,
+    TChannelBridge: TChannelBridge
 };

src/constants.js

@@ -61,7 +61,3 @@ export const TRACER_BAGGAGE_HEADER_PREFIX = 'uberctx-';
 
 // TRACER_STATE_HEADER_NAME is the header key used for a span's serialized context.
 export const TRACER_STATE_HEADER_NAME = 'uber-trace-id';
-
-// TCHANNEL_TRACING_PREFIX is the tracing header prefix that we give to all
-// tracing related tchannel headers.
-export const TCHANNEL_TRACING_PREFIX = '$tracing$';

src/propagators/text_map_codec.js

@@ -90,7 +90,7 @@ export default class TextMapCodec {
         return spanContext;
     }
 
-    inject(spanContext: SpanContext, carrier: any): void {
+    inject(spanContext: SpanContext, carrier: any = {}): void {
         let stringSpanContext = spanContext.toString();
         carrier[this._contextKey] = this._encodedValue(stringSpanContext);
 

src/tchannel_bridge.js

@@ -1,4 +1,3 @@
-// @flow
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -23,9 +22,12 @@ import * as constants from './constants';
 import Span from './span';
 import SpanContext from './span_context';
 import Utils from './util';
+import opentracing from 'opentracing';
 import Tracer from '../src/tracer';
 import TextMapCodec from '../src/propagators/text_map_codec';
 
+let TCHANNEL_TRACING_PREFIX = '$tracing$';
+
 export default class TChannelBridge {
     _tracer: Tracer;
     _codec: TextMapCodec;
@@ -34,11 +36,20 @@ export default class TChannelBridge {
         this._tracer = tracer;
         this._codec = new TextMapCodec({
             urlEncoding: false,
-            contextKey: constants.TCHANNEL_TRACING_PREFIX + constants.TRACER_STATE_HEADER_NAME,
-            baggagePrefix: constants.TCHANNEL_TRACING_PREFIX + constants.TRACER_BAGGAGE_HEADER_PREFIX
+            contextKey: TCHANNEL_TRACING_PREFIX + constants.TRACER_STATE_HEADER_NAME,
+            baggagePrefix: TCHANNEL_TRACING_PREFIX + constants.TRACER_BAGGAGE_HEADER_PREFIX
         });
     }
 
+    _tchannelCallbackWrapper(span, wrappedCallback, err, res) {
+        if (err) {
+            span.setTag(opentracing.Tags.ERROR, true);
+        }
+
+        span.finish();
+        return wrappedCallback(err, res);
+    }
+
     /**
      * Wraps a tchannel handler, and takes a context in order to populate the incoming context
      * with a span.
@@ -49,15 +60,62 @@ export default class TChannelBridge {
      * a the handler's context with a span.
      **/
     tracedHandler(handlerFunc: any, options: startSpanArgs = {}): Function {
-        return (context, request, headers, body, callback) => {
-            let operationName = request.arg1 || options.operationName;
-            let span = this._getSpanFromTChannelRequest(operationName, headers);
+        return (context, request, headers, body, wrappedCallback) => {
+            let operationName = options.operationName || request.arg1;
+            let span = this._extractSpan(operationName, headers);
+            span.setTag(opentracing.Tags.SPAN_KIND, opentracing.Tags.SPAN_KIND_RPC_SERVER);
+            span.setTag(opentracing.Tags.PEER_SERVICE, request.callerName);
+            let hostPort = request.remoteAddr.split(':');
+            span.setTag(opentracing.Tags.PEER_HOST_IPV4, Utils.ipToInt(hostPort[0]));
+            span.setTag(opentracing.Tags.PEER_PORT, parseInt(hostPort[1]));
+            if (request.headers && request.headers.as) {
+                span.setTag('as', request.headers.as);
+            }
             // In theory may overwrite tchannel span, but thats what we want anyway.
-            context.span = span;
+            context.openTracingSpan = span;
+
+            // remove headers prefixed with $tracing$
+            for (let key in headers) {
+                if (headers.hasOwnProperty(key) && Utils.startsWith(key, TCHANNEL_TRACING_PREFIX)) {
+                    delete headers[key];
+                }
+            }
+
+            let callback = this._tchannelCallbackWrapper.bind(null, span, wrappedCallback);
             handlerFunc(context, request, headers, body, callback);
         };
     }
 
+    _wrapTChannelSend(wrappedSend, channel, req, endpoint, headers, body, wrappedCallback) {
+        headers = headers || {};
+        let context = req.openTracingContext || {};
+        // if opentracingContext.openTracingSpan is null, then start a new root span
+        // else start a span that is the child of the context span.
+        let childOf = context.openTracingSpan;
+        context.openTracingSpan = this._tracer.startSpan(endpoint, {
+            childOf: childOf
+        });
+        context.openTracingSpan.setTag(opentracing.Tags.PEER_SERVICE, req.serviceName);
+        context.openTracingSpan.setTag(opentracing.Tags.SPAN_KIND, opentracing.Tags.SPAN_KIND_RPC_CLIENT);
+        headers = this._injectSpan(context.openTracingSpan, headers);
+
+        // wrap callback so that span can be finished as soon as the response is received
+        let callback = this._tchannelCallbackWrapper.bind(null, context.openTracingSpan, wrappedCallback);
+
+        return wrappedSend.call(channel, req, endpoint, headers, body, callback);
+    };
+
+    _wrapTChannelRequest(channel, wrappedRequest, requestOptions) {
+        // We set the parent to a span with trace_id zero, so that tchannel's
+        // outgoing tracing frame also has a trace id of zero.
+        // This forces other tchannel implementations to rely on the headers for the trace context.
+        requestOptions.parent = { span: this._getTChannelParentSpan() };
+
+        let tchannelRequest = wrappedRequest.call(channel, requestOptions);
+        tchannelRequest.openTracingContext = requestOptions.openTracingContext;
+        return tchannelRequest;
+    }
+
     /**
      * A function that wraps a json, or thrift encoded channel, in order to populate
      * the outgoing headers with trace context, and baggage information.
@@ -66,23 +124,14 @@ export default class TChannelBridge {
      * @param {Object} context - A context that contains the outgoing span to be seralized.  If the context does not contain a span then a new root span is created.
      * @returns {Object} channel - the trace wrapped channel.
      * */
-    tracedChannel(channel: any, context: any = {}): any {
-        let wrappedRequestFunc = channel.request.bind(channel);
-        channel.request = (requestOptions) => {
-            requestOptions.parent = { span: this._getTChannelParentSpan() };
-            let channelRequest = wrappedRequestFunc(requestOptions);
-            let wrappedSend = channelRequest.send.bind(channelRequest);
-            channelRequest.send = (endpoint, headers = {}, body, callback) => {
-                if (!context.span) {
-                    context.span = this._tracer.startSpan(endpoint);
-                }
-                this._saveSpanStateToCarrier(context.span, headers);
-                return wrappedSend(endpoint, headers, body, callback);
-            };
+    tracedChannel(channel: any): any {
+        let wrappedSend = channel.send;
+        let wrappedRequest = channel.channel.request;
 
-            return channelRequest;
-        };
+        // We are patching the top level channel request method, not the encoded request method.
+        channel.channel.request = this._wrapTChannelRequest.bind(this, channel.channel, wrappedRequest);
 
+        channel.send = this._wrapTChannelSend.bind(this, wrappedSend, channel);
         return channel;
     }
 
@@ -95,14 +144,16 @@ export default class TChannelBridge {
         };
     }
 
-    _getSpanFromTChannelRequest(operationName: string, headers: any = {}, options: any = {}): Span {
+    _extractSpan(operationName: string, headers: any): Span {
         let traceContext: ?SpanContext = this._codec.extract(headers);
-        options.childOf = traceContext;
+        let options = {
+            childOf: traceContext
+        }
         let span = this._tracer.startSpan(operationName, options);
         return span;
     }
 
-    _saveSpanStateToCarrier(span: Span, carrier: any = {}): any {
+    _injectSpan(span: Span, carrier: any): any {
         this._codec.inject(span.context(), carrier);
         return carrier;
     }

src/test_util.js

@@ -81,7 +81,13 @@ export default class TestUtils {
         }
 
         for (let tag in tags) {
-            if (tags.hasOwnProperty(tag) && (!(tag in expectedTags))) {
+            if (tags.hasOwnProperty(tag) && expectedTags.hasOwnProperty(tag)) {
+                if (expectedTags[tag] !== tags[tag]) {
+                    console.log('expected tag:', expectedTags[tag], ', actual tag: ', tags[tag]);
+                    return false;
+                }
+            } else {
+                // mismatch in tag keys
                 return false;
             }
         }

test/entrypoint.js

@@ -35,5 +35,6 @@ describe('entrypoint', () => {
         expect(entrypoint.RemoteReporter).to.be.a('function');
         expect(entrypoint.TestUtils).to.be.a('function');
         expect(entrypoint.SpanContext).to.be.a('function');
+        expect(entrypoint.TChannelBridge).to.be.a('function');
     });
 });