add propagation interface

packages/zipkin-instrumentation-express/src/wrapExpressHttpProxy.js

@@ -1,4 +1,4 @@
-const {Request, Annotation} = require('zipkin');
+const {Annotation} = require('zipkin');
 const url = require('url');
 
 function getPathnameFromPath(path) {
@@ -18,7 +18,7 @@ class ExpressHttpProxyInstrumentation {
       const clientTraceId = this.tracer.createChildId();
       this.tracer.setId(clientTraceId);
 
-      const proxyReqWithZipkinHeaders = Request.addZipkinHeaders(proxyReq, clientTraceId);
+      const proxyReqWithZipkinHeaders = this.tracer.injector(proxyReq);
       Object.defineProperty(serverReq, '_trace_id_proxy',
         {configurable: false, get: () => clientTraceId});
 

packages/zipkin-instrumentation-kafkajs/src/zipkin-instrumentation-kafkajs.js

@@ -1,4 +1,3 @@
-const {Request} = require('zipkin');
 const {
   recordConsumeStop, recordConsumeStart,
   recordProducerStart, recordProducerStop
@@ -49,7 +48,7 @@ const instrumentKafkaJs = (kafkaJs, {tracer, remoteServiceName}) => {
             id = recordProducerStart(tracer, 'send', remoteServiceName, {topic: params.topic});
 
             const withTraceHeaders = Object.assign({}, params, {
-              messages: params.messages.map(msg => Request.addZipkinHeaders(msg, id))
+              messages: params.messages.map(msg => tracer.injector(msg))
             });
 
             promise = obj[prop](withTraceHeaders);

packages/zipkin/index.d.ts

@@ -39,7 +39,8 @@ declare namespace zipkin {
       localServiceName?: string,
       localEndpoint?: model.Endpoint,
       log?: Console,
-      defaultTags?: {}
+      defaultTags?: {},
+      propagation?: propagation.Propagation
     });
 
     /** Returns the current trace ID or a sentinel value indicating its absence. */
@@ -61,6 +62,11 @@ declare namespace zipkin {
     recordLocalAddr(inetAddress: InetAddress): void;
     recordBinary(key: string, value: boolean | string | number): void;
     writeIdToConsole(message: any): void;
+    /** Extract propagation ctx from request */
+    extractId(readHeader: <T> (header: string) => option.IOption<T>): void;
+    /** Injector propagation ctx from request */
+    injector(request: any): object;
+
   }
 
   class TraceId {
@@ -271,14 +277,6 @@ declare namespace zipkin {
     toInt(): number;
   }
 
-  namespace HttpHeaders {
-    const TraceId: string;
-    const SpanId: string;
-    const ParentSpanId: string;
-    const Sampled: string;
-    const Flags: string;
-  }
-
   interface Record {
     traceId: TraceId;
     timestamp: number;
@@ -339,6 +337,7 @@ declare namespace zipkin {
   }
 
   namespace Instrumentation {
+
     class HttpServer {
       constructor(args: {
         tracer: Tracer,
@@ -357,7 +356,7 @@ declare namespace zipkin {
     }
 
     class HttpClient {
-      constructor(args: { tracer: Tracer, serviceName?: string, remoteServiceName?: string });
+      constructor(args: { tracer: Tracer, serviceName?: string, remoteServiceName?: string});
 
       recordRequest<T>(
         request: T,
@@ -368,6 +367,16 @@ declare namespace zipkin {
       recordError(traceId: TraceId, error: Error): void;
     }
   }
+  namespace propagation {
+      interface Propagation {
+        extractor<T>(tracer: Tracer, readHeader: <T> (header: string) => option.IOption<T>): TraceId;
+        injector<T, H>(req: T & { headers?: any }, traceId: TraceId): object;
+      }
+      class B3Propagation implements Propagation {
+        extractor<T>(tracer: Tracer, readHeader: <T> (header: string) => option.IOption<T>): TraceId;
+        injector<T, H>(req: T & { headers?: any }, traceId: TraceId): object;
+    }
+  }
 }
 
 export = zipkin;

packages/zipkin/src/index.ts

@@ -7,7 +7,6 @@ export { default as randomTraceId } from './tracer/randomTraceId';
 export { default as sampler } from './tracer/sampler';
 export { default as TraceId } from './tracer/TraceId';
 
-export { default as HttpHeaders } from './httpHeaders';
 export { default as InetAddress } from './InetAddress';
 
 export { default as BatchRecorder } from './batch-recorder';
@@ -16,7 +15,7 @@ export { default as ConsoleRecorder } from './console-recorder';
 export { default as ExplicitContext } from './explicit-context';
 
 export { default as Instrumentation } from './instrumentation';
-export { default as Request } from './request';
+export { default as B3Propagation } from './propagation/b3propagation';
 
 export { default as jsonEncoder } from './jsonEncoder';
 export { default as model } from './model';

packages/zipkin/src/instrumentation/httpClient.js

@@ -1,5 +1,4 @@
 const Annotation = require('../annotation');
-const Request = require('../request');
 const parseRequestUrl = require('../parseUrl');
 
 function requiredArg(name) {
@@ -19,7 +18,6 @@ class HttpClientInstrumentation {
 
   recordRequest(request, url, method) {
     this.tracer.setId(this.tracer.createChildId());
-    const traceId = this.tracer.id;
     const {path} = parseRequestUrl(url);
 
     this.tracer.recordServiceName(this.serviceName);
@@ -33,8 +31,7 @@ class HttpClientInstrumentation {
         serviceName: this.remoteServiceName
       }));
     }
-
-    return Request.addZipkinHeaders(request, traceId);
+    return this.tracer.injector(request);
   }
 
   recordResponse(traceId, statusCode) {

packages/zipkin/src/instrumentation/httpServer.js

@@ -1,25 +1,6 @@
 const Annotation = require('../annotation');
-const Header = require('../httpHeaders');
 const InetAddress = require('../InetAddress');
-const TraceId = require('../tracer/TraceId');
 const parseRequestUrl = require('../parseUrl');
-const {Some, None} = require('../option');
-
-function stringToBoolean(str) {
-  return str === '1' || str === 'true';
-}
-
-function stringToIntOption(str) {
-  try {
-    return new Some(parseInt(str));
-  } catch (err) {
-    return None;
-  }
-}
-
-function containsRequiredHeaders(readHeader) {
-  return readHeader(Header.TraceId) !== None && readHeader(Header.SpanId) !== None;
-}
 
 function requiredArg(name) {
   throw new Error(`HttpServerInstrumentation: Missing required argument ${name}.`);
@@ -38,34 +19,6 @@ class HttpServerInstrumentation {
     this.port = port;
   }
 
-  _createIdFromHeaders(readHeader) {
-    if (containsRequiredHeaders(readHeader)) {
-      const spanId = readHeader(Header.SpanId);
-      const parentId = spanId.map((sid) => {
-        const traceId = readHeader(Header.TraceId);
-        const parentSpanId = readHeader(Header.ParentSpanId);
-        const sampled = readHeader(Header.Sampled);
-        const flags = readHeader(Header.Flags).flatMap(stringToIntOption).getOrElse(0);
-        return new TraceId({
-          traceId: traceId.getOrElse(),
-          parentId: parentSpanId,
-          spanId: sid,
-          debug: flags === 1,
-          sampled: sampled.map(stringToBoolean),
-        });
-      });
-
-      return new Some(this.tracer.join(parentId.getOrElse()));
-    } else if (readHeader(Header.Flags) !== None || readHeader(Header.Sampled) !== None) {
-      const sampled = readHeader(Header.Sampled) === None
-        ? None : readHeader(Header.Sampled).map(stringToBoolean);
-      const flags = readHeader(Header.Flags).flatMap(stringToIntOption).getOrElse(0);
-      return new Some(this.tracer.createRootId(sampled, flags === 1));
-    } else {
-      return new Some(this.tracer.createRootId());
-    }
-  }
-
   spanNameFromRoute(method, route, code) { // eslint-disable-line class-methods-use-this
     if (code > 299 && code < 400) return `${method} redirected`;
     if (code === 404) return `${method} not_found`;
@@ -74,7 +27,7 @@ class HttpServerInstrumentation {
   }
 
   recordRequest(method, requestUrl, readHeader) {
-    this._createIdFromHeaders(readHeader).ifPresent(id => this.tracer.setId(id));
+    this.tracer.extractId(readHeader);
     const {id} = this.tracer;
     const {path} = parseRequestUrl(requestUrl);
 

packages/zipkin/src/propagation/b3propagation.js

@@ -0,0 +1,75 @@
+const {Some, None} = require('../option');
+const TraceId = require('../tracer/TraceId');
+
+function stringToBoolean(str) {
+  return str === '1' || str === 'true';
+}
+
+function stringToIntOption(str) {
+  try {
+    return new Some(parseInt(str));
+  } catch (err) {
+    return None;
+  }
+}
+
+class B3Propagation {
+  constructor() {
+    this.headers = {
+      TraceId: 'X-B3-TraceId',
+      SpanId: 'X-B3-SpanId',
+      ParentSpanId: 'X-B3-ParentSpanId',
+      Sampled: 'X-B3-Sampled',
+      Flags: 'X-B3-Flags'
+    };
+  }
+
+  extractor(tracer, readHeader) {
+    if (readHeader(this.headers.TraceId) !== None && readHeader(this.headers.SpanId) !== None) {
+      const spanId = readHeader(this.headers.SpanId);
+      const parentId = spanId.map((sid) => {
+        const traceId = readHeader(this.headers.TraceId);
+        const parentSpanId = readHeader(this.headers.ParentSpanId);
+        const sampled = readHeader(this.headers.Sampled);
+        const flags = readHeader(this.headers.Flags).flatMap(stringToIntOption).getOrElse(0);
+        return new TraceId({
+          traceId: traceId.getOrElse(),
+          parentId: parentSpanId,
+          spanId: sid,
+          debug: flags === 1,
+          sampled: sampled.map(stringToBoolean),
+        });
+      });
+
+      return new Some(tracer.join(parentId.getOrElse()));
+    } else if (readHeader(this.headers.Flags) !== None
+      || readHeader(this.headers.Sampled) !== None) {
+      const sampled = readHeader(this.headers.Sampled) === None
+        ? None : readHeader(this.headers.Sampled).map(stringToBoolean);
+      const flags = readHeader(this.headers.Flags).flatMap(stringToIntOption).getOrElse(0);
+      return new Some(tracer.createRootId(sampled, flags === 1));
+    } else {
+      return new Some(tracer.createRootId());
+    }
+  }
+
+  injector(request, traceId) {
+    const headers = request.headers || {};
+    headers[this.headers.TraceId] = traceId.traceId;
+    headers[this.headers.SpanId] = traceId.spanId;
+
+    traceId.parentSpanId.ifPresent((psid) => {
+      headers[this.headers.ParentSpanId] = psid;
+    });
+    traceId.sampled.ifPresent((sampled) => {
+      headers[this.headers.Sampled] = sampled ? '1' : '0';
+    });
+
+    if (traceId.isDebug()) {
+      headers[this.headers.Flags] = '1';
+    }
+    return headers;
+  }
+}
+
+module.exports = B3Propagation;

packages/zipkin/src/propagation/index.js

@@ -0,0 +1,5 @@
+const B3Propagation = require('./b3propagation');
+
+module.exports = {
+  B3Propagation
+};

packages/zipkin/src/tracer/index.js

@@ -1,3 +1,4 @@
+
 const isPromise = require('is-promise');
 const {None, Some} = require('../option');
 const {Sampler, alwaysSample} = require('./sampler');
@@ -8,6 +9,7 @@ const TraceId = require('./TraceId');
 const randomTraceId = require('./randomTraceId');
 const {now, hrtime} = require('../time');
 const {Endpoint} = require('../model');
+const {B3Propagation} = require('../propagation');
 
 
 function requiredArg(name) {
@@ -37,13 +39,15 @@ class Tracer {
     localEndpoint,
     /* eslint-disable no-console */
     log = console,
-    defaultTags
+    defaultTags,
+    propagation = new B3Propagation()
   }) {
     this.log = log;
     this.recorder = recorder;
     this.sampler = sampler;
     this.traceId128Bit = traceId128Bit;
     this.supportsJoin = supportsJoin;
+    this._propagation = propagation;
     if (localEndpoint) {
       this._localEndpoint = localEndpoint;
     } else {
@@ -265,6 +269,15 @@ class Tracer {
       }
     }
   }
+
+  extractId(readHeader) {
+    this._propagation.extractor(this, readHeader).ifPresent(id => this.setId(id));
+  }
+
+  injector(request) {
+    const headers = this._propagation.injector(request, this.id);
+    return Object.assign({}, request, {headers});
+  }
 }
 
 module.exports = Tracer;