Add URL health checks from config
This commit is contained in:
38
services/http_checks.py
Normal file
38
services/http_checks.py
Normal file
@@ -0,0 +1,38 @@
|
||||
import time
|
||||
from typing import Iterable, Tuple
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
|
||||
def get_url_checks(cfg) -> Iterable[Tuple[str, str]]:
|
||||
docker_cfg = cfg.get("docker", {})
|
||||
containers = docker_cfg.get("containers", {})
|
||||
for alias, value in containers.items():
|
||||
if isinstance(value, str):
|
||||
continue
|
||||
if isinstance(value, dict):
|
||||
url = value.get("url")
|
||||
if url:
|
||||
yield alias, url
|
||||
|
||||
|
||||
def check_url(url: str, timeout: int = 5) -> Tuple[bool, int | None, int | None, str | None]:
|
||||
start = time.time()
|
||||
req = Request(url, headers={"User-Agent": "tg-admin-bot"})
|
||||
try:
|
||||
with urlopen(req, timeout=timeout) as resp:
|
||||
status = int(resp.status)
|
||||
except HTTPError as e:
|
||||
status = int(e.code)
|
||||
elapsed_ms = int((time.time() - start) * 1000)
|
||||
return False, status, elapsed_ms, None
|
||||
except URLError as e:
|
||||
elapsed_ms = int((time.time() - start) * 1000)
|
||||
return False, None, elapsed_ms, str(e.reason)
|
||||
except Exception as e:
|
||||
elapsed_ms = int((time.time() - start) * 1000)
|
||||
return False, None, elapsed_ms, str(e)
|
||||
|
||||
elapsed_ms = int((time.time() - start) * 1000)
|
||||
ok = status < 400
|
||||
return ok, status, elapsed_ms, None
|
||||
Reference in New Issue
Block a user