Log Management and Analysis

1from elasticsearch import Elasticsearch, helpers
2from elasticsearch.exceptions import ElasticsearchException
3from datetime import datetime, timedelta
4import json
5import logging
6from typing import Dict, List, Optional, Any
7from dataclasses import dataclass, asdict
8import hashlib
9
10logger = logging.getLogger(__name__)
11
12@dataclass
13class LogEntry:
14    """Structured log entry"""
15    timestamp: str
16    level: str
17    service: str
18    message: str
19    logger_name: str
20    thread: str
21    host: str
22    environment: str
23    trace_id: Optional[str] = None
24    span_id: Optional[str] = None
25    user_id: Optional[str] = None
26    request_id: Optional[str] = None
27    metadata: Dict[str, Any] = None
28
29    def to_dict(self) -> Dict:
30        """Convert to dictionary for Elasticsearch"""
31        data = asdict(self)
32        if self.metadata:
33            data.update(self.metadata)
34        return data
35
36class ElasticsearchLogManager:
37    """
38    Manage logs with Elasticsearch
39    """
40
41    def __init__(self, hosts: List[str], index_prefix: str = 'logs'):
42        """
43        Initialize Elasticsearch client
44
45        Args:
46            hosts: List of Elasticsearch host URLs
47            index_prefix: Prefix for log indices
48        """
49        try:
50            self.es = Elasticsearch(
51                hosts,
52                # In production, add authentication
53                # basic_auth=('username', 'password'),
54                # use_ssl=True,
55                # verify_certs=True
56            )
57
58            # Test connection
59            if self.es.ping():
60                logger.info(f"✅ Connected to Elasticsearch: {hosts}")
61            else:
62                raise ConnectionError("Failed to connect to Elasticsearch")
63
64            self.index_prefix = index_prefix
65
66            # Create index template for consistent mapping
67            self._create_index_template()
68
69        except Exception as e:
70            logger.error(f"❌ Failed to initialize Elasticsearch: {e}")
71            raise
72
73    def _create_index_template(self):
74        """Create index template for log entries"""
75        template = {
76            'index_patterns': [f'{self.index_prefix}-*'],
77            'template': {
78                'settings': {
79                    'number_of_shards': 2,
80                    'number_of_replicas': 1,
81                    'index.lifecycle.name': 'logs-policy',
82                    'index.lifecycle.rollover_alias': f'{self.index_prefix}-alias'
83                },
84                'mappings': {
85                    'properties': {
86                        'timestamp': {'type': 'date'},
87                        'level': {'type': 'keyword'},
88                        'service': {'type': 'keyword'},
89                        'message': {'type': 'text', 'analyzer': 'standard'},
90                        'logger_name': {'type': 'keyword'},
91                        'thread': {'type': 'keyword'},
92                        'host': {'type': 'keyword'},
93                        'environment': {'type': 'keyword'},
94                        'trace_id': {'type': 'keyword'},
95                        'span_id': {'type': 'keyword'},
96                        'user_id': {'type': 'keyword'},
97                        'request_id': {'type': 'keyword'}
98                    }
99                }
100            }
101        }
102
103        try:
104            self.es.indices.put_index_template(
105                name=f'{self.index_prefix}-template',
106                body=template
107            )
108            logger.info(f"✅ Index template created: {self.index_prefix}-template")
109
110        except ElasticsearchException as e:
111            logger.warning(f"⚠️  Index template creation failed: {e}")
112
113    def _get_index_name(self, date: datetime = None) -> str:
114        """
115        Get time-based index name
116
117        Args:
118            date: Date for index (default: today)
119
120        Returns:
121            Index name like 'logs-2024-01-15'
122        """
123        if date is None:
124            date = datetime.now()
125
126        return f"{self.index_prefix}-{date.strftime('%Y-%m-%d')}"
127
128    def index_log(self, log_entry: LogEntry) -> bool:
129        """
130        Index a log entry
131
132        Args:
133            log_entry: Log entry to index
134
135        Returns:
136            True if successful
137        """
138        try:
139            index_name = self._get_index_name()
140
141            # Generate document ID from log content for deduplication
142            doc_id = hashlib.md5(
143                f"{log_entry.timestamp}{log_entry.service}{log_entry.message}".encode()
144            ).hexdigest()
145
146            self.es.index(
147                index=index_name,
148                id=doc_id,
149                document=log_entry.to_dict()
150            )
151
152            logger.debug(f"Indexed log: {log_entry.level} - {log_entry.message[:50]}")
153            return True
154
155        except ElasticsearchException as e:
156            logger.error(f"❌ Failed to index log: {e}")
157            return False
158
159    def bulk_index_logs(self, log_entries: List[LogEntry]) -> Dict:
160        """
161        Bulk index multiple log entries
162
163        Args:
164            log_entries: List of log entries
165
166        Returns:
167            Summary of bulk operation
168        """
169        try:
170            actions = []
171
172            for log_entry in log_entries:
173                index_name = self._get_index_name(
174                    datetime.fromisoformat(log_entry.timestamp)
175                )
176
177                doc_id = hashlib.md5(
178                    f"{log_entry.timestamp}{log_entry.service}{log_entry.message}".encode()
179                ).hexdigest()
180
181                actions.append({
182                    '_index': index_name,
183                    '_id': doc_id,
184                    '_source': log_entry.to_dict()
185                })
186
187            # Execute bulk operation
188            success, failed = helpers.bulk(
189                self.es,
190                actions,
191                stats_only=True
192            )
193
194            logger.info(f"✅ Bulk indexed: {success} successful, {failed} failed")
195
196            return {
197                'success': success,
198                'failed': failed,
199                'total': len(log_entries)
200            }
201
202        except Exception as e:
203            logger.error(f"❌ Bulk indexing failed: {e}")
204            return {'success': 0, 'failed': len(log_entries), 'total': len(log_entries)}
205
206    def search_logs(self, query: Dict, time_range: timedelta = None,
207                    size: int = 100) -> List[Dict]:
208        """
209        Search logs with query
210
211        Args:
212            query: Elasticsearch query DSL
213            time_range: Time range to search (default: last 24 hours)
214            size: Maximum results to return
215
216        Returns:
217            List of matching log entries
218        """
219        try:
220            if time_range is None:
221                time_range = timedelta(hours=24)
222
223            # Build time-based query
224            end_time = datetime.now()
225            start_time = end_time - time_range
226
227            full_query = {
228                'bool': {
229                    'must': [query],
230                    'filter': [
231                        {
232                            'range': {
233                                'timestamp': {
234                                    'gte': start_time.isoformat(),
235                                    'lte': end_time.isoformat()
236                                }
237                            }
238                        }
239                    ]
240                }
241            }
242
243            # Search across all indices
244            response = self.es.search(
245                index=f'{self.index_prefix}-*',
246                query=full_query,
247                size=size,
248                sort=[{'timestamp': {'order': 'desc'}}]
249            )
250
251            hits = response['hits']['hits']
252            results = [hit['_source'] for hit in hits]
253
254            logger.info(f"✅ Found {len(results)} logs matching query")
255
256            return results
257
258        except ElasticsearchException as e:
259            logger.error(f"❌ Search failed: {e}")
260            return []
261
262    def search_errors(self, service: str = None,
263                     time_range: timedelta = None) -> List[Dict]:
264        """
265        Search for error logs
266
267        Args:
268            service: Filter by service name
269            time_range: Time range to search
270
271        Returns:
272            List of error logs
273        """
274        query = {
275            'bool': {
276                'must': [
277                    {'terms': {'level': ['ERROR', 'CRITICAL', 'FATAL']}}
278                ]
279            }
280        }
281
282        if service:
283            query['bool']['must'].append({'term': {'service': service}})
284
285        return self.search_logs(query, time_range=time_range)
286
287    def get_log_statistics(self, time_range: timedelta = None) -> Dict:
288        """
289        Get log statistics
290
291        Args:
292            time_range: Time range for statistics
293
294        Returns:
295            Statistics summary
296        """
297        try:
298            if time_range is None:
299                time_range = timedelta(hours=24)
300
301            end_time = datetime.now()
302            start_time = end_time - time_range
303
304            # Aggregation query
305            query = {
306                'bool': {
307                    'filter': [
308                        {
309                            'range': {
310                                'timestamp': {
311                                    'gte': start_time.isoformat(),
312                                    'lte': end_time.isoformat()
313                                }
314                            }
315                        }
316                    ]
317                }
318            }
319
320            aggs = {
321                'by_level': {
322                    'terms': {'field': 'level', 'size': 10}
323                },
324                'by_service': {
325                    'terms': {'field': 'service', 'size': 20}
326                },
327                'error_rate': {
328                    'filter': {'terms': {'level': ['ERROR', 'CRITICAL', 'FATAL']}}
329                }
330            }
331
332            response = self.es.search(
333                index=f'{self.index_prefix}-*',
334                query=query,
335                aggs=aggs,
336                size=0  # We only want aggregations
337            )
338
339            # Extract aggregation results
340            total_logs = response['hits']['total']['value']
341            by_level = {
342                bucket['key']: bucket['doc_count']
343                for bucket in response['aggregations']['by_level']['buckets']
344            }
345            by_service = {
346                bucket['key']: bucket['doc_count']
347                for bucket in response['aggregations']['by_service']['buckets']
348            }
349            error_count = response['aggregations']['error_rate']['doc_count']
350
351            stats = {
352                'total_logs': total_logs,
353                'error_count': error_count,
354                'error_rate': (error_count / total_logs * 100) if total_logs > 0 else 0,
355                'by_level': by_level,
356                'by_service': by_service,
357                'time_range_hours': time_range.total_seconds() / 3600
358            }
359
360            logger.info(f"✅ Statistics: {total_logs} logs, {error_count} errors")
361
362            return stats
363
364        except ElasticsearchException as e:
365            logger.error(f"❌ Failed to get statistics: {e}")
366            return {}
367
368    def delete_old_logs(self, retention_days: int = 30) -> Dict:
369        """
370        Delete logs older than retention period
371
372        Args:
373            retention_days: Number of days to retain logs
374
375        Returns:
376            Deletion summary
377        """
378        try:
379            cutoff_date = datetime.now() - timedelta(days=retention_days)
380
381            # Delete indices older than cutoff
382            index_pattern = f"{self.index_prefix}-*"
383            indices = self.es.indices.get(index=index_pattern)
384
385            deleted_indices = []
386
387            for index_name in indices.keys():
388                # Extract date from index name
389                try:
390                    date_str = index_name.split('-')[-3:]  # ['2024', '01', '15']
391                    index_date = datetime(
392                        int(date_str[0]),
393                        int(date_str[1]),
394                        int(date_str[2])
395                    )
396
397                    if index_date < cutoff_date:
398                        self.es.indices.delete(index=index_name)
399                        deleted_indices.append(index_name)
400                        logger.info(f"🗑️  Deleted index: {index_name}")
401
402                except (ValueError, IndexError):
403                    logger.warning(f"⚠️  Could not parse date from index: {index_name}")
404
405            return {
406                'deleted_count': len(deleted_indices),
407                'deleted_indices': deleted_indices,
408                'retention_days': retention_days
409            }
410
411        except ElasticsearchException as e:
412            logger.error(f"❌ Failed to delete old logs: {e}")
413            return {'deleted_count': 0, 'error': str(e)}

1import re
2from datetime import datetime
3from typing import Dict, List, Optional, Pattern
4from dataclasses import dataclass
5import logging
6
7logger = logging.getLogger(__name__)
8
9@dataclass
10class LogPattern:
11    """Log pattern definition"""
12    name: str
13    regex: Pattern
14    fields: List[str]
15    sample: str
16
17class LogParser:
18    """
19    Parse various log formats into structured data
20    """
21
22    def __init__(self):
23        """Initialize log parser with common patterns"""
24        self.patterns: Dict[str, LogPattern] = {}
25
26        # Define common log patterns
27        self._register_common_patterns()
28
29        logger.info("📝 Log parser initialized with patterns")
30
31    def _register_common_patterns(self):
32        """Register common log format patterns"""
33
34        # Apache/Nginx access log
35        apache_pattern = LogPattern(
36            name='apache_access',
37            regex=re.compile(
38                r'(?P<ip>[d.]+) - (?P<user>[w-]+) [(?P<timestamp>[^]]+)] '
39                r'"(?P<method>w+) (?P<path>[^s]+) (?P<protocol>[^"]+)" '
40                r'(?P<status>d+) (?P<size>d+) "(?P<referrer>[^"]*)" "(?P<user_agent>[^"]*)"'
41            ),
42            fields=['ip', 'user', 'timestamp', 'method', 'path', 'protocol',
43                   'status', 'size', 'referrer', 'user_agent'],
44            sample='192.168.1.1 - admin [01/Jan/2024:12:00:00 +0000] "GET /api/users HTTP/1.1" 200 1024 "-" "Mozilla/5.0"'
45        )
46        self.register_pattern(apache_pattern)
47
48        # Python logging format
49        python_pattern = LogPattern(
50            name='python_logging',
51            regex=re.compile(
52                r'(?P<timestamp>d{4}-d{2}-d{2} d{2}:d{2}:d{2},d{3}) - '
53                r'(?P<logger>[w.]+) - (?P<level>w+) - (?P<message>.+)'
54            ),
55            fields=['timestamp', 'logger', 'level', 'message'],
56            sample='2024-01-15 12:00:00,123 - app.module - ERROR - Something went wrong'
57        )
58        self.register_pattern(python_pattern)
59
60        # Syslog format
61        syslog_pattern = LogPattern(
62            name='syslog',
63            regex=re.compile(
64                r'(?P<timestamp>w{3}s+d{1,2} d{2}:d{2}:d{2}) '
65                r'(?P<host>[w-]+) (?P<process>[w[]]+): (?P<message>.+)'
66            ),
67            fields=['timestamp', 'host', 'process', 'message'],
68            sample='Jan 15 12:00:00 server01 sshd[1234]: Failed password for user'
69        )
70        self.register_pattern(syslog_pattern)
71
72        # JSON log format
73        json_pattern = LogPattern(
74            name='json',
75            regex=re.compile(r'^(?P<json>{.+})$'),
76            fields=['json'],
77            sample='{"timestamp": "2024-01-15T12:00:00Z", "level": "ERROR", "message": "Failed"}'
78        )
79        self.register_pattern(json_pattern)
80
81        # Application error log with stack trace
82        error_pattern = LogPattern(
83            name='error_with_trace',
84            regex=re.compile(
85                r'(?P<timestamp>d{4}-d{2}-d{2} d{2}:d{2}:d{2}) '
86                r'[(?P<level>w+)] (?P<error_type>[w.]+): (?P<message>[^
87]+)'
88            ),
89            fields=['timestamp', 'level', 'error_type', 'message'],
90            sample='2024-01-15 12:00:00 [ERROR] ValueError: Invalid input'
91        )
92        self.register_pattern(error_pattern)
93
94    def register_pattern(self, pattern: LogPattern):
95        """Register a log pattern"""
96        self.patterns[pattern.name] = pattern
97        logger.debug(f"Registered pattern: {pattern.name}")
98
99    def parse_line(self, line: str, pattern_name: str = None) -> Optional[Dict]:
100        """
101        Parse a log line
102
103        Args:
104            line: Log line to parse
105            pattern_name: Specific pattern to use (auto-detect if None)
106
107        Returns:
108            Parsed fields or None
109        """
110        if pattern_name:
111            # Use specified pattern
112            if pattern_name in self.patterns:
113                pattern = self.patterns[pattern_name]
114                match = pattern.regex.match(line.strip())
115
116                if match:
117                    return match.groupdict()
118
119            logger.warning(f"⚠️  Pattern not found: {pattern_name}")
120            return None
121
122        # Auto-detect pattern
123        for pattern in self.patterns.values():
124            match = pattern.regex.match(line.strip())
125
126            if match:
127                result = match.groupdict()
128                result['_pattern'] = pattern.name
129                logger.debug(f"Matched pattern: {pattern.name}")
130                return result
131
132        logger.warning(f"⚠️  No pattern matched for line: {line[:50]}...")
133        return None
134
135    def parse_file(self, file_path: str, pattern_name: str = None) -> List[Dict]:
136        """
137        Parse an entire log file
138
139        Args:
140            file_path: Path to log file
141            pattern_name: Pattern to use
142
143        Returns:
144            List of parsed log entries
145        """
146        parsed_logs = []
147
148        try:
149            with open(file_path, 'r', encoding='utf-8') as f:
150                for line_num, line in enumerate(f, 1):
151                    parsed = self.parse_line(line, pattern_name)
152
153                    if parsed:
154                        parsed['_line_number'] = line_num
155                        parsed['_raw'] = line.strip()
156                        parsed_logs.append(parsed)
157
158            logger.info(f"✅ Parsed {len(parsed_logs)} logs from {file_path}")
159
160        except FileNotFoundError:
161            logger.error(f"❌ File not found: {file_path}")
162        except Exception as e:
163            logger.error(f"❌ Failed to parse file: {e}")
164
165        return parsed_logs
166
167    def extract_metrics(self, parsed_logs: List[Dict]) -> Dict:
168        """
169        Extract metrics from parsed logs
170
171        Args:
172            parsed_logs: List of parsed log entries
173
174        Returns:
175            Metrics summary
176        """
177        metrics = {
178            'total_logs': len(parsed_logs),
179            'by_pattern': {},
180            'by_level': {},
181            'by_status': {},
182            'error_count': 0,
183            'unique_ips': set(),
184            'unique_paths': set()
185        }
186
187        for log in parsed_logs:
188            # Count by pattern
189            pattern = log.get('_pattern', 'unknown')
190            metrics['by_pattern'][pattern] = metrics['by_pattern'].get(pattern, 0) + 1
191
192            # Count by level
193            level = log.get('level', log.get('status', 'unknown'))
194            metrics['by_level'][level] = metrics['by_level'].get(level, 0) + 1
195
196            # Track errors
197            if level in ['ERROR', 'CRITICAL', 'FATAL'] or                (log.get('status') and int(log.get('status', 0)) >= 400):
198                metrics['error_count'] += 1
199
200            # Track unique IPs
201            if 'ip' in log:
202                metrics['unique_ips'].add(log['ip'])
203
204            # Track unique paths
205            if 'path' in log:
206                metrics['unique_paths'].add(log['path'])
207
208        # Convert sets to counts
209        metrics['unique_ips'] = len(metrics['unique_ips'])
210        metrics['unique_paths'] = len(metrics['unique_paths'])
211
212        return metrics
213
214    def detect_anomalies(self, parsed_logs: List[Dict]) -> List[Dict]:
215        """
216        Detect anomalies in parsed logs
217
218        Args:
219            parsed_logs: List of parsed log entries
220
221        Returns:
222            List of detected anomalies
223        """
224        anomalies = []
225
226        # Detect high error rates
227        error_logs = [
228            log for log in parsed_logs
229            if log.get('level') in ['ERROR', 'CRITICAL', 'FATAL'] or
230               (log.get('status') and int(log.get('status', 0)) >= 500)
231        ]
232
233        error_rate = len(error_logs) / len(parsed_logs) * 100 if parsed_logs else 0
234
235        if error_rate > 5:  # More than 5% errors
236            anomalies.append({
237                'type': 'high_error_rate',
238                'severity': 'critical',
239                'message': f'Error rate is {error_rate:.2f}%',
240                'error_count': len(error_logs),
241                'total_count': len(parsed_logs)
242            })
243
244        # Detect repeated errors
245        error_messages = {}
246        for log in error_logs:
247            msg = log.get('message', '')
248            error_messages[msg] = error_messages.get(msg, 0) + 1
249
250        for msg, count in error_messages.items():
251            if count >= 10:  # Same error repeated 10+ times
252                anomalies.append({
253                    'type': 'repeated_error',
254                    'severity': 'warning',
255                    'message': f'Error repeated {count} times: {msg[:100]}',
256                    'count': count
257                })
258
259        # Detect suspicious patterns (example: many 404s from same IP)
260        ip_404s = {}
261        for log in parsed_logs:
262            if log.get('status') == '404' and 'ip' in log:
263                ip = log['ip']
264                ip_404s[ip] = ip_404s.get(ip, 0) + 1
265
266        for ip, count in ip_404s.items():
267            if count >= 20:  # 20+ 404s from same IP
268                anomalies.append({
269                    'type': 'suspicious_activity',
270                    'severity': 'warning',
271                    'message': f'IP {ip} generated {count} 404 errors',
272                    'ip': ip,
273                    'count': count
274                })
275
276        logger.info(f"✅ Detected {len(anomalies)} anomalies")
277
278        return anomalies

1import time
2from typing import Dict, List, Callable
3from collections import deque, defaultdict
4from datetime import datetime, timedelta
5from threading import Thread, Lock
6import logging
7
8logger = logging.getLogger(__name__)
9
10class LogAlert:
11    """Log-based alert definition"""
12
13    def __init__(self, name: str, condition: Callable,
14                 severity: str, cooldown_seconds: int = 300):
15        """
16        Initialize alert
17
18        Args:
19            name: Alert name
20            condition: Function that returns True if alert should fire
21            severity: Alert severity level
22            cooldown_seconds: Minimum time between alerts
23        """
24        self.name = name
25        self.condition = condition
26        self.severity = severity
27        self.cooldown_seconds = cooldown_seconds
28        self.last_fired = None
29
30    def should_fire(self) -> bool:
31        """Check if alert is in cooldown"""
32        if self.last_fired is None:
33            return True
34
35        elapsed = (datetime.now() - self.last_fired).total_seconds()
36        return elapsed >= self.cooldown_seconds
37
38    def fire(self):
39        """Mark alert as fired"""
40        self.last_fired = datetime.now()
41
42class RealTimeLogAnalyzer:
43    """
44    Analyze logs in real-time and trigger alerts
45    """
46
47    def __init__(self, window_size: int = 1000):
48        """
49        Initialize analyzer
50
51        Args:
52            window_size: Number of recent logs to keep in memory
53        """
54        self.window_size = window_size
55        self.log_buffer = deque(maxlen=window_size)
56        self.alerts: List[LogAlert] = []
57        self.alert_handlers: List[Callable] = []
58        self.stats = defaultdict(int)
59        self.lock = Lock()
60
61        # Register default alerts
62        self._register_default_alerts()
63
64        logger.info(f"📊 Real-time analyzer initialized (window: {window_size})")
65
66    def _register_default_alerts(self):
67        """Register default alert conditions"""
68
69        # High error rate alert
70        def high_error_rate():
71            recent_logs = list(self.log_buffer)
72            if len(recent_logs) < 100:
73                return False
74
75            errors = sum(
76                1 for log in recent_logs[-100:]
77                if log.get('level') in ['ERROR', 'CRITICAL', 'FATAL']
78            )
79
80            error_rate = errors / 100 * 100
81            return error_rate > 10  # More than 10% errors
82
83        self.register_alert(LogAlert(
84            name='High Error Rate',
85            condition=high_error_rate,
86            severity='critical',
87            cooldown_seconds=300
88        ))
89
90        # Repeated errors alert
91        def repeated_errors():
92            recent_logs = list(self.log_buffer)[-50:]
93            error_messages = defaultdict(int)
94
95            for log in recent_logs:
96                if log.get('level') == 'ERROR':
97                    msg = log.get('message', '')[:100]
98                    error_messages[msg] += 1
99
100            return any(count >= 5 for count in error_messages.values())
101
102        self.register_alert(LogAlert(
103            name='Repeated Errors',
104            condition=repeated_errors,
105            severity='warning',
106            cooldown_seconds=180
107        ))
108
109        # Service unavailable alert
110        def service_unavailable():
111            recent_logs = list(self.log_buffer)[-20:]
112
113            unavailable_count = sum(
114                1 for log in recent_logs
115                if log.get('status') in ['503', '504']
116            )
117
118            return unavailable_count >= 3
119
120        self.register_alert(LogAlert(
121            name='Service Unavailable',
122            condition=service_unavailable,
123            severity='critical',
124            cooldown_seconds=60
125        ))
126
127    def register_alert(self, alert: LogAlert):
128        """Register an alert condition"""
129        self.alerts.append(alert)
130        logger.info(f"✅ Registered alert: {alert.name}")
131
132    def register_alert_handler(self, handler: Callable):
133        """
134        Register alert handler function
135
136        Args:
137            handler: Function that accepts (alert_name, severity, details)
138        """
139        self.alert_handlers.append(handler)
140
141    def process_log(self, log_entry: Dict):
142        """
143        Process a single log entry
144
145        Args:
146            log_entry: Parsed log entry
147        """
148        with self.lock:
149            # Add to buffer
150            self.log_buffer.append(log_entry)
151
152            # Update statistics
153            level = log_entry.get('level', 'INFO')
154            self.stats[f'level_{level}'] += 1
155            self.stats['total_logs'] += 1
156
157            # Check alert conditions
158            for alert in self.alerts:
159                if alert.should_fire() and alert.condition():
160                    self._trigger_alert(alert, log_entry)
161                    alert.fire()
162
163    def _trigger_alert(self, alert: LogAlert, log_entry: Dict):
164        """Trigger alert"""
165        alert_data = {
166            'name': alert.name,
167            'severity': alert.severity,
168            'timestamp': datetime.now().isoformat(),
169            'triggering_log': log_entry,
170            'stats': dict(self.stats)
171        }
172
173        logger.warning(f"🚨 ALERT: {alert.name} ({alert.severity})")
174
175        # Call registered handlers
176        for handler in self.alert_handlers:
177            try:
178                handler(**alert_data)
179            except Exception as e:
180                logger.error(f"❌ Alert handler failed: {e}")
181
182    def get_statistics(self) -> Dict:
183        """Get current statistics"""
184        with self.lock:
185            buffer_size = len(self.log_buffer)
186
187            return {
188                'buffer_size': buffer_size,
189                'window_utilization': buffer_size / self.window_size * 100,
190                'stats': dict(self.stats),
191                'alerts_registered': len(self.alerts)
192            }
193
194    def tail_file(self, file_path: str, parser: 'LogParser'):
195        """
196        Tail a log file and process in real-time
197
198        Args:
199            file_path: Path to log file to tail
200            parser: LogParser instance for parsing
201        """
202        logger.info(f"👁️  Tailing log file: {file_path}")
203
204        try:
205            with open(file_path, 'r') as f:
206                # Move to end of file
207                f.seek(0, 2)
208
209                while True:
210                    line = f.readline()
211
212                    if line:
213                        # Parse and process log line
214                        parsed = parser.parse_line(line)
215
216                        if parsed:
217                            self.process_log(parsed)
218                    else:
219                        # No new line, wait briefly
220                        time.sleep(0.1)
221
222        except KeyboardInterrupt:
223            logger.info("⚠️  Stopped tailing file")
224        except Exception as e:
225            logger.error(f"❌ Tail failed: {e}")