Merge pull request #24 from ExpediaDotCom/http-collector-dispatcher

adding support for http collector dispatcher

Author: keshavpeswani

package.json

@@ -20,13 +20,15 @@
     "google-protobuf": "^3.0.0",
     "grpc": "^1.9.0",
     "opentracing": "^0.14.0",
-    "uuid": "^3.2.1"
+    "uuid": "^3.2.1",
+    "request": "^2.88.0"
   },
   "devDependencies": {
     "@types/node": "7.0.12",
     "@types/chai": "3.4.35",
     "@types/mocha": "2.2.40",
     "@types/kafka-node": "2.0.7",
+    "@types/request": "2.48.0",
     "kafka-node": "2.6.1",
     "chai": "^3.5.0",
     "grpc-tools": "1.6.6",

src/configuration.ts

@@ -15,10 +15,11 @@
  */
 
 import FileDispatcher from './dispatchers/file';
-import RemoteDispatcher from './dispatchers/remote';
-import {Dispatcher} from './dispatchers/dispatcher';
+import RemoteGrpcAgentDispatcher from './dispatchers/grpc_agent';
+import { Dispatcher } from './dispatchers/dispatcher';
 import InMemoryDispatcher from './dispatchers/in_memory';
 import NoopDispatcher from './dispatchers/noop';
+import HttpCollectorDispatcher from './dispatchers/http_collector';
 import { DispatcherConfig } from './dispatchers/dispatcher-config';
 import { TracerConfig } from './tracer-config';
 
@@ -31,7 +32,9 @@ export default class Configuration {
                 case 'file':
                     return new FileDispatcher(dispatcher.filePath);
                 case 'haystack_agent':
-                    return new RemoteDispatcher(dispatcher.agentHost, dispatcher.agentPort, config.logger);
+                    return new RemoteGrpcAgentDispatcher(dispatcher.agentHost, dispatcher.agentPort, config.logger);
+                case 'http_collector':
+                    return new HttpCollectorDispatcher(dispatcher.collectorUrl, dispatcher.collectorHttpHeaders, config.logger);
                 case 'in_memory':
                     return new InMemoryDispatcher();
                 default:

src/dispatchers/dispatcher-config.ts

@@ -19,4 +19,6 @@ export interface DispatcherConfig {
     filePath?: string;
     agentHost?: string;
     agentPort?: number;
+    collectorUrl?: string;
+    collectorHttpHeaders?: { [key: string]: any };
 }

src/dispatchers/remote.ts renamed to src/dispatchers/grpc_agent.ts

@@ -26,13 +26,13 @@ export default class RemoteDispatcher implements Dispatcher {
     _logger: any;
 
     constructor(agentHost: string = 'haystack-agent', agentPort: number = 35000, logger: Logger = new NullLogger()) {
-        logger.info(`Initializing the remote dispatcher, connecting at ${agentHost}:${agentPort}`);
+        logger.info(`Initializing the remote grpc agent dispatcher, connecting at ${agentHost}:${agentPort}`);
         this._client = new services.SpanAgentClient(`${agentHost}:${agentPort}`, grpc.credentials.createInsecure());
         this._logger = logger;
     }
 
     name(): string {
-        return 'RemoteDispatcher';
+        return 'GrpcAgentDispatcher';
     }
 
     dispatch(span: Span, callback: (error) => void): void {

src/dispatchers/http_collector.ts

@@ -0,0 +1,64 @@
+
+/*
+ *  Copyright 2018 Expedia, Inc.
+ *
+ *       Licensed under the Apache License, Version 2.0 (the "License");
+ *       you may not use this file except in compliance with the License.
+ *      You may obtain a copy of the License at
+ *
+ *           http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *       Unless required by applicable law or agreed to in writing, software
+ *       distributed under the License is distributed on an "AS IS" BASIS,
+ *       WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *       See the License for the specific language governing permissions and
+ *       limitations under the License.
+ */
+
+import {Dispatcher} from './dispatcher';
+import Span from '../span';
+import { Logger, NullLogger } from '../logger';
+import Utils from '../utils';
+import {CoreOptions, Response} from 'request';
+import * as request from 'request';
+
+export default class HttpCollectorDispatcher implements Dispatcher {
+    _collectorUrl: string;
+    _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;
+        this._logger = logger;
+        this._logger.info(`Initializing the http collector dispatcher, connecting at ${_collectorUrl}`);
+    }
+
+    name(): string {
+        return 'HttpCollectorDispatcher';
+    }
+
+    dispatch(span: Span, callback: (error) => void): void {
+        const serializedSpanBuffer = Utils.convertToProtoSpan(span).serializeBinary();
+        const options: CoreOptions = {};
+        options.headers = this._headers;
+        options.body = serializedSpanBuffer;
+        request.post(this._collectorUrl, options, (error: any, response: Response, body: any) => {
+            if (error) {
+                this._logger.error(`Fail to dipatch to http collector with error ${error}`);
+                return;
+            }
+            if (response.statusCode < 200 || response.statusCode >= 300) {
+                this._logger.error(`Fail to dispatch to http collector with statusCode ${response.statusCode} and response ${body}`);
+                return;
+            }
+            this._logger.debug(`successfully submitted the span to http collector with http response: ${body}`);
+        });
+    }
+
+    close(callback: () => void): void {
+        if (callback) {
+            callback();
+        }
+    }
+}

src/index.ts

@@ -21,7 +21,8 @@ import Tracer from './tracer';
 import NoopDispatcher from './dispatchers/noop';
 import InMemoryDispatcher from './dispatchers/in_memory';
 import FileDispatcher from './dispatchers/file';
-import AgentDispatcher from './dispatchers/remote';
+import AgentDispatcher from './dispatchers/grpc_agent';
+import HttpCollectorDispatcher from './dispatchers/http_collector';
 import { DispatcherConfig } from './dispatchers/dispatcher-config';
 import Configuration from './configuration';
 import { Logger } from './logger';
@@ -39,6 +40,7 @@ export {
     InMemoryDispatcher,
     FileDispatcher,
     AgentDispatcher,
+    HttpCollectorDispatcher,
     DispatcherConfig,
     opentracing,
     Logger