Logs 采集：城市环卫视角下的可观测性基石

可视化演示

三步展示云原生日志采集架构：Fluent Bit节点采集与转发、Fluent Bit+Loki轻量架构、以及Fluent Bit+Vector+Loki分层聚合架构。

日志采集使命

想象一座现代城市。每栋写字楼里，员工随手写下工作便条——“12:03 支付超时”“12:07 重试成功”。这些便条被塞进各个角落，没人收就会烂在纸堆里，直到某天系统出了大事故，才发现想找那几张关键便条比大海捞针还难。软件世界里的日志（Log）就是这些便条。

在可观测性体系里，有三个互相补充的视角：Metrics（指标）是城市的体检仪表，告诉你体温、血压是否正常；Logs（日志）是市民写下的事件记录，精确到几时几分发生了什么；Traces（链路追踪）是一张流转回执，记录一笔订单从下单到支付经过的每个环节。日志是排障时的第一手资料，也是事后审计的依据——所以“日志采集”不是随手捡几张纸，而是一套被刻意设计的市政工程。

云原生环境让这项工作比传统机房困难得多：容器像深夜的临时摊位，几分钟就收摊走人，日志随 Pod 销毁而消失；来源五花八门，标准输出、日志文件、系统事件、K8s 事件各有各的格式；大促时日志量能在几秒内翻十倍。日志采集的核心目标可以概括为八个字：不丢、不乱、不慢、可查。

于是，城市需要一支专业环卫队伍——轻便的回收车负责上门收集，集中分拣中心负责整理归类，中央仓库负责低成本归档。这就是本文将展开的 [Fluent Bit](0)、[Vector](1)、[Loki](2) 三者协奏的故事。

采集流水线

环卫系统的第一件事是上门收垃圾。回收车开到每个小区楼下，把垃圾桶里的袋子装上车——这就是采集（Input）。可垃圾袋里什么都有：矿泉水瓶、剩菜、电池，总不能在车上混着运到处理厂，所以收运员会在车上做初步分拣——这就是解析（Parser）与处理（Filter）。最后，分好类的垃圾被装进不同的编织袋，送往对应的处理厂——这就是发送（Output）。

具体到管道工具，每个环节都有标准动作。采集层面向三类常见来源：文件（tail 插件持续读文件尾部）、标准输出（容器 stdout/stderr）、网络转发（其他 Agent 通过 forward 协议送来的数据）。解析层把一行原始字符串拆成结构化字段——用正则匹配时间、级别、服务名，用 JSON 解析器直接展开对象。处理层则做减法：丢弃 DEBUG 噪音、对手机号做脱敏、给每条日志打上 cluster/env 等元数据标签。

处理完成后进入发送层：攒够一批、压缩、通过 HTTP/gRPC 转发到下游存储或聚合器。需要理解的是，一条日志在管道中不会凭空消失，除非被显式丢弃。这个“数据不灭”的约束可以用一个简单的守恒式描述：

$$\text{In} = P_{\text{ok}} + P_{\text{drop}} + B$$

1. 输入总量 In：进入采集管线的原始日志总条数，相当于回收车从各小区装上的垃圾袋总数。

$$\text{In}$$

2. 成功处理数 P_ok：被正确解析并通过过滤、进入后续发送阶段的日志条数，相当于被成功分拣打包的可回收物。

$$P_{\text{ok}}$$

3. 丢弃数 P_drop：因格式错误、过滤规则等原因被显式丢弃的日志条数，相当于被单独挑出的有害垃圾。

$$P_{\text{drop}}$$

4. 缓冲量 B：仍在缓冲队列中等待处理的日志条数，相当于滞留中转站的垃圾。

$$B$$

每一条日志都像一袋垃圾，从小区到处理厂必须经过收集、分拣、打包、清运四个站点。少任何一道，下游仓库都会出现“什么都收了、但什么都找不到”的混乱。

Fluent Bit 轻量采集

城市最基础的环卫单元是轻便的小型回收车。它车身小、油耗低，可以随意穿进每个小区门口，成本低到每个社区都能配一辆。[Fluent Bit](0) 就是日志采集领域的这辆小三轮：一个用 C 语言实现的轻量级日志采集器，也是 CNCF 毕业项目，内存占用通常只有几 MB 级别，在资源紧张的容器环境里非常友好。

小不代表能力弱。Fluent Bit 采用插件化架构，拥有覆盖输入、解析、过滤、输出的庞大插件生态：tail 读文件、stdin 读标准输出、systemd 读系统日志；正则、JSON、logfmt 等解析器；grep、modify、throttle 等过滤器；以及 HTTP、TLS 转发到 Loki、Kafka、S3 等数十种输出目标。在 Kubernetes 中，它常以 DaemonSet 方式在每个节点上运行一个实例，专门收集该节点所有 Pod 的日志。

轻量并非拿可靠性换来的。当下游不可达时，Fluent Bit 会按指数退避策略等待重连，而不是疯狂重试打爆下游；内置的背压保护会在缓冲接近上限时限制读取速度，避免把自己内存撑爆。对一个“贴在每个节点上跑”的组件来说，这种保守姿态极其重要。

小三轮虽小，但它是城市环卫的地基。正因为便宜可靠，它才能铺到城市的毛细血管里，把最细碎的声音先收起来。

Vector 高效管道

当城市变大，垃圾从各个社区涌进来，光靠小三轮已经不够。我们需要一座自动分拣处理厂：传送带高速运转，机械臂自动识别有害垃圾、压平纸箱，再按目的地把垃圾分到不同出口。[Vector](0) 就是日志采集链路中的这座处理厂——一个用 Rust 编写的高性能可观测性管道。

Rust 语言让 Vector 拥有两个突出优点：一是高吞吐，单实例每秒可处理数十万条日志事件，远高于多数通用脚本服务；二是内存效率高，没有 GC 停顿，在长时运行下内存曲线平稳。它的核心抽象是“管道拓扑”：sources（源）→ transforms（变换）→ sinks（汇）。你可以把多个源的日志汇聚进来，经过解析、过滤、采样、脱敏、路由等多个变换，再输出到多个目的地——一份日志同时送审计、送告警、送归档，互不干扰。

Vector 与 Fluent Bit 并不互斥，而是定位互补。Fluent Bit 更轻，适合贴地采集；Vector 更重，适合集中处理。常见配合是：节点上 Fluent Bit 负责收集，集群中心的 Vector 负责解析、过滤、打标签、路由，最后再交给 Loki 或 Kafka。在大规模多集群场景下，Vector 甚至可以作为跨区域的转发枢纽，承担流量削峰和格式统一的工作。

处理厂的出现，让环卫体系从“每个小区自己处理”升级为“分散收运、集中处理”。这正是分布式日志采集走向规模化之后必然的一步。

Loki 聚合存储

垃圾处理厂完工后，最后的问题来了：仓库怎么建？一种做法是把每张便签全文拍照、建目录，方便以后按关键词翻查——这是传统 ELK 的全文索引思路，成本高。还有一种做法：只在每个收纳箱上贴一张标签（来自哪个小区、哪栋楼、什么类型），箱子内部直接压紧打包，不逐字阅读——这就是 [Loki](0) 的思路。

Loki 是 Grafana 家族中的日志聚合系统，它只在标签上建立索引，日志内容被压缩成 chunk 块存放在对象存储中。查询时先用标签缩小范围（例如 app=order、env=prod、level=error），再扫描对应的小范围内容。这套设计让存储成本大幅下降，一台普通服务器就能支撑海量日志的长期留存。配合 Grafana 与 LogQL 查询语言，可以做到“指标、日志、链路”三合一排障。

当然，它有明确的取舍：Loki 不适合对日志正文做深度全文模糊搜索，Elasticsearch 在复杂文本检索上更灵活。但在 SRE 排障场景中，我们通常已经知道时间和服务范围，标签索引反而更高效。用公式来对比会更直观：

$$\text{Cost}_{\text{Loki}} \approx \sum_{k=1}^{K} \text{card}(l_k) + O(N), \qquad \text{Cost}_{\text{ELK}} = O(N \times L)$$

1. 日志总条数 N：时间窗口内入库的日志条数，对应仓库收到的便签总数。

$$N$$

2. 平均文本长度 L：每条日志正文的平均字符数，对应每张便签的平均字数。

$$L$$

3. 标签基数 card(l_k)：第 k 个标签的去重取值数量，例如 environment 标签只有 prod、staging、dev 三个取值，基数就是 3。

$$\text{card}(l_k)$$

4. 成本对比：Loki 的开销由“标签基数之和加内容压缩存储”构成，与正文长度解耦；ELK 的开销随 N×L 增长，等于给每张便签都建一份全文目录。

$$\sum_{k=1}^{K}\text{card}(l_k) + O(N) \ll O(N \times L)$$

Loki 就像一座只按“来源”归档的中央仓库，不收全文目录，却能用最低的成本承接整座城市的档案。在需要长期留存和快速定位的场景下，“标签先过滤、内容后扫描”的设计才是主流。

架构方案选型

环卫系统没有标准答案：小镇可以只配一辆回收车直送处理站，大城市必须分区收运加集中分拣，特大城市还要加跨区转运中心。日志采集架构也是按规模与需求演化的。

最简单的小型场景，一条链路就足够：节点上的 Fluent Bit 直接采集日志并推送到 Loki，Grafana 负责查询展示。部件少、链路短、故障点少，适合几十个服务以内的团队。当规模上升，链路会演化为 Fluent Bit → Vector → Loki：Fluent Bit 在每个节点收集，Vector 在中心负责解析、过滤、添加元数据、多路输出（例如日志同时送 Loki 排障、送 Kafka 离线分析），Loki 作为存储与查询入口。这个组合兼顾了轻量采集与集中处理，是目前生产环境中最流行的形态之一。

更大的场景，比如多集群、多云架构，则会在每个集群内部署一套 Fluent Bit 加轻量 Vector，再通过中心 Vector 把数据汇聚到统一的 Loki 集群。选型时的核心权衡包括：日志规模决定是否需要 Vector 做缓冲与聚合；查询方式决定用 Loki 还是 ES；合规要求决定是否需要在管道中插入脱敏与过滤；还有运维成本——多一个组件就多一份维护开销。

选型不是在工具之间做二选一，而是像规划城市交通一样，先看清城市的人口密度和出行结构，再决定铺地铁还是铺公交。轻量起步、按需加码，是日志采集架构演进的常态。

可靠性性能

垃圾运输路上总会有意外：某小区突然爆仓、处理厂短暂停机、道路堵车。一辆只顾踩油门的卡车会把垃圾顶进下水道；成熟的环卫系统会在小区门口排队、临时租用空地堆货、失败后回调度中心重新排队。日志采集的可靠性，就是这套“排队与排队重来”的机制。

技术上有三个关键机制。一是背压（backpressure）：采集器与下游构成生产者-消费者模型，当下游处理速度跟不上时，采集器不会无限加速接收，而是通过限制读取速度或丢弃策略把压力向上游传递，避免内存被打爆。二是缓冲：内存缓冲区响应快但容量有限，Vector 等工具还支持磁盘缓冲，把待发送数据落到磁盘，即使进程重启也不会丢。三是重试与退避：发送失败后按 1 秒、2 秒、4 秒的指数退避策略重试，并设置上限，防止下游恢复瞬间被重试请求淹没。

背压的数学本质非常简洁：设 Q(t) 为缓冲队列中的积压量，λ(t) 为日志到达速率，μ(t) 为发送速率，那么积压量如何变化？

$$\frac{dQ(t)}{dt} = \lambda(t) - \mu(t)$$

1. 积压量 Q(t)：t 时刻缓冲队列中待发送的日志条数，对应中转站里堆着的垃圾量。

$$Q(t)$$

2. 到达速率 λ(t)：单位时间内进入采集器的日志量，由业务系统的日志产生速度决定。

$$\lambda(t)$$

3. 发送速率 μ(t)：单位时间内被成功发送到下游的日志量，取决于网络与下游处理能力。

$$\mu(t)$$

4. 积压变化率 dQ/dt：积压的增速等于到达率减发送率；当 λ 大于 μ 时积压增长，必须启动背压控制。

$$\frac{dQ(t)}{dt} = \lambda(t) - \mu(t)$$

理解了这个公式，再看性能指标就心中有数：吞吐量（每秒处理事件数）、发送 P95 延迟、CPU/RSS 内存占用、丢包率。当观察到积压持续增长时，优先考虑扩容下游、调整批次大小，或降低上游采集速率，而不是无限加大缓冲。真正可靠的城市环卫，不是看地图上跑得多快，而是看暴雨之后城市还通不通——日志采集同理，背压、缓冲、重试是它的保险舱，让系统在洪峰来临时依然从容。

AI 应用场景

Python 演示

import jsonimport randomimport reimport timefrom collections import deque LOG_PATTERN = re.compile(    r'^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) '    r'\[(?P<level>[A-Z]+)\] '    r'(?P<service>[a-z]+): (?P<message>.+)$') LEVELS = ['INFO', 'WARN', 'ERROR', 'DEBUG']SERVICES = ['auth', 'order', 'gateway', 'pay']MESSAGES = ['request ok', 'timeout', 'db slow', 'cache miss']  def generate_log():    """模拟一条原始日志。"""    ts = time.strftime('%Y-%m-%d %H:%M:%S')    level = random.choice(LEVELS)    service = random.choice(SERVICES)    message = random.choice(MESSAGES)    return '{} [{}] {}: {}'.format(ts, level, service, message)  class MiniLogPipeline:    """迷你日志采集管道: 采集 -> 解析 -> 过滤 -> 缓冲 -> 批量发送。"""     def __init__(self, batch_size=10):        self.buffer = deque()        self.batch_size = batch_size        self.parse_count = 0        self.filter_count = 0        self.sent_count = 0        self.dropped_count = 0     def collect(self, raw_log):        """模拟采集器收到一条日志, 并推进管道。"""        parsed = self.parse(raw_log)        if parsed is None:            self.dropped_count += 1            return False        self.parse_count += 1        if not self.keep(parsed):            return False        self.filter_count += 1        self.buffer.append(parsed['message'])        if len(self.buffer) >= self.batch_size:            self.flush()        return True     def parse(self, raw_log):        """把一行原始日志解析成结构化字段。"""        m = LOG_PATTERN.match(raw_log)        return m.groupdict() if m else None     def keep(self, log):        """只保留 WARN 和 ERROR, INFO 与 DEBUG 是噪音被过滤。"""        return log['level'] in ('WARN', 'ERROR')     def flush(self):        """缓冲满时打包成一批 JSON 发送到下游(模拟)。"""        batch = list(self.buffer)        self.buffer.clear()        payload = json.dumps(batch)        self.sent_count += len(batch)        return payload  if __name__ == '__main__':    pipe = MiniLogPipeline(batch_size=5)    for _ in range(50):        pipe.collect(generate_log())    if pipe.buffer:        pipe.flush()    print('parsed={} kept={} sent={} dropped={}'.format(        pipe.parse_count, pipe.filter_count, pipe.sent_count, pipe.dropped_count))

1. LOG_PATTERN = re.compile(...)预编译解析规则，对应流水线里的“解析器”，正则把原始日志拆成时间、级别、服务名、消息四个字段。
2. self.buffer = deque()用双向队列做缓冲，对应管道里的“中转站”，攒够一批再发，避免每条日志都触发一次网络请求。
3. def parse(self, raw_log):格式解析阶段，返回结构化字段字典；解析失败返回 None，对应“格式错误被丢弃”。
4. def keep(self, log):过滤阶段，只保留 WARN 和 ERROR 等高等级日志，对应 Fluent Bit 中 filter 插件做的噪音裁剪。
5. def flush(self):批量发送模拟，把缓冲中的数据打包成 JSON 一次“发走”，对应采集器的 output 层。
6. if __name__ == '__main__':可执行入口，循环模拟日志产生并送入管道。

常见误区

面试题