腾讯云服务器换ip轻量级中央日志（Pi rsyslog SQLite）实战

独立服务器润信云

一、前言

想在小型办公室或家庭局域网里把所有 Linux 机器的日志收集到一台低功耗设备（如 Raspberry Pi）上做集中保存与查询？本篇给出最简单、可落地的方案：用 rsyslog 接收远程日志，把每台主机的日志写到 /var/log/remote/，再用一个简单的 Python 脚本把日志解析入 SQLite，方便后续按关键词/时间快速检索与备份。整个链路依次覆盖：硬件（Pi）→ 网络（syslog/UDP）→ 数据库（SQLite）→ 编程（Python 自动化）。

二、适用场景

小型运维环境、实验室、家庭服务器日志集中保存；需要长期留存并能快速按关键字段检索的轻量场景；想把日志保存在本地，不依赖外部云服务。

三、准备（非常简单）

一台 Raspberry Pi / 任意 Debian/Ubuntu 服务器（作为日志接收端）客户端若干台 Linux（会向 Pi 发送 syslog）在接收端安装：rsyslog、python3（通常系统已有）在接收端准备一个目录 /var/log/remote/ 存放每台主机的日志文件

四、部署步骤（直接复制粘贴可用）

1）在接收服务器（Pi）上：安装 rsyslog（若未安装）

sudo apt update

sudo apt install -y rsyslog python3

2）启用 rsyslog 接收 UDP/TCP（新增文件

/etc/rsyslog.d/50-remote.conf

）

适用于 rsyslog v8+（Ubuntu 20.04/22.04 通用）——把下面内容写入/etc/rsyslog.d/50-remote.conf：

/etc/rsyslog.d/50-remote.conf

启用 UDP 和 TCP 接收

module(load="imudp")

input(type="imudp" port="514")

module(load="imtcp")

input(type="imtcp" port="514")

按主机名写入不同日志文件

template(name="RemoteLogs" type="string" string="/var/log/remote/%HOSTNAME%.log")

*.* ?RemoteLogs

& stop

保存后，创建目录并重启 rsyslog：

sudo mkdir -p /var/log/remote

sudo chown syslog:adm /var/log/remote

sudo systemctl restart rsyslog

sudo systemctl status rsyslog

提示：默认 514 端口需要 root 权限监听；若你不想用 514，可改成高端口（如 10514），同时客户端端口对应改为该端口。

3）在客户端（每台发日志的 Linux 机器）配置 rsyslog 转发（示例

/etc/rsyslog.d/60-forward.conf

）

把下面替换成你的 Pi IP（例如 192.168.1.10）并保存：

/etc/rsyslog.d/60-forward.conf

*.* @@192.168.1.10:514

说明：@@ 表示 TCP；若使用 UDP 用单个 @。保存后重启 rsyslog：

sudo systemctl restart rsyslog

4）快速测试（在客户端）

用 logger 发送测试消息到远程：

logger -t mytest "hello from client $(hostname)"

或者指定远程

logger -n 192.168.1.10 -P 514 -T -t mytest "remote test"

在接收端查看文件 /var/log/remote/<客户端主机名>.log 应能看到消息：

sudo tail -n 50 /var/log/remote/your-client-hostname.log

五、把日志解析并入 SQLite（简单版 Python 脚本）

在接收端创建目录与脚本：

sudo mkdir -p /opt/central_log

sudo chown $(whoami):$(whoami) /opt/central_log

保存下面脚本为/opt/central_log/log_to_sqlite.py，并赋可执行权限 chmod +x/opt/central_log/log_to_sqlite.py。

!/usr/bin/env python3

/opt/central_log/log_to_sqlite.py

最简单的实现：扫描 /var/log/remote/*.log 最新行并写入 sqlite（适合低流量场景）

import glob, os, sqlite3, time, re

LOG_DIR = "/var/log/remote"

DB = "/opt/central_log/central_logs.db"

简单的 syslog 行解析（匹配： "Mmm dd HH:MM:SS hostname proc[pid]: msg"）

SYSLOG_RE = re.compile(r^(?Pw+s+d+s+d+:d+:d+)s+(?P[w.-]+)s+(?P[w/-.[]:]+)s*(?::)?s*(?P.*)$)

def init_db():

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute(CREATE TABLE IF NOT EXISTS logs (

id INTEGER PRIMARY KEY AUTOINCREMENT,

ts TEXT,

host TEXT,

proc TEXT,

msg TEXT,

source_file TEXT

电脑主机改装云服务器

))

c.execute(CREATE INDEX IF NOT EXISTS idx_logs_host_ts ON logs(host, ts))

conn.commit()

conn.close()

def parse_line(line):

m = SYSLOG_RE.match(line)

if m:

return m.groupdict()

else:

return {"ts": "", "host": "", "proc": "", "msg": line.strip()}

def ingest():

init_db()

conn = sqlite3.connect(DB)

c = conn.cursor()

记录已经读到的偏移（file->offset），保存在临时文件

offset_file = "/opt/central_log/.offsets"

offsets = {}

if os.path.exists(offset_file):

with open(offset_file,r) as f:

for l in f:

fn,off = l.strip().split( ,1)

offsets[fn]=int(off)

for filepath in glob.glob(os.path.join(LOG_DIR,"*.log")):

try:

size = os.path.getsize(filepath)

last_off = offsets.get(filepath, 0)

if size < last_off:

日志被轮转，重置

last_off = 0

if size == last_off:

continue

腾讯云服务器地址更改

with open(filepath,r,errors=ignore) as fh:

fh.seek(last_off)

for line in fh:

parsed = parse_line(line)

c.execute(INSERT INTO logs (ts,host,proc,msg,source_file) VALUES (?,?,?,?,?),

(parsed.get(ts), parsed.get(host), parsed.get(proc), parsed.get(msg), os.path.basename(filepath)))

offsets[filepath] = fh.tell()

conn.commit()

except Exception as e:

print("failed to read", filepath, e)

conn.close()

保存 offsets

with open(offset_file,w) as f:

for k,v in offsets.items():

f.write(f"{k} {v} ")

if __name__ == "__main__":

ingest()

说明与注意

该脚本为最简单可用版本：按文件增量读取并解析常见 syslog 行，适合小规模部署和低流量。它把每条日志的原始时间戳（如 Oct 11 12:00:01）、主机名、进程名与消息保存到 SQLite。若你有高日志量或需要实时入库，可改用 rsyslog 的 omprog、ommysql 或更专业的 ELK/Fluentd/Graylog。

5）把脚本设为定时任务（每天/每分钟跑一次）

用 cron（每分钟扫描新日志）：

sudo crontab -e

添加（每分钟运行一次）

* * * * * /usr/bin/python3 /opt/central_log/log_to_sqlite.py >/dev/null 2>&1

或者创建 systemd service + timer（可选）。

六、查询示例（用 sqlite3）

查看一台主机最近 20 条日志：

sqlite3 /opt/central_log/central_logs.db "SELECT datetime(now), host, ts, proc, substr(msg,1,80) FROM logs WHERE host=your-hostname ORDER BY id DESC LIMIT 20;"

按关键字搜索（比如 sshd 错误）：

sqlite3 /opt/central_log/central_logs.db "SELECT id,host,ts,proc,msg FROM logs WHERE msg LIKE %sshd% ORDER BY id DESC LIMIT 50;"

七、常见问题与安全提示

UDP 丢包/顺序问题：默认用 UDP（轻量），会有丢包；若需要可靠传输，推荐使用 TCP（@@）或 TLS（较复杂）。日志轮转：请为 /var/log/remote/*.log 配置 logrotate（系统默认可能不会），避免磁盘满。端口安全：不要直接把 514 暴露到公网。可在路由器上限制只有内网设备或 VPN 可访问。性能：本方案适合轻量场景（几十台主机、低到中等日志量）。高并发场景请使用专业日志平台（ELK、Loki、Graylog 等）。

八、进阶（可选，非必须）

用 rsyslog 的 omprog 直接把每条日志推给你自己的脚本（更实时，但实现稍复杂）；把 SQLite 导入到 Grafana/InfluxDB 做可视化；为脚本增加 Web 小接口（Flask）做搜索展示；用 rsyslog TLS + 客户端证书提升安全。

九、总结

本文给出了一套极简、实用、可复制的中央日志收集方案：

在 Pi 上用 rsyslog 接收远程日志并按主机写文件；用一个非常简单的 Python 脚本把文件逐行解析并入 SQLite，支持按主机/关键字检索；整个流程无复杂依赖，几条命令即可落地，适合想把日志本地化保存并轻量检索的场景。

江苏云服务器厂家直供