Merge pull request #25 from ExpediaDotCom/support-single-dual-span-modes

adding single(sharable) and dual-span(non-sharable) support, default …

Author: keshavpeswani

.gitignore

@@ -65,3 +65,4 @@ src/proto_idl_codegen/
 dist/
 .idea/typescript-compiler.xml
 *.iml
+ws2/

examples/index.js

@@ -45,6 +45,9 @@ const config = {
         // or
 
         // type: 'haystack_agent'
+
+        // or
+        // type: 'http_collector'
     },
     logger: logger
 };

integration-tests/docker-compose.yml

@@ -14,6 +14,15 @@ services:
     - /bin/sh
     - -c
     - 'sleep 10 && java -jar /app/bin/haystack-agent.jar --config-provider file --file-path /app/bin/dev.conf'
+   haystack_collector:
+    image: expediadotcom/haystack-http-span-collector:1.1
+    depends_on:
+    - zookeeper
+    - kafkasvc
+    environment:
+      HAYSTACK_PROP_KAFKA_PRODUCER_PROPS_BOOTSTRAP_SERVERS: "kafkasvc:9092"
+    ports:
+    - "8080:8080"
    zookeeper:
     image: wurstmeister/zookeeper
     ports:

src/dispatchers/http_collector.ts

@@ -27,11 +27,11 @@ export default class HttpCollectorDispatcher implements Dispatcher {
     _headers: { [key: string]: any };
     _logger: Logger;
 
-    constructor(_collectorUrl: string = 'http://localhost:8080/span', headers: { [key: string]: any } = {}, logger: Logger = new NullLogger()) {
-        this._collectorUrl = _collectorUrl;
-        this._headers = headers;
+    constructor(collectorUrl: string = 'http://haystack-collector:8080/span', headers: { [key: string]: any } = {}, logger: Logger = new NullLogger()) {
+        this._collectorUrl = collectorUrl;
+        this._headers = headers || {};
         this._logger = logger;
-        this._logger.info(`Initializing the http collector dispatcher, connecting at ${_collectorUrl}`);
+        this._logger.info(`Initializing the http collector dispatcher, connecting at ${this._collectorUrl}`);
     }
 
     name(): string {

src/span_context.ts

@@ -22,17 +22,20 @@ export default class SpanContext extends opentracing.SpanContext {
     spanId: string;
     parentSpanId: string;
     baggage: any;
+    extractedContext: boolean;
 
     constructor(
         traceId: string,
         spanId: string,
         parentSpanId: string,
-        baggage = {}) {
+        baggage = {},
+        extractedContext: boolean = false) {
         super();
         this.traceId = traceId;
         this.spanId = spanId;
         this.parentSpanId = parentSpanId;
         this.baggage = baggage;
+        this.extractedContext = extractedContext;
     }
 
     setTraceId(traceId: string): void {
@@ -56,7 +59,15 @@ export default class SpanContext extends opentracing.SpanContext {
             newBaggage);
     }
 
+    setExtractedContext(): void {
+        this.extractedContext = true;
+    }
+
     isValid(): boolean {
         return !!(this.traceId && this.spanId);
     }
+
+    isExtractedContext(): boolean {
+        return this.extractedContext;
+    }
 }

src/start_span_fields.ts

@@ -23,5 +23,4 @@ export default class StartSpanFields implements opentracing.SpanOptions {
     references?: opentracing.Reference[];
     tags?: any;
     startTime?: number;
-    callerSpanContext?: SpanContext;
 }

src/tracer.ts

@@ -38,12 +38,14 @@ export default class Tracer extends opentracing.Tracer {
     _commonTags: { [key: string]: any };
     _logger: any;
     _registry: PropagationRegistry;
+    _useDualSpanMode: boolean;
 
     constructor(serviceName: string,
                 dispatcher: Dispatcher = new NoopDispatcher(),
                 commonTags: { [key: string]: any } = {},
                 logger: Logger = new NullLogger(),
-                idGenerator: Generator = new UUIDGenerator()) {
+                idGenerator: Generator = new UUIDGenerator(),
+                useDualSpanMode: boolean = false) {
         super();
         this._commonTags = commonTags || {};
         this._serviceName = serviceName;
@@ -54,6 +56,7 @@ export default class Tracer extends opentracing.Tracer {
         this._registry.register(opentracing.FORMAT_BINARY, new BinaryPropagator());
         this._registry.register(opentracing.FORMAT_HTTP_HEADERS, new TextMapPropagator(new URLCodex()));
         this._idGenerator = idGenerator;
+        this._useDualSpanMode = useDualSpanMode;
     }
 
     startSpan(operationName: string, fields?: StartSpanFields): Span {
@@ -87,7 +90,7 @@ export default class Tracer extends opentracing.Tracer {
             }
         }
 
-        const ctx = this._createSpanContext(parent, fields.callerSpanContext);
+        const ctx = this._createSpanContext(parent, spanTags);
         return this._spanStart(operationName, ctx, startTime, references, spanTags);
     }
 
@@ -102,16 +105,29 @@ export default class Tracer extends opentracing.Tracer {
         return span;
     }
 
-    _createSpanContext(parent: SpanContext, callerContext: SpanContext): SpanContext {
+    private isServerSpan(spanTags: { [key: string]: any }): boolean {
+        const spanKind = spanTags[opentracing.Tags.SPAN_KIND];
+        return spanKind && (spanKind === 'server');
+    }
+
+    // This is a check to see if the tracer is configured to support single
+    // single span type (Zipkin style shared span id) or
+    // dual span type (client and server having their own span ids ).
+    // a. If tracer is not of dualSpanType and if it is a server span then we
+    // just return the parent context with the same shared span ids
+    // b. If tracer is not of dualSpanType and if the parent context is an extracted one from the wire
+    // then we assume this is the first span in the server and so just return the parent context
+    // with the same shared span ids
+    private _createSpanContext(parent: SpanContext, spanTags: { [key: string]: any }): SpanContext {
         if (!parent || !parent.isValid) {
-            if (callerContext) {
-                return new SpanContext(callerContext.traceId, callerContext.spanId, callerContext.parentSpanId, callerContext.baggage);
+            const parentBaggage = parent && parent.baggage;
+            return new SpanContext(this._idGenerator.generate(), this._idGenerator.generate(), '', parentBaggage);
+        } else {
+            if (!this._useDualSpanMode && (this.isServerSpan(spanTags) || parent.isExtractedContext())) {
+                return new SpanContext(parent.traceId, parent.spanId, parent.parentSpanId, parent.baggage);
             } else {
-                const parentBaggage = parent && parent.baggage;
-                return new SpanContext(this._idGenerator.generate(), this._idGenerator.generate(), parentBaggage);
+                return new SpanContext(parent.traceId, this._idGenerator.generate(), parent.spanId, parent.baggage);
             }
-        } else {
-            return new SpanContext(parent.traceId, this._idGenerator.generate(), parent.spanId, parent.baggage);
         }
     }
 
@@ -165,7 +181,11 @@ export default class Tracer extends opentracing.Tracer {
             throw new Error('extractor is not supported for format=' + format);
         }
 
-        return propagator.extract(carrier);
+        const ctx = propagator.extract(carrier);
+        if (ctx) {
+            ctx.setExtractedContext();
+        }
+        return ctx;
     }
 
     static initTracer(config: TracerConfig): opentracing.Tracer {

tests/integration/agent-integration.spec.ts

@@ -16,10 +16,10 @@
 
 import Tracer from '../../src/tracer';
 import * as opentracing from 'opentracing';
-import SpanContext from '../../src/span_context';
 import * as kafka from 'kafka-node';
 import { expect } from 'chai';
 import { Logger } from '../../src/logger'
+const messages  = require('../../src/proto_idl_codegen/span_pb');
 
 class ConsoleLogger implements Logger {
     log(msg) { console.log(msg); }
@@ -29,28 +29,24 @@ class ConsoleLogger implements Logger {
     error(msg: string): void { this.log(msg); }
 }
 
-describe('Haystack Integration Tests', () => {
-    describe('Tracer Test', () => {
-        return it("should generate spans and push to haystack-agent", function(done) {
-                this.timeout(6000);
-                const TraceId = '1848fadd-fa16-4b3e-8ad1-6d73339bbee7'
-                const tracer = Tracer.initTracer({
-                    serviceName: 'my-service',
-                    commonTags: {
-                        'my-service-version': '0.1.0'
-                    },
-                    dispatcher: {
-                        type: 'haystack_agent',
-                        agentHost: 'haystack_agent'
-                    },
-                    logger: new ConsoleLogger(),
-                });
+const TraceId = '1848fadd-fa16-4b3e-8ad1-6d73339bbee7';
+const SpanId = '7a7cc5bf-796e-4527-9b42-13ae5766c6fd';
+const ParentSpanId = 'e96de653-ad6e-4ad5-b437-e81fd9d2d61d';
+
+const options = {
+    groupId: 'integration-test',
+    kafkaHost: 'kafkasvc:9092',
+    fromOffset: 'earliest' as ("earliest" | "latest" | "none"),
+    encoding: 'buffer',
+    keyEncoding: 'utf8'
+};
 
+const consumer = new kafka.ConsumerGroup(options, 'proto-spans');
+
+const executeTest = (consumer: kafka.ConsumerGroup, tracer: opentracing.Tracer, done) => {
+    const carrier = {'Trace-ID': TraceId , 'Span-ID': SpanId, 'Parent-ID': ParentSpanId, 'Baggage-myKey': 'myVal'};
                 const serverSpan = tracer.startSpan('my-operation', {
-                    childOf: new SpanContext(
-                        TraceId,
-                        '7a7cc5bf-796e-4527-9b42-13ae5766c6fd',
-                        'e96de653-ad6e-4ad5-b437-e81fd9d2d61d')
+                    childOf: tracer.extract(opentracing.FORMAT_TEXT_MAP, carrier)
                 })
                 .setTag(opentracing.Tags.SPAN_KIND, 'server')
                 .setTag(opentracing.Tags.HTTP_METHOD, 'GET');
@@ -70,30 +66,75 @@ describe('Haystack Integration Tests', () => {
 
                 var serverSpanReceived = 0;
                 var clientSpanReceived = 0;
-                    
-                const options = {
-                    groupId: 'integration-test',
-                    kafkaHost: 'kafkasvc:9092',
-                    fromOffset: 'earliest' as ("earliest" | "latest" | "none")
-                };
-                const consumer = new kafka.ConsumerGroup(options, 'proto-spans');
+
                 consumer.on('message', (kafkaMessage) => {
                     expect(kafkaMessage.key).eq(TraceId);
 
                     const spanBuffer = kafkaMessage.value as Buffer;
-                    //TODO: figure out why Buffer is failing to deserialize into a proto object
-                    if (spanBuffer.includes('7a7cc5bf-796e-4527-9b42-13ae5766c6fd')) {
+                    const protoSpan = messages.Span.deserializeBinary(spanBuffer);
+                    
+                    var isServerSpan = false;
+                    protoSpan.getTagsList().forEach((tag) => {
+                        if (tag.getKey() === 'span.kind' && tag.getVstr() === 'server') {
+                            isServerSpan = true;
+                        }
+                    });
+                    if (isServerSpan) {
                         serverSpanReceived = serverSpanReceived + 1;
+                        expect(protoSpan.getTraceid()).eq(TraceId);
+                        expect(protoSpan.getSpanid()).eq(SpanId);
+                        expect(protoSpan.getParentspanid()).eq(ParentSpanId);
                     } else {
                         clientSpanReceived = clientSpanReceived + 1;
-                    }                    
+                        expect(protoSpan.getTraceid()).eq(TraceId);
+                        expect(protoSpan.getSpanid() === SpanId).eq(false);
+                        expect(protoSpan.getParentspanid()).eq(SpanId);
+                    }
                 });
 
                 setTimeout(() => {
                     expect(serverSpanReceived).eq(1);
                     expect(clientSpanReceived).eq(1);
                     done();
                 }, 5000);
+            }
+
+describe('Haystack Integration Tests', () => {
+    describe('Tracer Test with haystack agent', () => {
+        return it("should generate spans and push to haystack-agent", function(done) {
+                this.timeout(6000);
+            
+                const tracer = Tracer.initTracer({
+                    serviceName: 'my-service',
+                    commonTags: {
+                        'my-service-version': '0.1.0'
+                    },
+                    dispatcher: {
+                        type: 'haystack_agent',
+                        agentHost: 'haystack_agent'
+                    },
+                    logger: new ConsoleLogger(),
+                });
+                executeTest(consumer, tracer, done);
+            
+            });
+    });
+
+    describe('Tracer Test with haystack collector', () => {
+        return it("should generate spans and push to haystack-collector", function(done) {
+            this.timeout(6000);
+            const tracer = Tracer.initTracer({
+                serviceName: 'my-service',
+                commonTags: {
+                    'my-service-version': '0.1.0'
+                },
+                dispatcher: {
+                    type: 'http_collector',
+                    collectorUrl: 'http://haystack_collector:8080/span'
+                },
+                logger: new ConsoleLogger(),
             });
+            executeTest(consumer, tracer, done);
+        });
     });
 });