/**
 * Copyright 2018, OpenCensus Authors
 *
 * 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 { Exporter , ExporterConfig , Span } from '@opencensus/core' ; import { logger , Logger } from '@opencensus/core' ; import * as os from 'os' ; import { spanToThrift , Tag , TagValue , ThriftProcess , ThriftUtils , UDPSender , Utils , } from './jaeger-driver' ; const DEFAULT_BUFFER_FLUSH_INTERVAL_MILLIS = 1000 ; const DEFAULT_BUFFER_SIZE = 1000 ; /**
 * Options for Jaeger configuration
 */ export interface JaegerTraceExporterOptions extends ExporterConfig { serviceName : string ; tags ? : Tag [ ] ; host ? : string ; port ? : number ; maxPacketSize ? : number ; } let pjsonVersion ; try { pjsonVersion = require ( '../../package.json' ) . version ; } catch { pjsonVersion = require ( '../package.json' ) . version ; } /** Format and sends span information to Jaeger */ export class JaegerTraceExporter implements Exporter { static readonly JAEGER_OPENCENSUS_EXPORTER_VERSION_TAG_KEY = 'opencensus.exporter.jaeger.version' ; static readonly TRACER_HOSTNAME_TAG_KEY = 'opencensus.exporter.jaeger.hostname' ; static readonly PROCESS_IP = 'ip' ; private process : ThriftProcess ; private logger : Logger ; sender : typeof UDPSender ; queue : Span [ ] = [ ] ; successCount = 0 ; /** Maximum size of a buffer. */ private bufferSize : number ; /** Timeout control */ /** Max time for a buffer can wait before being sent */ private bufferTimeout : number ; /** Indicates when the buffer timeout is running */ private timeoutSet = false ; constructor ( options ) { this . logger = options . logger || logger . logger ( ) ; this . bufferTimeout = options . bufferTimeout || DEFAULT_BUFFER_FLUSH_INTERVAL_MILLIS ; this . bufferSize = options . bufferSize || DEFAULT_BUFFER_SIZE ; this . sender = new UDPSender ( options ) ; const tags = options . tags || [ ] ; const defaultTags = { } ; defaultTags [ JaegerTraceExporter . JAEGER_OPENCENSUS_EXPORTER_VERSION_TAG_KEY ] = `opencensus-exporter-jaeger-${ pjsonVersion }` ; defaultTags [ JaegerTraceExporter . TRACER_HOSTNAME_TAG_KEY ] = os . hostname ( ) ; defaultTags [ JaegerTraceExporter . PROCESS_IP ] = Utils . myIp ( ) ; const _tags = [ ... tags , ... Utils . convertObjectToTags ( defaultTags ) ] ; this . process = { serviceName : options . serviceName , tags : ThriftUtils . getThriftTags ( _tags ) , } ; this . sender . setProcess ( this . process ) ; } param span onEndSpan ( span ) { if ( ! span . isRootSpan ( ) ) return ; this . logger . debug ( 'onEndSpan: adding rootSpan: %s' , span . name ) ; this . addSpanToSenderBuffer ( span ) . then ( result => { this . addToBuffer ( span , result as number ) ; for ( const localSpan of span . spans ) { this . addSpanToSenderBuffer ( localSpan ) . then ( result => { this . addToBuffer ( localSpan , result as number ) ; } ) . catch ( err => { return ; } ) ; } } ) . catch ( err => { return ; } ) ; this . setBufferTimeout ( ) ; } /** Not used for this exporter */ onStartSpan ( span ) { } private addToBuffer ( span , numSpans ) { if ( numSpans > 0 ) { this . successCount += numSpans ; if ( numSpans === this . queue . length ) { this . queue = [ span ] ; } else { this . queue = [ ] ; } } else { this . logger . debug ( 'adding to buffer %s' , span . name ) ; this . queue . push ( span ) ; if ( this . queue . length > this . bufferSize ) { this . flush ( ) ; } } } private addSpanToSenderBuffer ( span ) { const thriftSpan = spanToThrift ( span ) ; return new Promise ( ( resolve , reject ) => { this . sender . append ( thriftSpan , ( numSpans , err ? ) => { if ( err ) { this . logger . error ( `failed to add span: ${ err }` ) ; reject ( err ) ; } else { this . logger . debug ( 'successful append for : %s' , numSpans ) ; resolve ( numSpans ) ; } } ) ; } ) ; } param spans publish ( spans ) { this . logger . debug ( 'JaegerExport publishing' ) ; for ( const span of spans ) { if ( this . queue . indexOf ( span ) === - 1 ) { this . onEndSpan ( span ) ; } } return this . flush ( ) ; } private flush ( ) { return new Promise ( ( resolve , reject ) => { try { this . sender . flush ( ( numSpans , err ? ) => { if ( err ) { this . logger . error ( `failed to flush span: ${ err }` ) ; reject ( err ) ; } else { this . logger . debug ( 'successful flush for : %s' , numSpans ) ; this . successCount += numSpans ; this . queue = [ ] ; resolve ( numSpans ) ; } } ) ; } catch ( err ) { this . logger . error ( `failed to flush span: ${ err }` ) ; } } ) ; } close ( ) { this . sender . close ( ) ; } /** Start the buffer timeout, when finished calls flush method */ private setBufferTimeout ( ) { this . logger . debug ( 'JaegerExporter: set timeout' ) ; if ( this . timeoutSet ) { return ; } this . timeoutSet = true ; setTimeout ( ( ) => { if ( this . queue . length === 0 ) { return ; } this . timeoutSet = false ; this . flush ( ) ; } , this . bufferTimeout ) ; } }