ELK 日志分析平台 —— Logstash

文章目录

ELK 日志分析平台 —— Logstash
- Logstash 简介
- - Logstash的工作原理
  - [注]：Logstash file插件 sincedb文件
- 实验
- - Logstash 安装
  - Logstash 插件使用
  - - 1、修改 swappiness 参数
    - 2、Logstash file 插件使用
    - 3、将数据存入 ES
    - 4、Logstash file插件 sincedb文件
    - 5、Syslog 输入插件
    - 6、多行过滤插件
    - 7、采集 ES 日志，并存储到 ES
    - 8、grok过滤插件
    - 9、apache 服务日志过滤实战
- 实验截图
- - 安装 Logstash
  - Logstash 插件使用
  - - 1、修改 swappiness 参数
    - 2、命令行使用插件
    - 3、file 插件的使用
    - 4、多行过滤
    - 5、采集ES 日志
    - 6、grok过滤插件
    - 7、apache服务日志过滤
    - 8、Logstash file插件 sincedb文件
    - 9、Syslog 输入插件

Logstash 简介

官网： https://www.elastic.co/guide/en/logstash/7.6/introduction.html
安装包下载地址： https://www.elastic.co/downloads/past-releases/logstash-7-6-2

Logstash 是具有实时流水线功能的开源数据收集引擎。Logstash可以动态统一来自不同来源的数据，并将数据标准化到您选择的目标位置。清除所有数据并使其民主化，以用于各种高级下游分析和可视化用例。

Logstash和Beats有助于收集，聚合和丰富您的数据并将其存储在Elasticsearch中。
Logstash是一个开源的服务器端数据处理管道。
logstash拥有200多个插件，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的 “存储库” 中。（大多都是 Elasticsearch。）
Logstash管道有两个必需的元素，输入和输出，以及一个可选元素过滤器。

Logstash的工作原理

官网：https://www.elastic.co/guide/en/logstash/7.6/pipeline.html

Logstash事件处理管道包括三个阶段：输入→过滤器→输出。输入会生成事件，过滤器会对其进行修改，输出会将它们发送到其他地方。输入和输出支持编解码器，使您可以在数据进入或退出管道时对其进行编码或解码，而不必使用单独的过滤器。

输入项
您可以使用输入将数据获取到Logstash。一些更常用的输入是：
- file：从文件系统上的文件读取，就像UNIX命令一样 tail -0F
- syslog：在众所周知的端口514上侦听syslog消息，并根据RFC3164格式进行解析
- redis：使用redis通道和redis列表从redis服务器读取。Redis经常在集中式Logstash安装中用作“代理”，该安装会将来自远程Logstash“托运人”的Logstash事件排队。
- beats：进程的事件发送的 beats。

筛选器
筛选器是Logstash管道中的中间处理设备。如果事件符合特定条件，则可以将过滤器与条件语句结合使用以对事件执行操作。一些有用的过滤器包括：
- grok：解析和构造任意文本。当前，Grok是Logstash中将非结构化日志数据解析为结构化和可查询内容的最佳方法。Logstash内置120种模式，很可能会找到满足您需求的模式！
- mutate：对事件字段执行常规转换。您可以重命名，删除，替换和修改事件中的字段。
- drop：完全删除事件，例如调试事件。
- clone：复制事件，可能会添加或删除字段。
- geoip：添加有关IP地址地理位置的信息（还在Kibana中显示惊人的图表！）

有关可用过滤器的更多信息，请参阅过滤器插件。

输出
输出是Logstash管道的最后阶段。一个事件可以通过多个输出，但是一旦完成所有输出处理，该事件就完成了执行。一些常用的输出包括：
- elasticsearch：将事件数据发送到Elasticsearch。如果您打算以一种高效，便捷且易于查询的格式保存数据，那么Elasticsearch是您的最佳选择。时期。是的，我们有偏见:)
- file：将事件数据写入磁盘上的文件。
- graphite：将事件数据发送到graphite，graphite是一种流行的开源工具，用于存储和绘制指标图形。http://graphite.readthedocs.io/en/latest/
- statsd：将事件数据发送到statsd，该服务“通过UDP侦听统计信息（如计数器和计时器），并将聚合发送到一个或多个可插拔后端服务”。如果您已经在使用statsd，这可能对您有用！

有关可用输出的更多信息，请参见输出插件。

编解码器
编解码器基本上是流过滤器，可以作为输入或输出的一部分进行操作。编解码器使您可以轻松地将消息的传输与序列化过程分开。流行的编解码器包括json，msgpack和plain （文本）。
- json：以JSON格式编码或解码数据。
- multiline：将多行文本事件（例如java异常和stacktrace消息）合并到单个事件中。

有关可用编解码器的更多信息，请参见编解码器插件。

[注]：Logstash file插件 sincedb文件

Logstash file插件 sincedb文件内容解释：
跟踪监视文件中的当前位置，该插件通过将其记录在一个名为sincedb的单独文件中来跟踪每个文件中的当前位置。这样就可以停止并重新启动Logstash，并使它从中断处开始取走，而不会丢失Logstash停止时添加到文件中的行。

inode编号
文件系统的主要设备号
文件系统的次要设备号
文件中的当前字节偏移量
最后一个活动时间戳（浮点数）
与此记录匹配的最后一个已知路径

实验

Logstash 安装

# 安装 openjdk，官方要求，安装 logstash 需要 Java8 或者 Java11 。 这里选用安装 11.
yum install -y java-11-openjdk.x86_64
# 安装 logstash
rpm -ivh logstash-7.6.1.rpm

Logstash 插件使用

1、修改 swappiness 参数

# swappiness参数值可设置范围在0到100之间。
# 此参数值越低,就会让Linux系统尽量少用swap分区,多用内存;
# 参数值越高就是反过来,使内核更多的去使用swap空间。
cat /proc/sys/vm/swappiness
echo 10 > /proc/sys/vm/swappiness
cat /proc/sys/vm/swappiness

2、Logstash file 插件使用


# 测试，插件使用。从标准输入读取事件，将事件打印到标准输出。
/usr/share/logstash/bin/logstash -e 'input { stdin {}} output { stdout {} }'# 以文件形式输出
# path => "/tmp/demo"   ： 输出路劲。
# codec => line { format => "custom format: %{message}"}    ： 自定义输出格式。vim /etc/logstash/conf.d/demo.conf
input {stdin {}
}
output {file {path => "/tmp/demo"codec => line { format => "custom format: %{message}"}}
}# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/demo.conf

3、将数据存入 ES

# 将日志存入 elasticsearch
# path => "/var/log/messages"   ：用作输入的文件的路径。路径必须是绝对路径，不能是相对路径。
# start_position => "beginning"   ：选择Logstash最初从开头开始读取文件。
# hosts => ["172.25.9.1:9200"]    ：设置远程存储的主机。如果给定一个数组，它将在hosts参数中指定的主机之间负载均衡请求。
# index => "syslog-%{+yyyy.MM.dd}"       : 写入事件的格式。vim /etc/logstash/conf.d/demo.conf
input {file {path => "/var/log/messages"start_position => "beginning"}
}output {stdout {}elasticsearch {hosts => ["172.25.9.1:9200"]index => "syslog-%{+yyyy.MM.dd}"}
}# web 界面访问head。
http://172.25.9.1:9100/

4、Logstash file插件 sincedb文件

# Logstash file插件 sincedb文件
cd /usr/share/logstash/data/plugins/inputs/file/
# 查看 sincedb 文件
l.
ll .sincedb_452905a167cf4509fd08acb964fdb20c
cat .sincedb_452905a167cf4509fd08acb964fdb20c
#  显示文件索引节点号（inode）。一个索引节点代表一个文件。
ls -i /var/log/messages

5、Syslog 输入插件

logstash 可以伪装成日志服务器，直接接受远程日志。

vim /etc/logstash/conf.d/demo.conf
input {#file {# path => "/var/log/messages"#  start_position => "beginning"#}syslog {port => 514}
}output {stdout {}#file {#  path => "/tmp/demo"#  codec => line { format => "custom format: %{message}"}#}elasticsearch {hosts => ["172.25.9.1:9200"]index => "rsyslog-%{+yyyy.MM.dd}"}
}# 配置远程 rsyslog 服务
# server2、3：
vim /etc/rsyslog.conf
……15 $ModLoad imudp16 $UDPServerRun 514
……
90 *.* @@172.25.9.1:514
……systemctl restart rsyslog# server1：
# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/demo.conf

6、多行过滤插件

多行过滤可以把多行日志记录合并为一行事件

cd /etc/logstash/conf.d/
vim test.conf
# pattern => "^EOF"  ： 多行结束的标志或者开始的标志。
# negate => true ： 是否对pattern的结果取反，该行表示没有遇到 EOF 开头的行都是前一行
# what => previous : previous 向上匹配，next 向下匹配。
# 官网 ：https://www.elastic.co/guide/en/logstash/7.6/multiline.html
input {stdin {codec => multiline {pattern => "^EOF"negate => truewhat => previous}}
}output {stdout {}
}# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/test.conf

7、采集 ES 日志，并存储到 ES

cd /etc/logstash/conf.d/
vim es.conf
# start_position => "beginning"   : 选择Logstash 最初从哪个位置读取信息，beginning 从总是开始位置读取，end 从结尾读取。将文件作为实时流，所以这里的结尾指的是上一次读取到的结尾。
# codec => multiline：编码方式为多行编码器，多行编解码器将折叠多行消息并将其合并为一个事件。
input {file {path => "/var/log/elasticsearch/my-es.log"start_position => "beginning"codec => multiline {pattern => "^\["negate => truewhat => previous}}
}output {stdout {}elasticsearch {hosts => ["172.25.9.1:9200"]index => "eslog-%{+yyyy.MM.dd}"}
}# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf

8、grok过滤插件

cd /etc/logstash/conf.d/
vim test1.conf
# 官网：https://www.elastic.co/guide/en/logstash/7.6/plugins-filters-grok.html
#  grok ：grok模式的语法是 %{SYNTAX:SEMANTIC}，SYNTAX 是与文本匹配的模式的名称，类似日志中的数据类型。SEMANTIC 是为匹配的文本段指定的标识符，类似数据的标题。
# match  ：  一个散列，它定义了查找位置以及使用哪些模式的映射。
input {stdin {}
}
filter {grok {match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}
output {stdout {}
}# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/test1.conf

9、apache 服务日志过滤实战

# 配置 httpd 服务
yum install -y httpd
systemctl start httpd
# 默认发布页
echo server1 > /var/www/html/index.html
# Logstash 采集用户对该目录有相应的权限
ll -d /var/log/httpd/
chmod 755 /var/log/httpd/# 编写配置文件
cd /etc/logstash/conf.d/
vim apache.conf
# match => { "message" => "%{HTTPD_COMBINEDLOG}" }
input {file {path => "/var/log/httpd/access_log"start_position => "beginning"}
}filter {grok {match => { "message" => "%{HTTPD_COMBINEDLOG}" }}
}output {stdout {}elasticsearch {hosts => ["172.25.9.1:9200"]index => "apachelog-%{+yyyy.MM.dd}"}
}# 指定文件使用插件
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/apache.conf

实验截图

安装 Logstash

Logstash 插件使用

1、修改 swappiness 参数

2、命令行使用插件

3、file 插件的使用

4、多行过滤

5、采集ES 日志

6、grok过滤插件

7、apache服务日志过滤

8、Logstash file插件 sincedb文件

9、Syslog 输入插件