# 限制输出数量避免卡顿
grep -m 1000 "ERROR" large.log

# 使用fgrep加速固定字符串匹配
fgrep -f error_patterns.txt application.log

# 并行处理加速搜索
find /var/log -name "*.log" -exec grep -l "ERROR" {} \;

# 时间范围筛选+错误统计
sed -n '/2023-11-15 14:00:00/,/2023-11-15 15:00:00/p' app.log | grep -c "ERROR"

# 多条件管道过滤
grep "ERROR" application.log | grep -v "预期错误" | head -50

# 根据traceId追踪完整请求链路
grep -A 10 "traceId:0a1b2c3d" service*.log | sort -k 3

# 多服务日志时间序列分析
grep "特定事务ID" service1.log service2.log service3.log | sort -t ' ' -k 2

# 查找慢查询（响应时间大于1秒）
awk '$9 > 1000 {print $0}' access.log

# 统计接口响应时间分布
awk '{print $9}' access.log | sort -n | uniq -c

# 找出最耗时的请求
awk '{print $7, $9}' access.log | sort -k2 -nr | head -20

# 统计每小时错误数量
grep "ERROR" application.log | awk '{print $2}' | cut -d: -f1 | sort | uniq -c

# 错误类型分类统计
grep "Exception" application.log | awk -F: '{print $4}' | sort | uniq -c | sort -nr

# TOP 10错误排名
grep "ERROR" application.log | awk '{print $5}' | sort | uniq -c | sort -nr | head -10

# 生成错误时间序列数据（CSV格式）
grep "ERROR" application.log | awk '{print $1" "$2","$5}' | cut -d, -f1 | head -100

# 准备监控系统导入数据
grep -c "ERROR" application.log | awk '{print "errors,host=server01 value=" $1}'

#!/usr/bin/env python3
import subprocess
import re
from collections import defaultdict
from typing import List, Dict, Pattern
import signal

class LogMonitor:
    """日志监控器，实时监控并统计错误"""
    
    def __init__(self, log_file: str, error_patterns: List[str]):
        self.log_file = log_file
        self.compiled_patterns = [re.compile(p, re.IGNORECASE) for p in error_patterns]
        self.error_counts: Dict[str, int] = defaultdict(int)
        self._running = False
        
    def _handle_signal(self, signum, frame):
        """处理中断信号"""
        self._running = False
        
    def monitor(self):
        """开始监控日志"""
        signal.signal(signal.SIGINT, self._handle_signal)
        self._running = True
        
        # 使用更安全的参数传递方式
        with subprocess.Popen(
            ['tail', '-f', self.log_file],
            stdout=subprocess.PIPE,
            universal_newlines=True
        ) as process:
            
            print(f"开始监控日志文件: {self.log_file}")
            try:
                for line in process.stdout:
                    if not self._running:
                        break
                    self._process_line(line.strip())
            except KeyboardInterrupt:
                pass
            finally:
                print("\n监控结束")
                self._display_summary()
    
    def _process_line(self, line: str):
        """处理单行日志"""
        for pattern in self.compiled_patterns:
            if pattern.search(line):
                pattern_str = pattern.pattern
                self.error_counts[pattern_str] += 1
                print(f"检测到错误: {pattern_str} - 总计: {self.error_counts[pattern_str]}")
    
    def _display_summary(self):
        """显示错误统计摘要"""
        if self.error_counts:
            print("\n错误统计:")
            for pattern, count in sorted(self.error_counts.items(), key=lambda x: x[1], reverse=True):
                print(f"{pattern}: {count}次")
        else:
            print("未检测到任何错误")

def main():
    # 监控常见的错误类型
    error_patterns = [
        "TimeoutException",
        "NullPointerException",
        "OutOfMemoryError",
        "Connection refused"
    ]
    
    monitor = LogMonitor('/var/log/application.log', error_patterns)
    monitor.monitor()

if __name__ == "__main__":
    main()

<?php
/**
 * 实时监控日志文件，匹配关键词并输出
 * 
 * @param string $logFile 日志文件路径
 * @param array $keywords 要监控的关键词数组
 * @param int $checkInterval 检查间隔时间(秒)
 * @throws RuntimeException 当文件无法访问时抛出异常
 */
function monitorLog(string $logFile, array $keywords, int $checkInterval = 1): void {
    // 验证文件可读性
    if (!is_readable($logFile)) {
        throw new RuntimeException("无法读取日志文件: {$logFile}");
    }

    $lastSize = filesize($logFile);
    $keywords = array_map('strtolower', $keywords); // 统一转为小写，便于不区分大小写匹配
    
    while (true) {
        clearstatcache();
        $currentSize = filesize($logFile);
        
        if ($currentSize < $lastSize) {
            // 处理日志轮转情况(文件被清空或重新创建)
            $lastSize = 0;
        }
        
        if ($currentSize > $lastSize) {
            try {
                $fh = fopen($logFile, 'r');
                if (!$fh) {
                    throw new RuntimeException("无法打开日志文件: {$logFile}");
                }
                
                fseek($fh, $lastSize);
                
                while (!feof($fh)) {
                    $line = trim(fgets($fh));
                    if (empty($line)) continue;
                    
                    $lowerLine = strtolower($line);
                    foreach ($keywords as $keyword) {
                        if (str_contains($lowerLine, $keyword)) {
                            echo sprintf("[%s] 发现 %s: %s\n", 
                                date('Y-m-d H:i:s'), 
                                $keyword, 
                                $line
                            );
                            break; // 找到任一关键词即可跳出循环
                        }
                    }
                }
                
                $lastSize = $currentSize;
            } finally {
                if (isset($fh) && is_resource($fh)) {
                    fclose($fh);
                }
            }
        }
        
        sleep($checkInterval);
    }
}

// 监控配置
try {
    $keywords = ['exception', 'error', 'failed', 'timeout'];
    monitorLog('/var/log/application.log', $keywords);
} catch (RuntimeException $e) {
    echo "监控失败: " . $e->getMessage() . PHP_EOL;
    exit(1);
}