spanManager.py (4742B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\publicGateway\grpc\spanManager.py 2 from __future__ import absolute_import 3 import logging 4 import uuid 5 from featuremanagement.client.flag import flag_int 6 import stackless_tracing 7 from .tracing import propagate_trace 8 from eveProto import get_message_type_url 9 from eveProto.tracing import bytes_to_hex 10 from stackless_tracing.attributes import SPAN_KIND, MESSAGE_ARRIVAL_TIME 11 from stackless_tracing.attributes import PAYLOAD_TYPE, PAYLOAD_BYTES, PRIMITIVE, DOMAIN 12 from stackless_tracing.attributes import EXTERNAL_ORIGIN, ISSUED, MESSAGE_ID, APPLICATION 13 from stackless_tracing.attributes import RESPONSE_PAYLOAD, RESPONSE_PAYLOAD_BYTES, INTERNAL_ORIGIN, DISPATCHED, GATEWAY 14 from stackless_tracing.attributes import STATUS_MESSAGE, STATUS_CODE, ERROR 15 from stackless_tracing.attributes import RPC_SYSTEM 16 logger = logging.getLogger(__name__) 17 _FALLBACK_SAMPLE_RATE = 1000 18 _SAMPLE_RATE_FLAG_KEY = 'eve-desktop-client-request-tracing-sample-rate' 19 20 class SpanManager(object): 21 service_name = 'publicGatewaySvc' 22 kind = 'client' 23 domain = 'eve_public' 24 primitive = 'request' 25 26 def __init__(self): 27 self._tracers = {} 28 self._trace_ids = {} 29 self._sample_rate = _FALLBACK_SAMPLE_RATE 30 flag_int(key=_SAMPLE_RATE_FLAG_KEY, default_value=_FALLBACK_SAMPLE_RATE, callback=self.update_sample_rate) 31 32 def update_sample_rate(self, evaluate_flag): 33 self._sample_rate = evaluate_flag() 34 35 def start_request_trace(self, request, payload): 36 span_name = '{} send'.format(payload.DESCRIPTOR.full_name) 37 tracer, span, context = stackless_tracing.start_trace(span_name=span_name, service_name=self.service_name, store_context=False, sample_rate=self._sample_rate) 38 trace_id = stackless_tracing.get_trace_id_from_context(context) 39 span_id = stackless_tracing.get_span_id_from_context(context) 40 sampled = stackless_tracing.get_sampled_from_context(context) 41 sample_rate = stackless_tracing.get_sample_rate_from_context(context) 42 propagate_trace(request, trace_id, span_id, sampled, sample_rate) 43 if not sampled: 44 return 45 payload_name = get_message_type_url(payload.DESCRIPTOR.full_name) 46 span.add_context({SPAN_KIND: self.kind, 47 PAYLOAD_TYPE: payload_name, 48 PAYLOAD_BYTES: payload.ByteSize(), 49 DOMAIN: self.domain, 50 PRIMITIVE: self.primitive, 51 RPC_SYSTEM: 'grpc', 52 EXTERNAL_ORIGIN: request.external_origin, 53 ISSUED: request.issued, 54 MESSAGE_ID: str(uuid.UUID(bytes=request.correlation_uuid)), 55 APPLICATION: str(uuid.UUID(bytes=request.application_instance_uuid))}) 56 self._tracers[trace_id] = (tracer, span) 57 self._trace_ids[request.correlation_uuid] = trace_id 58 59 def end_request_trace(self, response, arrival_datetime): 60 trace_id = bytes_to_hex(response.trace_context.trace_id) 61 correlation_uuid = response.correlation_uuid 62 if trace_id is None: 63 logger.warn('missing trace decision in eve_public response: {}'.format(response)) 64 return 65 if trace_id not in self._tracers: 66 if correlation_uuid not in self._trace_ids: 67 return 68 logger.error('unexpected trace_id for request {}\n\twant: {}\n\thave: {}'.format(bytes_to_hex(correlation_uuid), self._trace_ids.get(correlation_uuid), trace_id)) 69 trace_id = self._trace_ids.get(correlation_uuid) 70 tracer, span = self._tracers.get(trace_id) 71 if tracer is None: 72 return 73 span.add_context({RESPONSE_PAYLOAD: response.payload.TypeName(), 74 RESPONSE_PAYLOAD_BYTES: response.payload.ByteSize(), 75 INTERNAL_ORIGIN: response.internal_origin, 76 DISPATCHED: response.dispatched, 77 GATEWAY: str(uuid.UUID(bytes=response.gateway_instance_uuid)), 78 MESSAGE_ARRIVAL_TIME: arrival_datetime.isoformat('T') + 'Z'}) 79 stackless_tracing.end_trace(tracer) 80 del self._tracers[trace_id] 81 del self._trace_ids[response.correlation_uuid] 82 83 def timeout_request_trace(self, correlation_id): 84 if correlation_id is None: 85 return 86 if correlation_id not in self._trace_ids: 87 return 88 trace_id = self._trace_ids.get(correlation_id) 89 del self._trace_ids[correlation_id] 90 if trace_id not in self._tracers: 91 return 92 tracer, span = self._tracers.get(trace_id) 93 del self._tracers[span.trace_id] 94 if tracer is None: 95 return 96 span.add_context({STATUS_CODE: 2, 97 STATUS_MESSAGE: 'span timeout', 98 ERROR: True}) 99 stackless_tracing.end_trace(tracer)