Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/opentelemetry/trace/propagation/tracecontext.py: 29%

1# Copyright The OpenTelemetry Authors 

2# SPDX-License-Identifier: Apache-2.0 

3# 

4import re 

5 

6from opentelemetry import trace 

7from opentelemetry.context.context import Context 

8from opentelemetry.propagators import textmap 

9from opentelemetry.trace import format_span_id, format_trace_id 

10from opentelemetry.trace.span import TraceState 

11 

12 

13class TraceContextTextMapPropagator(textmap.TextMapPropagator): 

14 """Extracts and injects using w3c TraceContext's headers.""" 

15 

16 _TRACEPARENT_HEADER_NAME = "traceparent" 

17 _TRACESTATE_HEADER_NAME = "tracestate" 

18 _TRACEPARENT_HEADER_FORMAT = ( 

19 "^[ \t]*([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})" 

20 + "(-.*)?[ \t]*$" 

21 ) 

22 _TRACEPARENT_HEADER_FORMAT_RE = re.compile(_TRACEPARENT_HEADER_FORMAT) 

23 

24 def extract( 

25 self, 

26 carrier: textmap.CarrierT, 

27 context: Context | None = None, 

28 getter: textmap.Getter[textmap.CarrierT] = textmap.default_getter, 

29 ) -> Context: 

30 """Extracts SpanContext from the carrier. 

31 

32 See `opentelemetry.propagators.textmap.TextMapPropagator.extract` 

33 """ 

34 if context is None: 

35 context = Context() 

36 

37 header = getter.get(carrier, self._TRACEPARENT_HEADER_NAME) 

38 

39 if not header: 

40 return context 

41 

42 match = re.search(self._TRACEPARENT_HEADER_FORMAT_RE, header[0]) 

43 if not match: 

44 return context 

45 

46 version: str = match.group(1) 

47 trace_id: str = match.group(2) 

48 span_id: str = match.group(3) 

49 trace_flags: str = match.group(4) 

50 

51 if trace_id == "0" * 32 or span_id == "0" * 16: 

52 return context 

53 

54 if version == "00": 

55 if match.group(5): # type: ignore 

56 return context 

57 if version == "ff": 

58 return context 

59 

60 tracestate_headers = getter.get(carrier, self._TRACESTATE_HEADER_NAME) 

61 if tracestate_headers is None: 

62 tracestate = None 

63 else: 

64 tracestate = TraceState.from_header(tracestate_headers) 

65 

66 span_context = trace.SpanContext( 

67 trace_id=int(trace_id, 16), 

68 span_id=int(span_id, 16), 

69 is_remote=True, 

70 trace_flags=trace.TraceFlags(int(trace_flags, 16)), 

71 trace_state=tracestate, 

72 ) 

73 return trace.set_span_in_context( 

74 trace.NonRecordingSpan(span_context), context 

75 ) 

76 

77 def inject( 

78 self, 

79 carrier: textmap.CarrierT, 

80 context: Context | None = None, 

81 setter: textmap.Setter[textmap.CarrierT] = textmap.default_setter, 

82 ) -> None: 

83 """Injects SpanContext into the carrier. 

84 

85 See `opentelemetry.propagators.textmap.TextMapPropagator.inject` 

86 """ 

87 span = trace.get_current_span(context) 

88 span_context = span.get_span_context() 

89 if span_context == trace.INVALID_SPAN_CONTEXT: 

90 return 

91 traceparent_string = f"00-{format_trace_id(span_context.trace_id)}-{format_span_id(span_context.span_id)}-{span_context.trace_flags:02x}" 

92 setter.set(carrier, self._TRACEPARENT_HEADER_NAME, traceparent_string) 

93 if span_context.trace_state: 

94 tracestate_string = span_context.trace_state.to_header() 

95 setter.set( 

96 carrier, self._TRACESTATE_HEADER_NAME, tracestate_string 

97 ) 

98 

99 @property 

100 def fields(self) -> set[str]: 

101 """Returns a set with the fields set in `inject`. 

102 

103 See 

104 `opentelemetry.propagators.textmap.TextMapPropagator.fields` 

105 """ 

106 return {self._TRACEPARENT_HEADER_NAME, self._TRACESTATE_HEADER_NAME}