Coverage for qreward / utils / scheduler / config_watcher.py: 97.97%

1"""Configuration watcher for hot-reloading schedule parameters at runtime.""" 

2 

3import json (empty)

4import logging (empty)

5import os (empty)

6import threading (empty)

7import time (empty)

8from typing import Any, Callable, Optional (empty)

9 

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

11 

12_ENV_PREFIX = "QREWARD_SCHEDULE_" (empty)

13_UPDATABLE_FIELDS = frozenset({ (empty)

14 "timeout", "retry_times", "retry_interval", "limit_size", "limit_window", 

15 "hedged_request_time", "hedged_request_proportion", 

16 "hedged_request_max_times", "speed_up_max_multiply", "debug", 

17 "adaptive_limit", "adaptive_limit_min", "adaptive_limit_max", 

18 "adaptive_error_threshold", "adaptive_latency_threshold", 

19 "adaptive_window_seconds", "priority", 

20}) 

21 

22 

23class ConfigWatcher: (empty)

24 """Watches a configuration source and hot-reloads ScheduleConfig changes. 

25 

26 Supports three source modes: 

27 - **file**: Watches a JSON file for changes (polls by mtime). 

28 - **env**: Reads environment variables with ``QREWARD_SCHEDULE_`` prefix. 

29 - **callback**: Invokes a user-supplied callable that returns a dict. 

30 

31 Example: 

32 >>> watcher = ConfigWatcher( 

33 ... config, source="file", file_path="config.json" 

34 ... ) 

35 >>> watcher.start() 

36 >>> # ... later ... 

37 >>> watcher.stop() 

38 """ 

39 

40 def __init__( (empty)

41 self, 

42 config: Any, 

43 source: str = "callback", 

44 file_path: Optional[str] = None, 

45 callback: Optional[Callable[[], dict[str, Any]]] = None, 

46 poll_interval: float = 5.0, 

47 cooldown: float = 5.0, 

48 ) -> None: 

49 """Initialize the config watcher. 

50 

51 Args: 

52 config: The ScheduleConfig instance to update. 

53 source: Source mode - "file", "env", or "callback". 

54 file_path: Path to JSON config file (required for "file" mode). 

55 callback: Callable returning config dict (required for 

56 "callback" mode). 

57 poll_interval: Seconds between polls. 

58 cooldown: Minimum seconds between successive updates. 

59 """ 

60 if source not in ("file", "env", "callback"): (empty)

61 raise ValueError( (empty)

62 f"source must be 'file', 'env', or 'callback', got {source!r}" 

63 ) 

64 if source == "file" and not file_path: (empty)

65 raise ValueError("file_path is required for 'file' source") (empty)

66 if source == "callback" and callback is None: (empty)

67 raise ValueError("callback is required for 'callback' source") (empty)

68 

69 self._config = config (empty)

70 self._source = source (empty)

71 self._file_path = file_path (empty)

72 self._callback = callback (empty)

73 self._poll_interval = poll_interval (empty)

74 self._cooldown = cooldown (empty)

75 self._last_update_time: float = -float("inf") (empty)

76 self._last_mtime: float = 0 (empty)

77 self._stop_event = threading.Event() (empty)

78 self._thread: Optional[threading.Thread] = None (empty)

79 

80 def start(self) -> None: (empty)

81 """Start the watcher daemon thread.""" 

82 if self._thread is not None and self._thread.is_alive(): (empty)

83 return (empty)

84 self._stop_event.clear() (empty)

85 self._thread = threading.Thread(target=self._poll_loop, daemon=True) (empty)

86 self._thread.start() (empty)

87 logger.debug("ConfigWatcher started (source=%s)", self._source) (empty)

88 

89 def stop(self) -> None: (empty)

90 """Stop the watcher thread.""" 

91 self._stop_event.set() (empty)

92 if self._thread is not None: 92 ↛ 95line 92 didn't jump to line 95 because the condition on line 92 was always true(empty)

93 self._thread.join(timeout=self._poll_interval + 1) (empty)

94 self._thread = None (empty)

95 logger.debug("ConfigWatcher stopped") (empty)

96 

97 @property (empty)

98 def is_running(self) -> bool: (empty)

99 """Check if the watcher thread is alive.""" 

100 return self._thread is not None and self._thread.is_alive() (empty)

101 

102 def poll_once(self) -> bool: (empty)

103 """Perform a single poll and update if changed. 

104 

105 Returns: 

106 True if config was updated, False otherwise. 

107 """ 

108 now = time.monotonic() (empty)

109 if now - self._last_update_time < self._cooldown: (empty)

110 return False (empty)

111 

112 new_values = self._read_source() (empty)

113 if not new_values: (empty)

114 return False (empty)

115 

116 filtered = { (empty)

117 k: v 

118 for k, v in new_values.items() 

119 if k in _UPDATABLE_FIELDS 

120 } 

121 if not filtered: (empty)

122 return False (empty)

123 

124 self._config.update(**filtered) (empty)

125 self._last_update_time = now (empty)

126 logger.debug("Config updated: %s", filtered) (empty)

127 return True (empty)

128 

129 def _poll_loop(self) -> None: (empty)

130 """Background polling loop.""" 

131 while not self._stop_event.is_set(): (empty)

132 try: (empty)

133 self.poll_once() (empty)

134 except Exception as exc: (empty)

135 logger.warning("ConfigWatcher poll error: %s", exc) (empty)

136 self._stop_event.wait(self._poll_interval) (empty)

137 

138 def _read_source(self) -> Optional[dict[str, Any]]: (empty)

139 """Read configuration from the configured source.""" 

140 if self._source == "file": (empty)

141 return self._read_file() (empty)

142 elif self._source == "env": (empty)

143 return self._read_env() (empty)

144 elif self._source == "callback": 144 ↛ 146line 144 didn't jump to line 146 because the condition on line 144 was always true(empty)

145 return self._read_callback() (empty)

146 return None 

147 

148 def _read_file(self) -> Optional[dict[str, Any]]: (empty)

149 """Read config from a JSON file, only if mtime changed.""" 

150 if not self._file_path or not os.path.isfile(self._file_path): (empty)

151 return None (empty)

152 mtime = os.path.getmtime(self._file_path) (empty)

153 if mtime == self._last_mtime: (empty)

154 return None (empty)

155 self._last_mtime = mtime (empty)

156 with open(self._file_path, "r") as fh: (empty)

157 return json.load(fh) (empty)

158 

159 def _read_env(self) -> Optional[dict[str, Any]]: (empty)

160 """Read config from environment variables with QREWARD_SCHEDULE_ 

161 prefix.""" 

162 result: dict[str, Any] = {} (empty)

163 for key, value in os.environ.items(): (empty)

164 if not key.startswith(_ENV_PREFIX): (empty)

165 continue (empty)

166 field_name = key[len(_ENV_PREFIX):].lower() (empty)

167 if field_name not in _UPDATABLE_FIELDS: (empty)

168 continue (empty)

169 result[field_name] = _coerce_value(field_name, value) (empty)

170 return result or None (empty)

171 

172 def _read_callback(self) -> Optional[dict[str, Any]]: (empty)

173 """Read config from user callback.""" 

174 if self._callback is None: (empty)

175 return None (empty)

176 return self._callback() (empty)

177 

178 

179def _coerce_value(field_name: str, raw: str) -> Any: (empty)

180 """Coerce a string environment variable value to the correct type.""" 

181 bool_fields = {"debug", "adaptive_limit"} (empty)

182 int_fields = { (empty)

183 "retry_times", "limit_size", "hedged_request_max_times", 

184 "speed_up_max_multiply", "adaptive_limit_min", 

185 "adaptive_limit_max", "priority", 

186 } 

187 float_fields = { (empty)

188 "timeout", "retry_interval", "limit_window", 

189 "hedged_request_time", "hedged_request_proportion", 

190 "adaptive_error_threshold", "adaptive_latency_threshold", 

191 "adaptive_window_seconds", 

192 } 

193 if field_name in bool_fields: (empty)

194 return raw.lower() in ("true", "1", "yes", "on") (empty)

195 if field_name in int_fields: (empty)

196 return int(raw) (empty)

197 if field_name in float_fields: (empty)

198 return float(raw) (empty)

199 return raw (empty)