Linux教程FG399-日志处理与分析

内容简介：本文风哥教程参考Linux官方文档、Red Hat Enterprise Linux官方文档、Ansible Automation Platform官方文档、Docker官方文档、Kubernetes官方文档和Podman官方文档等内容，详细介绍了相关技术的配置和使用方法。

风哥提示：

本文档介绍日志处理与分析的方法。

Part01-日志分析脚本

1.1 Shell日志分析

# 创建日志分析脚本
[root@linux ~]# cat > /fglinux/scripts/log_analyzer.sh << 'EOF' #!/bin/bash # log_analyzer.sh # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn LOG_FILE="${1:-/var/log/nginx/access.log}" echo "==========================================" echo "日志分析报告 - $(date)" echo "日志文件: $LOG_FILE" echo "==========================================" # 总请求数 TOTAL=$(wc -l < "$LOG_FILE") echo "总请求数: $TOTAL" # 独立IP数 UNIQUE_IP=$(awk '{print $1}' "$LOG_FILE" | sort -u | wc -l) echo "独立IP数: $UNIQUE_IP" # HTTP状态码统计 echo "" echo "=== HTTP状态码统计 ===" awk '{print $9}' "$LOG_FILE" | sort | uniq -c | sort -rn | head -10 # 访问量Top10的IP echo "" echo "=== 访问量Top10的IP ===" awk '{print $1}' "$LOG_FILE" | sort | uniq -c | sort -rn | head -10 # 访问量Top10的URL echo "" echo "=== 访问量Top10的URL ===" awk '{print $7}' "$LOG_FILE" | sort | uniq -c | sort -rn | head -10 # 每小时访问量 echo "" echo "=== 每小时访问量 ===" awk '{print substr($4, 14, 2)}' "$LOG_FILE" | sort | uniq -c # 平均响应大小 echo "" echo "=== 平均响应大小 ===" awk '{sum+=$10; count++} END {print "平均响应: " int(sum/count) " bytes"}' "$LOG_FILE" # 错误请求统计 echo "" echo "=== 错误请求统计 ===" grep -E '" (4[0-9]{2}|5[0-9]{2}) ' "$LOG_FILE" | awk '{print $9}' | sort | uniq -c | sort -rn # 慢请求统计(响应时间>1秒)
echo “”
echo “=== 慢请求统计 ===”
if grep -q ‘request_time’ “$LOG_FILE”; then
awk -F'”‘ ‘{print $6}’ “$LOG_FILE” | awk ‘{if($1>1) print $1}’ | wc -l
echo “个请求响应时间超过1秒”
fi

echo “”
echo “==========================================”
echo “分析完成”
EOF

[root@linux ~]# chmod +x /fglinux/scripts/log_analyzer.sh

# 执行日志分析
[root@linux ~]# /fglinux/scripts/log_analyzer.sh /var/log/nginx/access.log
==========================================
日志分析报告 – Fri Apr 4 01:30:00 CST 2026
日志文件: /var/log/nginx/access.log
==========================================
总请求数: 100000
独立IP数: 1234

=== HTTP状态码统计 ===
85000 200
8000 304
3000 404
2000 500
1000 301

=== 访问量Top10的IP ===
5000 192.更多视频教程www.fgedu.net.cn168.from PG视频:www.itpux.com1.100
3000 192.168.1.101
2500 192.168.1.102
2000 192.168.1.103
1500 192.168.1.104
1200 192.168.1.105
1000 192.168.1.106
800 192.168.1.107
600 192.168.1.108
500 192.168.1.109

=== 访问量Top10的URL ===
10000 /index.html
8000 /a更多学习教程公众号风哥教程itpux_compi/v1/users
6000 /api/v1/products
5000 /static/css/main.css
4000 /static/js/main.js
3000 /api/v1/orders
2000 /favicon.ico
1500 /api/v1/login
1000 /api/v1/logout
800 /robots.txt

=== 每小时访问量 ===
5000 00
3000 01
2000 02
1500 03
1000 04
…

=== 平均响应大小 ===
平均响应: 2500 bytes

=== 错误请求统计 ===
3000 404
2000 500
500 403
200 400

=== 慢请求统计 ===
150
个请求响应时间超过1秒

==========================================
分析完成

Part02-Python日志分析

2.1 Python日志处理脚本

# 创建Python日志分析脚本
[root@linux ~]# cat > /fglinux/scripts/py_log_analyzer.py << 'EOF' #!/usr/bin/env python3 # -*- coding: utf-8 -*- # py_log_analyzer.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import re import sys from collections import Counter, defaultdict from datetime import datetime import json class LogAnalyzer: def __init__(self, log_file): self.log_file = log_file self.pattern = re.compile( r'(?P\S+) \S+ \S+ \[(?P[^\]]+)\] ‘
r'”(?P\S+) (?P\S+) \S+” ‘
r'(?P\d+) (?P\d+) ‘
r'”(?P[^”]*)” “(?P[^”]*)”‘
)
self.logs = []

def parse_log(self):
“””解析日志文件”””
with open(self.log_file, ‘r’) as f:
for line in f:
match = self.pattern.match(line)
if match:
self.logs.append(match.groupdict())

def get_statistics(self):
“””获取统计数据”””
stats = {
‘total_requests’: len(self.logs),
‘unique_ips’: len(set(log[‘ip’] for log in self.logs)),
‘status_codes’: Counter(log[‘status’] for log in self.logs),
‘top_ips’: Counter(log[‘ip’] for log in self.logs).most_common(10),
‘top_urls’: Counter(log[‘url’] for log in self.logs).most_common(10),
‘methods’: Counter(log[‘method’] for log in self.logs),
‘hourly_dist’: Counter(
datetime.strptime(log[‘time’].split()[0], ‘%d/%b/%Y:%H:%M:%S’).hour
for log in self.logs
),
‘error_requests’: [log for log in self.logs if log[‘status’].startswith((‘4’, ‘5’))]
}
return stats

def detect_anomalies(self):
“””检测异常”””
anomalies = []

# 检测高频访问IP
ip_counter = Counter(log[‘ip’] for log in self.logs)
for ip, count in ip_counter.items():
if count > 1000:
anomalies.append({
‘type’: ‘high_frequency’,
‘ip’: ip,
‘count’: count,
‘message’: f’IP {ip} 访问次数异常: {count}’
})

# 检测404攻击
url_404 = Counter(
log[‘url’] for log in self.logs
if log[‘status’] == ‘404’
)
for url, count in url_404.items():
if count > 100:
anomalies.append({
‘type’: ‘scan_attack’,
‘url’: url,
‘count’: count,
‘message’: f’URL {url} 404次数异常: {count}’
})

return anomalies

def generate_report(self):
“””生成报告”””
self.parse_log()
stats = self.get_statistics()
anomalies = self.detect_anomalies()

report = {
‘timestamp’: datetime.now().isoformat(),
‘log_file’: self.log_file,
‘statistics’: {
‘total_requests’: stats[‘total_requests’],
‘unique_ips’: stats[‘unique_ips’],
‘status_codes’: dict(stats[‘status_codes’]),
‘top_ips’: stats[‘top_ips’],
‘top_urls’: stats[‘top_urls’],
‘methods’: dict(stats[‘methods’]),
‘hourly_distribution’: dict(sorted(stats[‘hourly_dist’].items())),
‘error_count’: len(stats[‘error_requests’])
},
‘anomalies’: anomalies
}

return report

if __name__ == ‘__main__’:
log_file = sys.argv[1] if len(sys.argv) > 1 else ‘/var/log/nginx/access.log’

analyzer = LogAnalyzer(log_file)
report = analyzer.generate_report()

print(json.dumps(report, indent=2, ensure_ascii=False))
EOF

# 执行Python日志分析
[root@linux ~]# python3 /fglinux/scripts/py_log_analyzer.py /var/log/nginx/access.log
{
“timestamp”: “2026-04-04T01:35:00.123456”,
“log_file”: “/var/log/nginx/access.log”,
“statistics”: {
“total_requests”: 100000,
“unique_ips”: 1234,
“status_codes”: {
“200”: 85000,
“304”: 8000,
“404”: 3000,
“500”: 2000,
“301”: 1000
},
“top_ips”: [
[“192.168.1.100”, 5000],
[“192.168.1.101”, 3000],
[“192.168.1.102”, 2500]
],
“top_urls”: [
[“/index.html”, 10000],
[“/api/v1/users”, 8000],
[“/api/v1/products”, 6000]
],
“methods”: {
“GET”: 80000,
“POST”: 15000,
“PUT”: 3000,
“DELETE”: 2000
},
“hourly_distribution”: {
“0”: 5000,
“1”: 3000,
“2”: 2000
},
“error_count”: 5000
},
“anomalies”: [
{
“type”: “high_frequency”,
“ip”: “192.168.1.100”,
“count”: 5000,
“message”: “IP 192.168.1.100 访问次数异常: 5000”
}
]
}

风哥针对日志分析建议：

定期分析日志发现异常
配置日志轮转避免过大
使用正则表达式高效解析
建立异常检测规则
生成可视化报告

本文由风哥教程整理发布,仅用于学习测试使用,转载注明出处:http://www.fgedu.net.cn/10327.html