from typing import Tuple import requests from requests import RequestException from app.alarms import get_or_create_alarm from app.extensions import db from app.models.alarms import AlarmState from app.models.mirrors import Proxy from app.terraform import BaseAutomation class AlarmProxyHTTPStatusAutomation(BaseAutomation): short_name = "alarm_http_status" description = "Check all deployed proxies for HTTP status code" frequency = 45 def automate(self, full: bool = False) -> Tuple[bool, str]: proxies = Proxy.query.filter(Proxy.destroyed.is_(None)) for proxy in proxies: try: if proxy.url is None: continue r = requests.get(proxy.url, allow_redirects=False, timeout=5) r.raise_for_status() alarm = get_or_create_alarm(proxy.brn, "http-status") if r.is_redirect: alarm.update_state( AlarmState.CRITICAL, f"{r.status_code} {r.reason}" ) else: alarm.update_state(AlarmState.OK, f"{r.status_code} {r.reason}") except requests.HTTPError: alarm = get_or_create_alarm(proxy.brn, "http-status") alarm.update_state(AlarmState.CRITICAL, f"{r.status_code} {r.reason}") except RequestException as e: alarm = get_or_create_alarm(proxy.brn, "http-status") alarm.update_state(AlarmState.CRITICAL, repr(e)) db.session.commit() return True, ""