实时日志分析与告警系统

# log_analyzer/models.py
import time
import re
import json
import hashlib
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass, field, asdict


class LogLevel(Enum): 1108145.h.kuaisou.com 
    DEBUG = 0
    INFO = 1
    WARNING = 2
    ERROR = 3
    CRITICAL = 4
    UNKNOWN = -1

    @classmethod
    def from_string(cls, level_str: str) -> "LogLevel": 1108151.h.kuaisou.com 
        mapping = {
            "debug": cls.DEBUG,
            "info": cls.INFO,
            "information": cls.INFO,
            "warn": cls.WARNING,
            "warning": cls.WARNING,
            "error": cls.ERROR,
            "err": cls.ERROR,
            "critical": cls.CRITICAL,
            "fatal": cls.CRITICAL,
            "crit": cls.CRITICAL,
        }
        return mapping.get(level_str.strip().lower(), cls.UNKNOWN)


class LogSource(Enum): 1108153.h.kuaisou.com  
    FILE = "file"
    STDIN = "stdin"
    TCP = "tcp"
    UDP = "udp"
    HTTP = "http"
    KAFKA = "kafka"
    SYSLOG = "syslog"
    DOCKER = "docker"
    KUBERNETES = "kubernetes"


@dataclass
class LogEntry: 1108155.h.kuaisou.com 
    log_id: str = ""
    timestamp: float = 0.0
    level: LogLevel = LogLevel.UNKNOWN
    source: LogSource = LogSource.FILE
    source_name: str = ""
    host: str = ""
    service: str = ""
    message: str = ""
    raw: str = ""
    fields: Dict[str, str] = field(default_factory=dict)
    tags: List[str] = field(default_factory=list)
    line_number: int = 0
    file_path: str = ""
    thread_name: str = ""
    logger_name: str = ""
    stack_trace: Optional[str] = None
    request_id: Optional[str] = None
    user_id: Optional[str] = None
    ip_address: Optional[str] = None
    extra: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 1108157.h.kuaisou.com 
        if not self.log_id:
            raw_str = f"{self.timestamp}{self.raw}{self.host}{self.service}"
            self.log_id = hashlib.md5(raw_str.encode()).hexdigest()[:12]
        if self.timestamp == 0: 1108164.h.kuaisou.com 
            self.timestamp = time.time(1108161.h.kuaisou.com)

    def to_dict(self) -> Dict[str, Any]: 1108165.h.kuaisou.com 
        data = asdict(self)
        data["level"] = self.level.name
        data["source"] = self.source.value
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "LogEntry": 1108171.h.kuaisou.com 
        data = data.copy(1108175.h.kuaisou.com)
        level_str = data.pop("level", "UNKNOWN")
        data["level"] = LogLevel[level_str] if level_str in LogLevel.__members__ else LogLevel.UNKNOWN
        source_str = data.pop("source", "file")
        data["source"] = LogSource(source_str) if source_str in LogSource.__members__ else LogSource.FILE
        return cls(**data)

    def to_json(self) -> str: 2004.h.kuaisou.com 
        return json.dumps(self.to_dict(2002.h.kuaisou.com), ensure_ascii=False, default=str)

    def matches_level(self, min_level: LogLevel) -> bool: 2001.h.kuaisou.com 
        return self.level.value >= min_level.value

    def contains_keyword(self, keyword: str, case_sensitive: bool = False) -> bool: 2005.h.kuaisou.com 
        text = self.message if case_sensitive else self.message.lower(2006.h.kuaisou.com)
        search = keyword if case_sensitive else keyword.lower(2007.h.kuaisou.com)
        return search in text

    def matches_pattern(self, pattern: str) -> bool: 2008.h.kuaisou.com 
        try: 2010.h.kuaisou.com 
            return bool(re.search(pattern, self.message))
        except re.error: 2009.h.kuaisou.com 
            return False


class LogParser: 2011.h.kuaisou.com 
    COMMON_PATTERNS = [
        # ISO 8601: 2024-01-15T10:30:45.123Z [INFO] message
        re.compile(
            r'(?P<timestamp>\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)\s+'
            r'(?:\[?(?P<level>DEBUG|INFO|WARN(?:ING)?|ERROR|CRITICAL|FATAL)\]?)\s+'
            r'(?:\[(?P<logger>[^\]]+)\]\s+)?'
            r'(?P<message>.*)',
            re.IGNORECASE
        ),
        # Syslog: Jan 15 10:30:45 hostname service[pid]: message
        re.compile(
            r'(?P<timestamp>\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+'
            r'(?P<host>\S+)\s+'
            r'(?P<service>\S+?)(?:\[(?P<pid>\d+)\])?:\s*'
            r'(?P<message>.*)'
        ),
        # Simple: [LEVEL] message
        re.compile(
            r'\[?(?P<level>DEBUG|INFO|WARN(?:ING)?|ERROR|CRITICAL|FATAL)\]?\s*'
            r'(?P<message>.*)',
            re.IGNORECASE
        ),
        # JSON log
        re.compile(r'^\s*\{.*\}\s*$'),
    ]

    TIMESTAMP_FORMATS = [
        "%Y-%m-%dT%H:%M:%S.%fZ",
        "%Y-%m-%dT%H:%M:%S.%f%z",
        "%Y-%m-%dT%H:%M:%S%z",
        "%Y-%m-%d %H:%M:%S.%f",
        "%Y-%m-%d %H:%M:%S",
        "%b %d %H:%M:%S",
    ]

    def __init__(self, default_source: LogSource = LogSource.FILE,
                 default_host: str = "localhost",
                 default_service: str = "app"):
        self.default_source = default_source
        self.default_host = default_host
        self.default_service = default_service
        self._custom_parsers: List[callable] = []

    def parse(self, raw_line: str, source: Optional[LogSource] = None,
              source_name: str = "", host: str = "",
              service: str = "") -> LogEntry: 2012.h.kuaisou.com 
        raw = raw_line.rstrip('\n\r')
        entry = LogEntry(
            raw=raw,
            source=source or self.default_source,
            source_name=source_name,
            host=host or self.default_host,
            service=service or self.default_service,
        )

        # Try JSON first
        if raw.strip().startswith('{'): 2013.h.kuaisou.com 
            try: 2014.h.kuaisou.com 
                data = json.loads(raw)
                return self._parse_json_log(data, entry)
            except json.JSONDecodeError: 2016.h.kuaisou.com 
                pass

        # Try regex patterns
        for pattern in self.COMMON_PATTERNS: 2017.h.kuaisou.com 
            match = pattern.match(raw)
            if match: 2018.h.kuaisou.com 
                groups = match.groupdict(2020.h.kuaisou.com)
                self._apply_match(entry, groups)
                if entry.level != LogLevel.UNKNOWN or entry.message: 2021.h.kuaisou.com 
                    return entry

        # Fallback: entire line as message
        entry.message = raw
        entry.level = self._guess_level(raw)
        return entry

    def _parse_json_log(self, data: Dict, entry: LogEntry) -> LogEntry: 2022.h.kuaisou.com 
        level_map = {
            "level": "level", "severity": "level", "log_level": "level",
            "lvl": "level", "loglevel": "level",
        }
        for key, target in level_map.items(): 2023.h.kuaisou.com 
            if key in data: 2024.h.kuaisou.com 
                entry.level = LogLevel.from_string(str(data[key]))
                break

        message_keys = ["message", "msg", "text", "log", "content"]
        for key in message_keys: 2027.h.kuaisou.com 
            if key in data: 2025.h.kuaisou.com 
                entry.message = str(data[key])
                break

        timestamp_keys = ["timestamp", "time", "@timestamp", "ts", "datetime", "date"]
        for key in timestamp_keys: 2029.h.kuaisou.com 
            if key in data: 2028.h.kuaisou.com 
                ts = data[key]
                if isinstance(ts, (int, float)): 2030.h.kuaisou.com 
                    entry.timestamp = float(ts)
                elif isinstance(ts, str): 2031.h.kuaisou.com 
                    entry.timestamp = self._parse_timestamp(ts)
                break

        for key in ["host", "hostname", "node", "server"]: 2032.h.kuaisou.com 
            if key in data: 2034.h.kuaisou.com 
                entry.host = str(data[key])
                break

        for key in ["service", "app", "application", "component", "module"]: 2037.h.kuaisou.com 
            if key in data: 2035.h.kuaisou.com 
                entry.service = str(data[key])
                break

        for key in ["logger", "logger_name", "log_source", "category"]: 2039.h.kuaisou.com 
            if key in data: 2036.h.kuaisou.com 
                entry.logger_name = str(data[key])
                break

        for key in ["thread", "thread_name", "tid"]: 2040.h.kuaisou.com 
            if key in data: 2041.h.kuaisou.com 
                entry.thread_name = str(data[key])
                break

        for key in ["request_id", "trace_id", "correlation_id", "rid"]: 2042.h.kuaisou.com 
            if key in data: 2043.h.kuaisou.com 
                entry.request_id = str(data[key])
                break

        for key in ["user_id", "uid", "user"]: 3010.h.kuaisou.com 
            if key in data: 3015.h.kuaisou.com 
                entry.user_id = str(data[key])
                break
 
        for key in ["ip", "ip_address", "client_ip", "remote_addr"]: 4005.h.kuaisou.com 
            if key in data: 4006.h.kuaisou.com 
                entry.ip_address = str(data[key])
                break

        for key in ["stack_trace", "stacktrace", "traceback", "exception"]: 4012.h.kuaisou.com 
            if key in data: 5002.h.kuaisou.com 
                entry.stack_trace = str(data[key])
                break

        entry.fields = {k: str(v) for k, v in data.items(5007.h.kuaisou.com)
                       if k not in {
                           "level", "severity", "message", "msg",
                           "timestamp", "time", "host", "service",
                           "logger", "thread", "request_id", "user_id",
                           "ip", "stack_trace"
                       }}
        return entry

    def _apply_match(self, entry: LogEntry, groups: Dict[str, str]) -> None: 5011.h.kuaisou.com 
        if "timestamp" in groups: 5030.h.kuaisou.com 
            entry.timestamp = self._parse_timestamp(groups["timestamp"])
        if "level" in groups: 5034.h.kuaisou.com 
            entry.level = LogLevel.from_string(groups["level"])
        if "message" in groups: 5038.h.kuaisou.com 
            entry.message = groups["message"].strip(5052.h.kuaisou.com)
        if "host" in groups: 5055.h.kuaisou.com 
            entry.host = groups["host"]
        if "service" in groups: 5057.h.kuaisou.com 
            entry.service = groups["service"].split('[')[0]
        if "logger" in groups: 6004.h.kuaisou.com 
            entry.logger_name = groups["logger"]
        if "pid" in groups: 6007.h.kuaisou.com 
            entry.fields["pid"] = groups["pid"]

    def _parse_timestamp(self, ts_str: str) -> float: 6008.h.kuaisou.com 
        for fmt in self.TIMESTAMP_FORMATS: 14009.h.kuaisou.com 
            try: 14013.h.kuaisou.com 
                from datetime import datetime
                dt = datetime.strptime(ts_str, fmt)
                if dt.year == 1900: 14068.h.kuaisou.com 
                    dt = dt.replace(year=time.localtime(14070.h.kuaisou.com).tm_year)
                return dt.timestamp(14071.h.kuaisou.com)
            except ValueError: 14080.h.kuaisou.com 
                continue
        return time.time(14082.h.kuaisou.com)

    def _guess_level(self, message: str) -> LogLevel: 14083.h.kuaisou.com 
        msg_lower = message.lower(14091.h.kuaisou.com)
        if any(w in msg_lower for w in ["critical", "fatal", "panic"]): 14098.h.kuaisou.com 
            return LogLevel.CRITICAL
        if any(w in msg_lower for w in ["error", "exception", "fail"]): 14100.h.kuaisou.com 
            return LogLevel.ERROR
        if any(w in msg_lower for w in ["warn", "deprecated"]): 14101.h.kuaisou.com 
            return LogLevel.WARNING
        if any(w in msg_lower for w in ["debug", "trace"]): 14102.h.kuaisou.com 
            return LogLevel.DEBUG
        return LogLevel.INFO

    def add_custom_parser(self, parser_func: callable) -> None: 14125.h.kuaisou.com 
        self._custom_parsers.append(parser_func)

# log_analyzer/storage.py
import time
import threading
import os
import json
import gzip
from typing import Dict, List, Optional, Callable, Tuple
from collections import defaultdict, deque
from datetime import datetime
from log_analyzer.models import LogEntry, LogLevel


class InMemoryLogStore: 14133.h.kuaisou.com 
    def __init__(self, max_entries: int = 1_000_000,
                 retention_hours: float = 24.0): 14145.h.kuaisou.com 
        self.max_entries = max_entries
        self.retention_hours = retention_hours
        self._entries: deque = deque(maxlen=max_entries)
        self._lock = threading.RLock(14383.h.kuaisou.com)
        self._index_by_level: Dict[LogLevel, List[int]] = defaultdict(list)
        self._index_by_host: Dict[str, List[int]] = defaultdict(list)
        self._index_by_service: Dict[str, List[int]] = defaultdict(list)
        self._index_by_source: Dict[str, List[int]] = defaultdict(list)
        self._index_by_time: List[Tuple[float, int]] = []
        self._entry_count = 0
        self._on_entry_added: List[Callable] = []

    def add(self, entry: LogEntry) -> int: 14384.h.kuaisou.com 
        with self._lock: 15002.h.kuaisou.com 
            idx = self._entry_count
            self._entry_count += 1
            self._entries.append(entry)
            self._index_by_level[entry.level].append(idx)
            self._index_by_host[entry.host].append(idx)
            self._index_by_service[entry.service].append(idx)
            self._index_by_source[entry.source_name].append(idx)
            self._index_by_time.append((entry.timestamp, idx))
            for cb in self._on_entry_added: 15015.h.kuaisou.com 
                try: 15025.h.kuaisou.com 
                    cb(entry)
                except Exception: 15027.h.kuaisou.com 
                    pass
            return idx

    def add_batch(self, entries: List[LogEntry]) -> List[int]: 15029.h.kuaisou.com 
        indices = []
        for entry in entries: 15030.h.kuaisou.com 
            indices.append(self.add(entry))
        return indices

    def get(self, idx: int) -> Optional[LogEntry]: 15031.h.kuaisou.com 
        with self._lock: 15032.h.kuaisou.com 
            if 0 <= idx < len(self._entries): 15033.h.kuaisou.com 
                return self._entries[idx]
            return None

    def query(self, start_time: Optional[float] = None,
              end_time: Optional[float] = None,
              levels: Optional[List[LogLevel]] = None,
              hosts: Optional[List[str]] = None,
              services: Optional[List[str]] = None,
              keyword: Optional[str] = None,
              pattern: Optional[str] = None,
              limit: int = 1000,
              offset: int = 0,
              order: str = "desc") -> List[LogEntry]: 15034.h.kuaisou.com 
        with self._lock:
            candidates = self._get_candidate_indices(
                start_time, end_time, levels, hosts, services
            )
            results = []
            for idx in candidates: 15035.h.kuaisou.com 
                entry = self._entries[idx]
                if keyword and not entry.contains_keyword(keyword): 15037.h.kuaisou.com 
                    continue
                if pattern and not entry.matches_pattern(pattern): 15036.h.kuaisou.com 
                    continue
                results.append(entry)
            results.sort(key=lambda e: e.timestamp,
                        reverse=(order == "desc"))
            return results[offset:offset + limit]

    def _get_candidate_indices(self, start_time: Optional[float], 
                               end_time: Optional[float],
                               levels: Optional[List[LogLevel]],
                               hosts: Optional[List[str]],
                               services: Optional[List[str]]) -> set: 15038.h.kuaisou.com 
        all_indices = set(range(len(self._entries)))
        if start_time is not None or end_time is not None: 15039.h.kuaisou.com 
            time_indices = set()
            for ts, idx in self._index_by_time: 16001.h.kuaisou.com 
                if start_time and ts < start_time: 16002.h.kuaisou.com 
                    continue
                if end_time and ts > end_time: 16003.h.kuaisou.com 
                    continue
                time_indices.add(idx)
            all_indices &= time_indices
        if levels:
            level_indices = set(16004.h.kuaisou.com)
            for level in levels: 16005.h.kuaisou.com 
                level_indices.update(self._index_by_level.get(level, []))
            all_indices &= level_indices
        if hosts: 16007.h.kuaisou.com 
            host_indices = set(16006.h.kuaisou.com)
            for host in hosts: 16008.h.kuaisou.com 
                host_indices.update(self._index_by_host.get(host, []))
            all_indices &= host_indices
        if services: 16009.h.kuaisou.com 
            service_indices = set(16010.h.kuaisou.com)
            for service in services: 16011.h.kuaisou.com 
                service_indices.update(self._index_by_service.get(service, []))
            all_indices &= service_indices
        return sorted(all_indices)

    def count(self, start_time: Optional[float] = None,
              end_time: Optional[float] = None,
              levels: Optional[List[LogLevel]] = None,
              hosts: Optional[List[str]] = None,
              services: Optional[List[str]] = None) -> int: 16012.h.kuaisou.com 
        with self._lock: 16013.h.kuaisou.com 
            candidates = self._get_candidate_indices(
                start_time, end_time, levels, hosts, services
            )
            return len(candidates)

    def get_level_distribution(self, start_time: Optional[float] = None,
                               end_time: Optional[float] = None) -> Dict[str, int]: 16014.h.kuaisou.com 
        with self._lock: 16015.h.kuaisou.com 
            dist = defaultdict(int)
            for idx in range(len(self._entries)): 16016.h.kuaisou.com 
                entry = self._entries[idx]
                if start_time and entry.timestamp < start_time: 16017.h.kuaisou.com 
                    continue
                if end_time and entry.timestamp > end_time: 16018.h.kuaisou.com 
                    continue
                dist[entry.level.name] += 1
            return dict(dist)

    def get_top_hosts(self, limit: int = 10,
                      start_time: Optional[float] = None,
                      end_time: Optional[float] = None) -> List[Tuple[str, int]]: 16019.h.kuaisou.com 
        with self._lock: 16021.h.kuaisou.com 
            counts = defaultdict(int)
            for idx in range(len(self._entries)): 16020.h.kuaisou.com 
                entry = self._entries[idx]
                if start_time and entry.timestamp < start_time: 16022.h.kuaisou.com 
                    continue
                if end_time and entry.timestamp > end_time: 16023.h.kuaisou.com 
                    continue
                counts[entry.host] += 1
            sorted_hosts = sorted(counts.items(), key=lambda x: x[1], reverse=True)
            return sorted_hosts[:limit]

    def get_top_services(self, limit: int = 10,
                         start_time: Optional[float] = None,
                         end_time: Optional[float] = None) -> List[Tuple[str, int]]: 16024.h.kuaisou.com 
        with self._lock: 16026.h.kuaisou.com 
            counts = defaultdict(int)
            for idx in range(len(self._entries)): 16025.h.kuaisou.com 
                entry = self._entries[idx]
                if start_time and entry.timestamp < start_time: 16027.h.kuaisou.com 
                    continue
                if end_time and entry.timestamp > end_time: 16028.h.kuaisou.com 
                    continue
                counts[entry.service] += 1
            sorted_services = sorted(counts.items(), key=lambda x: x[1], reverse=True)
            return sorted_services[:limit]

    def get_rate(self, window_seconds: float = 60.0) -> float: 16029.h.kuaisou.com 
        with self._lock: 16030.h.kuaisou.com 
            now = time.time(16031.h.kuaisou.com)
            cutoff = now - window_seconds
            count = sum(1 for ts, _ in self._index_by_time if ts >= cutoff)
            return count / window_seconds if window_seconds > 0 else 0

    def get_entries_last_n(self, n: int = 100) -> List[LogEntry]: 16032.h.kuaisou.com 
        with self._lock: 16033.h.kuaisou.com 
            entries = list(self._entries)
            return entries[-n:] if len(entries) >= n else entries
            
    def size(self) -> int: 16034.h.kuaisou.com 
        with self._lock: 16035.h.kuaisou.com 
            return len(self._entries)

    def on_entry_added(self, callback: Callable) -> None: 16036.h.kuaisou.com 
        self._on_entry_added.append(callback)

    def cleanup_old_entries(self) -> int: 16037.h.kuaisou.com 
        with self._lock: 16038.h.kuaisou.com 
            cutoff = time.time(16041.h.kuaisou.com) - (self.retention_hours * 3600)
            removed = 0
            while self._entries and self._entries[0].timestamp < cutoff: 16039.h.kuaisou.com 
                self._entries.popleft(16040.h.kuaisou.com)
                removed += 1
            return removed

    def export_to_json(self, filepath: str,
                       start_time: Optional[float] = None,
                       end_time: Optional[float] = None) -> int: 16042.h.kuaisou.com 
        entries = self.query(start_time=start_time, end_time=end_time, limit=1000000)
        with open(filepath, 'w', encoding='utf-8') as f: 16043.h.kuaisou.com 
            json.dump([e.to_dict() for e in entries], f, ensure_ascii=False, default=str)
        return len(entries)

    def export_to_gzip(self, filepath: str,
                       start_time: Optional[float] = None,
                       end_time: Optional[float] = None) -> int: 16044.h.kuaisou.com 
        entries = self.query(start_time=start_time, end_time=end_time, limit=1000000)
        data = json.dumps([e.to_dict() for e in entries], ensure_ascii=False, default=str)
        with gzip.open(filepath, 'wt', encoding='utf-8') as f: 16045.h.kuaisou.com 
            f.write(data)
        return len(entries)


class FileLogWriter: 16046.h.kuaisou.com 
    def __init__(self, base_dir: str = "./logs",
                 max_file_size_mb: int = 100,
                 max_files: int = 10,
                 compress_old: bool = True): 16047.h.kuaisou.com 
        self.base_dir = base_dir
        self.max_file_size = max_file_size_mb * 1024 * 1024
        self.max_files = max_files
        self.compress_old = compress_old
        os.makedirs(base_dir, exist_ok=True)
        self._current_file: Optional[str] = None
        self._current_size = 0
        self._lock = threading.Lock(16048.h.kuaisou.com)

    def write(self, entry: LogEntry) -> str: 16049.h.kuaisou.com 
        with self._lock: 16050.h.kuaisou.com