102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
import logging
|
|
import time
|
|
|
|
from odoo import api, fields, models
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
requests = None
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
HTTP_CHECK_TIMEOUT = 10 # seconds
|
|
HTTP_RETRY_DELAY = 2 # seconds between pass 1 and pass 2
|
|
|
|
|
|
class ServiceInstance(models.Model):
|
|
_inherit = "service.instance"
|
|
|
|
last_http_status_code = fields.Integer(
|
|
string="Last HTTP Status Code",
|
|
readonly=True,
|
|
default=0,
|
|
)
|
|
last_http_check_date = fields.Datetime(
|
|
string="Last HTTP Check Date",
|
|
readonly=True,
|
|
)
|
|
http_status_ok = fields.Boolean(
|
|
string="HTTP Status OK",
|
|
readonly=True,
|
|
default=True,
|
|
)
|
|
|
|
def check_http_status(self):
|
|
"""
|
|
Perform HTTP check for each record and return the KO recordset.
|
|
|
|
Writes last_http_status_code, last_http_check_date and http_status_ok on every
|
|
checked record. Does NOT create maintenance.request — that decision belongs to
|
|
the caller (cron) after optional retry logic.
|
|
"""
|
|
ko_records = self.browse()
|
|
for rec in self:
|
|
if not rec.service_url or not rec.equipment_id:
|
|
continue
|
|
if rec.equipment_id.maintenance_mode:
|
|
continue
|
|
status_ok = False
|
|
status_code = -1
|
|
now = fields.Datetime.now()
|
|
url = rec.service_url
|
|
if not url.lower().startswith("https://"):
|
|
url = "https://" + url.removeprefix("http://").removeprefix("HTTP://")
|
|
try:
|
|
response = requests.get(url, timeout=HTTP_CHECK_TIMEOUT)
|
|
status_code = response.status_code
|
|
status_ok = status_code == 200
|
|
except requests.exceptions.RequestException as e:
|
|
_logger.warning("HTTP check failed for %s: %s", rec.service_url, e)
|
|
rec.write(
|
|
{
|
|
"last_http_status_code": status_code,
|
|
"last_http_check_date": now,
|
|
"http_status_ok": status_ok,
|
|
}
|
|
)
|
|
if not status_ok:
|
|
ko_records |= rec
|
|
return ko_records
|
|
|
|
@api.model
|
|
def cron_check_http_services(self):
|
|
"""
|
|
Check all active services with a URL, with one retry on failure.
|
|
|
|
Pass 1: test every eligible service.
|
|
If any fail, wait HTTP_RETRY_DELAY seconds then retest only the KO ones.
|
|
maintenance.request is created only for services that fail both passes,
|
|
reducing noise from transient HTTP errors.
|
|
"""
|
|
domain = [
|
|
("active", "=", True),
|
|
("service_url", "!=", False),
|
|
("equipment_id", "!=", False),
|
|
]
|
|
services = self.search(domain).filtered(
|
|
lambda s: not s.equipment_id.maintenance_mode
|
|
)
|
|
|
|
ko_after_pass1 = services.check_http_status()
|
|
|
|
if not ko_after_pass1:
|
|
return
|
|
|
|
time.sleep(HTTP_RETRY_DELAY)
|
|
|
|
ko_confirmed = ko_after_pass1.check_http_status()
|
|
|
|
for service in ko_confirmed:
|
|
service.equipment_id.create_http_maintenance_request([service])
|