ELK日志分析平台之Logstash数据采集

Posted 2022-11-27 S4061222

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ELK日志分析平台之Logstash数据采集相关的知识，希望对你有一定的参考价值。

logstash简介

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

数据采集三要素：输入，过滤和输出

输入：采集各种样式、大小和来源的数据

Logstash 支持各种输入选择，同时从众多常用来源捕捉事件。
能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。

过滤器：实时解析和转换数据

数据从源传输到存储库的过程中，Logstash
过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。

- 利用 Grok 从非结构化数据中派生出结构
- 从 IP 地址破译出地理坐标
- 将 PII 数据匿名化，完全排除敏感字段
- 简化整体处理，不受数据源、格式或架构的影响

输出：选择您的存储库，导出您的数据

尽管 Elasticsearch 是我们的首选输出方向，能够为我们的搜索和分析带来无限可能，但它并非唯一选择。 Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例。

一 Logstash安装与配置

实验环境：
接上篇，server7、server8、server9，是Elasticsearch集群
logstash数据采集 (重量级！！！)，占内存，，需要java
通过logstash来往里面存数据，耗费资源，不建议在集群中，新建虚拟机server10

三种解决内存占用方法：
不锁定内存1G，jvm缩小，swap打开

安jdk和logstask

可以作软链接，此处不再演示

二 Logstash的输入输出

1 命令行方式：标准输入到标准输出

标准输入hello word，标准输出hello word；标准输入jiaojaio，标准输出jiaojiao

2 conf文件方式：标准输入，输出到文件

编辑 test.conf文件

-f 指定文件

-e编辑（必备输入输出，过滤可选），调用标准输入和标准输出
%message :接收变量,接受stdin的输入内容

命令行标准输入hello world，输出到了/tmp/testfile文件中

查看 /tmp/testfile 文件

3 conf文件方式：标准输入，输出到ES和标准输出到终端

编写es.conf文件

标准输出终端
输出到ES(server8)中

终端有输出（为了回显）

ES集群内部也有输出, 默认是localhost，修改ip方式连接9200，任意节点都可，多个同时都可，索引：默认值（自动录入年月日）
（访问9100的，使用 es-head 作可视化，数据浏览！！！）

可以加多个插件，输入和输出都可多种模块

三 Logstash采集日志

以用户logstack读取文件，修改文件权限

编写es.conf文件，从日志文件输入，输出到终端和ES(server8)中

执行es.conf文件

已经可以查看到终端有日志内容输出

ES(server8)中有日志生成

ES端删除logstash索引(es.conf文件中定义的名称)，上篇设定server7不存储数据，只有server8和server9有副本

删除文件后，不会重写一份数据

每个文件都有inode编号，删除索引，删除sincedb，会重新从头读取
只会输入变更，不会数据冗余，重新读取

logger 写日志，会有数据
删除文件后，不会重写一份数据，此处只有新产生的日志
ES中查看输出

只有最新的一条日志

每台服务器上都部署logstash，过于重量级！！！
logstash可以伪装成日志服务器，直接接受远程日志：

默认514日志收集端口，server10监听514端口，server10伪装成日志服务器

编辑es.conf 文件

[root@server10 conf.d]# cat es.conf 
input 
	#file 
	#	path => "/var/log/messages"
	#	start_position => "beginning"
	#
	
	syslog 		%伪装syslog，开放端口514
		port => 514
	


output 
	stdout 

	elasticsearch 
		hosts => ["172.25.28.10:9200"]
		index => "syslog-%+yyyy.MM.dd"	%索引为syslog-年月日

执行es.conf 文件后，查看server10的端口514已开放

远程主机server8编辑/etc/rsyslog.conf文件
打开打开udp端口，514端口

所有的日志发送给172.25.28.10:514一份，连接远端主机server10

重启日志服务

server10上查看端口，出现本机和server8主机的日志端口

ES中查看有syslog标签的索引生成

server7同server8一样操作

server10端口查看

终端中已经输出，可以看到server7和server8的日志

ES中查看server7和server8的日志syslog

四过滤

1 多行过滤插件

编写测试文件：test.conf
遇到定义的EOF：向上合并，整体输出

输出到文件/tmp/testfile，格式为custom format:数据

执行test.conf文件

测试：多行输入，以EOF结束输入，可以看到标准输出是一条
/tmp/testfile文件中也是一条

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

使用es集群的server7的日志（之前有一些报错日志），把my-es.log发给server10的/var/log

先看一下正确的日志都是以时间开头的，并且被中括号[ ]括起来的一行，而错误日志有很多行，比如下图的at org开头的这些，他们合起来应该是一条错误日志。

修改test.conf文件，先不加多行输入模块，看效果

终端查看

es中查看

发现错误日志，只显示一行！！！不是连续的全部显示一整条数据

解决方法：加入多行模块

先删除生成的ES的输出索引eslog

删除sincedb文件，重新读取

修改文件test.conf文件，如何判断一个完整的日志条目，以为时间开始，以【开头的行认为一个完整的日志条目，
以 [ 开头的行，\\ 为转义

执行test.conf

ES中查看

是一条完成的数据，错误日志！！！（java中常用）

2 grok切片过滤插件

与录入信息保持一致，可以切出来不同的信息（client/method/request），理解为ES前的预处理

编辑grok.conf文件，把输入切片成五块，分别对应

执行grok.conf文件

输入一串数据，根据设定的切片方法，一一对应

apache服务日志过滤

server10安装apache，启动服务

修改apache的默认发布页面

查看apache的日志格式

在/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns目录下，有很多软件的日志的输出形式，查看apache的日志输出格式

我们只需要按照这个规定切片就好了

编辑 grok.conf 文件

把 apache 的日志作为 grok的输入，日志文件需要给读的权限，日志文件的目录需要给读和执行的权限，读的时候是logstash的身份，按照默认的HTTPD_COMBINEDLOG方式切片