AI 帮你分析日志：从排查到根因，比翻 grep 快 100 倍

import re
from collections import Counter, defaultdict
from datetime import datetime

log_pattern = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<url>\S+) \S+" (?P<status>\d+) (?P<size>\d+)'
)

hourly_status = defaultdict(lambda: Counter())
ip_counter = Counter()
slow_requests = 0
total_requests = 0
not_found_urls = Counter()

with open("access.log", "r") as f:
    for line in f:
        match = log_pattern.match(line)
        if not match:
            continue
        
        total_requests += 1
        data = match.groupdict()
        
        # 状态码统计
        status_class = f"{data['status'][0]}xx"
        hour = data['time'].split(":")[1]
        hourly_status[hour][status_class] += 1
        
        # IP 统计
        ip_counter[data['ip']] += 1
        
        # 404 统计
        if data['status'] == '404':
            not_found_urls[data['url']] += 1

# 输出结果
print("=== 每小时状态码分布 ===")
for hour in sorted(hourly_status.keys()):
    print(f"  {hour}:00 -> {dict(hourly_status[hour])}")

print("\n=== Top 10 IP ===")
for ip, count in ip_counter.most_common(10):
    print(f"  {ip}: {count} 次请求")

print("\n=== Top 10 404 URL ===")
for url, count in not_found_urls.most_common(10):
    print(f"  {url}: {count} 次")

import re
from datetime import datetime

# 假设每个容器的日志文件
log_files = {
    "order": "order-service.log",
    "payment": "payment-service.log",
    "user": "user-service.log",
}

trace_pattern = re.compile(r'trace_id[=: ]*([a-f0-9-]+)', re.IGNORECASE)
time_pattern = re.compile(r'(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})')

entries = []

for service, filepath in log_files.items():
    try:
        with open(filepath, "r") as f:
            for line in f:
                time_match = time_pattern.search(line)
                trace_match = trace_pattern.search(line)
                
                entry = {
                    "time": time_match.group(1) if time_match else "",
                    "service": service,
                    "trace_id": trace_match.group(1) if trace_match else "",
                    "message": line.strip()
                }
                entries.append(entry)
    except FileNotFoundError:
        print(f"跳过 {filepath}，文件不存在")

# 按时间排序
entries.sort(key=lambda x: x["time"])

# 按 trace_id 分组
from collections import defaultdict
trace_groups = defaultdict(list)
for entry in entries:
    if entry["trace_id"]:
        trace_groups[entry["trace_id"]].append(entry)

# 找出有错误的 trace
print("=== 包含错误的请求追踪 ===")
for trace_id, group in trace_groups.items():
    has_error = any("ERROR" in e["message"] or "error" in e["message"].lower() for e in group)
    if has_error:
        print(f"\nTrace ID: {trace_id}")
        for e in group:
            print(f"  [{e['time']}] {e['service']}: {e['message']}")

# 方法一：直接用 grep + awk
grep '" 5[0-9][0-9] ' access.log | awk '{print $1, $7, $9}'

# 方法二：结合时间过滤（假设日志有时间戳）
awk '/Jul\/15\/2026:1[3-4]:/' access.log | grep '" 5[0-9][0-9] ' | awk '{print $1, $7, $9}'

# 方法三：用 rg（ripgrep）更快
rg '" 5\d\d "' access.log --no-heading -r '$1\t$7\t$9'

# 用 sed 脱敏
sed -E 's/(token=)[^&]*/\1****/g; s/(password=)[^&]*/\1****/g' app.log > sanitized.log