From a5337c3d2953546a15cc7847db5e695eb050d647 Mon Sep 17 00:00:00 2001 From: thomas Date: Sun, 30 Aug 2026 23:22:38 +0200 Subject: [PATCH] feat: Refactor log-alert application into modular structure - Introduced alerters package with base Alerter class and implementations for LogAlerter and GotifyAlerter. - Created fetchers package with abstract LogFetcher class and implementations for FileLogFetcher, LokiLogFetcher, and ParseableLogFetcher. - Added filters package with base Filter class and implementations for RegexpFilter and GeolocationFilter. - Implemented rules package with base AlertRule class and SimpleAlertRule for alerting logic. - Enhanced configuration handling with utilities for loading and validating JSON configuration. - Updated logging setup for better logging management. - Modified main application logic to utilize the new modular structure, improving maintainability and readability. - Updated config.json and config.schema.json to reflect changes in alerting and fetching configurations. - Added python-dateutil to requirements for date parsing functionality. --- log-alert/alerters/__init__.py | 7 + log-alert/alerters/base.py | 17 ++ log-alert/alerters/gotify.py | 30 ++++ log-alert/alerters/log.py | 20 +++ log-alert/config.json | 34 ++-- log-alert/config.schema.json | 12 +- log-alert/fetchers/__init__.py | 8 + log-alert/fetchers/base.py | 36 +++++ log-alert/fetchers/file.py | 81 ++++++++++ log-alert/fetchers/loki.py | 60 +++++++ log-alert/fetchers/parseable.py | 64 ++++++++ log-alert/filters/__init__.py | 7 + log-alert/filters/base.py | 20 +++ log-alert/filters/geolocation.py | 34 ++++ log-alert/filters/regexp.py | 29 ++++ log-alert/log-alert.py | 266 +++---------------------------- log-alert/requirements.txt | 1 + log-alert/rules/__init__.py | 5 + log-alert/rules/base.py | 16 ++ log-alert/rules/simple.py | 50 ++++++ log-alert/utils/__init__.py | 12 ++ log-alert/utils/config.py | 62 +++++++ log-alert/utils/logging.py | 17 ++ 23 files changed, 627 insertions(+), 261 deletions(-) create mode 100644 log-alert/alerters/__init__.py create mode 100644 log-alert/alerters/base.py create mode 100644 log-alert/alerters/gotify.py create mode 100644 log-alert/alerters/log.py create mode 100644 log-alert/fetchers/__init__.py create mode 100644 log-alert/fetchers/base.py create mode 100644 log-alert/fetchers/file.py create mode 100644 log-alert/fetchers/loki.py create mode 100644 log-alert/fetchers/parseable.py create mode 100644 log-alert/filters/__init__.py create mode 100644 log-alert/filters/base.py create mode 100644 log-alert/filters/geolocation.py create mode 100644 log-alert/filters/regexp.py create mode 100644 log-alert/rules/__init__.py create mode 100644 log-alert/rules/base.py create mode 100644 log-alert/rules/simple.py create mode 100644 log-alert/utils/__init__.py create mode 100644 log-alert/utils/config.py create mode 100644 log-alert/utils/logging.py diff --git a/log-alert/alerters/__init__.py b/log-alert/alerters/__init__.py new file mode 100644 index 0000000..2cbf9bc --- /dev/null +++ b/log-alert/alerters/__init__.py @@ -0,0 +1,7 @@ +"""Alerters package for log-alert.""" + +from .base import Alerter +from .log import LogAlerter +from .gotify import GotifyAlerter + +__all__ = ["Alerter", "LogAlerter", "GotifyAlerter"] diff --git a/log-alert/alerters/base.py b/log-alert/alerters/base.py new file mode 100644 index 0000000..bab7229 --- /dev/null +++ b/log-alert/alerters/base.py @@ -0,0 +1,17 @@ +"""Abstract base class for alerters.""" + +from abc import ABC, abstractmethod + + +class Alerter(ABC): + """Abstract base class for alerting implementations.""" + + @abstractmethod + def send_alert(self, title: str, message: str) -> None: + """Send an alert. + + Args: + title: Alert title + message: Alert message + """ + pass diff --git a/log-alert/alerters/gotify.py b/log-alert/alerters/gotify.py new file mode 100644 index 0000000..a3d5b43 --- /dev/null +++ b/log-alert/alerters/gotify.py @@ -0,0 +1,30 @@ +"""Gotify alerter implementation.""" + +import logging +import requests +from typing import Dict, Any + +from .base import Alerter + +logger = logging.getLogger("log-alert") + + +class GotifyAlerter(Alerter): + """Concrete implementation for Gotify alert manager.""" + + def __init__(self, config: Dict[str, Any]): + self.url = config["url"] + self.token = config.get("token") + + def send_alert(self, title: str, message: str) -> None: + """Send an alert to Gotify.""" + payload = { + "title": title, + "message": message, + "priority": 5 + } + try: + response = requests.post(f"{self.url}?token={self.token}", json=payload) + response.raise_for_status() + except requests.exceptions.RequestException as e: + logger.error(f"Error sending alert to Gotify: {e}") diff --git a/log-alert/alerters/log.py b/log-alert/alerters/log.py new file mode 100644 index 0000000..581c3bd --- /dev/null +++ b/log-alert/alerters/log.py @@ -0,0 +1,20 @@ +"""Standard output alerter implementation.""" + +import logging +import requests +from typing import Dict, Any + +from .base import Alerter + +logger = logging.getLogger("log-alert") + + +class LogAlerter(Alerter): + """Concrete implementation for log alert manager.""" + + def __init__(self, config: Dict[str, Any]): + pass + + def send_alert(self, title: str, message: str) -> None: + """Send an alert to the log.""" + logger.info(f"[ALERT] {title} / {message}") diff --git a/log-alert/config.json b/log-alert/config.json index 6b477e2..2eef5c4 100644 --- a/log-alert/config.json +++ b/log-alert/config.json @@ -1,14 +1,17 @@ { "log-fetchers": { - "loki-fileserver": { - "type": "loki", + "parseable": { + "type": "parseable", "config": { - "url-from-env": "{LOKI_URL}" + "url-from-env": "{PARSEABLE_URL}", + "dataset-from-env": "{PARSEABLE_DATASET}", + "user-from-env": "{PARSEABLE_USER}", + "password-from-env": "{PARSEABLE_PASSWORD}" } } }, - "alert-managers": { - "gotify-paris": { + "alerters": { + "gotify": { "type": "gotify", "config": { "url-from-env": "{GOTIFY_URL}", @@ -20,10 +23,10 @@ "ssh-outside": { "check-interval": 30, "log-fetcher": { - "name": "loki-fileserver", + "name": "parseable", "filters": { "labels": { - "container": "openssh-server" + "container_name": "/openssh-server" }, "text": "Accepted" } @@ -42,19 +45,20 @@ } } ], - "alert-manager": { - "name": "gotify-paris", + "alerter": { + "name": "gotify", "title": "Outside SSH login", - "message": "New SSH login for {username} on {instance} from ip {ip} (country: {country}, provider: {isp}, method: {method})" + "message": "New SSH login for {username} on {hostname} from ip {ip} (country: {country}, provider: {isp}, method: {method})" } }, "ssh-local": { "check-interval": 30, "log-fetcher": { - "name": "loki-fileserver", + "name": "parseable", "filters": { "labels": { - "filename": "/var/log/host/auth.log" + "filename": "/var/log/host/auth.log", + "process": "sshd" }, "text": "Accepted" } @@ -67,10 +71,10 @@ } } ], - "alert-manager": { - "name": "gotify-paris", + "alerter": { + "name": "gotify", "title": "Local SSH login", - "message": "New SSH login for {username} on {instance} from ip {ip} (method: {method})" + "message": "New SSH login for {username} on {hostname} from ip {ip} (method: {method})" } } } diff --git a/log-alert/config.schema.json b/log-alert/config.schema.json index 1c279e8..8466849 100644 --- a/log-alert/config.schema.json +++ b/log-alert/config.schema.json @@ -2,7 +2,7 @@ "$schema": "http://json-schema.org/draft-07/schema#", "title": "Log Alert configuration schema", "type": "object", - "required": ["log-fetchers", "alert-managers", "alerting-rules"], + "required": ["log-fetchers", "alerters", "alerting-rules"], "properties": { "log-fetchers": { "type": "object", @@ -10,18 +10,18 @@ "type": "object", "required": ["type", "config"], "properties": { - "type": { "type": "string", "enum": ["loki"] }, + "type": { "type": "string", "enum": ["file", "loki", "parseable"] }, "config": { "type": "object" } } } }, - "alert-managers": { + "alerters": { "type": "object", "additionalProperties": { "type": "object", "required": ["type", "config"], "properties": { - "type": { "type": "string", "enum": ["gotify"] }, + "type": { "type": "string", "enum": ["log", "gotify"] }, "config": { "type": "object" } } } @@ -30,7 +30,7 @@ "type": "object", "additionalProperties": { "type": "object", - "required": ["check-interval", "log-fetcher", "filters", "alert-manager"], + "required": ["check-interval", "log-fetcher", "filters", "alerter"], "properties": { "check-interval": { "type": "number", "minimum": 0 }, "log-fetcher": { @@ -53,7 +53,7 @@ } } }, - "alert-manager": { + "alerter": { "type": "object", "required": ["name"], "properties": { diff --git a/log-alert/fetchers/__init__.py b/log-alert/fetchers/__init__.py new file mode 100644 index 0000000..a6cd8a9 --- /dev/null +++ b/log-alert/fetchers/__init__.py @@ -0,0 +1,8 @@ +"""Fetchers package for log-alert.""" + +from .base import LogFetcher +from .file import FileLogFetcher +from .loki import LokiLogFetcher +from .parseable import ParseableLogFetcher + +__all__ = ["LogFetcher", "FileLogFetcher", "LokiLogFetcher", "ParseableLogFetcher"] diff --git a/log-alert/fetchers/base.py b/log-alert/fetchers/base.py new file mode 100644 index 0000000..c9b3ce3 --- /dev/null +++ b/log-alert/fetchers/base.py @@ -0,0 +1,36 @@ +"""Abstract base class for log fetchers.""" + +from abc import ABC, abstractmethod +from typing import Dict, Any, List + + +class LogFetcher(ABC): + """Abstract base class for log fetchers.""" + + """Fetch logs.""" + @abstractmethod + def fetch_logs(self, filters: Dict[str, Any]) -> List[Dict[str, Any]]: + """Fetch logs within the specified time range. + + Args: + filters: Dictionary of filters to apply + + Returns: + List of log entries + """ + pass + + """ Fetch logs with time range.""" + @abstractmethod + def fetch_logs_time_range(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: + """Fetch logs within the specified time range. + + Args: + filters: Dictionary of filters to apply + start_time: Start time in seconds since epoch + end_time: End time in seconds since epoch + + Returns: + List of log entries + """ + pass diff --git a/log-alert/fetchers/file.py b/log-alert/fetchers/file.py new file mode 100644 index 0000000..8733400 --- /dev/null +++ b/log-alert/fetchers/file.py @@ -0,0 +1,81 @@ +"""File log fetcher implementation.""" + +import logging +import requests +from typing import Dict, Any, List +from dateutil.parser import parse + +from .base import LogFetcher + +logger = logging.getLogger("log-alert") + + +class FileLogFetcher(LogFetcher): + """Concrete implementation for fetching logs from a file.""" + + type = "TAIL" + + def __init__(self, config: Dict[str, Any]): + try: + self.file = open(config["file"], "r") + except FileNotFoundError: + logger.error(f"Log file not found: {config['file']}") + raise + self.position = 0 # To keep track of the last read position + + def fetch_logs(self, filters: Dict[str, Any]) -> List[Dict[str, Any]]: + """Fetch logs from the file.""" + logs = [] + + self.file.seek(self.position) # Move to the last read position + while True: + self.position = self.file.tell() + line = self.file.readline() + if not line: + break # End of file reached, exit loop + + logger.debug(f"Read line: {line.strip()} at position {self.position}") + + parts = line.strip().split(maxsplit=1) + timestamp_str, rest = parts[0], parts[1] + + # Parse the timestamp string into a datetime object + dt = parse(timestamp_str) + # Convert to seconds since epoch + epoch_seconds = int(dt.timestamp()) + + log_entry = {"timestamp": epoch_seconds, "log": rest} + + logs.append(log_entry) + + return logs + + def fetch_logs_time_range(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: + """Fetch logs from the file within the specified time range.""" + logs = [] + + self.file.seek(0) + while True: + line = self.file.readline() + if not line: + break # End of file reached, exit loop + + logger.debug(f"Read line: {line.strip()} at position {self.position}") + + parts = line.strip().split(maxsplit=1) + timestamp_str, rest = parts[0], parts[1] + + dt = parse(timestamp_str) + epoch_seconds = int(dt.timestamp()) + + if epoch_seconds < start_time: + continue + if epoch_seconds > end_time: + break + + log_entry = {"timestamp": epoch_seconds, "log": rest} + + logs.append(log_entry) + + return logs + \ No newline at end of file diff --git a/log-alert/fetchers/loki.py b/log-alert/fetchers/loki.py new file mode 100644 index 0000000..c460b4c --- /dev/null +++ b/log-alert/fetchers/loki.py @@ -0,0 +1,60 @@ +"""Loki log fetcher implementation.""" + +import logging +from time import time +import requests +from typing import Dict, Any, List + +from .base import LogFetcher + +logger = logging.getLogger("log-alert") + + +class LokiLogFetcher(LogFetcher): + """Concrete implementation for fetching logs from Loki.""" + + def __init__(self, config: Dict[str, Any]): + self.url = config["url"] + self.last_fetched_time = int(time.time()) + + def fetch_logs(self, filters: Dict[str, Any]) -> List[Dict[str, Any]]: + """Fetch logs from Loki without time range.""" + old_time = self.last_fetched_time + self.last_fetched_time = int(time.time()) + return self.fetch_logs_time_range(filters, old_time, self.last_fetched_time) + + def fetch_logs_time_range(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: + """Fetch logs from Loki within the specified time range.""" + query = '{' + for label in filters.get("labels", {}): + if len(query) > 1: + query += ',' + query += f'{label}="{filters["labels"][label]}"' + query += '}' + if "text" in filters: + query += f' |= "{filters["text"]}"' + logger.debug(f"Executing Loki query: {query}") + payload = { + "query": query, + "limit": 1000, + "start": str(int(start_time) * 1000000000), # Convert to nanoseconds + "end": str(int(end_time) * 1000000000), + "direction": "forward" + } + try: + response = requests.get(f"{self.url}/loki/api/v1/query_range", params=payload) + response.raise_for_status() + data = response.json() + logs = [] + for stream in data.get("data", {}).get("result", []): + for value in stream.get("values", []): + timestamp, log = value + logs.append({ + "timestamp": timestamp, + "log": log, + "labels": stream.get("stream", {}) + }) + return logs + except requests.exceptions.RequestException as e: + logger.error(f"Error fetching logs from Loki: {e}") + return [] diff --git a/log-alert/fetchers/parseable.py b/log-alert/fetchers/parseable.py new file mode 100644 index 0000000..34fa588 --- /dev/null +++ b/log-alert/fetchers/parseable.py @@ -0,0 +1,64 @@ +"""Parseable log fetcher implementation.""" + +import datetime +import logging +from time import time +import requests +from typing import Dict, Any, List + +from .base import LogFetcher + +logger = logging.getLogger("log-alert") + + +class ParseableLogFetcher(LogFetcher): + """Concrete implementation for fetching logs from Parseable.""" + + def __init__(self, config: Dict[str, Any]): + self.url = config["url"] + self.dataset = config["dataset"] + self.user = config.get("user") + self.password = config.get("password") + self.last_fetched_time = int(time()) # Initialize with current time + + def fetch_logs(self, filters: Dict[str, Any]) -> List[Dict[str, Any]]: + """Fetch logs from Parseable without time range.""" + old_time = self.last_fetched_time + self.last_fetched_time = int(time()) + return self.fetch_logs_time_range(filters, old_time, self.last_fetched_time) + + def fetch_logs_time_range(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: + """Fetch logs from Parseable within the specified time range.""" + query = 'SELECT * FROM \'' + query += f'{self.dataset}\' WHERE ' + labelNum = 0 + for label in filters.get("labels", {}): + if labelNum > 0: + query += ' AND ' + query += f'{label}=\'{filters["labels"][label]}\'' + labelNum += 1 + if "text" in filters: + if labelNum > 0: + query += ' AND ' + query += f'log LIKE \'%{filters["text"]}%\'' + logger.debug(f"Executing Parseable query: {query}") + payload = { + "query": query, + "startTime": datetime.datetime.fromtimestamp(start_time).strftime("%Y-%m-%dT%H:%M:%SZ"), + "endTime": datetime.datetime.fromtimestamp(end_time).strftime("%Y-%m-%dT%H:%M:%SZ") + } + try: + response = requests.post(f"{self.url}/api/v1/query", json=payload, auth=(self.user, self.password), headers = { "Content-Type": "application/json" }) + response.raise_for_status() + data = response.json() + logs = [] + for item in data: + logs.append({ + "timestamp": item.get("p_timestamp"), + "log": item.get("log"), + "labels": {k: v for k, v in item.items() if k not in ["p_timestamp", "log"]} + }) + return logs + except requests.exceptions.RequestException as e: + logger.error(f"Error fetching logs from Parseable: {e}") + return [] diff --git a/log-alert/filters/__init__.py b/log-alert/filters/__init__.py new file mode 100644 index 0000000..cd15e6f --- /dev/null +++ b/log-alert/filters/__init__.py @@ -0,0 +1,7 @@ +"""Filters package for log-alert.""" + +from .base import Filter +from .regexp import RegexpFilter +from .geolocation import GeolocationFilter + +__all__ = ["Filter", "RegexpFilter", "GeolocationFilter"] diff --git a/log-alert/filters/base.py b/log-alert/filters/base.py new file mode 100644 index 0000000..99adc40 --- /dev/null +++ b/log-alert/filters/base.py @@ -0,0 +1,20 @@ +"""Abstract base class for filters.""" + +from abc import ABC, abstractmethod +from typing import Dict, Any, Optional + + +class Filter(ABC): + """Abstract base class for filters.""" + + @abstractmethod + def filter(self, log: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Apply filter to a log entry. + + Args: + log: Log entry dictionary + + Returns: + Modified log entry, or None if the log should be filtered out + """ + pass diff --git a/log-alert/filters/geolocation.py b/log-alert/filters/geolocation.py new file mode 100644 index 0000000..61e280d --- /dev/null +++ b/log-alert/filters/geolocation.py @@ -0,0 +1,34 @@ +"""Geolocation filter implementation.""" + +import logging +import requests +from typing import Dict, Any, Optional + +from .base import Filter + +logger = logging.getLogger("log-alert") + + +class GeolocationFilter(Filter): + """Concrete implementation for Geolocation filter.""" + + def __init__(self, config: Dict[str, Any]): + self.source_field = config["source-field"] + + def filter(self, log: Dict[str, Any]) -> Optional[Dict[str, Any]]: + ip_address = log.get("labels", {}).get(self.source_field) + if not ip_address: + logger.warning("No IP address found in log labels for geolocation") + else: + try: + response = requests.get(f"http://ip-api.com/json/{ip_address}").json() + if response["status"] == "success": + logger.debug(f"Found info {response} for IP {ip_address}") + del response["status"] + del response["query"] + log.setdefault("labels", {}).update(response) + else: + logger.warning("No info found for IP {ip_address}") + except requests.exceptions.RequestException as e: + logger.error(f"Error fetching geolocation for IP {ip_address}: {e}") + return log diff --git a/log-alert/filters/regexp.py b/log-alert/filters/regexp.py new file mode 100644 index 0000000..4098f69 --- /dev/null +++ b/log-alert/filters/regexp.py @@ -0,0 +1,29 @@ +"""Regexp filter implementation.""" + +import logging +import re +from typing import Dict, Any, Optional + +from .base import Filter + +logger = logging.getLogger("log-alert") + + +class RegexpFilter(Filter): + """Concrete implementation for Regexp filter.""" + + def __init__(self, config: Dict[str, Any]): + self.match = config["match"] + + def filter(self, log: Dict[str, Any]) -> Optional[Dict[str, Any]]: + match = re.search(self.match, log["log"]) + if match: + # Only call groupdict() when there is a match + groups = match.groupdict() + logger.debug(f"Regex match for '{self.match}' in log: {groups}") + if groups: + log.setdefault("labels", {}).update(groups) + return log + # no match + logger.debug(f"Regex did not match for pattern '{self.match}' in log: {log.get('log')}") + return None diff --git a/log-alert/log-alert.py b/log-alert/log-alert.py index 30ee507..fdc8600 100755 --- a/log-alert/log-alert.py +++ b/log-alert/log-alert.py @@ -1,270 +1,56 @@ #!/usr/bin/env python3 import argparse -import json -import jsonschema import logging -import os -import re -import requests -import sys import time -from abc import ABC, abstractmethod -from typing import Dict, Any, List, Optional +from typing import Dict, Any + +from alerters.log import LogAlerter +from utils.logging import setup_logging, get_logger +from utils.config import load_config, validate_config_with_schema, update_config_from_env +from fetchers import LogFetcher, FileLogFetcher, LokiLogFetcher, ParseableLogFetcher +from filters import Filter, RegexpFilter, GeolocationFilter +from alerters import Alerter, GotifyAlerter # Configure root logger -logging.basicConfig( - level=os.environ.get("LOGLEVEL", "INFO"), - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", -) -logger = logging.getLogger("log-alert") +setup_logging() +logger = get_logger("log-alert") -# Base Classes -class LogFetcher(ABC): - """Abstract base class for log fetchers.""" - - @abstractmethod - def fetch_logs(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: - pass - -class Filter(ABC): - """Abstract base class for filters.""" - - @abstractmethod - def filter(self, log: Dict[str, Any]) -> Dict[str, Any]: - pass - -class AlertManager(ABC): - """Abstract base class for alert managers.""" - - @abstractmethod - def send_alert(self, title: str, message: str) -> None: - pass - -# Loki Log Fetcher -class LokiLogFetcher(LogFetcher): - """Concrete implementation for fetching logs from Loki.""" - - def __init__(self, config: Dict[str, Any]): - self.url = config["url"] - - def fetch_logs(self, filters: Dict[str, Any], start_time: int, end_time: int) -> List[Dict[str, Any]]: - """Fetch logs from Loki within the specified time range.""" - query = '{' - for label in filters.get("labels", {}): - if len(query) > 1: - query += ',' - query += f'{label}="{filters["labels"][label]}"' - query += '}' - if "text" in filters: - query += f' |= "{filters["text"]}"' - logger.debug(f"Executing Loki query: {query}") - payload = { - "query": query, - "limit": 1000, - "start": str(int(start_time) * 1000000000), # Convert to nanoseconds - "end": str(int(end_time) * 1000000000), - "direction": "forward" - } - try: - response = requests.get(f"{self.url}/loki/api/v1/query_range", params=payload) - response.raise_for_status() - data = response.json() - logs = [] - for stream in data.get("data", {}).get("result", []): - for value in stream.get("values", []): - timestamp, log = value - logs.append({ - "timestamp": timestamp, - "log": log, - "labels": stream.get("stream", {}) - }) - return logs - except requests.exceptions.RequestException as e: - logger.error(f"Error fetching logs from Loki: {e}") - return [] - -# Regexp Filter -class RegexpFilter(Filter): - """Concrete implementation for Regexp filter.""" - - def __init__(self, config: Dict[str, Any]): - self.match = config["match"] - - def filter(self, log: Dict[str, Any]) -> Dict[str, Any]: - match = re.search(self.match, log["log"]) - if match: - # Only call groupdict() when there is a match - groups = match.groupdict() - logger.debug(f"Regex match for '{self.match}' in log: {groups}") - if groups: - log.setdefault("labels", {}).update(groups) - return log - # no match - logger.debug(f"Regex did not match for pattern '{self.match}' in log: {log.get('log')}") - return None - -# Geolocation Filter -class GeolocationFilter(Filter): - """Concrete implementation for Geolocation filter.""" - - def __init__(self, config: Dict[str, Any]): - self.source_field = config["source-field"] - - def filter(self, log: Dict[str, Any]) -> Dict[str, Any]: - ip_address = log.get("labels", {}).get(self.source_field) - if not ip_address: - logger.warning("No IP address found in log labels for geolocation") - else: - try: - response = requests.get(f"http://ip-api.com/json/{ip_address}").json() - if response["status"] == "success": - logger.debug(f"Found info {response} for IP {ip_address}") - del response["status"] - del response["query"] - log.setdefault("labels", {}).update(response) - else: - logger.warning("No info found for IP {ip_address}") - except requests.exceptions.RequestException as e: - logger.error(f"Error fetching geolocation for IP {ip_address}: {e}") - return log - -# Gotify Alert Manager -class GotifyAlertManager(AlertManager): - """Concrete implementation for Gotify alert manager.""" - - def __init__(self, config: Dict[str, Any]): - self.url = config["url"] - self.token = config.get("token") - - def send_alert(self, title: str, message: str) -> None: - """Send an alert to Gotify.""" - payload = { - "title": title, - "message": message, - "priority": 5 - } - try: - response = requests.post(f"{self.url}?token={self.token}", json=payload) - response.raise_for_status() - except requests.exceptions.RequestException as e: - logger.error(f"Error sending alert to Gotify: {e}") - -# Alert Rule -class AlertRule: - """Represents an alert rule with filters and alert template.""" - - def __init__(self, log_fetchers: LogFetcher, alert_managers: AlertManager, config: Dict[str, Any]): - self.log_fetcher = log_fetchers[config["log-fetcher"]["name"]] - self.fetcher_filters = config["log-fetcher"].get("filters", {}) - self.check_interval = config.get("check-interval", 60) - self.filters = [] - for filter in config.get("filters", []): - if filter["type"] == "regexp": - self.filters.append(RegexpFilter(filter["config"])) - elif filter["type"] == "geolocation": - self.filters.append(GeolocationFilter(filter["config"])) - else: - raise ValueError(f"Unsupported filter type: {filter['type']}") - self.alert_manager = alert_managers[config["alert-manager"]["name"]] - self.alert_title = config["alert-manager"]["title"] - self.alert_message = config["alert-manager"]["message"] - self.last_run = time.time() - self.check_interval - self.next_run = time.time() - - def run(self) -> None: - logs = self.log_fetcher.fetch_logs(self.fetcher_filters, self.last_run, self.next_run) - for log_entry in logs: - logger.debug(f"Checking log: {log_entry['log']}") - for filter in self.filters: - log_entry = filter.filter(log_entry) - if log_entry is None: - break - if log_entry is None: - continue - title = self.alert_title.format_map(log_entry.get("labels", {})) - message = self.alert_message.format_map(log_entry.get("labels", {})) - logger.info(f"Sending message: {message}, title: {title}, with params: {log_entry}") - self.alert_manager.send_alert(title, message) - self.last_run = self.next_run - self.next_run = time.time() + self.check_interval +from rules import SimpleAlertRule as AlertRule # Main Application class LogAlertApp: """Main application class to manage log fetching and alerting.""" def __init__(self, config_path: str): - self.config = self._load_config(config_path) + self.config = load_config(config_path) logger.debug(f"Configuration loaded: {self.config}") self.log_fetchers = {} for key, fetcher in self.config["log-fetchers"].items(): self.log_fetchers[key] = self._init_log_fetcher(fetcher) - self.alert_managers = {} - for key, manager in self.config["alert-managers"].items(): - self.alert_managers[key] = self._init_alert_manager(manager) + self.alerters = {} + for key, alerter in self.config["alerters"].items(): + self.alerters[key] = self._init_alerter(alerter) self.alert_rules = {} for key, rule in self.config["alerting-rules"].items(): - self.alert_rules[key] = AlertRule(self.log_fetchers, self.alert_managers, rule) - - def _load_config(self, config_path: str) -> Dict[str, Any]: - """Load the configuration from a JSON file and validate it with JSON Schema.""" - try: - with open(config_path, 'r') as config_file: - # read JSON first - config = json.load(config_file) - # Perform schema validation if jsonschema is available - self._validate_config_with_schema(config) - # Update config to load env variable where required - return self._update_config_from_env(config) - except FileNotFoundError: - logger.error(f"Error: Configuration file '{config_path}' not found.") - sys.exit(1) - except json.JSONDecodeError: - logger.error(f"Error: Invalid JSON in configuration file '{config_path}'.") - sys.exit(1) - - def _validate_config_with_schema(self, config: Dict[str, Any]) -> None: - """Validate a loaded config dict against log-alert/config.schema.json if jsonschema is installed.""" - schema_path = os.path.join(os.path.dirname(__file__), 'config.schema.json') - try: - with open(schema_path, 'r') as sf: - schema = json.load(sf) - jsonschema.validate(instance=config, schema=schema) - except FileNotFoundError: - logger.error(f"Schema file '{schema_path}' not found") - sys - except jsonschema.exceptions.ValidationError as e: - logger.error(f"Configuration validation error: {e.message}") - logger.error("Detailed error:", e) - sys.exit(1) - except Exception as e: - logger.error(f"Unexpected error while validating configuration: {e}") - sys.exit(1) - - def _update_config_from_env(self, config: Dict[str, Any]) -> Dict[str, Any]: - """Update config values from environment variables if specified.""" - for key, value in list(config.items()): - if isinstance(value, dict): - config[key] = self._update_config_from_env(value) - elif isinstance(value, list): - config[key] = [self._update_config_from_env(item) for item in value] - elif isinstance(value, str) and key.endswith("-from-env"): - new_key = key[:-9] # Remove '-from-env' - config[new_key] = value.format_map(os.environ) - del config[key] - return self._update_config_from_env(config) # re-evaluate in case of nested env vars - return config + self.alert_rules[key] = AlertRule(self.log_fetchers, self.alerters, rule) def _init_log_fetcher(self, fetcher_config: Dict[str, Any]) -> LogFetcher: """Initialize the log fetcher based on config.""" - if fetcher_config["type"] == "loki": + if fetcher_config["type"] == "file": + return FileLogFetcher(fetcher_config["config"]) + elif fetcher_config["type"] == "loki": return LokiLogFetcher(fetcher_config["config"]) + elif fetcher_config["type"] == "parseable": + return ParseableLogFetcher(fetcher_config["config"]) else: raise ValueError(f"Unsupported log fetcher type: {fetcher_config['type']}") - def _init_alert_manager(self, manager_config: Dict[str, Any]) -> AlertManager: + def _init_alerter(self, manager_config: Dict[str, Any]) -> Alerter: """Initialize the alert manager based on config.""" - if manager_config["type"] == "gotify": - return GotifyAlertManager(manager_config["config"]) + if manager_config["type"] == "log": + return LogAlerter(manager_config["config"]) + elif manager_config["type"] == "gotify": + return GotifyAlerter(manager_config["config"]) else: raise ValueError(f"Unsupported alert manager type: {manager_config['type']}") diff --git a/log-alert/requirements.txt b/log-alert/requirements.txt index 1d66c4f..2ba4dde 100644 --- a/log-alert/requirements.txt +++ b/log-alert/requirements.txt @@ -1,2 +1,3 @@ requests +python-dateutil jsonschema>=4.0.0 \ No newline at end of file diff --git a/log-alert/rules/__init__.py b/log-alert/rules/__init__.py new file mode 100644 index 0000000..5b78b7c --- /dev/null +++ b/log-alert/rules/__init__.py @@ -0,0 +1,5 @@ +"""Rules package for alerting.""" +from .base import AlertRule +from .simple import SimpleAlertRule + +__all__ = ["AlertRule", "SimpleAlertRule"] diff --git a/log-alert/rules/base.py b/log-alert/rules/base.py new file mode 100644 index 0000000..a36488d --- /dev/null +++ b/log-alert/rules/base.py @@ -0,0 +1,16 @@ +"""Base classes for alert rules.""" +from abc import ABC, abstractmethod +from typing import Dict, Any + + +class AlertRule(ABC): + """Abstract base class for alert rules. + + Concrete implementations should implement the `run()` method and manage + their own scheduling (`last_run`, `next_run`). + """ + + @abstractmethod + def run(self) -> None: + """Execute the alert rule logic.""" + raise NotImplementedError diff --git a/log-alert/rules/simple.py b/log-alert/rules/simple.py new file mode 100644 index 0000000..ab90b86 --- /dev/null +++ b/log-alert/rules/simple.py @@ -0,0 +1,50 @@ +"""Simple AlertRule implementation copied from the previous monolithic implementation.""" +import logging +import time +from typing import Dict, Any, Optional + +from .base import AlertRule +from fetchers import LogFetcher +from filters import RegexpFilter, GeolocationFilter +from alerters import Alerter + +logger = logging.getLogger("log-alert") + + +class SimpleAlertRule(AlertRule): + """Represents an alert rule with filters and alert template.""" + + def __init__(self, log_fetchers: Dict[str, LogFetcher], alerters: Dict[str, Alerter], config: Dict[str, Any]): + self.log_fetcher = log_fetchers[config["log-fetcher"]["name"]] + self.fetcher_filters = config["log-fetcher"].get("filters", {}) + self.check_interval = config.get("check-interval", 60) + self.filters = [] + for filter in config.get("filters", []): + if filter["type"] == "regexp": + self.filters.append(RegexpFilter(filter["config"])) + elif filter["type"] == "geolocation": + self.filters.append(GeolocationFilter(filter["config"])) + else: + raise ValueError(f"Unsupported filter type: {filter['type']}") + self.alerter = alerters[config["alerter"]["name"]] + self.alert_title = config["alerter"]["title"] + self.alert_message = config["alerter"]["message"] + self.last_run = time.time() - self.check_interval + self.next_run = time.time() + + def run(self) -> None: + logs = self.log_fetcher.fetch_logs_time_range(self.fetcher_filters, start_time=self.last_run, end_time=self.next_run) + for log_entry in logs: + logger.debug(f"Checking log: {log_entry['log']}") + for filter in self.filters: + log_entry = filter.filter(log_entry) + if log_entry is None: + break + if log_entry is None: + continue + title = self.alert_title.format_map(log_entry.get("labels", {})) + message = self.alert_message.format_map(log_entry.get("labels", {})) + logger.info(f"Sending message: {message}, title: {title}, with params: {log_entry}") + self.alerter.send_alert(title, message) + self.last_run = self.next_run + self.next_run = time.time() + self.check_interval diff --git a/log-alert/utils/__init__.py b/log-alert/utils/__init__.py new file mode 100644 index 0000000..5a948f5 --- /dev/null +++ b/log-alert/utils/__init__.py @@ -0,0 +1,12 @@ +"""Utilities package for log-alert.""" + +from .config import load_config, validate_config_with_schema, update_config_from_env +from .logging import setup_logging, get_logger + +__all__ = [ + "load_config", + "validate_config_with_schema", + "update_config_from_env", + "setup_logging", + "get_logger", +] diff --git a/log-alert/utils/config.py b/log-alert/utils/config.py new file mode 100644 index 0000000..b53f2c9 --- /dev/null +++ b/log-alert/utils/config.py @@ -0,0 +1,62 @@ +"""Configuration loading and validation utilities.""" + +import json +import jsonschema +import logging +import os +import sys +from typing import Dict, Any + +logger = logging.getLogger("log-alert") + + +def load_config(config_path: str) -> Dict[str, Any]: + """Load the configuration from a JSON file and validate it with JSON Schema.""" + try: + with open(config_path, 'r') as config_file: + # read JSON first + config = json.load(config_file) + # Perform schema validation if jsonschema is available + validate_config_with_schema(config) + # Update config to load env variable where required + return update_config_from_env(config) + except FileNotFoundError: + logger.error(f"Error: Configuration file '{config_path}' not found.") + sys.exit(1) + except json.JSONDecodeError: + logger.error(f"Error: Invalid JSON in configuration file '{config_path}'.") + sys.exit(1) + + +def validate_config_with_schema(config: Dict[str, Any]) -> None: + """Validate a loaded config dict against config.schema.json if jsonschema is installed.""" + # Find the schema file relative to the log-alert.py location + schema_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config.schema.json') + try: + with open(schema_path, 'r') as sf: + schema = json.load(sf) + jsonschema.validate(instance=config, schema=schema) + except FileNotFoundError: + logger.error(f"Schema file '{schema_path}' not found") + except jsonschema.exceptions.ValidationError as e: + logger.error(f"Configuration validation error: {e.message}") + logger.error("Detailed error:", e) + sys.exit(1) + except Exception as e: + logger.error(f"Unexpected error while validating configuration: {e}") + sys.exit(1) + + +def update_config_from_env(config: Dict[str, Any]) -> Dict[str, Any]: + """Update config values from environment variables if specified.""" + for key, value in list(config.items()): + if isinstance(value, dict): + config[key] = update_config_from_env(value) + elif isinstance(value, list): + config[key] = [update_config_from_env(item) for item in value] + elif isinstance(value, str) and key.endswith("-from-env"): + new_key = key[:-9] # Remove '-from-env' + config[new_key] = value.format_map(os.environ) + del config[key] + return update_config_from_env(config) # re-evaluate in case of nested env vars + return config diff --git a/log-alert/utils/logging.py b/log-alert/utils/logging.py new file mode 100644 index 0000000..4623a0f --- /dev/null +++ b/log-alert/utils/logging.py @@ -0,0 +1,17 @@ +"""Logging configuration for log-alert application.""" + +import logging +import os + + +def setup_logging(): + """Configure root logger for the application.""" + logging.basicConfig( + level=os.environ.get("LOGLEVEL", "INFO"), + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + + +def get_logger(name: str) -> logging.Logger: + """Get a logger instance with the given name.""" + return logging.getLogger(name)