mirror of
https://github.com/napnap75/multiarch-docker-images.git
synced 2026-09-25 20:31:52 +02:00
47 lines
2.1 KiB
Python
47 lines
2.1 KiB
Python
"""timestamp filter implementation."""
|
|
|
|
import logging
|
|
import requests
|
|
import datetime
|
|
from typing import Dict, Any, Optional
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from .base import Filter
|
|
|
|
logger = logging.getLogger("log-alert")
|
|
|
|
|
|
class TimestampFilter(Filter):
|
|
"""Concrete implementation for Timestamp filter."""
|
|
|
|
def __init__(self, config: Dict[str, Any]):
|
|
self.source_field = config["source-field"]
|
|
self.timezone = config.get("timezone", "UTC")
|
|
self.timestamp_format = config.get("timestamp-format")
|
|
|
|
def filter(self, log: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
timestamp_string = log.get("labels", {}).get(self.source_field)
|
|
if timestamp_string is None:
|
|
logger.warning(f"No timestamp found in log for field {self.source_field}")
|
|
return None
|
|
else:
|
|
try:
|
|
if self.timestamp_format:
|
|
logger.debug(f"Parsing timestamp {timestamp_string} with format {self.timestamp_format}")
|
|
parsed_time = datetime.datetime.strptime(timestamp_string, self.timestamp_format)
|
|
if parsed_time.year == 1900:
|
|
# If the year is not specified, assume the current year
|
|
parsed_time = parsed_time.replace(year=datetime.datetime.now().year)
|
|
timestamp_value = int(parsed_time.replace(tzinfo=ZoneInfo(self.timezone)).timestamp())
|
|
else:
|
|
timestamp_value = int(datetime.datetime.fromisoformat(timestamp_string).timestamp())
|
|
except Exception as e:
|
|
logger.error(f"Error parsing timestamp {timestamp_value} with format {self.timestamp_format}: {e}")
|
|
|
|
if timestamp_value >= log.get("fetcher-start-time", 0) and timestamp_value < log.get("fetcher-end-time", float('inf')):
|
|
logger.debug(f"Log timestamp {timestamp_value} is within the fetcher time range.")
|
|
return log
|
|
else:
|
|
logger.debug(f"Log timestamp {timestamp_value} is outside the fetcher time range.")
|
|
return None
|