Coverage for qreward / utils / scheduler / telemetry.py: 97.44%

1"""Optional OpenTelemetry integration for schedule metrics export. 

2 

3All OTel imports are guarded by try/except so this module works 

4even when opentelemetry packages are not installed. 

5""" 

6 

7import os (empty)

8import logging (empty)

9from typing import Any, Optional (empty)

10 

11logger = logging.getLogger(__name__) (empty)

12 

13_OTEL_AVAILABLE = False (empty)

14_meter_mod: Any = None (empty)

15_trace_mod: Any = None (empty)

16 

17try: (empty)

18 from opentelemetry import metrics as _meter_mod # type: ignore[no-redef] (empty)

19 from opentelemetry import trace as _trace_mod # type: ignore[no-redef] 

20 _OTEL_AVAILABLE = True 

21except ImportError: (empty)

22 pass (empty)

23 

24_ENV_ENABLED_KEY = "QREWARD_OTEL_ENABLED" (empty)

25 

26 

27class TelemetryExporter: (empty)

28 """Exports ScheduleMetrics to OpenTelemetry counters, histograms and spans. 

29 

30 When OTel SDK is not installed or disabled via environment variable, 

31 all methods are no-ops. 

32 

33 Example: 

34 >>> exporter = TelemetryExporter() 

35 >>> exporter.record(metrics) # records counters + histogram 

36 >>> with exporter.span("my_func") as span: 

37 ... span.set_attribute("retry_count", 3) 

38 """ 

39 

40 def __init__( (empty)

41 self, 

42 meter_name: str = "qreward.scheduler", 

43 tracer_name: str = "qreward.scheduler", 

44 ) -> None: 

45 self._enabled = self._check_enabled() (empty)

46 self._meter: Any = None (empty)

47 self._tracer: Any = None (empty)

48 self._total_calls_counter: Any = None (empty)

49 self._success_counter: Any = None (empty)

50 self._failure_counter: Any = None (empty)

51 self._latency_histogram: Any = None (empty)

52 

53 if self._enabled: (empty)

54 self._meter = _meter_mod.get_meter(meter_name) (empty)

55 self._tracer = _trace_mod.get_tracer(tracer_name) (empty)

56 self._total_calls_counter = self._meter.create_counter( (empty)

57 "qreward.schedule.total_calls", 

58 description="Total number of schedule invocations", 

59 ) 

60 self._success_counter = self._meter.create_counter( (empty)

61 "qreward.schedule.success_count", 

62 description="Number of successful schedule completions", 

63 ) 

64 self._failure_counter = self._meter.create_counter( (empty)

65 "qreward.schedule.failure_count", 

66 description="Number of failed schedule attempts", 

67 ) 

68 self._latency_histogram = self._meter.create_histogram( (empty)

69 "qreward.schedule.latency_ms", 

70 unit="ms", 

71 description="Schedule invocation latency in milliseconds", 

72 ) 

73 

74 @classmethod (empty)

75 def is_available(cls) -> bool: (empty)

76 """Check if OpenTelemetry SDK is installed and enabled.""" 

77 return _OTEL_AVAILABLE and cls._check_enabled() (empty)

78 

79 @staticmethod (empty)

80 def _check_enabled() -> bool: (empty)

81 """Check if OTel is available and not disabled by env var.""" 

82 if not _OTEL_AVAILABLE: (empty)

83 return False (empty)

84 env_val = os.environ.get(_ENV_ENABLED_KEY, "true").lower() (empty)

85 return env_val not in ("false", "0", "no", "off") (empty)

86 

87 def record(self, metrics: Any) -> None: (empty)

88 """Record schedule metrics to OTel counters and histogram. 

89 

90 Args: 

91 metrics: A ScheduleMetrics instance. 

92 """ 

93 if not self._enabled: (empty)

94 return (empty)

95 self._total_calls_counter.add(metrics.total_calls) (empty)

96 self._success_counter.add(metrics.success_count) (empty)

97 self._failure_counter.add(metrics.failure_count) (empty)

98 self._latency_histogram.record(metrics.total_latency_ms) (empty)

99 

100 def start_span( (empty)

101 self, name: str, attributes: Optional[dict[str, Any]] = None 

102 ) -> Any: 

103 """Start a new OTel span. 

104 

105 Args: 

106 name: Span name (typically the decorated function name). 

107 attributes: Optional span attributes. 

108 

109 Returns: 

110 A span object, or a no-op context if OTel is disabled. 

111 """ 

112 if not self._enabled or self._tracer is None: (empty)

113 return _NoOpSpan() (empty)

114 return self._tracer.start_span(name, attributes=attributes) (empty)

115 

116 def end_span(self, span: Any, metrics: Any) -> None: (empty)

117 """End a span and attach metrics as attributes. 

118 

119 Args: 

120 span: The span to end. 

121 metrics: A ScheduleMetrics instance. 

122 """ 

123 if not self._enabled or isinstance(span, _NoOpSpan): (empty)

124 return (empty)

125 span.set_attribute("qreward.total_calls", metrics.total_calls) (empty)

126 span.set_attribute( (empty)

127 "qreward.success_count", metrics.success_count 

128 ) 

129 span.set_attribute( (empty)

130 "qreward.failure_count", metrics.failure_count 

131 ) 

132 span.set_attribute("qreward.retry_count", metrics.retry_count) (empty)

133 span.set_attribute( (empty)

134 "qreward.total_latency_ms", metrics.total_latency_ms 

135 ) 

136 span.end() (empty)

137 

138 

139class _NoOpSpan: (empty)

140 """No-op span used when OTel is not available.""" 

141 

142 def set_attribute(self, key: str, value: Any) -> None: (empty)

143 pass (empty)

144 

145 def end(self) -> None: (empty)

146 pass (empty)

147 

148 def __enter__(self) -> "_NoOpSpan": (empty)

149 return self (empty)

150 

151 def __exit__(self, *args: Any) -> None: (empty)

152 pass (empty)