__init__.py (8619B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\machonet_tracing\__init__.py 2 from stackless_tracing.attributes import SPAN_NAME, SPAN_KIND, SERVICE_NAME, STATUS_MESSAGE, STATUS_CODE, ERROR, MACHO_CACHE_OK 3 from stackless_tracing.attributes import RPC_SYSTEM, RPC_SERVICE, RPC_METHOD 4 from stackless_tracing.attributes import SESSION_CALL_MS, WRITE_MS, PROXY 5 from stackless_tracing.attributes import BIND_PARAMS 6 import stackless_tracing 7 import monolithconfig 8 import logging 9 import time 10 from contextlib import contextmanager 11 from cluster import CLUSTER_INGRESS_ORIGIN_CLIENT, MACHONETMSG_TYPE_CALL_REQ 12 from carbon.common.lib.const import ADDRESS_TYPE_CLIENT 13 TRACED_REQUEST_TYPES = {MACHONETMSG_TYPE_CALL_REQ} 14 CALL_SERVICE_NAME = CALL_RPC_SYSTEM = 'machoNet' 15 RPC_METHOD_BIND = 'MachoBindObject' 16 logger = logging.getLogger(__name__) 17 18 def propagate_trace_context(packet): 19 if monolithconfig.on_proxy(): 20 return 21 tracer, trace_id, parent_id, sampled, ingress_id, sample_rate = stackless_tracing.get_trace_from_store() 22 cut_parent = None 23 if tracer is not None: 24 cut_parent = tracer.has_cut_parent() 25 _set_packet_trace_context(packet, trace_id, parent_id, ingress_id, sampled, sample_rate, cut_parent) 26 27 28 def _get_span_name(rpc_service, rpc_method): 29 return '{}.{}'.format(rpc_service, rpc_method) 30 31 32 def _set_packet_trace_context(packet, trace_id, parent_span_id, ingress_id, sampled, sample_rate, cut_parent): 33 if trace_id is not None: 34 packet.trace_id = trace_id 35 packet.parent_span_id = parent_span_id 36 packet.ingress_id = ingress_id 37 packet.sampled = sampled 38 packet.sample_rate = sample_rate 39 packet.cut_parent = cut_parent 40 packet.Changed() 41 42 43 def _get_active_span(): 44 tracer = stackless_tracing.get_tracer_from_store() 45 if tracer is None: 46 return 47 span = tracer.get_active_span() 48 return span 49 50 51 def _is_macho_bind_method(rpc_method): 52 return rpc_method == RPC_METHOD_BIND 53 54 55 def _get_bind_inner_payload(packet): 56 try: 57 return packet.payload[2][-1] 58 except Exception: 59 return None 60 61 62 class EgressTracer(object): 63 span_kind = 'client' 64 65 def __init__(self, packet, rpc_service, rpc_method, force_trace = False): 66 self.packet = packet 67 self.force_trace = force_trace 68 self.rpc_service = rpc_service 69 self.rpc_method = rpc_method 70 self.tracer = self.span = None 71 self.sampled = False 72 73 def __enter__(self): 74 if not self._is_tracable(): 75 return self 76 span_name = _get_span_name(self.rpc_service, self.rpc_method) 77 self.tracer, self.span, trace_context = stackless_tracing.start_trace(span_name=span_name, service_name=CALL_SERVICE_NAME, ingress_id=CLUSTER_INGRESS_ORIGIN_CLIENT, sampled=None, store_context=True) 78 self._set_packet_trace_context(trace_context) 79 self.sampled = stackless_tracing.get_sampled_from_context(trace_context) 80 if not self.sampled: 81 return self 82 if self._is_macho_bind_method(): 83 inner_payload = self._get_bind_inner_payload() 84 self.rpc_method = inner_payload[0] 85 self.span.add_context({SPAN_KIND: self.span_kind, 86 RPC_SYSTEM: CALL_RPC_SYSTEM, 87 RPC_SERVICE: self.rpc_service, 88 RPC_METHOD: self.rpc_method}) 89 return self 90 91 def __exit__(self, exc_type, exc_val, exc_tb): 92 if not self.sampled: 93 return 94 stackless_tracing.end_trace(self.tracer) 95 96 def _is_tracable(self): 97 if not monolithconfig.on_client(): 98 return False 99 if self.packet.command not in TRACED_REQUEST_TYPES: 100 return False 101 if self._is_macho_bind_method(): 102 if self._get_bind_inner_payload() is None: 103 return False 104 if _get_active_span() is not None: 105 return False 106 return True 107 108 def set_bind_parameters(self, bind_params): 109 if self.span is None: 110 return 111 self.span.add_context_field(BIND_PARAMS, bind_params) 112 113 def _set_packet_trace_context(self, trace_context): 114 propagate_trace_context(self.packet) 115 116 def _get_bind_inner_payload(self): 117 return _get_bind_inner_payload(self.packet) 118 119 def _is_macho_bind_method(self): 120 return _is_macho_bind_method(self.rpc_method) 121 122 123 class IngressTracer(object): 124 span_kind = 'server' 125 126 def __init__(self, packet): 127 self.packet = packet 128 self.sampled = False 129 130 def __enter__(self): 131 return self 132 133 def __exit__(self, exc_type, exc_val, exc_tb): 134 self.end_trace() 135 136 @staticmethod 137 def start_macho_bind_trace(packet, object_name, method_name, bind_start, bind_end): 138 inner_payload = _get_bind_inner_payload(packet) 139 if inner_payload is None: 140 return 141 tracer, span = IngressTracer.start_trace(packet, object_name, method_name) 142 if span is not None: 143 span.add_context_field('eve.machonet.object_bind_ms', (bind_end - bind_start) * 1000) 144 145 @staticmethod 146 def start_trace(packet, rpc_service, rpc_method): 147 context = {} 148 tracer = span = None 149 if not IngressTracer._is_traceable(packet): 150 return (tracer, span) 151 span_name = _get_span_name(rpc_service, rpc_method) 152 trace_id, parent_span_id, sampled, ingress_id, sample_rate = IngressTracer.extract_trace_context(packet) 153 if IngressTracer.has_cut_parent(packet): 154 context['eve.tracing.has_removed_parents'] = True 155 if sampled is None and packet.source.addressType != ADDRESS_TYPE_CLIENT: 156 return (tracer, span) 157 tracer, span, trace_context = stackless_tracing.start_trace(context=context, span_name=span_name, service_name=CALL_SERVICE_NAME, trace_id=trace_id, parent_span_id=parent_span_id, sampled=sampled, ingress_id=ingress_id, sample_rate=sample_rate, store_context=True) 158 IngressTracer._set_packet_trace_context(packet, trace_context) 159 if not stackless_tracing.get_sampled_from_context(trace_context): 160 return (tracer, span) 161 span.event.created_at = packet.read_time 162 span.add_context({SPAN_KIND: IngressTracer.span_kind, 163 RPC_SYSTEM: CALL_RPC_SYSTEM, 164 RPC_METHOD: rpc_method, 165 RPC_SERVICE: rpc_service}) 166 return (tracer, span) 167 168 @staticmethod 169 def end_trace(): 170 tracer, _trace_id, _parent_id, sampled, ingress_id, _sample_rate = stackless_tracing.get_trace_from_store() 171 if not sampled or tracer is None: 172 return 173 stackless_tracing.end_trace(tracer) 174 175 @staticmethod 176 def _is_traceable(packet): 177 if not monolithconfig.on_server(): 178 return False 179 if packet.command not in TRACED_REQUEST_TYPES: 180 return False 181 return True 182 183 @staticmethod 184 def extract_trace_context(packet): 185 trace_id = packet.trace_id 186 parent_span_id = packet.parent_span_id 187 sampled = packet.sampled 188 ingress_id = packet.ingress_id or CLUSTER_INGRESS_ORIGIN_CLIENT 189 sample_rate = packet.sample_rate 190 return (trace_id, 191 parent_span_id, 192 sampled, 193 ingress_id, 194 sample_rate) 195 196 @staticmethod 197 def has_cut_parent(packet): 198 return packet.cut_parent 199 200 @staticmethod 201 def _set_packet_trace_context(packet, trace_context): 202 propagate_trace_context(packet) 203 204 @staticmethod 205 def set_error_state(description): 206 span = _get_active_span() 207 if span is None: 208 return 209 span.add_context({STATUS_MESSAGE: description, 210 STATUS_CODE: 2, 211 ERROR: True}) 212 213 @staticmethod 214 def set_object_cache_ok(): 215 span = _get_active_span() 216 if span is None: 217 return 218 span.add_context({MACHO_CACHE_OK: True}) 219 220 @staticmethod 221 def record_ingress_proxy_id(proxy_node_id): 222 span = _get_active_span() 223 if span is None: 224 return 225 span.add_context_field(PROXY, proxy_node_id) 226 227 @staticmethod 228 @contextmanager 229 def record_write_response_ms(): 230 start_time = time.time() 231 yield 232 span = _get_active_span() 233 if span is None: 234 return 235 span.add_context_field(WRITE_MS, (time.time() - start_time) * 1000) 236 237 @staticmethod 238 @contextmanager 239 def record_process_call_ms(): 240 start_time = time.time() 241 yield 242 span = _get_active_span() 243 if span is None: 244 return 245 span.add_context_field(SESSION_CALL_MS, (time.time() - start_time) * 1000)