92 lines
2.1 KiB
Python
92 lines
2.1 KiB
Python
# checks.py
|
|
import socket
|
|
import threading
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
|
|
class SystemChecks:
|
|
def __init__(self, network_targets, ziti_target):
|
|
self.network_targets = network_targets
|
|
self.ziti_target = ziti_target
|
|
|
|
self.network_online = None
|
|
self.ziti_online = None
|
|
|
|
self._stop_event = threading.Event()
|
|
self._thread = None
|
|
|
|
@property
|
|
def ready(self):
|
|
"""Return True when network & ziti are both available"""
|
|
return (
|
|
self.network_online is True
|
|
and self.ziti_online is True
|
|
)
|
|
|
|
def start(self):
|
|
"""Start background status polling"""
|
|
if self._thread is not None:
|
|
return
|
|
|
|
self._thread = threading.Thread(
|
|
target=self._poll,
|
|
daemon=True
|
|
)
|
|
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
"""Stop background status polling"""
|
|
self._stop_event.set()
|
|
|
|
def _poll(self):
|
|
"""Continuously check system status"""
|
|
while not self._stop_event.is_set():
|
|
self.network_online = self.check_network_status()
|
|
self.ziti_online = self.check_ziti_status()
|
|
|
|
# print(
|
|
# f"System status => "
|
|
# f"Network: {self.network_online}, "
|
|
# f"Ziti: {self.ziti_online}"
|
|
# )
|
|
|
|
if self.ready:
|
|
interval = 30
|
|
else:
|
|
interval = 3
|
|
|
|
self._stop_event.wait(interval)
|
|
|
|
def check_network_status(self):
|
|
"""Loops through configured network targets and return True if any respond"""
|
|
for host, port in self.network_targets:
|
|
if self._tcp_ping(host, int(port), timeout=1.5):
|
|
return True
|
|
|
|
return False
|
|
|
|
def check_ziti_status(self):
|
|
"""Makes HTTP request to Ziti endpoint"""
|
|
if not self.ziti_target:
|
|
return False
|
|
|
|
try:
|
|
request = urllib.request.Request(
|
|
self.ziti_target,
|
|
method="GET",
|
|
)
|
|
|
|
with urllib.request.urlopen(request, timeout=2.0) as response:
|
|
return 200 <= response.status < 300
|
|
|
|
except (urllib.error.URLError, TimeoutError, OSError):
|
|
return False
|
|
|
|
def _tcp_ping(self, host, port, timeout=1):
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
return True
|
|
except OSError:
|
|
return False |