<s> import sys <EOL> from StringIO import StringIO <EOL> from collections import defaultdict <EOL> from zope . interface import implements <EOL> from twisted . internet import reactor <EOL> from twisted . web . client import FileBodyProducer <EOL> from twisted . web . http_headers import Headers <EOL> from tryfer import log <EOL> from tryfer . interfaces import ITracer <EOL> from tryfer . _thrift . zipkinCore import constants <EOL> from tryfer . formatters import json_formatter , base64_thrift_formatter <EOL> class EndAnnotationTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> DEFAULT_END_ANNOTATIONS = ( constants . CLIENT_RECV , constants . SERVER_SEND ) <EOL> def __init__ ( self , tracer , end_annotations = None ) : <EOL> self . _tracer = tracer <EOL> self . _end_annotations = end_annotations or self . DEFAULT_END_ANNOTATIONS <EOL> self . _annotations_for_trace = defaultdict ( list ) <EOL> def record ( self , traces ) : <EOL> for ( trace , annotations ) in traces : <EOL> trace_key = ( trace . trace_id , trace . span_id ) <EOL> self . _annotations_for_trace [ trace_key ] . extend ( annotations ) <EOL> for annotation in annotations : <EOL> if annotation . name in self . _end_annotations : <EOL> saved_annotations = self . _annotations_for_trace [ trace_key ] <EOL> del self . _annotations_for_trace [ trace_key ] <EOL> log . debug ( format = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> system = self . __class__ . __name__ , <EOL> trace_key = trace_key , <EOL> annotations = annotations ) <EOL> self . _tracer . record ( [ ( trace , saved_annotations ) ] ) <EOL> break <EOL> class RawZipkinTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , scribe_client , category = None ) : <EOL> self . _scribe = scribe_client <EOL> self . _category = category or '<STR_LIT>' <EOL> def record ( self , traces ) : <EOL> d = self . _scribe . log ( <EOL> self . _category , <EOL> [ base64_thrift_formatter ( trace , annotations ) <EOL> for ( trace , annotations ) in traces ] ) <EOL> d . addErrback ( <EOL> log . err , <EOL> "<STR_LIT>" . format ( <EOL> self . _category ) ) <EOL> class ZipkinTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , scribe_client , category = None , end_annotations = None , <EOL> max_traces = <NUM_LIT:50> , max_idle_time = <NUM_LIT:10> , _reactor = None ) : <EOL> self . _tracer = EndAnnotationTracer ( <EOL> BufferingTracer ( <EOL> RawZipkinTracer ( scribe_client , category ) , <EOL> max_traces = max_traces , <EOL> max_idle_time = max_idle_time , <EOL> _reactor = _reactor ) , <EOL> end_annotations = end_annotations <EOL> ) <EOL> def record ( self , traces ) : <EOL> return self . _tracer . record ( traces ) <EOL> class RawRESTkinHTTPTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , agent , trace_url ) : <EOL> self . _agent = agent <EOL> self . _trace_url = trace_url <EOL> def record ( self , traces ) : <EOL> producer = FileBodyProducer ( StringIO ( json_formatter ( traces ) ) ) <EOL> d = self . _agent . request ( '<STR_LIT:POST>' , self . _trace_url , Headers ( { } ) , producer ) <EOL> d . addErrback ( <EOL> log . err , <EOL> "<STR_LIT>" . format ( self . _trace_url ) ) <EOL> class RESTkinHTTPTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , agent , trace_url , end_annotations = None , <EOL> max_traces = <NUM_LIT:50> , max_idle_time = <NUM_LIT:10> , _reactor = None ) : <EOL> self . _tracer = EndAnnotationTracer ( <EOL> BufferingTracer ( <EOL> RawRESTkinHTTPTracer ( agent , trace_url ) , <EOL> max_traces = max_traces , <EOL> max_idle_time = max_idle_time , <EOL> _reactor = _reactor ) , <EOL> end_annotations = end_annotations <EOL> ) <EOL> def record ( self , traces ) : <EOL> self . _tracer . record ( traces ) <EOL> class RawRESTkinScribeTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , scribe_client , category = None ) : <EOL> self . _scribe_client = scribe_client <EOL> self . _category = category or '<STR_LIT>' <EOL> def record ( self , traces ) : <EOL> d = self . _scribe_client . log ( <EOL> self . _category , <EOL> [ json_formatter ( traces ) ] ) <EOL> d . addErrback ( <EOL> log . err , <EOL> "<STR_LIT>" . format ( <EOL> self . _category ) ) <EOL> class RESTkinScribeTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , scribe_client , category = None , end_annotations = None , <EOL> max_traces = <NUM_LIT:50> , max_idle_time = <NUM_LIT:10> , _reactor = None ) : <EOL> self . _tracer = EndAnnotationTracer ( <EOL> BufferingTracer ( <EOL> RawRESTkinScribeTracer ( scribe_client , category ) , <EOL> max_traces = max_traces , <EOL> max_idle_time = max_idle_time , <EOL> _reactor = _reactor ) , <EOL> end_annotations = end_annotations <EOL> ) <EOL> def record ( self , traces ) : <EOL> return self . _tracer . record ( traces ) <EOL> class DebugTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , destination = None ) : <EOL> self . destination = destination or sys . stdout <EOL> def record ( self , traces ) : <EOL> self . destination . write ( json_formatter ( traces , indent = <NUM_LIT:2> ) ) <EOL> self . destination . write ( '<STR_LIT:\n>' ) <EOL> self . destination . flush ( ) <EOL> class BufferingTracer ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ITracer ) <EOL> def __init__ ( self , tracer , max_traces = <NUM_LIT:50> , max_idle_time = <NUM_LIT:10> , _reactor = None ) : <EOL> self . _max_traces = max_traces <EOL> self . _max_idle_time = max_idle_time <EOL> self . _reactor = _reactor or reactor <EOL> self . _tracer = tracer <EOL> self . _buffer = [ ] <EOL> self . _idle_dc = None <EOL> self . _flush_dc = None <EOL> def _reset ( self ) : <EOL> if self . _idle_dc and self . _idle_dc . active ( ) : <EOL> self . _idle_dc . reset ( self . _max_idle_time ) <EOL> else : <EOL> self . _idle_dc = self . _reactor . callLater ( <EOL> self . _max_idle_time , self . _flush ) <EOL> def _flush ( self ) : <EOL> if self . _idle_dc and self . _idle_dc . active ( ) : <EOL> self . _idle_dc . cancel ( ) <EOL> if self . _flush_dc is not None : <EOL> self . _flush_dc = None <EOL> flushable = self . _buffer <EOL> self . _buffer = [ ] <EOL> if flushable : <EOL> self . _tracer . record ( flushable ) <EOL> def record ( self , traces ) : <EOL> self . _buffer . extend ( traces ) <EOL> if len ( self . _buffer ) >= self . _max_traces : <EOL> if self . _flush_dc is None : <EOL> self . _flush_dc = self . _reactor . callLater ( <NUM_LIT:0> , self . _flush ) <EOL> else : <EOL> self . _reset ( ) <EOL> _globalTracers = [ ] <EOL> def set_tracers ( tracers ) : <EOL> global _globalTracers <EOL> _globalTracers = tracers <EOL> def push_tracer ( tracer ) : <EOL> global _globalTracers <EOL> _globalTracers . append ( tracer ) <EOL> def get_tracers ( ) : <EOL> return _globalTracers </s>
