这是我在香港机房里干过的一次“救火”。凌晨两点，外面挂着 8 号风球，冷通道的风像刀子一样吹在手背上。客服打来电话说支付系统又间歇超时，合规审计还催着要 90 天全量日志。散落在不同业务节点上的日志已经让团队头大，我当场决定：把这摊日志彻底收拢，用 rsyslog + Elastic Stack 建一套能扛量、可回溯、好检索的集中式平台。下面是我从零到一的完整落地过程——不光能让新手照着跑起来，也足够让老手看到细节和坑位。

1. 目标与整体设计

目标

• 集中采集：系统日志（journald/syslog）、应用文本日志、Nginx 访问/错误日志、审计日志。
• 实时检索：秒级可查；查询时延 < 2s（常规 24h 范围）。
• 成本可控：冷热分层 + ILM，90 天在线，180 天冷存。
• 可用可靠：链路端到端加密，断链不丢（磁盘队列兜底），支持灰度扩容。

架构（两种模式，优先 B，A 作扩展）

模式B（推荐，轻量/低维护）：
[各业务主机 rsyslog]  --HTTPS JSON-->  [Elasticsearch ingest pipeline]  -->  [索引按ILM生命周期]
                                           |
                                           +--> [Kibana 可视化/告警]

模式A（重解析/脏日志多时用）：
[各业务主机 rsyslog]  --TCP/TLS-->  [Logstash]  --HTTPS-->  [Elasticsearch]
                                            |
                                            +--> 清洗/脱敏/分流

我最终选 模式 B：rsyslog 直接用 omelasticsearch 发 JSON 到 ES，解析放到 ingest pipeline，减少一跳、降低维护复杂度。对于特别脏、格式多变的日志，再单独上 Logstash。

2. 现场环境与硬件/网络参数

位置：香港葵涌机房（HK）
操作系统：Debian 12 (bookworm)
时区：系统统一 UTC（显示到 Kibana 再按用户时区/HKT）
网络：专线内网 10G，跨可用区冗余；公网出口独立（做审计/外部拉取快照）

2.1 集群与主机规格（实配）

吞吐估算（经验法）：

• 峰值 20k 事件/秒（EPS），日增约 800–1000GB（含访问日志）。
• 热层保留 7 天；温层（如有）30 天；冷层 90 天；归档到对象存储 180 天。
• JVM 堆按磁盘大小 1:16 粗略估算；ES 单节点堆 16–24GB 区间足矣（谨慎上 32GB 以上）。

3. 日志分类与索引策略（命名 & ILM）

索引命名（统一前缀便于跨空间管理）：

• logs-sys-YYYY.MM.DD （系统日志）
• logs-nginx-YYYY.MM.DD （Nginx 访问/错误）
• logs-app-<app>-YYYY.MM.DD （应用，按 app 分索引）
• logs-audit-YYYY.MM.DD （审计）

ILM（示例）

• hot：7 天（rollover by 50GB 或 1d）
• warm（可选）：30 天（降低副本、合并段）
• cold：其后到 90 天（冻结/搜慢但便宜）
• delete：>90 天删除，另有 snapshot 到对象存储保 180 天

4. 安全与端口

MTU 提示：香港某些专线叠了 VXLAN/EVPN，有效 MTU 1472/1460 是我踩过的坑。ES 客户端超时/握手失败时，别忘了排查 MTU/DF 位；必要时在 rsyslog 侧启用小包或让网络侧做 TCPMSS 调整。

5. 部署步骤（端到端）

5.1 准备基础（所有节点）

sudo timedatectl set-timezone UTC
sudo apt update && sudo apt install -y ca-certificates curl gnupg jq

5.2 安装 Elasticsearch & Kibana（核心摘录）

这里不写死版本，只用 8.x 语义；生产务必固定版本并在灰度后再推广。

# 导入签名 & 源（以官方 APT 仓为例）
curl -fsSL https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elastic.gpg
echo "deb [signed-by=/usr/share/keyrings/elastic.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | \
  sudo tee /etc/apt/sources.list.d/elastic-8.x.list

sudo apt update
sudo apt install -y elasticsearch kibana

Elasticsearch 关键配置 /etc/elasticsearch/elasticsearch.yml（热节点示例）

cluster.name: hk-logs
node.name: es-hot-1
node.roles: [ data_hot, ingest, master ]
path.data: /data/esdata
path.logs: /var/log/elasticsearch
network.host: 0.0.0.0
discovery.seed_hosts: ["10.10.0.11","10.10.0.12","10.10.0.13"]
cluster.initial_master_nodes: ["es-hot-1","es-hot-2","es-hot-3"]

xpack.security.enabled: true
xpack.security.http.ssl:
  enabled: true
  keystore.path: certs/http.p12
xpack.security.transport.ssl:
  enabled: true
  verification_mode: certificate
  keystore.path: certs/transport.p12
  truststore.path: certs/transport.p12

# JVM 堆（/etc/elasticsearch/jvm.options.d/heap.options）
# -Xms16g
# -Xmx16g

启动与健康检查

sudo systemctl enable --now elasticsearch
curl -k -u elastic:临时密码 https://es-hot-1:9200/_cluster/health?pretty

Kibana /etc/kibana/kibana.yml

server.host: "0.0.0.0"
elasticsearch.hosts: ["https://es-hot-1:9200","https://es-hot-2:9200","https://es-hot-3:9200"]
elasticsearch.ssl.verificationMode: certificate
server.publicBaseUrl: "https://kibana.hk.example.com"

sudo systemctl enable --now kibana

我把 Kibana 放在内网，通过公司 VPN/零信任入口访问；公网只暴露跳板/SSO，避免 5601 直接裸奔。

5.3 创建写入凭证（API Key 或最小权限用户）

示例：创建 role + user（也可用 API Key）

# 1) 创建 role ，允许对 logs-* 写入和管道执行
cat <<'JSON' > role.json
{
  "indices": [
    { "names": ["logs-*"], "privileges": ["create_index","write","create","view_index_metadata"] }
  ],
  "cluster": ["monitor"]
}
JSON

curl -k -u elastic:*** -H "Content-Type: application/json" \
  -X PUT https://es-hot-1:9200/_security/role/log_writer -d @role.json

# 2) 创建用户
curl -k -u elastic:*** -H "Content-Type: application/json" \
  -X POST https://es-hot-1:9200/_security/user/rsyslog_writer -d '{
    "password": "强口令！",
    "roles": ["log_writer"],
    "full_name": "rsyslog client"
}'

实战里我更多用 API Key：在 Kibana → Stack Management → API Keys 生成（限制索引前缀、有效期），然后分发到 rsyslog。

5.4 配置 Ingest Pipeline（解析/脱敏）

示例：Nginx 访问日志解析管道

curl -k -u elastic:*** -H 'Content-Type: application/json' \
  https://es-hot-1:9200/_ingest/pipeline/nginx-access -X PUT -d '{
  "description": "Parse nginx access log",
  "processors": [
    { "grok": {
        "field": "message",
        "patterns": ["%{IPORHOST:clientip} - %{DATA:ident} %{DATA:auth} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status:int} %{NUMBER:bytes:int} \"%{DATA:referrer}\" \"%{DATA:agent}\""]
    }},
    { "date": { "field": "timestamp", "formats": ["dd/MMM/yyyy:H:m:s Z"] }},
    { "user_agent": { "field": "agent", "target_field": "ua" }},
    { "remove": { "field": ["ident","auth"] }},
    { "set": { "field": "log_type", "value": "nginx_access" }}
  ],
  "on_failure": [{ "set": { "field": "ingest_failure", "value": true }}]
}'

系统日志管道（示例）

curl -k -u elastic:*** -H 'Content-Type: application/json' \
  https://es-hot-1:9200/_ingest/pipeline/syslog -X PUT -d '{
  "processors": [
    { "grok": { "field": "message", "patterns": ["%{SYSLOGTIMESTAMP:sys_ts} %{HOSTNAME:host} %{DATA:program}(?:\\[%{NUMBER:pid}\\])?: %{GREEDYDATA:msg}"] }},
    { "set": { "field": "log_type", "value": "syslog" }},
    { "set": { "field": "event.ingested", "value": "{{_ingest.timestamp}}" }}
  ]
}'

5.5 rsyslog（客户端）安装与配置

sudo apt install -y rsyslog rsyslog-gnutls
rsyslogd -v

启用 JSON 模板 + 发送到 Elasticsearch（omelasticsearch）

rsyslog 8.32+ 支持 omelasticsearch；Debian 12 自带模块足够用。

我把系统日志（imjournal）与文本日志（imfile）统一成 JSON，直发 ES 指定 pipeline。

/etc/rsyslog.d/00-global.conf

module(load="imuxsock")     # 本地 UNIX socket
module(load="imjournal" StateFile="imjournal.state")  # systemd 日志
module(load="imfile")       # 采集文本文件
module(load="omelasticsearch")
module(load="mmjsonparse")  # 如需从应用已是JSON的 message 里二次解析

# 发送队列（磁盘兜底，断链不丢）
queue.type="LinkedList"
queue.filename="es_queue"
queue.maxdiskspace="10g"
queue.highwatermark="500000"
queue.lowwatermark="200000"
queue.discardmark="800000"
queue.discardseverity="7"    # 仅在极端情况下丢最低等级

# 通用 JSON 模板（包含主机、程序、消息、时间等）
template(name="jsonfmt" type="list") {
  constant(value="{")
    constant(value="\"@timestamp\":\"")     property(name="timereported" dateFormat="rfc3339")
    constant(value="\",\"host\":\"")        property(name="hostname")
    constant(value="\",\"program\":\"")     property(name="programname")
    constant(value="\",\"pid\":")           property(name="procid")
    constant(value=",\"severity\":\"")      property(name="syslogseverity-text")
    constant(value="\",\"facility\":\"")    property(name="syslogfacility-text")
    constant(value="\",\"message\":")       property(name="msg" format="json")
  constant(value="}")
}

# Elasticsearch 目标（建议用 API Key）
set $!es.apiKey = "base64EncodedId:apiKey==";  # Kibana 里复制

action(
  type="omelasticsearch"
  server="es-hot-1.hk.local"
  serverport="9200"
  usehttps="on"
  searchIndex="logs-sys-%$YEAR%.$MONTH%.$DAY%"
  action.resubmitOnFailure="on"
  template="jsonfmt"
  dynSearchIndex="on"
  # 指定 ingest pipeline
  pipeline="syslog"
  # 认证
  authmode="apikey"
  apikey="$!es.apiKey"
  # TLS 验证
  tls.cacert="/etc/rsyslog.d/ca.crt"
  errorfile="/var/log/rsyslog_es_error.log"
  bulkmode="on"
  bulksize="1000"
  timeout="10"
)

采集 Nginx 文本日志（imfile） /etc/rsyslog.d/10-nginx.conf

input(type="imfile"
  File="/var/log/nginx/access.log"
  Tag="nginx-access"
  Severity="info"
  Facility="local6"
  addMetadata="on")

# 单独目标，送到 nginx pipeline & 索引
template(name="nginxIndex" type="string" string="logs-nginx-%$YEAR%.$MONTH%.$DAY%")

if ($programname == "nginx-access") then {
  action(
    type="omelasticsearch"
    server="es-hot-1.hk.local"
    serverport="9200"
    usehttps="on"
    searchIndex="logs-nginx-%$YEAR%.$MONTH%.$DAY%"
    pipeline="nginx-access"
    authmode="apikey"
    apikey="$!es.apiKey"
    tls.cacert="/etc/rsyslog.d/ca.crt"
    template="jsonfmt"
    bulkmode="on"
    bulksize="2000"
  )
  stop
}

注意：应用本身若输出 JSON（例如结构化日志），可启用 mmjsonparse 对 message 二次展开到字段，少走 grok。

证书：我把 ES 的 CA 拿到客户端验证（tls.cacert），API Key 通过 Ansible Vault 分发。若要 mTLS，也可在 rsyslog 配置 tls.myCert/tls.myPrivKey。

重启/验证

sudo systemctl restart rsyslog
tail -f /var/log/rsyslog_es_error.log
# ES 端看数据
curl -k -u elastic:*** 'https://es-hot-1:9200/logs-sys-*/_count?pretty'

6. 索引模板与 ILM 配置（示例）

ILM policy

curl -k -u elastic:*** -H 'Content-Type: application/json' \
  -X PUT https://es-hot-1:9200/_ilm/policy/logs-default -d '{
  "policy": {
    "phases": {
      "hot":  { "actions": { "rollover": { "max_age": "1d", "max_size": "50gb" } } },
      "warm": { "min_age": "7d", "actions": { "set_priority": { "priority": 50 }, "forcemerge": {"max_num_segments": 1}, "shrink": {"number_of_shards": 1} } },
      "cold": { "min_age": "30d", "actions": { "set_priority": { "priority": 0 } } },
      "delete": { "min_age": "90d", "actions": { "delete": {} } }
    }
  }
}'

模板（匹配 logs-*）

curl -k -u elastic:*** -H 'Content-Type: application/json' \
  -X PUT https://es-hot-1:9200/_index_template/logs-template -d '{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "index.lifecycle.name": "logs-default",
      "index.lifecycle.rollover_alias": "logs-roll"
    },
    "mappings": {
      "dynamic": true,
      "properties": {
        "@timestamp": { "type": "date" },
        "host": { "type": "keyword" },
        "program": { "type": "keyword" },
        "status": { "type": "integer" },
        "bytes":  { "type": "long" },
        "log_type": { "type": "keyword" }
      }
    }
  },
  "priority": 100
}'

生产中我还会把 logs-sys-*、logs-nginx-* 做不同模板，精细字段类型，减少 mapping 冲突。

7. Kibana：空间、角色、仪表盘与告警

空间：按业务线开 Space（支付、内容、平台），最小权限分配。

索引模式：logs-*、logs-nginx-* 分开建。

可视化：HTTP 状态码分布、Top URL、P95/P99 响应时间（如果埋点有时长字段）、错误趋势。

检测 & 告警：Watcher/Rules——如 5 分钟 5xx 超过基线即发 Slack/飞书；ingest_failure:true 升级处理。

8. 运维侧数据与控制台小表

8.1 采集与吞吐（某日峰值观测）

8.2 存储曲线（热/冷占比）

9. 灰度与扩容策略

客户端灰度：Ansible/SSM 批次下发 rsyslog 配置，先 10%，观测错误率与 ES ingest 压力，再扩大。

集群扩容：

吞吐升高：先加 ingest 角色（或给热节点增加 ingest），再加热数据节点。

存储紧张：增加冷节点或缩短热层天数，ILM 自动迁移。

回滚：rsyslog 可同时配置两路 action（新旧 ES 集群），必要时改权重回切。

10. 快照备份（S3 兼容对象存储，香港区）

注册仓库

curl -k -u elastic:*** -H 'Content-Type: application/json' \
  -X PUT https://es-hot-1:9200/_snapshot/hk-s3 -d '{
    "type": "s3",
    "settings": {
      "bucket": "hk-log-snapshots",
      "endpoint": "s3.hk.example.com",
      "protocol": "https",
      "region": "ap-east-1",
      "path_style_access": true
    }
}'

定期快照（管理机 crontab 或 Kibana UI 任务）

curl -k -u elastic:*** -X PUT \
  "https://es-hot-1:9200/_snapshot/hk-s3/daily-$(date +%F)?wait_for_completion=true"

有次冷节点盘打满就是靠快照+删冷层救回来的。香港的对象存储延迟低，回存也很快。

11. 真实坑与解决过程（我在机房踩过的点）

MTU/分片问题：rsyslog 批量写 ES，TLS 握手偶发失败。抓包看到 Fragmentation needed，确认专线 VXLAN 有效 MTU 1472。解决：网络侧设 TCPMSS clamp，并把 rsyslog bulksize 从 5000 降到 2000，错误即消失。

时间戳乱飞：部分主机还在 HKT，ES 用 UTC，Kibana 展示错 8 小时。解决：所有 Linux 统一 UTC；Kibana 用户在界面切换时区。

Mapping 冲突：应用字段 status 有时是字符串（"OK"），有时整数。解决：前置 ingest 做 rename，把应用里的 status 改名 app_status，保留 HTTP 的 status:int。

API Key 泄漏风险：一开始直接写在 rsyslog 配置。解决：Ansible Vault 加密变量分发，rsyslog 配置读取环境文件；另开最小权限 Key，定期轮转。

高峰丢日志担忧：晚高峰 EPS 飙升时 ES 拒写（429）。解决：开启 rsyslog 磁盘队列 10GB，action.resubmitOnFailure=on；ES 调整线程池与队列，峰值过后自动回补。

Kibana 权限：合规同事只需要审计索引。解决：Space + Role 细分到 logs-audit-*，满足最小授权。

Nginx Log 花样多：有些 upstream 缺字段导致 grok fail。解决：on_failure 打 ingest_failure:true，Kibana 建仪表盘监控失败率，超过阈值触发告警并回滚 Nginx 日志格式变更。

12. 可复制的一套最小化清单（Checklist）

 ES/Kibana 安装、TLS 打开、创建 log_writer 或 API Key

 建立 ingest pipeline：syslog、nginx-access

 下发 rsyslog：imjournal + imfile、JSON 模板、omelasticsearch、磁盘队列

 建立索引模板 & ILM，确认 rollover 正常

 Kibana 创建索引模式、基础可视化与告警

 压测：logger/wrk 模拟，观察 ingest CPU、段合并、GC

 生产灰度 10% → 50% → 100%，全程监控错误率与时延

 快照仓库绑定与每日快照任务

 文档化：接入规范、字段约定、索引命名、变更回滚方案

13. 扩展：什么时候上 Logstash？

复杂清洗：需要多表 Join、GeoIP、脱敏（手机号/卡号掩码）、复杂条件路由。

协议多样：除了 syslog，还有 Kafka、Beats、HTTP 多来源汇聚。

团队分工：把复杂逻辑固化在 Logstash pipeline，客户端 rsyslog 只负责可靠传输（RELP/TLS）。

最小 Logstash 输入（TCP/TLS）到 ES（仅示例）

input {
  tcp {
    port => 5001
    codec => json_lines
    ssl_enable => true
    ssl_cert => "/etc/logstash/certs/logstash.crt"
    ssl_key  => "/etc/logstash/certs/logstash.key"
  }
}
filter {
  # grok / mutate / geoip / fingerprint ...
}
output {
  elasticsearch {
    hosts => ["https://es-hot-1:9200","https://es-hot-2:9200"]
    index => "logs-app-%{+YYYY.MM.dd}"
    # user/password 或 api_key
  }
}

14. 收尾：那一夜之后

风在机房窗外呼呼直响，Kibana 的趋势图却稳稳地贴着预期。凌晨四点，我们把支付超时的根因钉住——某个上游在整点前后抖了一次，错误集中在两台边缘节点上，Nginx 日志里显眼得很。合规的同事第二天早上来，又要拉 30 天内所有涉及特定 IP 的访问记录，我打开可视化把条件一套，导出 CSV，邮件回发，十分钟不到。

这套 rsyslog + Elastic Stack 并不神秘。关键在于：

• 结构化、前后一致的字段；
• 从网络到队列的韧性设计；
• 从热到冷再到快照的成本曲线；
• 以及面对突发时，敢在机房里把每一个环节都“按住”的那点狠劲。

希望这篇“有汗渍”的实操笔记，能让你在下一个风雨夜里，也把日志这摊事儿收拾得干净利落。