Logstash input输入 beats插件和 syslog插件

Posted 2022-12-03 周天祥

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Logstash input输入 beats插件和 syslog插件相关的知识，希望对你有一定的参考价值。

Logstash input输入 beats插件和 syslog插件

Logstash input多个输入插件同时使用

Logstash -7.2.0

filter使用的插件：grok、kv、urldecode、date、mutate、geoip

1、先看总体配置logstash.conf

（执行时请去除所有中文注释）

input 
   #beats输入插件
   beats 
	#绑定主机
     host => "0.0.0.0"
	 #绑定端口
     port => 5044
	 #额外添加字段，这里是为了区分来自哪一个插件
	 add_field => "[fields][class]" => "beats"
   
   
   syslog 
   #绑定端口
	port => 514
	#额外添加字段，这里是为了区分来自哪一个插件
	add_field => "[fields][class]" => "json"
   


filter  
	#处理来自beats插件的日志,beats这里收集的是tomcat的日志
	# 样例：192.168.68.88 - - [16/Mar/2020:11:22:08 +0800] "GET /esws/testService/test?name=天道酬勤&size=50 HTTP/1.1" 200 15315
	if [fields][class] == "beats"
		#grop过滤插件，在编写grop时，可以使用kibana，kibana上有编写工具，无需自己搭建（官方grok速度太慢）
		grok 
			#解析Apache日志，自动分割
			match =>  "message" => "%COMMONAPACHELOG" 
		
		#键值过滤器
		kv 
			#对request字段操作
			source => "request"
			按照& ？ 分割
			field_split => "&?"
			value_split => "="
			#选取自己需要的分割后的字段
			include_keys => ["op","reportlet","formlet"] 
		 
		#解码
		urldecode 
			#解码全部字段
			all_fields => true
		
		#日期处理插件
		date 
			#日期匹配，匹配格式可以有多个
			match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
			#匹配的日期存储到字段中
			target => "@timestamp"
		
			
		# 数据修改
		mutate
			#移除指定字段
			remove_field => ["agent","beat","offset","tags","prospector","log","ident","[host][name]","[host][hostname]","[host][architecture]","[host][os]","[host][id]","auth","[input][type]"]
			#复制字段
			copy =>  "@timestamp" => "timestamp" 
			#copy =>  "[fields][fields_type]" => "fields_type" 
			copy =>  "formlet" => "reportlet" 
		
		
		mutate
			#替换
			gsub => ["reportlet", "%2F", "/"]
		
		
		if ! [fields_type]  
			mutate
				copy =>  "[fields][fields_type]" => "fields_type" 
			
		

		date
			match => [ "timestamp", "yyyy-MM-dd-HH:mm:ss" ]
			locale => "cn"
		
		
		#ip解析，分析IP的位置
		geoip
			source => "clientip"
		
	
	#处理来自syslog插件的日志
	if [fields][class] == "json"
		json 
			source => "message"
		
		
		if [host] == "192.168.68.100" 
			mutate
				add_field => "fields_type" => "firewall"
			
		
	
	
	  if ! [fields_type] 
                        mutate
				add_field => "fields_type" => "error-221"
                               
                 


output 
	elasticsearch 
		action => "index"
		#填写ES集群
    hosts => ["http://node-01:9200","http://node-02:9200","http://node-03:9200"]
#ES如果有登陆验证，要配置用户名和密码
#	user => "admin"
#	password => "123456"
		#按字段值，存入不同的索引中
		index => "%fields_type-%+YYYY-MM"