Linux日志聚合与统计方法指南

一、日志聚合方案

1. 基础工具组合

• rsyslog/syslog-ng：系统日志收集与转发
• logrotate：日志轮转管理
• grep/awk/sed：基础文本处理工具

2. 专业日志聚合工具

• ELK Stack (Elasticsearch + Logstash + Kibana)

  • Logstash：日志收集、过滤和转发
  • Elasticsearch：日志存储和索引
  • Kibana：可视化展示

• Fluentd：轻量级日志收集器，支持多种输入输出

• Graylog：一体化日志管理解决方案
• Loki (Grafana Labs)：轻量级日志聚合系统，专为云原生设计

3. 实时日志监控

• tail -f + grep：基础实时监控
• multitail：多文件实时监控
• lnav：高级日志文件查看器

二、日志统计方法

1. 命令行统计技巧

# 统计错误日志数量
grep -c "ERROR" /var/log/app.log

# 按小时统计日志量
awk '{print $1}' /var/log/nginx/access.log | cut -d: -f1 | sort | uniq -c

# 统计IP访问量
awk '{print $1}' /var/log/nginx/access.log | sort | uniq -c | sort -nr

# 统计HTTP状态码
awk '{print $9}' /var/log/nginx/access.log | sort | uniq -c

2. 高级分析工具

• GoAccess：实时Web日志分析器
• AWStats：高级Web统计工具
• Logwatch：系统日志分析报告工具

3. 自定义脚本分析

#!/usr/bin/env python3
from collections import Counter
import re

error_pattern = re.compile(r'ERROR')
ip_pattern = re.compile(r'\d+\.\d+\.\d+\.\d+')

error_count = 0
ip_counter = Counter()

with open('/var/log/app.log') as f:
    for line in f:
        if error_pattern.search(line):
            error_count += 1
        ip_match = ip_pattern.search(line)
        if ip_match:
            ip_counter[ip_match.group()] += 1

print(f"Total errors: {error_count}")
print("Top 10 IPs:")
for ip, count in ip_counter.most_common(10):
    print(f"{ip}: {count}")

三、最佳实践

1. 日志标准化：使用统一格式(如JSON)记录日志
2. 日志分级：合理使用DEBUG, INFO, WARN, ERROR等级别
3. 日志轮转：配置合理的日志轮转策略防止磁盘爆满
4. 敏感信息过滤：避免记录密码等敏感信息
5. 集中化管理：将多台服务器的日志集中存储分析
6. 告警机制：对关键错误设置实时告警

四、性能优化

1. 日志缓冲：使用缓冲写入减少I/O压力
2. 异步处理：日志聚合工具采用异步传输
3. 采样策略：高流量时考虑采样而非全量收集
4. 索引优化：合理设置Elasticsearch索引策略
5. 冷热分离：将老旧日志移至冷存储

五、云原生方案

1. Kubernetes日志收集：

   • 使用Fluent Bit作为DaemonSet收集节点日志
   • 应用容器输出到stdout/stderr
   • 通过Sidecar容器收集特定路径日志

2. Serverless日志：

   • AWS Lambda + CloudWatch Logs
   • Google Cloud Functions + Stackdriver
   • Azure Functions + Application Insights

通过合理选择和组合上述工具与方法，可以构建高效可靠的Linux日志聚合与统计分析系统。