prepare config for healthchecks integration

src/lego_monitoring/__init__.py

@@ -49,11 +49,6 @@ async def async_main():
 
     logging.basicConfig(level=config.log_level)
 
-    tg_client = await sender.get_client()
-
-    cvars.tg_client.set(tg_client)
-    my_username = (await tg_client.get_me()).username
-    logging.info(f"Logged in as @{my_username}")
     check_sets = config_enums.CheckSet
 
     checker_sets: dict[config_enums.CheckSet, list[Coroutine | BaseChecker]] = {
@@ -94,8 +89,18 @@ async def async_main():
 
     checker_sets[check_sets.REMIND][0].check_args = [checkers]
 
-    command_manager = CommandHandlerManager(checkers)
-    await command_manager.attach_handlers(tg_client)
+    if config.alert_channels.telegram is not None:
+        tg_client = await sender.get_client()
+        my_username = (await tg_client.get_me()).username
+        logging.info(f"Logged in as @{my_username}")
+
+        command_manager = CommandHandlerManager(checkers)
+        await command_manager.attach_handlers(tg_client)
+    else:
+        logging.info("Telegram integration is disabled")
+        tg_client = None
+
+    cvars.tg_client.set(tg_client)
 
     signal.signal(signal.SIGTERM, stop_gracefully)
 

src/lego_monitoring/alerting/commands.py

@@ -13,7 +13,7 @@ from .sender import format_message
 def admin_chat_only(
     handler: Callable[[events.NewMessage.Event], Awaitable[None]],
 ) -> Callable[[events.NewMessage.Event], Awaitable[None]]:
-    admin_room_id = cvars.config.get().telegram.room_id
+    admin_room_id = cvars.config.get().alert_channels.telegram.room_id
 
     async def safe_handler(event: events.NewMessage.Event) -> None:
         if event.chat_id == admin_room_id:

src/lego_monitoring/alerting/sender.py

@@ -1,5 +1,3 @@
-from dataclasses import dataclass
-
 from telethon import TelegramClient
 from telethon.sessions import MemorySession
 
@@ -10,7 +8,7 @@ from .enum import SEVERITY_TO_EMOJI, AlertType, Severity
 
 async def get_client() -> TelegramClient:
     config = cvars.config.get()
-    api_id, api_hash, bot_token = config.telegram.creds.split(",")
+    api_id, api_hash, bot_token = config.alert_channels.telegram.creds.split(",")
     client = await TelegramClient(MemorySession(), api_id, api_hash, connection_retries=None).start(bot_token=bot_token)
     client.parse_mode = "html"
     return client
@@ -38,13 +36,17 @@ async def send_alert(alert: Alert, note: str = "") -> None:
         raise NotImplementedError  # TODO
     else:
         ...  # temp_client = False
-    room_id = cvars.config.get().telegram.room_id
-    message = format_message(alert, note)
-    await client.send_message(entity=room_id, message=message)
-    # if temp_client:
-    #     await client.close()
+    if client is not None:
+        room_id = cvars.config.get().alert_channels.telegram.room_id
+        message = format_message(alert, note)
+        await client.send_message(entity=room_id, message=message)
+        # if temp_client:
+        #     await client.close()
+    # TODO ping healthchecks if enabled
 
 
+# TODO service itself has to be monitored like everything else - with regular pinging - if we're
+# using healthchecks
 async def send_start_alert() -> None:
     config = cvars.config.get()
     await send_alert(

src/lego_monitoring/config/__init__.py

@@ -5,6 +5,7 @@ from typing import Optional
 from alt_utils import NestedDeserializableDataclass
 
 from . import enums
+from .alert_channels import AlertChannelsConfig
 from .checks.cpu import CpuCheckConfig
 from .checks.net import NetCheckConfig
 from .checks.ram import RamCheckConfig
@@ -21,16 +22,10 @@ class ChecksConfig(NestedDeserializableDataclass):
     net: NetCheckConfig = field(default_factory=NetCheckConfig)
 
 
-@dataclass
-class TelegramConfig:
-    creds: str
-    room_id: int
-
-
 @dataclass
 class Config(NestedDeserializableDataclass):
     checks: ChecksConfig
-    telegram: TelegramConfig
+    alert_channels: AlertChannelsConfig
     enabled_check_sets: list[enums.CheckSet] = field(default_factory=list)
     log_level: enums.LogLevelName = enums.LogLevelName.INFO
 

src/lego_monitoring/config/alert_channels.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass, field
+from typing import Optional
+
+from alt_utils import NestedDeserializableDataclass
+
+
+@dataclass
+class TelegramConfig:
+    creds: str
+    room_id: int
+
+
+@dataclass
+class HealthchecksConfig:
+    pinging_keys: str | dict[str, str]
+
+    def __post_init__(self):
+        lines = self.pinging_keys.split()
+        self.pinging_keys = {}
+        for l in lines:
+            slug, key = l.split(":")
+            self.pinging_keys[slug] = key
+
+
+@dataclass
+class AlertChannelsConfig(NestedDeserializableDataclass):
+    telegram: Optional[TelegramConfig] = None
+    healthchecks: Optional[HealthchecksConfig] = None

src/lego_monitoring/core/cvars.py

@@ -1,4 +1,5 @@
 from contextvars import ContextVar
+from typing import Optional
 
 from telethon import TelegramClient
 
@@ -7,5 +8,5 @@ from lego_monitoring.alerting.current import CurrentAlerts
 from ..config import Config
 
 config: ContextVar[Config] = ContextVar("config")
-tg_client: ContextVar[TelegramClient] = ContextVar("tg_client")
+tg_client: ContextVar[Optional[TelegramClient]] = ContextVar("tg_client")
 current_alerts: ContextVar[list[CurrentAlerts]] = ContextVar("current_alerts", default=[])