我来帮你创建一个自动收集日志的脚本方案，根据不同的场景,我提供以下几种常见实现：
基础Shell脚本（Linux/Unix系统）
#!/bin/bash
# 日志收集脚本 log_collector.sh
# 配置参数
LOG_DIR="/var/log" # 日志目录
OUTPUT_DIR="/tmp/collected_logs" # 输出目录
DATE=$(date +%Y%m%d_%H%M%S)
HOSTNAME=$(hostname)
# 需要收集的日志文件列表
LOG_FILES=(
"/var/log/syslog"
"/var/log/auth.log"
"/var/log/nginx/access.log"
"/var/log/mysql/error.log"
)
# 创建输出目录
mkdir -p ${OUTPUT_DIR}/${DATE}
# 收集日志函数
collect_logs() {
echo "开始收集日志..."
for log_file in "${LOG_FILES[@]}"; do
if [ -f "$log_file" ]; then
cp "$log_file" "${OUTPUT_DIR}/${DATE}/"
echo "已收集: $log_file"
else
echo "警告: $log_file 不存在"
fi
done
}
# 压缩打包
compress_logs() {
echo "正在压缩日志..."
cd ${OUTPUT_DIR}
tar -czf "${HOSTNAME}_logs_${DATE}.tar.gz" ${DATE}/
rm -rf ${DATE}/
echo "日志已打包: ${OUTPUT_DIR}/${HOSTNAME}_logs_${DATE}.tar.gz"
}
# 主函数
main() {
collect_logs
compress_logs
}
# 执行
main
Python脚本（跨平台，功能更强大）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志自动收集脚本
支持：本地收集、远程收集、过滤、压缩
"""
import os
import sys
import gzip
import shutil
import logging
import argparse
from datetime import datetime
from pathlib import Path
import subprocess
import json
class LogCollector:
def __init__(self, config_file=None):
self.config = self.load_config(config_file)
self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
self.output_dir = Path(self.config.get('output_dir', './collected_logs'))
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def load_config(self, config_file):
"""加载配置文件"""
default_config = {
'local_logs': [
'/var/log/syslog',
'/var/log/auth.log',
'/var/log/nginx/access.log'
],
'remote_servers': [],
'output_dir': './collected_logs',
'compress': True,
'max_size': 100 * 1024 * 1024 # 100MB
}
if config_file and Path(config_file).exists():
with open(config_file, 'r') as f:
loaded_config = json.load(f)
default_config.update(loaded_config)
return default_config
def collect_local_logs(self):
"""收集本地日志"""
self.logger.info("开始收集本地日志...")
collected_files = []
for log_path in self.config['local_logs']:
log_file = Path(log_path)
if log_file.exists():
try:
# 检查文件大小
if log_file.stat().st_size > self.config['max_size']:
self.logger.warning(f"{log_path} 超过大小限制，将截取最后部分")
self._tail_log(log_file)
else:
# 复制日志文件
dest = self.output_dir / f"{self.timestamp}_{log_file.name}"
shutil.copy2(log_file, dest)
collected_files.append(dest)
self.logger.info(f"已收集: {log_path}")
except Exception as e:
self.logger.error(f"收集 {log_path} 失败: {e}")
else:
self.logger.warning(f"日志文件不存在: {log_path}")
return collected_files
def _tail_log(self, log_file):
"""截取日志文件尾部"""
try:
# 使用tail命令获取最后部分
result = subprocess.run(
['tail', '-n', '10000', str(log_file)],
capture_output=True,
text=True
)
dest = self.output_dir / f"{self.timestamp}_{log_file.name}.tail"
with open(dest, 'w') as f:
f.write(result.stdout)
self.logger.info(f"已截取尾部日志: {dest}")
except Exception as e:
self.logger.error(f"截取日志失败: {e}")
def collect_remote_logs(self, server_config):
"""收集远程服务器日志（使用SSH）"""
self.logger.info(f"连接远程服务器: {server_config['host']}")
try:
# 创建临时目录
remote_temp = f"/tmp/log_collect_{self.timestamp}"
# SSH命令收集日志
ssh_cmd = f"""
mkdir -p {remote_temp}
for log in {' '.join(server_config['logs'])}; do
if [ -f "$log" ]; then
cp "$log" {remote_temp}/
fi
done
tar -czf /tmp/remote_logs_{self.timestamp}.tar.gz -C {remote_temp} .
rm -rf {remote_temp}
"""
# 执行远程命令
subprocess.run([
'ssh',
f"{server_config['user']}@{server_config['host']}",
ssh_cmd
], check=True)
# 下载压缩包
local_path = self.output_dir / f"remote_{server_config['host']}_{self.timestamp}.tar.gz"
subprocess.run([
'scp',
f"{server_config['user']}@{server_config['host']}:/tmp/remote_logs_{self.timestamp}.tar.gz",
str(local_path)
], check=True)
# 清理远程临时文件
subprocess.run([
'ssh',
f"{server_config['user']}@{server_config['host']}",
f"rm -f /tmp/remote_logs_{self.timestamp}.tar.gz"
])
self.logger.info(f"远程日志收集完成: {local_path}")
except Exception as e:
self.logger.error(f"远程收集失败: {e}")
def compress_output(self):
"""压缩输出目录"""
if not self.config.get('compress', True):
return
self.logger.info("正在压缩日志文件...")
try:
# 打包所有文件
archive_name = f"logs_{self.timestamp}.tar.gz"
archive_path = self.output_dir.parent / archive_name
# 使用tar命令
subprocess.run([
'tar', '-czf', str(archive_path),
'-C', str(self.output_dir.parent),
self.output_dir.name
], check=True)
# 删除原始目录
shutil.rmtree(self.output_dir)
self.logger.info(f"日志已压缩: {archive_path}")
return archive_path
except Exception as e:
self.logger.error(f"压缩失败: {e}")
def filter_logs(self, keyword):
"""过滤日志内容"""
self.logger.info(f"正在过滤包含 '{keyword}' 的日志...")
filtered_dir = self.output_dir / "filtered"
filtered_dir.mkdir(exist_ok=True)
for log_file in self.output_dir.glob("*"):
if log_file.is_file():
filtered_file = filtered_dir / log_file.name
try:
with open(log_file, 'r') as src, open(filtered_file, 'w') as dst:
for line in src:
if keyword in line:
dst.write(line)
self.logger.info(f"过滤完成: {log_file.name}")
except Exception as e:
self.logger.error(f"过滤 {log_file} 失败: {e}")
def create_summary(self):
"""生成收集报告"""
summary_file = self.output_dir / "collection_summary.txt"
with open(summary_file, 'w') as f:
f.write(f"日志收集报告\n")
f.write(f"时间: {datetime.now()}\n")
f.write(f"收集目录: {self.output_dir}\n\n")
f.write("收集的文件:\n")
for log_file in self.output_dir.glob("*"):
if log_file.is_file() and log_file.name != "collection_summary.txt":
size = log_file.stat().st_size
f.write(f" - {log_file.name} ({self._format_size(size)})\n")
self.logger.info(f"报告已生成: {summary_file}")
def _format_size(self, size):
"""格式化文件大小"""
for unit in ['B', 'KB', 'MB', 'GB']:
if size < 1024:
return f"{size:.2f} {unit}"
size /= 1024
return f"{size:.2f} TB"
def execute(self):
"""执行收集流程"""
try:
# 创建输出目录
self.output_dir.mkdir(parents=True, exist_ok=True)
# 收集本地日志
local_files = self.collect_local_logs()
# 收集远程日志
for server in self.config.get('remote_servers', []):
self.collect_remote_logs(server)
# 生成报告
self.create_summary()
# 压缩
result = self.compress_output()
self.logger.info("日志收集完成！")
return result
except Exception as e:
self.logger.error(f"收集过程出错: {e}")
return None
def main():
parser = argparse.ArgumentParser(description='日志自动收集工具')
parser.add_argument('-c', '--config', help='配置文件路径')
parser.add_argument('-k', '--keyword', help='过滤关键词')
parser.add_argument('--no-compress', action='store_true', help='不压缩')
args = parser.parse_args()
collector = LogCollector(args.config)
if args.no_compress:
collector.config['compress'] = False
result = collector.execute()
if args.keyword and result:
collector.filter_logs(args.keyword)
if result:
print(f"日志已保存至: {result}")
else:
print("日志收集失败")
sys.exit(1)
if __name__ == "__main__":
main()
配置文件示例 (config.json)
{
"local_logs": [
"/var/log/syslog",
"/var/log/auth.log",
"/var/log/nginx/access.log",
"/var/log/mysql/error.log",
"/var/log/apache2/access.log"
],
"remote_servers": [
{
"host": "192.168.1.100",
"user": "admin",
"logs": ["/var/log/syslog", "/var/log/nginx/access.log"]
}
],
"output_dir": "/data/collected_logs",
"compress": true,
"max_size": 104857600
}
定时执行（Crontab配置）
# 每天凌晨2点执行日志收集 0 2 * * * /usr/local/bin/log_collector.sh # 使用Python脚本，每周一执行 0 2 * * 1 /usr/bin/python3 /path/to/log_collector.py -c /path/to/config.json # 每小时收集一次（需要调整配置） 0 * * * * /usr/local/bin/log_collector.sh
日志轮转插件（可选）
# log_rotation.py - 日志轮转管理
import os
import time
from pathlib import Path
class LogRotation:
def __init__(self, log_dir, max_age_days=30):
self.log_dir = Path(log_dir)
self.max_age = max_age_days * 86400 # 转换为秒
def cleanup_old_logs(self):
"""清理过期的日志文件"""
current_time = time.time()
for log_file in self.log_dir.glob("*.tar.gz"):
file_age = current_time - log_file.stat().st_mtime
if file_age > self.max_age:
log_file.unlink()
print(f"已删除过期日志: {log_file}")
使用说明
-
安装依赖（Python版本）：
pip install paramiko # 如果需要SSH连接
-
赋予执行权限：
chmod +x log_collector.sh chmod +x log_collector.py
-
运行脚本：
# Shell版本 ./log_collector.sh
Python版本
python3 log_collector.py python3 log_collector.py -c config.json # 使用配置文件 python3 log_collector.py -k "ERROR" # 过滤ERROR级别日志
4. **设置定时任务**：
```bash
crontab -e
# 添加定时任务这个脚本方案具有以下特点：
- 支持本地和远程日志收集
- 可配置日志文件列表
- 自动压缩打包
- 日志过滤功能
- 生成收集报告
- 支持定时执行
- 错误处理和日志记录
你可以根据具体需求调整配置和功能。
