《Logstash 安装与管道配置》

Logstash 是 Elastic 官方的服务端数据处理管道，负责采集、转换并发送数据到 Elasticsearch 等目标。典型链路为 input → filter → output 三段式管道（pipeline）。

1 安装

官方安装文档：https://www.elastic.co/guide/en/logstash/current/installing-logstash.html

        
      
# 下载 tar.gz（以 7.x 为例）
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.15-linux-x86_64.tar.gz
tar -xzf logstash-7.17.15-linux-x86_64.tar.gz
cd logstash-7.17.15

2 第一个管道

        
# 命令行快速验证：stdin 输入，stdout 输出
bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

输入 hello logstash，输出即结构化 JSON。

3 配置文件

管道保存在 config/ 目录下的 .conf 文件中，结构如下：

        
      
input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

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

4 常用插件

阶段	常用插件	用途
input	`beats`	接收 Filebeat 上报的数据（默认 5044 端口）
input	`file`	直接读取本地日志文件
input	`stdin` / `tcp`	标准输入 / TCP 数据
filter	`grok`	正则解析非结构化日志（内置大量日志模式）
filter	`date`	解析时间戳并替换 `@timestamp`
filter	`mutate`	增删改字段、类型转换
filter	`geoip`	按 IP 解析地理位置
output	`elasticsearch`	写入 ES，指定 `index` 模板
output	`stdout`	调试输出（建议配合 `codec => rubydebug`）

5 启动与调试

        
      
# 指定管道文件启动（-t 用于测试配置语法）
bin/logstash -f /etc/logstash/conf.d/my.conf -t
bin/logstash -f /etc/logstash/conf.d/my.conf

# 通过 apt 安装后，用 systemd 管理
sudo systemctl enable logstash
sudo systemctl start logstash

6 常见问题

Filebeat 数据没有进入 ES：确认 output.logstash.hosts 与 Logstash beats 插件的 port 一致（默认 5044），防火墙放行该端口。
时间不对：日志时间默认是 Logstash 处理时间，需用 date filter 将日志自身时间写入 @timestamp。
grok 匹配失败：grok { match => {...} } 失败时日志会带 _grokparsefailure 标签，可用 bin/logstash 的调试输出查看原始 message。

参考文档

https://www.elastic.co/guide/en/logstash/current/installing-logstash.html

阅读 — · 全站 —