Coverage for merco/observability/tracing.py: 0%

1"""链路追踪""" 

2 

3import time 

4import uuid 

5from contextvars import ContextVar 

6 

7# 当前追踪 ID 

8current_trace: ContextVar[str] = ContextVar("trace_id", default="") 

9 

10 

11class TraceSpan: 

12 """追踪跨度""" 

13 

14 def __init__(self, operation: str, parent_id: str = None): 

15 self.operation = operation 

16 self.trace_id = current_trace.get() or str(uuid.uuid4())[:8] 

17 self.span_id = str(uuid.uuid4())[:8] 

18 self.parent_id = parent_id 

19 self.start_time = time.time() 

20 self.end_time = None 

21 self.attributes: dict = {} 

22 

23 def set_attribute(self, key: str, value): 

24 """设置属性""" 

25 self.attributes[key] = value 

26 

27 def end(self): 

28 """结束跨度""" 

29 self.end_time = time.time() 

30 

31 @property 

32 def duration(self) -> float: 

33 """获取耗时（秒）""" 

34 if self.end_time: 

35 return self.end_time - self.start_time 

36 return time.time() - self.start_time 

37 

38 def to_dict(self) -> dict: 

39 return { 

40 "trace_id": self.trace_id, 

41 "span_id": self.span_id, 

42 "parent_id": self.parent_id, 

43 "operation": self.operation, 

44 "duration": self.duration, 

45 "attributes": self.attributes, 

46 } 

47 

48 

49class Tracer: 

50 """链路追踪器""" 

51 

52 def __init__(self): 

53 self._spans: list[TraceSpan] = [] 

54 

55 def start_span(self, operation: str, parent_id: str = None) -> TraceSpan: 

56 """开始新的跨度""" 

57 span = TraceSpan(operation, parent_id) 

58 current_trace.set(span.trace_id) 

59 self._spans.append(span) 

60 return span 

61 

62 def get_spans(self, trace_id: str = None) -> list[dict]: 

63 """获取跨度列表""" 

64 spans = self._spans 

65 if trace_id: 

66 spans = [s for s in spans if s.trace_id == trace_id] 

67 return [s.to_dict() for s in spans]