ELK日志分析平台（二）----logstash数据采集

目录

一、logstash简介

二、安装

三、与elasticsearch连用

 四、将logstash伪装成日志服务器，接受远程日志

五、logstash的多行过滤

六、grok切片过滤

---

一、logstash简介

官网：Logstash Introduction | Logstash Reference [7.6] | Elastic

        Logstash是一个开源的服务器端数据处理管道，拥有200多个插件，能够同时从多个来源采集数据， 转换数据，然后将数据发送到您最喜欢的 “存储库” 中。（大多 都是 Elasticsearch）；

         Logstash管道有两个必需的元素，输入和输出，以及一个可选 元素过滤器。

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

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

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

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

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

二、安装

官网安装文档：Installing Logstash | Logstash Reference [7.6] | Elastic

下载地址：Download Logstash Free | Get Started Now | Elastic

环境：需要Java环境，主机的内存1G以上

##可以采用离线安装和在线安装
----------------离线--------------
##分别安装jdk和logstash包
rpm -ivh jdk-8u171-linux-x64.rpm
rpm -ivh logstash-7.6.1.rmp     ----需要和elasticsearch的版本匹配
 
----------------在线--------------
##下载并安装密钥验证包
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
 
##配置yum源
vim /etc/yum.repos.d/logstash.repo
---------------------------------------------------
[logstash-7.x]
name=Elastic repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
---------------------------------------------------
 
##安装
yum install logstash

##验证
-------------------------命令行-------------------------------
cd logstash-7.6.2            --切换到bin脚本的目录下，执行脚本文件
bin/logstash -e 'input { stdin { } } output { stdout {} }'
 
-------------------------文件-------------------------------
##在/etc/logstash/conf.d/目录下以.conf结尾的文件都可以被读到
 
vim test.conf
--------------------------------------------------------
input {
	stdin {}		--输入来自命令行标准输入
}
 
output {
    stdout {}       --在终端看到结果
 file {				--输出到/tmp/testfile文件中（输出的文件路径），格式为custom format: {输入内容}
   path => "/tmp/testfile"
   codec => line { format => "custom format: %{message}"}    --message变量可以将输入的值显示出来（在stdin中输入的数据都可以被messag截获）；
 }
}
--------------------------------------------------------
 
##运行
/usr/share/logstash/bin/logstash -f test.conf
hello world   --命令行输入
cat /tmp/testfile

三、与elasticsearch连用

##命令行输入 elasticsearch输出
vim /etc/logstash/conf.d/es.conf
------------------------------------------------
input {
	stdin {}
}
 
output {
	stdout {}			--标准输出一份
 
	elasticsearch {		--给elasticsearch输出一份
		hosts => ["192.168.19.71:9200"]	--目标elasticsearch主机ip与端口
		index => "logstash-%{+yyyy.MM.dd}"	--定制**索引名称**，索引格式为logstash-年月日
	}
}
------------------------------------------------
 
##执行
/usr/share/logstash/bin/logstash -f es.conf
 
##网页查看结果
192.168.19.71：9100
->点击数据浏览->指定索引logstash-2021.08.14，可以看到elasticsearch输出一份->查看数据
 

##文件输入到elasticsearch
 
##把日志文件作为输入：/var/log/messages
 
##修改权限使日志文件可被logstash可读
chmod 644 /var/log/messages
 
##修改/etc/logstash/conf.d/es.conf文件
------------------------------------------------
input {
	file {				--从文件/var/log/messages输入，从头开始输入
		path => "/var/log/messages"
		start_position => "beginning"    --起始位置
	}
}
 
output {
	stdout {}			--标准输出
 
	elasticsearch {		--输出elasticsearch
		hosts => ["192.168.19.71:9200"]	
		index => "logstash-%{+yyyy.MM.dd}"	
	}
}
------------------------------------------------
 
##执行文件
/ur/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf
 
##网页端查看
192.168.19.71:9100-->192.168.19.71:9200(连接到这里)

##完整删除
##在删除时不仅要删除索引，也需要删除数据库
logstash会把查询的进度按照分类保存到/usr/share/logstash/data/plugins/inputs/file/sincedb文件中；该文件主要负责记录数据在文件中查询的偏移量，所以在删除索引后，想全部查询文件，需要清除该文件
rm -f .sincedb_

sincedb文件一共6个字段：

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

 四、将logstash伪装成日志服务器，接受远程日志

##避免每个节点都安装logstash进行数据采集，可将含有logstash的主机，作为日志服务器，收集所有日志，并交给logstarsh进行处理
vim /etc/logstash/conf.d/es.conf
-------------------------------------------------
input {
	#file {
	#	path => "/var/log/messages"
	#	start_position => "beginning"
	#}
	
	syslog {		--伪装syslog，开放端口514
		port => 514     --日志收集端口（默认）
	}
}
 
output {
	stdout {}
 
	elasticsearch {
		hosts => ["192.168.19.71:9200"]
		index => "syslog-%{+yyyy.MM.dd}"	--索引为syslog-年月日
	}
}
-------------------------------------------------
 
##执行该文件
/use/share/logstash/bin/logstash -f es.conf
 
 
##查看端口是否开放
netstat -anltupe | grep 514
 
##配置服务器上的配置日志，允许远程备份
vim /etc/rsyslog.conf
-------------------------------------------------
$ModLoad imudp    --打开端口
$UDPServerRun 514  
 
*.* @@192.168.19.71：514  --日志发送到71主机上  
-------------------------------------------------
 
##在logstash主机上查看
192.168.19.71:9200--网页查看

五、logstash的多行过滤

##命令行多行输入
vim /etc/logstash/conf.d/multest.conf
------------------------------------------------------
input {
      stdin {
          codec => multiline {	--多行输入
            pattern => "EOF"	--结束标志词为EOF
            negate => "true"
            what => "previous"
          }
        }
}
 
output {
	stdout {}
 
 file {				
   path => "/tmp/multest"		--输出到文件/tmp/multest
   codec => line { format => "custom format: %{message}"}	
   								--格式为custom format:{数据}
 }
}
------------------------------------------------------
 
##执行文件
/usr/share/logstash/bin/logstash -f multest.conf
 
##命令行输入
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
1
2
3
4
EOF
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

##文件多行输入
##使用日志文件
vim multest.conf 
-------------------------------------------------------------------
input {
      file {		
        path => "/var/log/my-es.log"	--文件/var/log/my-es.log作为输入
        start_position => "beginning"	--从头开始输入
          codec => multiline {
            pattern => "^\["
            negate => "true"
            what => "previous"
          }
      }
}
 
output {
	stdout {}
 
	elasticsearch {			--输出到es
		hosts => ["192.168.19.71:9200"]
		index => "eslog-%{+yyyy.MM.dd}"
	}
}
-------------------------------------------------------------------
 
##执行
/usr/share/logstash/bin/logstash -f multest.conf
 
##查看
web->192.168.19.71：9100->header

对输入文件进行简单分析：正确的日志都是以时间开头的，并且被中括号[ ]括起来的一行，而错误日志有很多行，比如下图的at org开头的这些，他们合起来应该是一条错误日志。

六、grok切片过滤

     查看日志时，一般都有一定的规律。比如查看apache的日志，先是访问的ip，时间等等。在查看时我们需要查看其中一组数据，比如只想要ip这一列。此时就可以使用logstash的切片功能，根据日志的特征自定义grok的书写，得到想要的切片。

##对命令行输入的过滤
vim /etc/logstash/conf.d/grok.conf
----------------------------------------------------
input {
	stdin {}
}
 
filter {
	grok {						--把输入切片成五块，分别对应
	match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
	}
}
 
output {
	stdout {}
}
----------------------------------------------------
 
##执行
/etc/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf
 
##终端可以直接查看

##文件输入
##文件输入时需要注意权限的改变，使用户可读：644
##修改grok.conf文件
---------------------------------------------------
input {
	file {
		path => "/var/log/httpd/access_log"	--/var/log/httpd/access_log文件作为输入
		start_position => "beginning"		--从头开始
	}
 
}
 
filter {
	grok {
	match => { "message" => "%{HTTPD_COMBINEDLOG}" }	--按照默认的HTTPD_COMBINEDLOG方式切片
	}
}
 
output {
	stdout {}
 
	elasticsearch {
		hosts => ["192.168.19.71:9200"]
		index => "apachelog-%{+yyyy.MM.dd}"		--索引名字叫apachelog
	}
	
}
 
---------------------------------------------------
 
##执行
/etc/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf
 
##查看
终端和web端均可查看
 
##压力测试：
ab -n 100 -cl http://192.169.19.71/index.html
 
##再次查看数据
终端和web端均可查看

【编写切片格式】官方在：/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns提供了较多的模板，可以进行参考；

上述的文件输入，我们用的是Apche的文件输入，所以查看httpd的规定，如何写日志已经提前用变量的方法定义了按照这个规定切片就好了