例子

• Pipeline

  • input-filter-output的3阶段处理流程
  • 队列管理
  • 插件生命周期管理

• Logstash Event

  • 内部流转的数据表现形式
  • 原始数据在 Input被转换为event,在 output event被转换为目标格式数据
  • 在配置文件中可以对 Event中的属性进行增删改查

#第一个logstash示例
bin/logstash -e 'input { stdin { } } output { stdout {} }'

其实原来的logstash的作用，就是为了做数据的采集，但是因为logstash的速度比较慢，所以后面使用beats来代替了Logstash，当我们使用上面的命令进行启动的时候，就可以发现了，因为logstash使用java写的，首先需要启动虚拟机

我们在控制台输入 hello，马上就能看到它的输出信息

工作原理

Logstash 是 Elasticsearch 的最佳数据管道。
Logstash 是插件式管理模式，在输入、过滤、输出以及编码过程中都可以使用插件进行定制。Logstash 社区有超过 200 种可用插件。

Logstash 有两个必要元素：input 和 output ，一个可选元素：filter。
这三个元素，分别代表 Logstash 事件处理的三个阶段：输入 > 过滤器 > 输出。
Logstash的配置有三部分，如下所示

input { #输入
    stdin { ... } #标准输入
}
filter { #过滤，对数据进行分割、截取等处理
    ...
}
output { #输出
    stdout { ... } #标准输出
}

• input 负责从数据源采集数据。
• filter 将数据修改为你指定的格式或内容。
• output 将数据传输到目的地。

在实际应用场景中，通常输入、输出、过滤器不止一个。Logstash 的这三个元素都使用插件式管理方式，用户可以根据应用需要，灵活的选用各阶段需要的插件，并组合使用。

Logstash 有多个 input，每个 input 都会有自己的 codec。
数据会先存放到 Queue 中，Logstash 会把 Queue 中的数据分发到不同的 pipeline 中。
然后每一个 pipeline 由 Batcher、filter、output 组成
Batcher 的作用是批量地从 Queue 中取数据。Batcher 可以配置为一次取一百个数据。

启动

通过命令行启动 logstash 的方式如下：

bin/logstash [options]

其中 [options] 是您可以指定用于控制 Logstash 执行的命令行标志。
在命令行上设置的任何标志都会覆盖 Logstash 设置文件（logstash.yml）中的相应设置，但设置文件本身不会更改。

注 虽然可以通过指定命令行参数的方式，来控制 logstash 的运行方式，但显然这么做很麻烦。 建议通过指定配置文件的方式，来控制 logstash 运行，启动命令如下：

bin/logstash -f logstash.conf

若想了解更多的命令行参数细节，请参考 https://www.elastic.co/guide/en/logstash/current/running-logstash-command-line.html

核心

数据输入

input配置定义了数据的来源。其主要支持下面方式

• file

监控文件内容输出

• stdin

监控控制台输入

• syslogs

从syslogs读取数据

• beats

从Elastic beats接收数据

• kafka

从kafka topic中读取数据

数据处理

filter plugin 过滤器插件,主要是对数据进行处理。其支持下面几种处理策略

• grok

解析文本并构造 。把非结构化日志数据通过正则解析成结构化和可查询化

• date

日期解析 解析字段中的日期，然后转存到@timestamp

• mutate

对字段做处理 重命名、删除、替换和修改字段。

1. covert：类型转换。类型包括：integer，float，integer_eu，float_eu，string和boolean
2. split：使用分隔符把字符串分割成数组
3. merge：合并字段 。数组和字符串 ，字符串和字符串
4. rename：对字段重命名
5. remove_field：移除字段
6. join：用分隔符连接数组，如果不是数组则不做处理
7. gsub：用正则或者字符串替换字段值。仅对字符串有效
8. update：更新字段。如果字段不存在，则不做处理
9. replace：更新字段。如果字段不存在，则创建

• geoip

根据来自Maxmind GeoLite2数据库的数据添加有关IP地址的地理位置的信息

• ruby

ruby插件可以执行任意Ruby代码

• urldecode

用于解码被编码的字段,可以解决URL中 中文乱码的问题

• kv

通过指定分隔符将字符串分割成key/value

• useragent

添加有关用户代理(如系列,操作系统,版本和设备)的信息

数据输出

output配置定义了数据输出目标

• stdout

将数据输出到屏幕上

• file

将数据写入文件

• kafka

数据发送到kafka

• elasticseach

数据存储到elasticseach中

Logstash配置支持的运算符