157 lines
5.2 KiB
Python
157 lines
5.2 KiB
Python
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from .constants import (CHECKMK_CA_FILE, CHECKMK_SECRET, CHECKMK_SITE, CHECKMK_URL,
|
|
CHECKMK_USER, CHECKMK_VERIFY_TLS, FLAG_OFF, FLAG_ON)
|
|
|
|
API_PATH = "/check_mk/api/1.0"
|
|
SERVICES_PATH = "/domain-types/service/collections/all"
|
|
VERSION_PATH = "/version"
|
|
DOWNTIMES_PATH = "/domain-types/downtime/collections/all"
|
|
DEFAULT_TIMEOUT = 15.0
|
|
STATE_OK = 0
|
|
|
|
COLUMNS = [
|
|
"host_name",
|
|
"description",
|
|
"state",
|
|
"state_type",
|
|
"plugin_output",
|
|
"last_state_change",
|
|
"scheduled_downtime_depth",
|
|
"is_flapping",
|
|
]
|
|
|
|
NOT_OK_QUERY = {"op": "!=", "left": "state", "right": str(STATE_OK)}
|
|
|
|
|
|
class CheckmkError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServiceState:
|
|
host: str
|
|
service: str
|
|
state: int
|
|
hard: bool
|
|
output: str
|
|
last_state_change: int
|
|
in_downtime: bool
|
|
flapping: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Downtime:
|
|
host: str
|
|
service: str | None
|
|
start: str
|
|
end: str
|
|
comment: str
|
|
author: str
|
|
|
|
|
|
def _parse_downtime(extensions: dict) -> Downtime:
|
|
return Downtime(
|
|
host=extensions.get("host_name", ""),
|
|
service=extensions.get("service_description") or None,
|
|
start=str(extensions.get("start_time", "")),
|
|
end=str(extensions.get("end_time", "")),
|
|
comment=extensions.get("comment", ""),
|
|
author=extensions.get("author", ""),
|
|
)
|
|
|
|
|
|
def _equals(column: str, value: str) -> dict:
|
|
return {"op": "=", "left": column, "right": value}
|
|
|
|
|
|
def _is_hard(value) -> bool:
|
|
return str(value).lower() in ("1", "hard")
|
|
|
|
|
|
def _parse_service(extensions: dict) -> ServiceState:
|
|
return ServiceState(
|
|
host=extensions["host_name"],
|
|
service=extensions["description"],
|
|
state=int(extensions["state"]),
|
|
hard=_is_hard(extensions.get("state_type", 1)),
|
|
output=extensions.get("plugin_output", ""),
|
|
last_state_change=int(extensions.get("last_state_change", 0)),
|
|
in_downtime=int(extensions.get("scheduled_downtime_depth", 0)) > 0,
|
|
flapping=bool(int(extensions.get("is_flapping", 0))),
|
|
)
|
|
|
|
|
|
def is_configured(store) -> bool:
|
|
return all(store.get(key) for key in (CHECKMK_URL, CHECKMK_SITE, CHECKMK_USER, CHECKMK_SECRET))
|
|
|
|
|
|
class CheckmkClient:
|
|
def __init__(self, base_url: str, site: str, user: str, secret: str,
|
|
verify: bool | str = True, timeout: float = DEFAULT_TIMEOUT,
|
|
transport: httpx.AsyncBaseTransport | None = None):
|
|
self._client = httpx.AsyncClient(
|
|
base_url=f"{base_url.rstrip('/')}/{site}{API_PATH}",
|
|
headers={"Authorization": f"Bearer {user} {secret}", "Accept": "application/json"},
|
|
verify=verify,
|
|
timeout=timeout,
|
|
transport=transport,
|
|
)
|
|
|
|
@classmethod
|
|
def from_settings(cls, store) -> "CheckmkClient | None":
|
|
if not is_configured(store):
|
|
return None
|
|
url, site = store.get(CHECKMK_URL), store.get(CHECKMK_SITE)
|
|
user, secret = store.get(CHECKMK_USER), store.get(CHECKMK_SECRET)
|
|
if store.get(CHECKMK_VERIFY_TLS, FLAG_ON) == FLAG_OFF:
|
|
verify: bool | str = False
|
|
else:
|
|
verify = store.get(CHECKMK_CA_FILE) or True
|
|
return cls(url, site, user, secret, verify=verify)
|
|
|
|
async def _request(self, method: str, path: str, **kwargs) -> dict:
|
|
try:
|
|
response = await self._client.request(method, path, **kwargs)
|
|
except httpx.HTTPError as exc:
|
|
raise CheckmkError(f"{type(exc).__name__}: {exc}") from exc
|
|
if response.status_code >= 400:
|
|
raise CheckmkError(f"HTTP {response.status_code} on {path}: {response.text[:200]}")
|
|
return response.json()
|
|
|
|
async def version(self) -> dict:
|
|
return await self._request("GET", VERSION_PATH)
|
|
|
|
async def problem_services(self) -> list[ServiceState]:
|
|
body = await self._request(
|
|
"POST",
|
|
SERVICES_PATH,
|
|
json={"query": NOT_OK_QUERY, "columns": COLUMNS},
|
|
)
|
|
return [_parse_service(item["extensions"]) for item in body.get("value", [])]
|
|
|
|
async def problems(self) -> list[ServiceState]:
|
|
return await self.problem_services()
|
|
|
|
async def _services(self, query: dict) -> list[ServiceState]:
|
|
body = await self._request("POST", SERVICES_PATH, json={"query": query, "columns": COLUMNS})
|
|
return [_parse_service(item["extensions"]) for item in body.get("value", [])]
|
|
|
|
async def service_status(self, host: str, service: str) -> ServiceState | None:
|
|
query = {"op": "and", "expr": [_equals("host_name", host), _equals("description", service)]}
|
|
services = await self._services(query)
|
|
return services[0] if services else None
|
|
|
|
async def host_services(self, host: str) -> list[ServiceState]:
|
|
return await self._services(_equals("host_name", host))
|
|
|
|
async def downtimes(self, host: str | None = None) -> list[Downtime]:
|
|
params = {"host_name": host} if host else None
|
|
body = await self._request("GET", DOWNTIMES_PATH, params=params)
|
|
return [_parse_downtime(item["extensions"]) for item in body.get("value", [])]
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|