mirror of
https://github.com/napnap75/multiarch-docker-images.git
synced 2026-09-25 20:31:52 +02:00
thomas
a5337c3d29
- 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.
35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
"""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
|