如何写一个脚本自动收集日志

wen 实用脚本 2026-07-26 179

我来帮你创建一个自动收集日志的脚本方案，根据不同的场景,我提供以下几种常见实现：

如何写一个脚本自动收集日志

基础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
# 添加定时任务

这个脚本方案具有以下特点：

支持本地和远程日志收集
可配置日志文件列表
自动压缩打包
日志过滤功能
生成收集报告
支持定时执行
错误处理和日志记录

你可以根据具体需求调整配置和功能。

本文地址： https://www.bmjup.com/post/31056.html