patrickandClaude Sonnet 5.5
453c077386
Lesender Selbsttest (Version, Services, Hosts, Downtimes) beim Start und täglich 04:00, Seite /selftest, Teilcheck checkmk_api in /health, Chat-Meldung bei Wechsel. Toleranter Antwort-Parser (value/members, state_type int/String), Fallbacks für Host-Abfrage und Downtime-Liste, Fehlerhinweise je HTTP-Status. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
from collections import Counter
|
|
from datetime import datetime
|
|
from html import escape
|
|
from typing import Sequence
|
|
|
|
from ..checkmk import FAILURE_UNREACHABLE
|
|
from ..models import HOST_SERVICE_NAME, STATE_OK, StateRow, display_state, severity
|
|
from ..poller.outage import OutageNotice, RestoredNotice
|
|
from .delta import CLOCK_FORMAT, format_duration
|
|
|
|
SYSTEM_HOST = "checkbot"
|
|
SERVICE_CHECKMK = "Checkmk-Verbindung"
|
|
SERVICE_BASELINE = "Baseline"
|
|
SERVICE_SEND_LIMIT = "Sendelimit"
|
|
SERVICE_SELFTEST = "Selbsttest"
|
|
TOP_COUNT = 10
|
|
BULLET = "•"
|
|
|
|
|
|
def _clock(ts: int) -> str:
|
|
return datetime.fromtimestamp(ts).strftime(CLOCK_FORMAT)
|
|
|
|
|
|
def outage_plain(notice: OutageNotice) -> str:
|
|
since = _clock(notice.since)
|
|
if notice.kind == FAILURE_UNREACHABLE:
|
|
return f"Checkmk nicht erreichbar seit {since}"
|
|
return f"Antwort von Checkmk unerwartet seit {since}: {notice.brief}"
|
|
|
|
|
|
def restored_plain(notice: RestoredNotice) -> str:
|
|
return f"Checkmk wieder erreichbar nach {format_duration(notice.now - notice.since)}"
|
|
|
|
|
|
def wrap(plain: str) -> str:
|
|
return f"<b>{escape(SYSTEM_HOST)}</b>\n{escape(plain)}"
|
|
|
|
|
|
def _rank(row: StateRow) -> tuple:
|
|
return (row.service != HOST_SERVICE_NAME, -severity(row.state), row.host, row.service)
|
|
|
|
|
|
def _count_line(title: str, items: Sequence[StateRow], service: str) -> str:
|
|
counts = Counter(r.state for r in items)
|
|
parts = [f"{display_state(service, state)} {number}"
|
|
for state, number in sorted(counts.items(), key=lambda item: -severity(item[0]))]
|
|
return f"{title}: {', '.join(parts) or 'keine'}"
|
|
|
|
|
|
def _counts(rows: Sequence[StateRow]) -> list[str]:
|
|
hosts = [r for r in rows if r.service == HOST_SERVICE_NAME]
|
|
services = [r for r in rows if r.service != HOST_SERVICE_NAME]
|
|
return [_count_line("Hosts", hosts, HOST_SERVICE_NAME), _count_line("Services", services, "")]
|
|
|
|
|
|
def _top_line(row: StateRow) -> str:
|
|
if row.service == HOST_SERVICE_NAME:
|
|
return f"{BULLET} <b>{escape(row.host)}</b> Host {escape(display_state(row.service, row.state))}"
|
|
state = escape(display_state(row.service, row.state))
|
|
return f"{BULLET} <b>{escape(row.host)}</b> {escape(row.service)}: {state}"
|
|
|
|
|
|
def baseline_plain(rows: Sequence[StateRow]) -> str:
|
|
return f"Baseline aufgenommen: {len([r for r in rows if r.state != STATE_OK])} offene Probleme"
|
|
|
|
|
|
def baseline_message(rows: Sequence[StateRow]) -> str:
|
|
open_ = sorted((r for r in rows if r.state != STATE_OK), key=_rank)
|
|
blocks = [f"<b>{escape(SYSTEM_HOST)}</b>",
|
|
f"<b>{escape(baseline_plain(open_))}</b>",
|
|
"Still aufgenommen, bestehende Probleme werden nicht einzeln gemeldet; Recoveries kommen später.",
|
|
*(escape(line) for line in _counts(open_))]
|
|
if open_:
|
|
blocks.append(f"<b>Die {min(TOP_COUNT, len(open_))} kritischsten:</b>")
|
|
blocks += [_top_line(row) for row in open_[:TOP_COUNT]]
|
|
if len(open_) > TOP_COUNT:
|
|
blocks.append(escape(f"… und {len(open_) - TOP_COUNT} weitere"))
|
|
return "\n".join(blocks)
|
|
|
|
|
|
def overflow_plain(count: int) -> str:
|
|
return f"{count} weitere Alarme in der letzten Minute, siehe Verlauf" if count != 1 else \
|
|
"1 weiterer Alarm in der letzten Minute, siehe Verlauf"
|
|
|
|
|
|
def overflow_message(count: int, link: str | None = None) -> str:
|
|
text = wrap(overflow_plain(count))
|
|
return f"{text}\n{escape(link)}" if link else text
|