maintenance-tools/maintenance_service_http_monitoring/models/service_instance.py

import logging
from datetime import timedelta

from odoo import api, fields, models

try:
    import requests
except ImportError:
    requests = None

_logger = logging.getLogger(__name__)

HTTP_CHECK_TIMEOUT = 20  # seconds
HTTP_KO_CONFIRMATION_DELAY = timedelta(minutes=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,
    )
    http_maintenance_request = fields.Many2one(
        "maintenance.request",
        string="HTTP Maintenance Request",
        readonly=True,
    )
    http_first_ko_at = fields.Datetime(
        string="First HTTP KO Check (current streak)",
        readonly=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, and maintains http_first_ko_at (the start of the current KO
        streak, reset as soon as the service is OK again). Does NOT create
        maintenance.request — that decision belongs to cron_confirm_http_ko_services,
        once the service has been continuously KO for at least
        HTTP_KO_CONFIRMATION_DELAY, to avoid flagging transient outages (e.g. a short
        server overload) as real incidents.
        """
        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)
            vals = {
                "last_http_status_code": status_code,
                "last_http_check_date": now,
                "http_status_ok": status_ok,
            }
            if status_ok:
                vals["http_first_ko_at"] = False
            elif not rec.http_first_ko_at:
                vals["http_first_ko_at"] = now
            rec.write(vals)
            if not status_ok:
                ko_records |= rec
        return ko_records

    def _close_http_maintenance_request(self):
        """
        Close the open maintenance.request for each recovered service.

        Moves the request to the first done stage, posts a chatter note as OdooBot, and
        clears http_maintenance_request on the service instance.
        """
        done_stage = self.env["maintenance.stage"].search(
            [("done", "=", True)], limit=1
        )
        if not done_stage:
            return
        odoobot = self.env.ref("base.partner_root")
        for rec in self:
            request = rec.http_maintenance_request
            if not request or request.stage_id.done:
                continue
            request.sudo().write({"stage_id": done_stage.id})
            request.sudo().message_post(
                body=(
                    f"Service {rec.service_url} is back online. "
                    "This request has been automatically closed by the monitoring cron."
                ),
                author_id=odoobot.id,
                message_type="comment",
                subtype_xmlid="mail.mt_note",
            )
            rec.http_maintenance_request = False

    @api.model
    def cron_check_http_services(self):
        """
        Discovery sweep: check all active services with a URL.

        This cron only detects outages — it flags the start of a KO streak through
        http_first_ko_at (via check_http_status) and auto-resolves services that
        recovered. Maintenance requests are created by the separate, faster
        cron_confirm_http_ko_services once the outage is confirmed.
        """
        domain = [
            ("active", "=", True),
            ("service_url", "!=", False),
            ("equipment_id", "!=", False),
        ]
        services = self.search(domain).filtered(
            lambda s: not s.equipment_id.maintenance_mode
        )

        # Snapshot services that currently have an open request before the check
        services_with_open_request = services.filtered(
            lambda s: s.http_maintenance_request
            and not s.http_maintenance_request.stage_id.done
        )

        services.check_http_status()

        # Auto-resolve services that recovered
        recovered = services_with_open_request.filtered(lambda s: s.http_status_ok)
        if recovered:
            recovered._close_http_maintenance_request()

    @api.model
    def cron_confirm_http_ko_services(self):
        """
        Confirmation pass: re-check services currently flagged KO and create their
        maintenance.request once the outage is confirmed.

        Only services without an open request are considered (re-checking a service
        already under an open request is pointless). A service must be continuously
        KO for at least HTTP_KO_CONFIRMATION_DELAY: a service that recovered in the
        meantime resets http_first_ko_at via check_http_status and no request is
        created. Services already under an open request are handled by the slower
        discovery cron, which also auto-resolves them when they recover.
        """
        domain = [
            ("active", "=", True),
            ("service_url", "!=", False),
            ("equipment_id", "!=", False),
            ("http_first_ko_at", "!=", False),
            ("http_maintenance_request", "=", False),
        ]
        services = self.search(domain).filtered(
            lambda s: not s.equipment_id.maintenance_mode
        )

        ko_services = services.check_http_status()

        confirmed_ko = ko_services.filtered(
            lambda s: s.last_http_check_date - s.http_first_ko_at
            >= HTTP_KO_CONFIRMATION_DELAY
        )
        for service in confirmed_ko:
            service.equipment_id.create_http_maintenance_request(service)