ITWorx Pulse release export
bd774932d5
Public source validation / validate (push) Failing after 3m8s
48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent, dependency-free Pulse liveness check for an external monitor."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlparse
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
|
|
class NoRedirectHandler(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
def check(url: str, timeout: float = 5.0) -> tuple[bool, str]:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
|
|
return False, "endpoint must be an http(s) URL without embedded credentials"
|
|
if timeout < 1 or timeout > 30:
|
|
return False, "timeout must be between 1 and 30 seconds"
|
|
request = Request(url, headers={"Accept": "text/plain", "User-Agent": "pulse-deadman/1"}, method="GET")
|
|
try:
|
|
with build_opener(NoRedirectHandler).open(request, timeout=timeout) as response:
|
|
body = response.read(64).decode("utf-8", errors="replace").strip()
|
|
if response.status != 200 or body != "ok":
|
|
return False, f"unexpected liveness response ({response.status})"
|
|
except HTTPError as error:
|
|
return False, f"liveness returned HTTP {error.code}"
|
|
except (URLError, TimeoutError, OSError) as error:
|
|
return False, f"liveness request failed ({error.__class__.__name__})"
|
|
return True, "ok"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check Pulse /healthz from an independent monitor")
|
|
parser.add_argument("url", help="HTTPS URL to Pulse /healthz")
|
|
parser.add_argument("--timeout", type=float, default=5.0)
|
|
args = parser.parse_args()
|
|
ok, message = check(args.url, args.timeout)
|
|
print(message)
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|