mirrors/lego-monitoring
1
0
Fork 0
mirror of https://forgejo.altau.su/lego/lego-monitoring.git synced 2026-03-09 20:31:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'dev'

Browse source
This commit is contained in:
Alex Tau 2025-08-16 13:43:47 +03:00
commit 6a545df533
3 changed files with 14 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/lego_monitoring/__init__.py
View file

@ -56,7 +56,11 @@ async def async_main():
sender.send_alert(checks.generate_start_alert()), sender.send_alert(checks.generate_start_alert()),
IntervalChecker(checks.self_check, interval=datetime.timedelta(minutes=5), persistent=False), IntervalChecker(checks.self_check, interval=datetime.timedelta(minutes=5), persistent=False),
], ],
check_sets.CPU: [IntervalChecker(checks.cpu_check, interval=datetime.timedelta(minutes=3), persistent=True)], check_sets.CPU: [
IntervalChecker(
checks.cpu_check, interval=datetime.timedelta(minutes=3), persistent=True, ignore_first_run=True
)
],
check_sets.RAM: [IntervalChecker(checks.ram_check, interval=datetime.timedelta(minutes=1), persistent=True)], check_sets.RAM: [IntervalChecker(checks.ram_check, interval=datetime.timedelta(minutes=1), persistent=True)],
check_sets.TEMP: [IntervalChecker(checks.temp_check, interval=datetime.timedelta(minutes=5), persistent=True)], check_sets.TEMP: [IntervalChecker(checks.temp_check, interval=datetime.timedelta(minutes=5), persistent=True)],
check_sets.VULNIX: [ check_sets.VULNIX: [

8
src/lego_monitoring/checks/vulnix/__init__.py
View file

@ -2,7 +2,6 @@ from socket import gethostname
from lego_monitoring.alerting.alert import Alert from lego_monitoring.alerting.alert import Alert
from lego_monitoring.alerting.enum import AlertType, Severity from lego_monitoring.alerting.enum import AlertType, Severity
from lego_monitoring.alerting.sender import send_alert
from ..utils import format_for_healthchecks_slug from ..utils import format_for_healthchecks_slug
from .vulnix import get_vulnix_output from .vulnix import get_vulnix_output
@ -16,15 +15,14 @@ async def vulnix_check() -> list[Alert]:
try: try:
vulnix_output = get_vulnix_output(IS_TESTING) vulnix_output = get_vulnix_output(IS_TESTING)
except Exception as e: except Exception as e:
await send_alert( return [
Alert( Alert(
alert_type=AlertType.ERROR, AlertType.VULN,
message=f"Exception {type(e).__name__} while calling vulnix: {e}", message=f"Exception {type(e).__name__} while calling vulnix: {e}",
severity=Severity.CRITICAL, severity=Severity.CRITICAL,
healthchecks_slug=slug, healthchecks_slug=slug,
) )
) ]
return []
for finding in vulnix_output: for finding in vulnix_output:
if not IS_TESTING: if not IS_TESTING:
non_whitelisted_cves = [k for k in finding.description if k not in finding.whitelisted] non_whitelisted_cves = [k for k in finding.description if k not in finding.whitelisted]

7
src/lego_monitoring/core/checkers.py
View file

@ -87,14 +87,19 @@ class BaseChecker:
class IntervalChecker(BaseChecker): class IntervalChecker(BaseChecker):
_: KW_ONLY _: KW_ONLY
interval: datetime.timedelta interval: datetime.timedelta
ignore_first_run: bool = False
async def run_checker(self) -> None: async def run_checker(self) -> None:
interval_secs = self.interval.total_seconds() interval_secs = self.interval.total_seconds()
ignore_first_run = self.ignore_first_run
while True: while True:
logging.info(f"Calling {self.check.__name__}") logging.info(f"Calling {self.check.__name__}")
result = await self._call_check() result = await self._call_check()
logging.info(f"Got {len(result)} alerts") logging.info(f"Got {len(result)} alerts")
await self._handle_alerts(result) if ignore_first_run:
ignore_first_run = False
else:
await self._handle_alerts(result)
await asyncio.sleep(interval_secs) await asyncio.sleep(interval_secs)