__init__.py (11405B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\beeline\__init__.py 2 import functools 3 import logging 4 import os 5 import socket 6 from contextlib import contextmanager 7 from libhoney import Client 8 from beeline.trace import SynchronousTracer 9 from beeline.version import VERSION 10 from beeline import internal 11 import beeline.propagation.honeycomb 12 import sys 13 USER_AGENT_ADDITION = 'beeline-python/%s' % VERSION 14 _GBL = None 15 _INITPID = None 16 try: 17 import asyncio 18 try: 19 asyncio.get_running_loop() 20 except RuntimeError: 21 pass 22 23 from beeline.aiotrace import AsyncioTracer, traced_impl, untraced 24 25 def in_async_code(): 26 try: 27 asyncio.get_running_loop() 28 return True 29 except RuntimeError: 30 return False 31 32 33 except (ImportError, AttributeError): 34 from beeline.trace import traced_impl 35 36 def in_async_code(): 37 return False 38 39 40 class Beeline(object): 41 42 def __init__(self, writekey = '', dataset = '', service_name = '', tracer = None, sample_rate = 1, api_host = 'https://api.honeycomb.io', max_concurrent_batches = 10, max_batch_size = 100, send_frequency = 0.25, block_on_send = False, block_on_response = False, transmission_impl = None, sampler_hook = None, presend_hook = None, http_trace_parser_hook = beeline.propagation.honeycomb.http_trace_parser_hook, http_trace_propagation_hook = beeline.propagation.honeycomb.http_trace_propagation_hook, debug = False): 43 self.client = None 44 self.tracer_impl = None 45 self.presend_hook = None 46 self.sampler_hook = None 47 self.http_trace_parser_hook = None 48 self.http_trace_propagation_hook = None 49 self.debug = debug 50 if debug: 51 self._init_logger() 52 if not writekey: 53 writekey = os.environ.get('HONEYCOMB_WRITEKEY', '') 54 if not dataset: 55 dataset = os.environ.get('HONEYCOMB_DATASET', '') 56 if not service_name: 57 service_name = os.environ.get('HONEYCOMB_SERVICE', dataset) 58 self.client = Client(writekey=writekey, dataset=dataset, sample_rate=sample_rate, api_host=api_host, max_concurrent_batches=max_concurrent_batches, max_batch_size=max_batch_size, send_frequency=send_frequency, block_on_send=block_on_send, block_on_response=block_on_response, transmission_impl=transmission_impl, user_agent_addition=USER_AGENT_ADDITION, debug=debug) 59 self.log('initialized honeycomb client: writekey=%s dataset=%s service_name=%s', writekey, dataset, service_name) 60 if not writekey: 61 self.log('writekey not set! set the writekey if you want to send data to honeycomb') 62 if not dataset: 63 self.log('dataset not set! set a value for dataset if you want to send data to honeycomb') 64 self.client.add_field('service_name', service_name) 65 self.client.add_field('meta.beeline_version', VERSION) 66 self.client.add_field('meta.local_hostname', socket.gethostname()) 67 if in_async_code(): 68 self.tracer_impl = AsyncioTracer(self.client) 69 else: 70 self.tracer_impl = SynchronousTracer(self.client) 71 self.tracer_impl.register_hooks(presend=presend_hook, sampler=sampler_hook, http_trace_parser=http_trace_parser_hook, http_trace_propagation=http_trace_propagation_hook) 72 self.sampler_hook = sampler_hook 73 self.presend_hook = presend_hook 74 self.http_trace_parser_hook = http_trace_parser_hook 75 self.http_trace_propagation_hook = http_trace_propagation_hook 76 77 def send_now(self, data): 78 ev = self.client.new_event() 79 if data: 80 ev.add(data) 81 self._run_hooks_and_send(ev) 82 83 def add_field(self, name, value): 84 span = self.tracer_impl.get_active_span() 85 if span is None: 86 return 87 span.add_context_field(name, value) 88 89 def add(self, data): 90 span = self.tracer_impl.get_active_span() 91 if span is None: 92 return 93 span.add_context(data) 94 95 def tracer(self, name, trace_id = None, parent_id = None): 96 return self.tracer_impl(name=name, trace_id=trace_id, parent_id=parent_id) 97 98 def new_event(self, data = None, trace_name = ''): 99 if trace_name: 100 data['name'] = trace_name 101 if self.tracer_impl.get_active_trace_id(): 102 self.tracer_impl.start_span(context=data) 103 else: 104 self.tracer_impl.start_trace(context=data) 105 106 def send_event(self): 107 span = self.tracer_impl.get_active_span() 108 if span: 109 if span.is_root(): 110 self.tracer_impl.finish_trace(span) 111 return 112 self.tracer_impl.finish_span(span) 113 114 def send_all(self): 115 span = self.tracer_impl.get_active_span() 116 while span: 117 if span.is_root(): 118 self.tracer_impl.finish_trace(span) 119 return 120 self.tracer_impl.finish_span(span) 121 span = self.tracer_impl.get_active_span() 122 123 def traced(self, name, trace_id = None, parent_id = None): 124 return traced_impl(tracer_fn=self.tracer, name=name, trace_id=trace_id, parent_id=parent_id) 125 126 def traced_thread(self, fn): 127 trace_copy = self.tracer_impl._trace.copy() 128 129 @functools.wraps(fn) 130 def wrapped(*args, **kwargs): 131 self.tracer_impl._trace = trace_copy 132 return fn(*args, **kwargs) 133 134 return wrapped 135 136 def _run_hooks_and_send(self, ev): 137 presampled = False 138 if self.sampler_hook: 139 self.log('executing sampler hook on event ev = %s', ev.fields()) 140 keep, new_rate = self.sampler_hook(ev.fields()) 141 if not keep: 142 self.log('skipping event due to sampler hook sampling ev = %s', ev.fields()) 143 return 144 ev.sample_rate = new_rate 145 presampled = True 146 if self.presend_hook: 147 self.log('executing presend hook on event ev = %s', ev.fields()) 148 self.presend_hook(ev.fields()) 149 if presampled: 150 self.log('enqueuing presampled event ev = %s', ev.fields()) 151 ev.send_presampled() 152 else: 153 self.log('enqueuing event ev = %s', ev.fields()) 154 ev.send() 155 156 def _init_logger(self): 157 self._logger = logging.getLogger('honeycomb-beeline') 158 self._logger.setLevel(logging.DEBUG) 159 ch = logging.StreamHandler() 160 ch.setLevel(logging.DEBUG) 161 formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') 162 ch.setFormatter(formatter) 163 self._logger.addHandler(ch) 164 165 def log(self, msg, *args, **kwargs): 166 if self.debug: 167 self._logger.debug(msg, *args, **kwargs) 168 169 def get_responses_queue(self): 170 return self.client.responses() 171 172 def close(self): 173 if self.client: 174 self.client.close() 175 176 177 def init(writekey = '', dataset = '', service_name = '', tracer = None, sample_rate = 1, api_host = 'https://api.honeycomb.io', transmission_impl = None, sampler_hook = None, presend_hook = None, debug = False, *args, **kwargs): 178 global _INITPID 179 global _GBL 180 pid = os.getpid() 181 if _GBL: 182 if pid == _INITPID: 183 _GBL.log('beeline already initialized! skipping initialization') 184 return 185 _GBL.log('beeline already initialized, but process ID has changed (was {}, now {}). Reinitializing.'.format(_INITPID, pid)) 186 _GBL.close() 187 _GBL = Beeline(writekey=writekey, dataset=dataset, sample_rate=sample_rate, api_host=api_host, transmission_impl=transmission_impl, debug=debug, presend_hook=presend_hook, sampler_hook=sampler_hook, service_name=service_name, *args, **kwargs) 188 _INITPID = pid 189 190 191 def send_now(data): 192 bl = get_beeline() 193 if bl: 194 bl.send_now(data) 195 196 197 def add_field(name, value): 198 if _GBL: 199 _GBL.add_field(name, value) 200 201 202 def add(data): 203 bl = get_beeline() 204 if bl: 205 bl.add(data) 206 207 208 def add_context(data): 209 bl = get_beeline() 210 if bl: 211 bl.tracer_impl.add_context(data=data) 212 213 214 def add_context_field(name, value): 215 bl = get_beeline() 216 if bl: 217 bl.tracer_impl.add_context_field(name=name, value=value) 218 219 220 def remove_context_field(name): 221 bl = get_beeline() 222 if bl: 223 bl.tracer_impl.remove_context_field(name=name) 224 225 226 def add_rollup_field(name, value): 227 bl = get_beeline() 228 if bl: 229 bl.tracer_impl.add_rollup_field(name=name, value=value) 230 231 232 def add_trace_field(name, value): 233 bl = get_beeline() 234 if bl: 235 bl.tracer_impl.add_trace_field(name=name, value=value) 236 237 238 def remove_trace_field(name): 239 bl = get_beeline() 240 if bl: 241 bl.tracer_impl.remove_trace_field(name=name) 242 243 244 def tracer(name, trace_id = None, parent_id = None): 245 bl = get_beeline() 246 if bl: 247 return bl.tracer(name=name, trace_id=trace_id, parent_id=parent_id) 248 249 @contextmanager 250 def _noop_cm(): 251 yield 252 253 return _noop_cm() 254 255 256 def start_trace(context = None, trace_id = None, parent_span_id = None): 257 bl = get_beeline() 258 if bl: 259 return bl.tracer_impl.start_trace(context=context, trace_id=trace_id, parent_span_id=parent_span_id) 260 261 262 def finish_trace(span): 263 bl = get_beeline() 264 if bl: 265 bl.tracer_impl.finish_trace(span=span) 266 267 268 def start_span(context = None, parent_id = None): 269 bl = get_beeline() 270 if bl: 271 return bl.tracer_impl.start_span(context=context, parent_id=parent_id) 272 273 274 def finish_span(span): 275 bl = get_beeline() 276 if bl: 277 bl.tracer_impl.finish_span(span=span) 278 279 280 def propagate_and_start_trace(context, request): 281 bl = get_beeline() 282 if bl: 283 return bl.tracer_impl.propagate_and_start_trace(context, request) 284 285 286 def http_trace_parser_hook(headers): 287 bl = get_beeline() 288 if bl: 289 return bl.tracer_impl.http_trace_parser_hook(headers) 290 291 292 def http_trace_propagation_hook(): 293 bl = get_beeline() 294 if bl: 295 try: 296 return bl.tracer_impl.http_trace_propagation_hook(bl.tracer_impl.get_propagation_context()) 297 except Exception: 298 err = sys.exc_info() 299 bl.log('error: http_trace_propagation_hook returned exception: %s', err) 300 301 302 def marshal_trace_context(): 303 bl = get_beeline() 304 if bl: 305 return bl.tracer_impl.marshal_trace_context() 306 307 308 def new_event(data = None, trace_name = ''): 309 bl = get_beeline() 310 if bl: 311 bl.new_event(data=data, trace_name=trace_name) 312 313 314 def send_event(): 315 bl = get_beeline() 316 if bl: 317 bl.send_event() 318 319 320 def send_all(): 321 bl = get_beeline() 322 if bl: 323 bl.send_all() 324 325 326 def get_beeline(): 327 return _GBL 328 329 330 def get_responses_queue(): 331 bl = get_beeline() 332 if bl: 333 return bl.get_responses_queue() 334 335 336 def close(): 337 global _GBL 338 if _GBL: 339 _GBL.close() 340 _GBL = None 341 342 343 def traced(name, trace_id = None, parent_id = None): 344 return traced_impl(tracer_fn=tracer, name=name, trace_id=trace_id, parent_id=parent_id) 345 346 347 def traced_thread(fn): 348 bl = get_beeline() 349 if bl is None or bl.tracer_impl.get_active_trace_id() is None: 350 351 @functools.wraps(fn) 352 def noop(*args, **kwargs): 353 return fn(*args, **kwargs) 354 355 return noop 356 trace_copy = bl.tracer_impl._trace.copy() 357 358 @functools.wraps(fn) 359 def wrapped(*args, **kwargs): 360 bl.tracer_impl._trace = trace_copy 361 return fn(*args, **kwargs) 362 363 return wrapped