📚 课程目录第4章 Logstash入门

4. Logstash入门：安装与基础配置

Logstash是ELK全家桶里的“管道工”。它负责把日志从源头拽过来，经过清洗、整理，再送给Elasticsearch。说白了，你的日志能不能进入搜索库，就看Logstash的第一步踩得稳不稳。

我个人习惯把Logstash想象成一个工厂流水线：输入 → 过滤 → 输出。这个思想贯穿整个配置过程，你记在心里就好。

4.1 安装Logstash

安装这事，其实很简单。我推荐直接去官网下载对应操作系统的包。记得版本要和Elasticsearch对齐，否则后面会遇到一些奇奇怪怪的兼容问题。

版本对齐很重要：我用7.17系列举例，这是目前最稳定的长期支持版本。千万别Elasticsearch是7.x，Logstash装8.x。我之前摔过这个跟头，日志能送过去，但索引模板映射全乱套。

在Ubuntu上，你可以这样操作：

# 导入GPG密钥
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

# 添加仓库
echo "deb https://artifacts.elastic.co/packages/7.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-7.x.list

# 安装
sudo apt-get update && sudo apt-get install logstash

# 验证安装
/usr/share/logstash/bin/logstash --version

安装完后，二进制文件在 /usr/share/logstash/bin/logstash，配置文件在 /etc/logstash/conf.d/。这两个路径你最好记住，后面天天跟它们打交道。

4.2 第一个配置：读取文件

安装好了，总得让它干活。咱们从一个最简单的场景开始：读取一个日志文件。这是审计日志巡检的起点，那些系统日志、应用日志都躺在磁盘上等你呢。

你想想看，Logstash怎么读文件？它用 file 输入插件。这个插件特别聪明，它会记住上次读到了哪一行——靠的是一个叫做 sincedb 的文件。嗯，这里要注意：如果你删了 sincedb，它会从头重新读一遍。

sincedb的位置：默认在 ~/.sincedb_xxxxx。我在项目中遇到过：重启Logstash后，因为路径权限问题，它没写进去，结果重复发送了全部历史日志。这坑踩过一次，就永远忘不了。

创建一个配置文件 /etc/logstash/conf.d/file-input.conf：

input {
  file {
    path => "/var/log/syslog"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/sincedb"
    mode => "tail"
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

这段配置干了三件事：监听 /var/log/syslog，从头开始读，把结果打印到控制台。运行试一下：

sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file-input.conf

终端会刷刷刷打印出解析后的日志结构。我建议你先跑通这个，再去做后面的。小步快跑，永远是最稳的策略。

4.3 解析日志：用 grok 抽字段

读进来的是单行字符串，这可不行。审计日志需要把时间戳、IP、用户、行为这些字段单独提取出来。这就轮到 grok 出场了。

grok 说白了就是正则匹配的“高级封装”。它内置了很多模式，比如 %{IP:client_ip}、%{TIMESTAMP_ISO8601:timestamp}。你定义好模版，它就把字段给你抽出来。

举个例子，假设你这么一行日志：

2025-04-07 10:15:30, ERROR [auth-001] 192.168.1.10 - 登录失败: 用户 admin

用 grok 匹配：

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}, %{LOGLEVEL:level} \[%{DATA:session_id}\] %{IP:client_ip} - %{GREEDYDATA:action}" }
  }
}

匹配成功后，你会发现几个新字段：timestamp、level、session_id、client_ip、action。这些字段可以直接往Elasticsearch里送，后面做搜索和聚合就方便多了。

我的经验：一开始别追求一条grok匹配所有格式。我建议：先配一条最宽泛的，等日志进来看看实际内容，再慢慢调整。用 kibana 的 Grok Debugger 可以省不少时间。

4.4 输出到Elasticsearch

字段抽好了，该存了。输出到Elasticsearch的配置也简单：

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "syslog-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "your_password"
    ssl => false
  }
}

这段配置里的 index 参数，建议按天分索引。一来方便后续按日期清理，二来搜索性能也好拆。

完整配置文件长这样：

input {
  file {
    path => "/var/log/syslog"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/sincedb"
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}, %{LOGLEVEL:level} \[%{DATA:session_id}\] %{IP:client_ip} - %{GREEDYDATA:action}" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "syslog-%{+YYYY.MM.dd}"
  }
  stdout { codec => rubydebug }
}

保存为 /etc/logstash/conf.d/syslog-to-es.conf，重启 Logstash：

sudo systemctl restart logstash

启动后，去Kibana的Index Management看看，应该已经出现 syslog-2025.04.07 这类索引了。点进去，字段都在里面。

4.5 核心流程速览

为了让你一眼看清Logstash做了什么，我画了一张图：

4.6 常见问题与调试技巧

聊到调试，Logstash有个好用的开关：--config.test_and_exit。你写配置文件时，不确定格式对不对？直接运行：

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog-to-es.conf --config.test_and_exit

如果配置文件没问题，它输出"Configuration OK"然后退出。我用这个命令不下百次，省了很多重启服务的功夫。

还有两个常用调试技巧：

stdout输出：调试阶段，在output里加一个stdout { codec => rubydebug }，所有数据打印到终端，结构一目了然。
日志级别调整：通过 --debug 参数可以看到Logstash内部每次处理日志的详细信息。我曾经用它定位一个grok匹配超时的问题——本质是某条日志太长，默认超时时间不够。

核心原则：每次只改一个配置，看效果，再改下一个。别想着一步配完整套流水线。我见过太多同事一次配了10个filter，结果出问题根本不知道是哪个环节挂了。

4.7 本章配置参数速查表

组件	配置参数	说明	推荐值/示例
input(file)	`path`	要读取的文件路径	`/var/log/audit/*.log`
input(file)	`start_position`	首次读取的起始位置	`beginning` 或 `end`
input(file)	`sincedb_path`	偏移记录文件的路径	`/var/lib/logstash/sincedb`
filter(grok)	`match`	定义匹配规则	`"%{IP:client_ip}"`
filter(grok)	`break_on_match`	匹配成功是否停止	`true`（默认）
output(elasticsearch)	`hosts`	ES集群地址	`["http://localhost:9200"]`
output(elasticsearch)	`index`	目标索引名称（支持变量）	`syslog-%{+YYYY.MM.dd}`
output(elasticsearch)	`user/password`	ES认证凭据	`elastic` / `your_password`

你拿着这个表，对照着配置文件检查，基本不会漏参数。

4.8 打个样：一个完整的审计日志案例

我拿一条审计日志举个例子。假设日志长这样：

Apr  7 10:25:00 server01 sshd[12345]: Failed password for admin from 10.0.0.1 port 22 ssh2

配置如下：

input {
  file {
    path => "/var/log/auth.log"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/auth.sincedb"
  }
}

filter {
  grok {
    match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} sshd\[%{NUMBER:pid}\]: %{GREEDYDATA:message_detail}" }
  }
  mutate {
    add_field => { "log_type" => "ssh_audit" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "audit-ssh-%{+YYYY.MM.dd}"
  }
}

配置好后，重启Logstash。你可以用 curl 验证数据是否到达Elasticsearch：

curl -X GET "localhost:9200/audit-ssh-*/_search?pretty"

看到返回的 hits 里有日志内容，就说明一切正常。这个案例帮你处理常见的SSH登录审计场景，后面章节我们会在这个基础上叠加上更多的过滤规则和报警逻辑。