先说一下背景。上个月一个朋友找到我说他们公司两台服务器上面跑着六七个Java和Python服务每次线上出问题只能人肉登录每一台机器去翻日志。他们也试过网上的开源方案结果Agent一启动两台只有2G内存的机器差点直接宕掉。于是问我能不能“自己搓一个”。我说可以就是这套东西从空目录到跑通差不多花了半天。今天把整个过程完整记录下来给同样在中小规模环境里被日志问题折磨的运维和开发者做个参考。这里说的日志系统不是要跟ELK比性能而是做一个最小可用的轻量方案采集、传输、存储、查询这四件事全部自己实现零第三方依赖部署时间控制在半小时以内。目标很明确多台机器的日志统一收到一个地方按时间、按服务分好出问题时能快速检索到上下文。1. 动工前先把需求边界画清楚1.1 从痛点出发一台机器搞不定的日志到底在哪我在不少小公司里见过同一个场景日志不是没有而是“养在深闺人未识”。服务分散在几台机器上每个应用的日志都在自己的/var/log目录或者应用目录下面文件名五花八门格式更是随心所欲。有人用INFO前缀有人用一排空格做分隔还有人直接把异常堆栈整个打进去。排查问题的时候运维要把所有机器挨个登一遍用tail -f盯半天才能定位。这套流程不光慢还特别容易漏。比如一个三天前的问题当时在日志里翻不到最后发现日志早在rotate的时候被清掉了。我自己就踩过这种坑业务反馈一个接口偶尔超时翻了一天日志没结果后来才发现日志落在另一台机器上一个按周切割的旧文件里连文件名都差点没看出来。所以做日志系统的第一件事不是选型而是把场景想清楚。根据我自己的实践经验真正适合手搓的场景有几个特征机器数量在十台以内总日志量一天不超过十个GB没有专职的大数据团队服务器资源紧巴巴。如果你的规模超过这个量级老老实实上成熟方案更划算。边界画清楚之后能省掉后面大量的无用功。1.2 为什么不自找麻烦引入ELK有些朋友一看“日志系统”四个字第一反应是装一套ELK。我承认ELK是优秀的产品但在这类小场景里它更像是一把牛刀。Filebeat加Logstash加Elasticsearch加Kibana整套东西吃内存非常夸张组件之间的兼容性、升级维护都是成本。我见过一台2G内存的虚拟机光是Elasticsearch的JVM堆就被分配了1G再跑业务程序基本没法动弹。还要考虑一个现实问题日志系统本身也是一种软件也会出故障。ELK出问题时你面对的是一套更复杂的软件调试成本不会低。相比之下自己手搓的方案因为透明逻辑就那么几行代码出问题找起来非常直接。这里并不是否定成熟方案而是在强调“合适”二字。低于某个规模阈值时手搓带来的可维护性收益往往比功能全面性更重要。我实际操作下来的体会是如果你只是希望“日志能集中、能查、能保留一段时间”那么一个Python写的Agent加一个接收端比一套ELK轻两个数量级部署时间短两个数量级。后面我会给出完整代码目标是零第三方依赖装完Python就能直接跑。1.3 技术选型只做四件事不选一个大杂烩手搓一个日志系统本质上就是把这四件事落地采集、传输、存储、查询。你可以把采集端想象成一群环卫工人传输层是垃圾车服务端的存储是垃圾中转站查询是垃圾站里的分拣员。每个环节都可以独立替换这也是我推荐模块化实现的原因。采集语言选型上我在Shell、Python、Go之间犹豫过。Shell写胶水逻辑确实方便一行tail -F就能盯文件但遇到断线重连、offset维护、多线程这些就非常痛苦。Go部署干净性能好可代码量比Python多不少对于不到十万行的日志需求而言有点杀鸡用牛刀。最后选了Python理由很直接内建的socket、json、os模块已经足够完成几乎所有功能代码易读后续同事接手也快。传输协议我选了TCP后面会详细讲。存储第一版直接落成按天、按服务分目录的文本文件查询用rg或者grep。之所以不上数据库是因为文件检索在TB级以内配合索引工具足够用了而且文件天然便于备份和清理。为了避免有人说这不专业我想说明一下技术选型没有绝对的“正确”只有“这个阶段适合什么”。先把这套跑通将来量真的大了再把存储层换成SQLite或ClickHouse采集端和格式都不用动。2. 核心机制拆解日志的采集、传输与落盘2.1 统一日志格式是手搓系统的地基我见过不少项目上来就写采集器结果被一个反斜杠、一个换行折磨得欲仙欲死。根源就在于日志格式没有提前统一。先说结论单条日志使用JSON格式一行一条。例如{timestamp:2025-06-12T10:30:0008:00,level:ERROR,service:order,host:web-01,message:连接数据库超时重试3次失败}为什么非要用JSON因为JSON自带转义规则。那些“多行堆栈”“特殊字符把日志搞乱”的问题只要在生成端用json.dumps处理接收端用json.loads还原底层怎么换行都不用担心。你看到文件里的原始内容会是这样{timestamp:...,message:Traceback (most recent call last):\n File \app.py\, line 12\n run()\n}对堆栈里的换行符在JSON里自动变成了转义符号一条日志只有一行采集端就不会把一条日志拆成好几块。字段怎么定我建议最少包含timestamp、level、service、host、message这五个字段。如果业务需要可以再加trace_id、user_id、request_id这些链路字段。有一个我给团队的约定时间字段必须带时区不能只写2025-06-12 10:30:00。没有时区的时间在跨机房或者容器场景下就是灾难。2.2 采集端核心怎么做到不丢行、不重复手搓采集的时候最容易写出来的版本是“照搬tail -F加grep”简单但问题很多。比如tail -F虽然能跟随新文件但它不保存位置一旦进程重启就会从头开始读把历史日志重新发一遍。再比如网络抖动发送失败之后那一行的数据就凭空没了。真正的采集端要解决三件事定位、续读、重发。定位是指在目标文件里记住已经读到哪个字节偏移量同时记录那个文件的inode。续读指进程重启后先检查inode是否变化如果文件被logrotate切走了就放弃旧inode并重新从新文件的开头读取如果inode没变就从保存的offset继续。重发则指发送失败时不直接丢弃先把消息写入本地spool目录后面再补发。我这里给一个用Python实现的采集端核心框架重点代码都加了注释。别小看这几十行采集端最关键的位置就是维护offset和inodeimport os, json, socket, time, glob class FilePoller: def __init__(self, path, server_addr, spool_dir): self.path path self.server_addr server_addr self.spool_dir spool_dir self.offset_file path .offset self.inode_file path .inode self._load_state() def _load_state(self): # 每次启动时读回上次的位置保证进程重启后不从头开始 try: self.offset int(open(self.offset_file).read().strip()) self.inode int(open(self.inode_file).read().strip()) except Exception: self.offset 0 self.inode 0 def _save_state(self): # 每条数据处理完就保存状态尽量减少丢失窗口 open(self.offset_file, w).write(str(self.offset)) open(self.inode_file, w).write(str(self.inode)) def poll(self): stat os.stat(self.path) new_inode stat.st_ino if new_inode ! self.inode: # 文件被切走重新从新文件读 self.inode new_inode self.offset 0 with open(self.path, r, encodingutf-8) as f: f.seek(self.offset) for line in f: line line.strip() if not line: continue self._send(line) # 发送成功后记录当前offset self.offset f.tell() self._save_state()这里有个容易被忽略的点为什么要用文件本身的字节偏移量而不是行号因为行号在不同编码、不同换行符环境下不稳定而offset是文件指针的真实位置。只要文件没有被截断offset就是唯一可靠的续读依据。发送失败的时候我没在poll里直接处理而是把逻辑拆出来留给队列和spool文件真正的生产消费模型可以用queue.Queue承载。你可以自己扩展这一段把spool文件和Queue串起来这样在网络抖动时最多造成延迟不会造成单条日志永久丢失。2.3 传输层选型对比与实现要点传输层我在TCP、UDP、HTTP之间做过一个对比直接列成表格传输方式优点缺点适用场景UDP速度快服务端实现简单丢包后无法恢复日志会少能容忍丢日志的监控类日志TCP可靠不会丢网络库成熟维护长连接相对复杂连接数多时有压力绝大多数业务日志HTTP语义清晰跨防火墙方便每条日志都有大量HTTP头开销性能一般外部系统对接日志量小最终我选了TCP长连接。理由很简单日志最怕的就是“缺数据”排查问题时少一条日志可能就要花几个小时去猜。TCP本身带ACK和重传能提供“不丢包”的保证。代价是采集端多维护一下连接状态但这个复杂度在一个Agent里完全可以承受。实现上有几个细节值得注意。第一服务端的Socket监听队列backlog要设大一点默认值5在网络抖动时不够用我会显式设成128。第二读取消息时要“按行读取”不要用read()一把梭否则一粘包代码全乱。第三发送端要设send timeout防止服务端假死时Agent本地线程卡死。这些细节在代码里都有体现一个健壮的传输层不靠玄学靠的是把边界条件提前写清楚。2.4 服务端落盘该考虑什么服务端拿到日志之后第一件事是校验数据结构。我会用json.loads解析解析失败就丢弃并计数同时在本地记录一条告警。这个校验不是可选的因为你的Agent可能会被升级可能版本不一致生产环境里经常出现“发过来的是旧格式”的情况。然后才是落盘。目录结构我强烈建议按“服务/日期/级别”划分/data/minilog/store/ ├── order/ │ ├── 2025-06-12/ │ │ ├── INFO.log │ │ └── ERROR.log └── user/ └── 2025-06-12/ └── INFO.log这样做的最大好处是查询路径非常自然。想查某一天的某服务错误直接进对应目录rg即可。还能顺手用logrotate对单个日志文件做切割后续可以自动清理旧数据。落盘的写入方式也有讲究。业务日志量不大时我倾向于每次写入后不立即flush由Python的缓冲机制攒一批再落盘吞吐会好很多但对于需要实时观测的错误日志则单独对ERROR.log做一次flush。这两者折中后的效果是正常情况一批日志批量写异常情况错误日志实时落盘。3. 实操搭建从零到能用的完整步骤3.1 环境准备与目录规划操作前先明确环境一台Ubuntu 20.04虚拟机Python 3.8以上不需要安装任何第三方包。我会把整套系统放在/opt/minilog下面创建专用用户minilog避免用root直接跑服务。sudo useradd -r -s /usr/sbin/nologin minilog sudo mkdir -p /opt/minilog/{agent,server,store,spool} sudo chown -R minilog:minilog /opt/minilog注意我用的mkdir -p和花括号展开这是Linux下很常见的目录批量创建技巧。临时测试可以不用服务账号直接用当前用户挺方便但生产环境一定得用独立账号权限隔离能少很多审计问题。另外这些日志文件尽量不要和其他业务数据混在同一个分区最理想是单独挂一个数据盘防止日志写满根分区把系统搞挂。3.2 服务端接收日志并分目录落盘服务端我直接用Python标准库socketserver的ThreadingTCPServer实现。它的线程模型比多进程轻一百多个连接完全顶得住。完整代码大概一百行核心部分就是自定义handler里的handle方法import socket, json, os, datetime, socketserver class LogHandler(socketserver.BaseRequestHandler): def handle(self): # 单个连接按行读取防止粘包导致json解析失败 self.request.settimeout(10) while True: try: line self.request.recv(10240).decode(utf-8, replace) except socket.timeout: break if not line: break for item in line.split(\n): if not item.strip(): continue self.dispatch(item) def dispatch(self, item): try: data json.loads(item) service data.get(service, unknown) level data.get(level, INFO) date data.get(timestamp, )[:10] date date or datetime.datetime.now().strftime(%Y-%m-%d) log_dir f/opt/minilog/store/{service}/{date} os.makedirs(log_dir, exist_okTrue) target os.path.join(log_dir, f{level}.log) with open(target, a, encodingutf-8) as f: f.write(item \n) if level ERROR: f.flush() except json.JSONDecodeError: # 校验失败单独存脏数据 with open(/opt/minilog/store/bad.log, a, encodingutf-8) as f: f.write(item \n) def main(): server socketserver.ThreadingTCPServer((0.0.0.0, 9001), LogHandler) server.allow_reuse_address True server.request_queue_size 128 server.serve_forever() if __name__ __main__: main()这个版本足够当一个MVP使用。接收端用recv后split(\n)来处理粘包是简化做法实际生产环境最好用socket.makefile()逐行读取。为了不让篇幅变成源码课这里不再展开关键是让你理解每条日志的流动路径。3.3 采集端盯文件、传数据、带本地缓冲采集端的运行逻辑是这样的每秒钟检查一次目标文件是否有新增内容有就读取并发送。我建议把被采集的文件放在一个配置文件里这样加文件不需要改代码。以下是一个简化但能跑的Agentimport json, os, socket, time def send_line(host, port, line, spool_path): try: s socket.create_connection((host, port), timeout3) s.sendall(line.encode(utf-8) b\n) s.close() return True except Exception: # 失败先落盘等下一轮再发 with open(spool_path, a, encodingutf-8) as f: f.write(line \n) return False def poll_once(path, offset, inode, host, port, spool): stat os.stat(path) if stat.st_ino ! inode: offset 0 with open(path, r, encodingutf-8) as f: f.seek(offset) for line in f: line line.strip() if line: send_line(host, port, line, spool) offset f.tell() return offset, stat.st_ino if __name__ __main__: log_file /var/log/demo.log spool /opt/minilog/spool/demo.spool offset, inode 0, 0 while True: try: offset, inode poll_once(log_file, offset, inode, 127.0.0.1, 9001, spool) except FileNotFoundError: pass time.sleep(1)你可以先用这段代码跑通流程然后我建议生产环境再加两个东西一是把offset落盘避免Agent重启后重复发送二是把单线程改成多线程让不同文件的采集互不影响。这两个优化点直接对应了前面讲的状态恢复和并发模型。3.4 验证效果与日志检索启动顺序是关键先启动Server再启动Agent。不然Agent连不上主机所有日志会先落到spool目录。cd /opt/minilog/server python3 server.py cd /opt/minilog/agent python3 agent.py echo {timestamp:2025-06-12T12:00:0008:00,level:INFO,service:order,host:dev-01,message:hello minilog} /var/log/demo.log几秒钟后去/opt/minilog/store/order/2025-06-12/INFO.log查看那条日志应该已经躺着等你了。实测下来从写入文件到出现在store目录延迟基本低于两秒足够满足绝大多数场景。查询侧最直接的就是rg和grep。我常用这么几个组合查某个服务的所有错误用rg level:ERROR /opt/minilog/store/order/查某个时间段的日志先过滤日期目录再按小时过滤如果服务端要监听外部机器就别把bind地址写死成127.0.0.1但要注意防火墙。这里顺便提一句真正生产环境我还会在服务端前面加一层简单认证避免日志被任意主机写入。4. 上线后的真实问题与排查经验4.1 遇到的第一个坑时区与时间乱跳第一版上线当天我按时间查日志居然怎么都对不上。后来发现Agent发送的timestamp是UTC服务端又按这个时间落盘导致所有日志都比本地时间慢了8小时。这个问题的教训是日志系统里所有时间字段要么统一存UTC并在查询时转换要么统一存本地时间并在字段里标明时区绝对不能混着来。我最终选了后者timestamp固定为东八区且带08:00后缀。排查办法也不复杂直接在存下来的日志文件里抽样看几十条再和业务接口的响应时间对比。如果偏差有8小时基本就是时区问题。有些老系统还会遇到服务器时钟漂移导致的时间乱跳这种只能靠ntp同步或者至少在采集端给每台机器的时间偏差做一个统计辅助定位。4.2 多行堆栈日志把一条记录“劈成两半”这是手搓日志系统最经典的坑。应用异常时抛出的堆栈在原始日志文件里是十几行甚至是二十几行。如果Agent傻傻地按行读取堆栈会被拆成十几条单行日志每条单独发送服务端解析时必然出现json解析失败的情况存下来的文件变得支离破碎。解决办法有两个层次。第一从源头控制要求所有业务日志按JSON格式输出堆栈通过json.dumps把换行符转义掉。第二在采集端兜底如果发现这一行不是合法JSON就临时存储成一个buffer继续读下一行一直读到能拼成一个完整的JSON为止。第二种方案复杂一些需要依赖“JSON永远是一行”这个前提。实际效果其实是靠第一层次解决的所以我才反复强调统一格式的重要性。4.3 logrotate切割后采集端重复读取日志文件一大了就要切。logrotate默认会把旧日志改成类似demo.log.1的文件并在同目录下生成新的demo.log。如果Agent只记住了旧的offset却不知道inode已经变化就会在一个新文件上从旧的offset继续读。比如旧文件有5000字节新文件只有200字节Agent可能会seek到200之后直接跳过整个新文件或者从头读起造成重复。我们的绕坑方式就是前面代码里写的不停比对inode发现变化立刻把offset重置为0。这看起来简单却是整个Agent里最不能省的一块。4.4 磁盘与性能日志系统的“职业病”日志系统跑久了磁盘被写满是迟早的事。如果权限规划不好日志把根目录塞满最轻的后果是业务程序写不了文件严重的直接引发连锁故障。我这边定了几个规矩日志目录挂独立磁盘logrotate按天切割并保留30天ERROR.log和INFO.log分别设置不同的大小阈值定期用crontab跑一个磁盘水位检查超过阈值自动告警。清理磁盘空间时有一个非常实用的小技巧先找大文件再按目录统计。比如用du -sh /opt/minilog/store/列出每个服务的占用用find /opt/minilog/store -name .log -size 100M找到超大日志。等确认哪些数据可以删我再动rm整个过程能避免误删重要文件。有时候日志目录膨胀是因为某些服务把debug日志也发了过来这种问题的解法是在Agent配置里增加级别过滤而不是单纯靠存储端硬扛。4.5 问题速查表我整理了一份日志系统日常维护速查表遇到问题可以直接对照现象可能原因解决方法store目录里没有今天的日志Agent没启动或offset错误查看agent日志确认进程还在删除offset文件重新测试日志重复出现logrotate切文件后inode比对失败修复Agent的inode逻辑重置offset日志全部解析失败格式不是JSON回看2.1节统一生成端格式服务端连接失败端口没放通或服务没起来用ss -lntp查看监听端口时间全部慢8小时时区不统一修改Agent时间字段统一带时区启动时占用大量CPU日志量过大单线程循环震荡优化采集循环的sleep策略或引入多线程这张表是我在实际运维中反复填充出来的里面每一条都是踩过坑之后写下来的。新手遇到问题先对照这张表能省下大量翻论坛的时间。最后说一点我个人实操后的体会。手搓日志系统最大的收获不是代码本身而是彻底理解了日志从产生、采集、传输到存储的完整链路。以前用现成框架的时候组件之间出了问题只能靠猜现在每一层逻辑都是自己写的排查问题的思路反而清晰了很多。这套方案并不是要挑战成熟框架而是适合在中小规模环境里快速搭建、快速见效。如果你想继续扩展我建议下一步把查询端换成简单的Web页面或者给Agent加上多级缓冲都是顺着这个思路很自然就能长出来的功能。踩过一次坑你就会发现日志系统并没有那么神秘无非是把每一行日志当成消息来对待罢了。