__init__.py (13245B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\stackless_tracing\__init__.py 2 from __future__ import absolute_import 3 from . import clonable_tracer 4 from . import attributes 5 import beeline 6 import monolithconfig 7 import monolithhoneycomb 8 import stackless 9 import blue 10 import logging 11 from datetimeutils.RFC3339 import blue_to_datetime_micro 12 from journey.tracker import get_journey_id_for_event 13 from cluster import CLUSTER_INGRESS_ORIGIN_NAME 14 CLUSTER_INGRESS_ORIGIN_IDS = {v:k for k, v in CLUSTER_INGRESS_ORIGIN_NAME.iteritems()} 15 CONFIG_GROUP = 'gc' 16 CONFIG_ENABLED = 'stackless_tracing_enabled' 17 CONFIG_SAMPLE_RATE = 'stackless_tracing_sample_rate' 18 CONFIG_SUPPRESSED_LIST = 'tracing_machonet_suppressed_traces' 19 CONFIG_CUT_LIST = 'tracing_machonet_cut_spans' 20 CONFIG_CLIENT_PARTICIPATION = 'stackless_tracing_client_participation' 21 SETTINGS_GROUP = 'tracing' 22 SETTING_SERVER_PARTICIPATION = 'server_participation' 23 CLIENT_NAME = 'monolithhoneycomb_tracing' 24 _client = None 25 _enabled = False 26 _sample_rate = monolithhoneycomb.SAFE_SAMPLE_RATE 27 _participation = False 28 _suppressed_trace_names = set() 29 logger = logging.getLogger(__name__) 30 31 def init(): 32 monolithconfig.add_global_config_callback(update_config) 33 monolithconfig.add_watch_group_callback(update_config, SETTINGS_GROUP) 34 35 36 def update_config(): 37 global _suppressed_trace_names 38 global _participation 39 global _enabled 40 global _sample_rate 41 _enabled = monolithconfig.enabled(CONFIG_ENABLED, CONFIG_GROUP) 42 _sample_rate = monolithconfig.get_value(CONFIG_SAMPLE_RATE, CONFIG_GROUP) 43 _participation = _get_participation() 44 _suppressed_trace_names = _get_suppressed_trace_names() 45 clonable_tracer.set_cut_span_names(_get_cut_span_names()) 46 try: 47 _sample_rate = int(_sample_rate) 48 except (TypeError, ValueError): 49 _sample_rate = monolithhoneycomb.SAFE_SAMPLE_RATE 50 51 _ensure_client() 52 53 54 def get_trace_from_store(): 55 return _get_trace_from_store() 56 57 58 def get_tracer_from_store(): 59 return _get_tracer_from_store() 60 61 62 def get_sampled_from_context(context): 63 return context.get(attributes.SAMPLED, None) 64 65 66 def get_ingress_id_from_context(context): 67 ingress_name = context.get(attributes.INGRESS, None) 68 return CLUSTER_INGRESS_ORIGIN_IDS.get(ingress_name) 69 70 71 def get_trace_id_from_context(context): 72 return context.get(attributes.TRACE_ID, None) 73 74 75 def get_span_id_from_context(context): 76 return context.get(attributes.SPAN_ID, None) 77 78 79 def get_sample_rate_from_context(context): 80 return context.get(attributes.SAMPLE_RATE, None) 81 82 83 def _get_participation_from_context(context): 84 return context.get(attributes.PARTICIPATION, True) 85 86 87 def is_suppressed(span_name): 88 return span_name in _suppressed_trace_names 89 90 91 def start_trace(context = None, span_name = None, service_name = None, trace_id = None, parent_span_id = None, sampled = None, ingress_id = None, sample_rate = None, store_context = True, create_new_tracer = False, force_sample = False): 92 global _client 93 tracer = span = None 94 trace_context = _create_trace_context(None, None, None, None, None) 95 if not _tracing_active(): 96 return (tracer, span, trace_context) 97 if context is None: 98 context = {} 99 if force_sample and not sampled and not _emission_suppressed(span_name): 100 context['eve.sampling_strategy'] = SAMPLING_STRATEGY_FORCED 101 if trace_id: 102 context[FORCED_FROM_TRACE_ID] = trace_id 103 if parent_span_id: 104 context[FORCED_FROM_SPAN_ID] = parent_span_id 105 sampled = True 106 trace_id = beeline.trace.generate_trace_id() 107 parent_span_id = None 108 sample_rate = FORCED_SAMPLE_RATE 109 elif sampled is None or not trace_id: 110 context['eve.sampling_strategy'] = 'deterministic' 111 if _decision_suppressed(span_name): 112 return (tracer, span, trace_context) 113 trace_id = beeline.trace.generate_trace_id() 114 if sample_rate is None: 115 sample_rate = getattr(_client, 'sample_rate', monolithhoneycomb.SAFE_SAMPLE_RATE) 116 sampled = _should_trace(trace_id, span_name, sample_rate) 117 else: 118 context['eve.sampling_strategy'] = 'sampled parent' 119 if sampled: 120 tracer = _get_tracer_from_store() 121 if tracer is None or create_new_tracer: 122 tracer = clonable_tracer.CloneableTracer(_client) 123 context.update(_get_stackless_context(ingress_id, sample_rate)) 124 context.update({attributes.SPAN_NAME: span_name, 125 attributes.SERVICE_NAME: attributes.SERVICE_NAME_PREFIX + service_name}) 126 tracer.sampler_hook = lambda _fields: (sampled, sample_rate) 127 span = tracer.start_or_cut_trace(span_name=span_name, context=context, trace_id=trace_id, parent_id=parent_span_id) 128 if span is None: 129 span_id = beeline.trace.generate_span_id() 130 else: 131 span_id = span.id 132 if store_context: 133 _store_trace(tracer, trace_id, span_id, sampled, ingress_id, sample_rate) 134 trace_context = _create_trace_context(trace_id, span_id, sampled, ingress_id, sample_rate) 135 return (tracer, span, trace_context) 136 137 138 def end_trace(tracer = None, remove_from_store = True): 139 if tracer is None: 140 tracer = _get_tracer_from_store() 141 if tracer is None: 142 return 143 span = tracer.get_active_span() 144 _set_ending_attributes(span) 145 tracer.finish_trace(span) 146 if remove_from_store: 147 _remove_trace() 148 149 150 def start_span(span_name = None, service_name = None, span_context = None): 151 span = None 152 if not _tracing_active(): 153 trace_context = _create_trace_context(None, None, None, None, None) 154 return (span, trace_context) 155 tracer, trace_id, parent_id, sampled, ingress_id, sample_rate = _get_trace_from_store() 156 if tracer is None: 157 trace_context = _create_trace_context(trace_id, parent_id, sampled, ingress_id, sample_rate) 158 return (span, trace_context) 159 if span_context is None: 160 span_context = {} 161 span_context['eve.sampling_strategy'] = 'sampled parent' 162 span_context.update(_get_stackless_context(ingress_id, sample_rate)) 163 span_context.update({attributes.SPAN_NAME: span_name, 164 attributes.SERVICE_NAME: attributes.SERVICE_NAME_PREFIX + service_name}) 165 span = tracer.start_or_cut_span(span_name=span_name, context=span_context, trace_id=trace_id, parent_id=parent_id) 166 _store_trace(tracer, trace_id, span.id, sampled, ingress_id, sample_rate) 167 trace_context = _create_trace_context(trace_id, span.id, sampled, ingress_id, sample_rate) 168 return (span, trace_context) 169 170 171 def end_span(span): 172 if span is None: 173 return 174 tracer, trace_id, parent_id, sampled, ingress_id, sample_rate = _get_trace_from_store() 175 if tracer is None: 176 return 177 _set_ending_attributes(span) 178 tracer.finish_span(span) 179 active_span_id = None 180 active_span = tracer.get_active_span() 181 if active_span is not None: 182 active_span_id = active_span.id 183 _store_trace(tracer, trace_id, active_span_id, sampled, ingress_id, sample_rate) 184 185 186 FORCED_SAMPLE_RATE = 1 187 SAMPLING_STRATEGY_FORCED = 'forced' 188 FORCED_FROM_TRACE_ID = 'app.trace.forced_from_trace_id' 189 FORCED_FROM_SPAN_ID = 'app.trace.forced_from_span_id' 190 MAX_STATUS_MESSAGE_LENGTH = 256 191 192 def add_context(span, context): 193 if span is None or not context: 194 return 195 span.add_context(context) 196 197 198 def mark_error(span, exception): 199 if span is None: 200 return 201 status_message = str(exception)[:MAX_STATUS_MESSAGE_LENGTH] 202 span.add_context({attributes.STATUS_CODE: 2, 203 attributes.STATUS_MESSAGE: status_message or exception.__class__.__name__, 204 attributes.ERROR: True}) 205 206 207 class Span(object): 208 209 def __init__(self, span_name, service_name, span_kind = 'internal'): 210 self.span_name = span_name 211 self.service_name = service_name 212 self.span_kind = span_kind 213 self.span = None 214 215 def __enter__(self): 216 self.span, trace_context = start_span(self.span_name, self.service_name) 217 if not get_sampled_from_context(trace_context): 218 return self 219 if self.span_kind is not None: 220 self.span.add_context({attributes.SPAN_KIND: self.span_kind}) 221 return self 222 223 def __exit__(self, exc_type, exc_value, exc_tb): 224 if exc_type is not None: 225 mark_error(self.span, exc_value) 226 end_span(self.span) 227 228 def add_context(self, context): 229 add_context(self.span, context) 230 231 def set_error(self, exception): 232 mark_error(self.span, exception) 233 234 235 def _set_ending_attributes(span): 236 current = stackless.getcurrent() 237 if not hasattr(current, 'GetRunTime'): 238 return 239 run_time_ms = current.GetRunTime() * 1000 240 span.add_context({attributes.TASKLET_RUNTIME_MS: run_time_ms}) 241 242 243 def _should_trace(trace_id, span_name, sample_rate): 244 if is_suppressed(span_name): 245 return False 246 return _deterministic_sample(trace_id, sample_rate) 247 248 249 def _deterministic_sample(trace_id, sample_rate): 250 return beeline.trace._should_sample(trace_id, sample_rate) 251 252 253 def _emission_suppressed(span_name): 254 return is_suppressed(span_name) or clonable_tracer.is_cut(span_name) 255 256 257 def _decision_suppressed(span_name): 258 if not _participation: 259 return True 260 if clonable_tracer.is_cut(span_name): 261 return True 262 return False 263 264 265 def _create_trace_context(trace_id, parent_id, sampled, ingress_id, sample_rate): 266 return {attributes.TRACE_ID: trace_id, 267 attributes.SPAN_ID: parent_id, 268 attributes.SAMPLED: sampled, 269 attributes.INGRESS_ID: ingress_id, 270 attributes.SAMPLE_RATE: sample_rate} 271 272 273 def _store_trace(tracer, trace_id, parent_id, sampled, ingress_id, sample_rate): 274 current = stackless.getcurrent() 275 try: 276 setattr(current, 'tracer', tracer) 277 setattr(current, 'trace_id', trace_id) 278 setattr(current, 'parent_id', parent_id) 279 setattr(current, 'sampled', sampled) 280 setattr(current, 'ingress_id', ingress_id) 281 setattr(current, 'sample_rate', sample_rate) 282 except Exception: 283 pass 284 285 286 def _remove_trace(): 287 current = stackless.getcurrent() 288 try: 289 setattr(current, 'tracer', None) 290 setattr(current, 'trace_id', None) 291 setattr(current, 'parent_id', None) 292 setattr(current, 'sampled', None) 293 setattr(current, 'ingress_id', None) 294 setattr(current, 'sample_rate', None) 295 except Exception: 296 pass 297 298 299 def _get_trace_from_store(): 300 current = stackless.getcurrent() 301 tracer = trace_id = parent_id = sampled = ingress_id = sample_rate = None 302 try: 303 tracer = getattr(current, 'tracer', None) 304 trace_id = getattr(current, 'trace_id', None) 305 parent_id = getattr(current, 'parent_id', None) 306 sampled = getattr(current, 'sampled', None) 307 ingress_id = getattr(current, 'ingress_id', None) 308 sample_rate = getattr(current, 'sample_rate', None) 309 except Exception: 310 pass 311 312 return (tracer, 313 trace_id, 314 parent_id, 315 sampled, 316 ingress_id, 317 sample_rate) 318 319 320 def _get_tracer_from_store(): 321 trace_data = _get_trace_from_store() 322 return trace_data[0] 323 324 325 def _ensure_client(): 326 global _client 327 if _client is None: 328 _client = monolithhoneycomb.new_client(CLIENT_NAME, _sample_rate) 329 else: 330 _client.sample_rate = _sample_rate 331 332 333 def _get_participation(): 334 if monolithconfig.on_client(): 335 return monolithconfig.enabled(CONFIG_CLIENT_PARTICIPATION, CONFIG_GROUP) 336 if monolithconfig.on_proxy(): 337 return False 338 if monolithconfig.on_server(): 339 return monolithconfig.enabled(SETTING_SERVER_PARTICIPATION, SETTINGS_GROUP) 340 341 342 def _get_suppressed_trace_names(): 343 return _get_list_from_config(CONFIG_SUPPRESSED_LIST) 344 345 346 def _get_cut_span_names(): 347 return _get_list_from_config(CONFIG_CUT_LIST) 348 349 350 def _get_list_from_config(config_name): 351 span_names = set() 352 raw = monolithconfig.get_value(config_name, CONFIG_GROUP) 353 if raw is None: 354 return span_names 355 raw_list = raw.split(';') 356 for span_name in raw_list: 357 if len(span_name) == 0: 358 continue 359 span_names.add(span_name) 360 361 return span_names 362 363 364 def _tracing_active(): 365 return _enabled & (_client is not None) 366 367 368 def _get_stackless_context(ingress_id, sample_rate): 369 current = stackless.getcurrent() 370 tasklet_id = getattr(current, 'tasklet_id', None) 371 wallclock = blue.os.GetWallclockTime() 372 journey_id = get_journey_id_for_event() 373 creation_datetime = getattr(current, 'creation_datetime', None) 374 stackless_context = {attributes.TIME_DILATION: blue.os.simDilation, 375 attributes.INGRESS: CLUSTER_INGRESS_ORIGIN_NAME.get(ingress_id, None)} 376 if creation_datetime is not None: 377 stackless_context[attributes.TASKLET_CREATION_DATETIME] = creation_datetime.isoformat('T') + 'Z' 378 if sample_rate is not None: 379 stackless_context[attributes.SAMPLE_RATE] = int(sample_rate) 380 if tasklet_id is not None: 381 stackless_context[attributes.TASKLET] = int(tasklet_id) 382 if wallclock is not None: 383 stackless_context[attributes.BLUE_TIME] = blue_to_datetime_micro(wallclock) 384 if journey_id is not None: 385 stackless_context[attributes.JOURNEY] = journey_id 386 return stackless_context