Logstash
数据搜集处理引擎，支持 200 多个插件。
Logstash.png
Logstash | 架构简介
Codec（Code / Decode）：将原始数据 decode 成 Event，再将 Event encode 成目标数据。
Logstash 架构.png
Logstash | 相关概念
Pipeline
- 包含了 input-filter-output 三个阶段的处理流程；
- Logstash 支持 200 多个插件，并支持插件的生命周期管理；
- Logstash 有自己的队列；
Event
- Event 是数据在内部流转时的具体表现形式，数据在 input 阶段被转换为 Event，在 output 被转换成目标格式数据；
- Event 其实是一个 Java Object，在 filter 阶段可以对 Event 的属性进行增删改查；
Input Plugins
一个 Pipeline 可以有多个 input 插件：
- Stdin / File
- Beats / Log4J / Elasticsearch / JDBC / Kafka / Rabbitmq / Redis
- JMX / HTTP / Websocket / UDP / TCP
- Google Cloud Storage / S3
- Github / Twitter
Input Plugin | File
- 支持从文件中读取数据；
- 文件读取需要解决的问题：文件只能被读取一次，重启后，需要从上次读取的位置继续，通过 sincedb 实现；
- 读取到文件新内容，发现新文件；
- 文件发生归档操作（文档位置发生变化，日志 Rotation），不能影响当前的内容读取；
Output Plugin
将 Event 发送到特定的目的地，是 Pipeline 的最后一个阶段：
- ElasticSearch
- Email / Pageduty
- Influxdb / Kafka / MongoDB / Opensdb / Zabbix
- HTTP / TCP / Websocket
Codec Plugin
将原始数据 decode 成 Event；将 Event encode 成目标数据：
- Line / Multiline
- JSON / Avro / Cef(ArcSight Common Event Format)
- Dots / Rubydebug
Filter Plugin
Filter Plugin 可以对 Logstash Event 进行各种处理，例如解析字段、删除字段、类型转换：
- Date - 日期解析；
- Dissect - 分隔符解析；
- Grok - 正则匹配解析；
- Mutate - 处理字段、重命名、删除、替换；
- Metircs - 聚合 Metrics；
- Ruby - 利用 Ruby 代码来动态修改 Event；
Filter Plugin | Mutate
对字段做各种操作：
- Convert - 类型转换；
- Gsub - 字符串替换；
- Split / Join / Merge - 切割 / 数组合并成字符串 / 数组合并；
- Rename - 字段重命名；
- Update / Replace - 字段内容更新替换；
- Remove_field - 字段删除；
Queue
Queue.png
In Memory Queue
进程 Crash，机器宕机都会引起数据的丢失。
Persistent Queue
进程 Crash，机器宕机也不会丢失数据；数据保证会被消费，可以替代 Kafka 等消息队列缓冲区的作用。
Codec Plugins - Single Line | 举几个栗子
sudo bin/logstash -e "input{stdin{codec=>line}}output{stdout{codec=> rubydebug}}"sudo bin/logstash -e "input{stdin{codec=>json}}output{stdout{codec=> rubydebug}}"sudo bin/logstash -e "input{stdin{codec=>line}}output{stdout{codec=> dots}}"
Codec Plugin - Multiline | 举个栗子
运行 multiline-exception.conf
sudo bin/logstash -f multiline-exception.conf
multiline-exception.conf 中的内容
- 以字母开头的输入会被匹配到；
- 匹配到的内容属于上一个 Event，需要再输入一个以字母开头的文本，Logstash 才会有输出；
input {
stdin {
codec => multiline {
pattern => "^\s"
what => "previous"
}
}
}
filter {}
output {
stdout { codec => rubydebug }
}
输入一段 Java 的堆栈异常作为多行数据
Exception in thread "main" java.lang.NullPointerException
at com.example.myproject.Book.getTitle(Book.java:16)
at com.example.myproject.Author.getBookTitles(Author.java:25)
at com.example.myproject.Bootstrap.main(Bootstrap.java:14)
再输入以字母开头的文本
hello word
输出上一段匹配到的 Java 的堆栈信息；
拿个 logstash.conf 出来分析一下
- 没有用到 sincedb，所以
sincedb_path => "/dev/null"；
input {
file {
path => "/home/lixinlei/data/movielens/ml-latest-small/movies.csv"
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
- movies.csv 大概长这样：
movieId,title,genres
1,Toy Story (1995),Adventure|Animation|Children|Comedy|Fantasy
2,Jumanji (1995),Adventure|Children|Fantasy
3,Grumpier Old Men (1995),Comedy|Romance
4,Waiting to Exhale (1995),Comedy|Drama|Romance
5,Father of the Bride Part II (1995),Comedy
6,Heat (1995),Action|Crime|Thriller
7,Sabrina (1995),Comedy|Romance
...
- 每行的数据用
,分割，分割出来的内容，分别归属到列 id，content，genre 下； - 列 genre 下的数据用
|分割； - 列 content 下的数据用
(分割，分割出来的第 1 个字符串存到新的列下，新的列叫 title；分割出来的第 2 个字符串存到新的列下，新的列叫 year； - 把列 year 的数据类型转为 integer；
- 去掉列 title 中的空格；
filter {
csv {
separator => ","
columns => ["id","content","genre"]
}
mutate {
split => { "genre" => "|" }
remove_field => ["path", "host","@timestamp","message"]
}
mutate {
split => ["content", "("]
add_field => { "title" => "%{[content][0]}"}
add_field => { "year" => "%{[content][1]}"}
}
mutate {
convert => {
"year" => "integer"
}
strip => ["title"]
remove_field => ["path", "host","@timestamp","message","content"]
}
}
- 把读取到的，转换过的信息写入 ElasticSearch 中；
output {
elasticsearch {
hosts => "http://localhost:9200"
index => "movies"
document_id => "%{id}"
}
stdout {}
}
