(service_name: str, timeout: int = 30, poll_interval: int = 1)
| 2654 | tmp = _wt_seen_path() + '.tmp' |
| 2655 | with open(tmp, 'w') as f: |
| 2656 | json.dump(_wt_seen, f) |
| 2657 | os.replace(tmp, _wt_seen_path()) |
| 2658 | except Exception as exc: |
| 2659 | logger.debug(f"[watchtower] seen save failed: {exc}") |
| 2660 | |
| 2661 | |
| 2662 | def _watchtower_check_once(): |
| 2663 | """One poll cycle: read the delta from every watcher log, page the new |
| 2664 | high/critical findings (deduped + merged), and refresh the summary.""" |
| 2665 | import watchtower as _wtmod |
| 2666 | cfg = shared_data.config |
| 2667 | wt = _wt_get() |
| 2668 | new = wt.poll() |
| 2669 | |
| 2670 | floor = _wtmod.SEV_RANK.get(cfg.get('watchtower_notify_min_severity', 'high'), 3) |
| 2671 | try: |
no test coverage detected