diff --git a/sdk/nexent/monitor/monitoring.py b/sdk/nexent/monitor/monitoring.py index e9381665d1..36e8ba3e11 100644 --- a/sdk/nexent/monitor/monitoring.py +++ b/sdk/nexent/monitor/monitoring.py @@ -2496,11 +2496,13 @@ def __init__(self): os.getenv("MODEL_MONITORING_BATCH_SIZE", "100")) self._flush_interval: int = int( os.getenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "30")) + self._error_retry_delay: float = min(1.0, float(self._flush_interval)) self._consecutive_failures: int = 0 self._max_failures: int = 3 self._degraded_until: float = 0.0 self._last_flush_time: float = time.time() self._running: bool = False + self._stop_event = threading.Event() self._flush_thread: Optional[threading.Thread] = None self._lock = threading.Lock() @@ -2511,6 +2513,7 @@ def _start_flush_thread(self) -> None: with self._lock: if self._running: return + self._stop_event.clear() self._running = True self._flush_thread = threading.Thread( target=self._flush_loop, @@ -2526,6 +2529,7 @@ def add_record(self, record: dict) -> None: self._buffer.append(record) def _flush_loop(self) -> None: + retry_delay = 0.0 while self._running: try: now = time.time() @@ -2537,13 +2541,20 @@ def _flush_loop(self) -> None: if should_flush: self._flush_to_db() self._last_flush_time = now + retry_delay = 0.0 except Exception as e: logger.error(f"Error in monitoring flush loop: {e}") + # Back off from a short delay instead of always waiting a full + # flush interval, so a transient error does not stall retries. + retry_delay = min( + self._flush_interval, + self._error_retry_delay if retry_delay <= 0.0 + else retry_delay * 2, + ) - for _ in range(10): - if not self._running: - return - time.sleep(self._flush_interval / 10) + wait_time = retry_delay if retry_delay > 0.0 else self._flush_interval + if self._stop_event.wait(timeout=wait_time): + return def _flush_to_db(self) -> None: now = time.time() @@ -2623,6 +2634,7 @@ def _write_batch(self, batch: List[dict]) -> None: def stop(self) -> None: self._running = False + self._stop_event.set() if self._flush_thread and self._flush_thread.is_alive(): self._flush_thread.join(timeout=5) logger.info("Monitoring buffer flush thread stopped") diff --git a/test/sdk/monitor/test_monitoring_buffer.py b/test/sdk/monitor/test_monitoring_buffer.py new file mode 100644 index 0000000000..d6878c0349 --- /dev/null +++ b/test/sdk/monitor/test_monitoring_buffer.py @@ -0,0 +1,46 @@ +import time + +from sdk.nexent.monitor.monitoring import MonitoringRecordBuffer + + +def test_stop_interrupts_flush_interval(monkeypatch): + monkeypatch.setenv("ENABLE_MODEL_MONITORING", "true") + monkeypatch.setenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "20") + buffer = MonitoringRecordBuffer() + + time.sleep(0.1) + started = time.monotonic() + buffer.stop() + + # The legacy worst case was ~2s (one tenth of the 20s interval); a 2s bound + # still fails the old behaviour while leaving headroom on a loaded runner. + assert time.monotonic() - started < 2.0 + assert not buffer._flush_thread.is_alive() + + +def test_flush_error_does_not_delay_retry_by_a_full_interval(monkeypatch): + monkeypatch.setenv("ENABLE_MODEL_MONITORING", "false") + monkeypatch.setenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "20") + buffer = MonitoringRecordBuffer() + + waits = [] + + def fake_wait(timeout=None): + waits.append(timeout) + return len(waits) >= 3 + + monkeypatch.setattr(buffer._stop_event, "wait", fake_wait) + + def boom(): + raise RuntimeError("transient") + + monkeypatch.setattr(buffer, "_flush_to_db", boom) + buffer._buffer.append({"record": 1}) + buffer._last_flush_time = 0.0 + buffer._running = True + + buffer._flush_loop() + + # Retries back off from a short delay instead of always waiting the full + # 20s flush interval. + assert waits == [1.0, 2.0, 4.0]