3. 日志采集与传输：常用日志采集工具、传输协议与可靠性保障

日志采集与传输是自助服务日志检索系统的数据入口层，决定了日志数据的完整性、实时性与安全性。本节围绕三大主流采集工具（Filebeat、Logstash、Fluentd）、三种传输协议（TCP/UDP/TLS）以及可靠性保障机制展开，帮助学员建立从采集端到传输层的完整技术认知。

3.1 常用日志采集工具对比

在生产环境中，日志采集工具需要兼顾轻量级、高性能、可扩展性。以下对 Filebeat、Logstash、Fluentd 进行核心特性对比：

3.2 传输协议：TCP / UDP / TLS

日志传输协议的选择直接影响数据送达率、延迟与安全性。三种协议的核心差异如下：

3.3 日志采集与传输架构图

下图展示了一个典型的准入门禁日志采集与传输架构，涵盖从边缘采集到中心处理的完整链路：

3.4 可靠性保障机制

日志传输的可靠性是准入门禁审计合规的基础。以下从三个层面保障数据不丢、不重、不乱：

3.4.1 采集端可靠性

• 内置缓冲： Filebeat 使用内存队列 + 磁盘 spill 机制，当输出端不可用时自动落盘，恢复后重发。
• 确认机制（ACK）： 每批日志发送后等待接收端 ACK，未确认数据自动重试。
• 断点续传： Filebeat 记录日志文件读取偏移量（registry file），重启后从断点继续采集。

3.4.2 传输层可靠性

• TCP 重传： 基于 TCP 协议的内置超时重传机制，保障数据包顺序到达。
• TLS 握手保活： 配置 keepalive 参数，检测连接中断后自动重建加密通道。
• 应用层心跳： Logstash 与 Filebeat 之间支持 Lumberjack 协议心跳，检测对端存活状态。

3.4.3 服务端可靠性

• 持久化队列： Logstash 可启用持久化队列（PQ），将待处理数据写入磁盘，防止进程崩溃丢数据。
• 死信队列（DLQ）： 解析失败的日志写入死信队列，供后续人工排查，不阻塞主流程。
• 幂等写入： 目标存储（如 Elasticsearch）通过文档 ID 去重，避免重复写入导致数据膨胀。

3.5 Filebeat 采集配置示例

以下为面向门禁日志采集的 Filebeat 配置片段，包含 TLS 传输与多行日志合并：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/access_control/*.log
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2}'
    negate: true
    match: after
  fields:
    log_type: access_control
    facility: door_controller

output.logstash:
  hosts: ["logstash-center.example.com:5044"]
  ssl.enabled: true
  ssl.certificate_authorities: ["/etc/filebeat/certs/ca.crt"]
  ssl.certificate: "/etc/filebeat/certs/filebeat.crt"
  ssl.key: "/etc/filebeat/certs/filebeat.key"

queue.mem:
  events: 4096
  flush.min_events: 512
  flush.timeout: 1s

3.6 Logstash 解析与脱敏配置示例

Logstash 管道配置示例，展示如何解析门禁日志、脱敏人员姓名并注入权限标签：

input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{DATA:door_id} %{DATA:person_name} %{DATA:action}" }
  }
  mutate {
    gsub => ["person_name", "(?<=.).*(?=.)", "***"]
  }
  mutate {
    add_field => { "security_level" => "confidential" }
  }
}

output {
  elasticsearch {
    hosts => ["https://es-cluster:9200"]
    index => "access-control-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${ES_PASSWORD}"
    ssl => true
    cacert => "/etc/logstash/certs/es-ca.crt"
  }
}