5.EFLK(ELK+filebeat)+filter过滤

这篇具有很好参考价值的文章主要介绍了5.EFLK(ELK+filebeat)+filter过滤。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

EFLK(ELK+filebeat)

elsaticsearch:
node1:192.168.242.66
node2:192.168.242.67

nginx+logstash+kibana:
192.168.242.68

filebeat:192.168.242.69

部署filebeat

systemctl stop firewalld
systemctl disable firewalld
setenforce 0

vim /etc/selinux/config

SELINUX=disabled
##安装 Filebeat

#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录

tar xf filebeat-6.7.2-linux-x86_64.tar.gz

mv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat
##安装  apache  服务,用来提供访问日志

yum -y install httpd

systemctl start httpd
systemctl enable httpd

修改配置文件

###设置 filebeat 的主配置文件

cd /usr/local/filebeat

cp filebeat.yml{,.bak}
vim filebeat.yml


filebeat.inputs:
  - type: log         
#指定 log 类型,从日志文件中读取消息

  enabled: true
  paths:
    - /var/log/httpd/access_log       
    #指定监控的日志文件
 
  tags: ["filebeat"]		#设置索引标签
  
  
  
  
#可以使用 fields 配置选项设置一些参数字段添加到 output 

  fields:          
    server_name: httpd
    log_type: access
    from: 192.168.242.69

--------------Elasticsearch output-------------------
(全部注释掉)

----------------Logstash output---------------------
output.logstash:
  hosts: ["192.168.242.68:5044"]      
  #指定 logstash 的 IP 和端口
###启动 filebeat

cd /usr/local/filebeat

nohup 
./filebeat -e -c filebeat.yml 

#-e:输出到标准输出,禁用syslog/文件输出
#-c:指定配置文件
#nohup:在系统后台不挂断地运行命令,退出终端不会影响程序的运行

logstash配置

###在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d

vim filebeat.conf
input {
     beats {
       port => "5044"
   }
}


output {
    elasticsearch {
        hosts => ["192.168.242.66:9200","192.168.242.67:9200"]
        index => "%{[fields][server_name]}-%{+YYYY.MM.dd}"
    }

}
###启动 logstash

logstash -f filebeat.conf

###如果配置失败,清空缓存文件

rm -rf /var/lib/logstash/.lock
###浏览器访问 
--》http://192.168.242.68:5601 登录 Kibana
--》单击“Create Index Pattern”按钮添加索引
--》“filebeat-*”
--》单击 “create” 按钮创建
--》单击 “Discover” 按钮可查看图表信息及日志信息。

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

logstash的filter过滤

grok(正则捕获插件)

  • grok 使用文本片段切分的方式来切分日志事件
  • logstash 官方也给了一些常用的常量来表达那些正则表达式,可以到这个 Github 地址查看有哪些常用的常量:
    内置的grok正则表达式

内置正则表达式调用

%{SYNTAX:SEMANTIC}

##SYNTAX代表匹配值的类型,例如,0.11可以NUMBER类型所匹配,10.222.22.25可以使用IP匹配。

##SEMANTIC表示存储该值的一个变量声明,它会存储在elasticsearch当中方便kibana做字段搜索和统计,你可以将一个IP定义为客户端IP地址client_ip_address,如%{IP:client_ip_address},所匹配到的值就会存储到client_ip_address这个字段里边,类似数据库的列名,也可以把 event log 中的数字当成数字类型存储在一个指定的变量当中,比如响应时间http_response_time,假设event log record如下:

message: 192.168.80.10 GET /index.html 15824 0.043

可以使用如下grok pattern来匹配这种记录
%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}
###在logstash.conf  配置文件中使用filter过滤规则

filter {
  grok {
    match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
  }
}
192.168.80.1 - - [11/Jul/2023:09:48:24 +0800] "GET / HTTP/1.1" 403 4897 "-" "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.5359.125 Safari/537.36"

192.168.80.1 - - [11/Jul/2023:10:24:52 +0800] "GET /test.html HTTP/1.1" 200 32 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36"

192.168.80.20 - - [11/Jul/2023:10:24:36 +0800] "GET /test.html HTTP/1.1" 200 32 "-" "curl/7.29.0"



###通过grok过滤规则过滤上面的三段数据


%{IP:remote_addr} - - \[%{HTTPDATE:log_time}\] \"%{WORD:http_method} %{URIPATH:request_url} (?<http_var>.+)\" %{NUMBER:response_code} .+ \".+\" \"(?<user_agent>.+)\"

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

###在配置文件中添加filter


vim /etc/logstash/conf.d/filebeat.conf

##添加

filter {
  grok {
    match => { "message" => "%{IP:remote_addr} - - \[%{HTTPDATE:log_time}\] \"%{WORD:http_method} %{URIPATH:request_url} (?<http_var>.+)\" %{NUMBER:response_code} .+ \".+\" \"(?<user_agent>.+)\"" }
  }
}

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

自定义表达式

语法:(?<field_name>pattern)


举例:捕获10或11和长度的十六进制数的queue_id可以使用表达式(?<queue_id>[0-9A-F]{10,11})
message: 192.168.80.10 GET /index.html 15824 0.043


(?<remote_addr>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) (?<http_method>[A-Z]+) (?<request_uri>/.*) (?<response_bytes>[0-9]+) (?<response_time>[0-9\.]+)

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

如果表达式匹配失败,会生成一个tags字段,字段值为 _grokparsefailure,需要重新检查上边的match配置解析是否正确。

ipv4 的正则表达式

0-9   [0-9]
10-99      [1-9][0-9]
100-199    1[0-9][0-9]
200-249    2[0-4][0-9]
250-255     25[0-5]

mutate(数据修改插件)

  • 它提供了丰富的基础类型数据处理能力。可以重命名,删除,替换和修改事件中的字段
###Mutate 过滤器常用的配置选项

add_field	        向事件添加新字段,也可以添加多个字段
remove_field        从事件中删除任意字段
add_tag	            向事件添加任意标签,在tag字段中添加一段自定义的内容,当tag字段中超过一个内容的时候会变成数组	
remove_tag	        从事件中删除标签(如果存在)
convert	            将字段值转换为另一种数据类型
id	                向现场事件添加唯一的ID
lowercase	        将字符串字段转换为其小写形式
replace	            用新值替换字段
strip	            删除开头和结尾的空格
uppercase	        将字符串字段转换为等效的大写字母
update	            用新值更新现有字段
rename	            重命名事件中的字段
gsub	            通过正则表达式替换字段中匹配到的值
merge	            合并数组或 hash 事件
split               通过指定的分隔符分割字段中的字符串为数组

重命名字段

filter {
    mutate {
	    #写法1,使用中括号括起来
        rename => ["old_field" => "new_field"]

        #写法2,使用大括号{}括起来
	    rename => { "old_field" => "new_field" }		
    }
}

添加字段

filter {
    mutate {
        add_field => {
        	"f1" => "field1"
        	"f2" => "field2"
        }
    }
}

删除字段

将字段删除
filter {
    mutate {
        remove_field  =>  ["message", "@version", "tags"]
    }
}

转换数据类型

###将filedName1字段数据类型转换成string类型,filedName2字段数据类型转换成float类型

filter {
    mutate {
        #写法1,使用中括号括起来
        convert  =>  ["filedName1", "string"]
		
        #写法2,使用大括号{}括起来
		convert => { "filedName2" => "float" }
    }
}

替换字段内容

filter {
    mutate {
        gsub => ["filedName", "/" , "_"]
    }
}
###将filedName字段中所有",“字符后面添加空格

filter {
    mutate {
        gsub => ["filedName", "," , ", "]
    }
}

以"|"为分割符拆分数据成为数组

filter {
    mutate {
        split => ["filedName", "|"]
    }
}

合并字段

filter {
    merge  { "filedName2" => "filedName1" }
}

用新值替换字段的值

filter {
    mutate {
        replace => { "filedName" => "new_value" }
    }
}

添加字段first,值为message数组的第一个元素的值

filter {
    mutate {
        split => ["message", "|"]
        add_field => {
            "first" => "%{[message][0]}"    
        } 
    }
}

有条件的添加标签

filter {
    #在日志文件路径包含 access 的条件下添加标签内容
    if [path] =~ "access" {
        mutate {
            add_tag => ["Nginx Access Log"]
        }
    }
    
    #在日志文件路径是 /var/log/nginx/error.log 的条件下添加标签内容
    if [path] == "/var/log/nginx/error.log" {
        mutate {
            add_tag => ["Nginx Error Log"]
        }
    }
}

multiline(多行合并插件)

  • java错误日志一般都是一条日志很多行的,会把堆栈信息打印出来,当经过 logstash 解析后,每一行都会当做一条记录存放到 ES, 那这种情况肯定是需要处理的。
  • 这里就需要使用 multiline 插件,对属于同一个条日志的记录进行拼接。

安装multiline插件

###在线安装插件

cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline
###离线安装插件

先在有网的机器上在线安装插件,然后打包,拷贝到服务器,执行安装命令

bin/logstash-plugin prepare-offline-pack --overwrite --output logstash-filter-multiline.zip logstash-filter-multiline


###上传压缩包文件   logstash-offline-plugins-6.7.2.zip 

./bin/logstash-plugin install file:///usr/share/logstash/logstash-offline-plugins-6.7.2.zip

###检查下插件是否安装成功,可以执行以下命令查看插件列表

bin/logstash-plugin list

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

  • 使用 multiline 插件
    • 第一步:每一条日志的第一行开头都是一个时间,可以用时间的正则表达式匹配到第一行。
    • 第二步:然后将后面每一行的日志与第一行合并。
    • 第三步:当遇到某一行的开头是可以匹配正则表达式的时间的,就停止第一条日志的合并,开始合并第二条日志。
    • 第四步:重复第二步和第三步。
###合并下面的java日志

vim /var/log/java.log



2022-11-11 17:09:19.774[XNIo-1 task-1]ERROR com.passjava.controlle .NembercController-查询用户 活动数据失败,异常信息为:
    com.passjava.exception.MemberException: 当前没有配置活动规则
    at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)
    at com.passjava.service.impl.dailyserviceImpL$$FastcLass
2022-11-11 17:10:56.256][KxNIo-1 task-1] ERROR com.passjava.controlle .NemberControl1er-查询员工 饭活动数据失败,异常信息为:
    com.passjava.exception.MemberException: 当前没有配置活动规则
    at com.passjava.service.impL.queryAdmin(DailyServiceImpl.java:1444)
    at com.passjava.service.impL.daiLyserviceImpL$$FastcLass
    
    
    
    
##过滤规则
##创建配置文件

vim /etc/logstash/conf.d/java.conf


input {
    file {
       path => "/var/log/java.log"
       type => "java"
       start_position => "beginning"
    }

}



filter {
  multiline {
    pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}"
    negate => true
    what => "previous"
  }
}


output {
    elasticsearch {
        hosts => ["192.168.242.66:9200","192.168.242.67:9200"]
       index => "java-%{+YYYY.MM.dd}"
    }

}
pattern:用来匹配文本的表达式,也可以是grok表达式

what:如果pattern匹配成功的话,那么匹配行是归属于上一个事件,还是归属于下一个事件。
           previous: 归属于上一个事件,向上合并。
            ext: 归属于下一个事件,向下合并

negate:是否对pattern的结果取反。false:不取反,是默认值。true:取反。
                将多行事件扫描过程中的行匹配逻辑取反(如果pattern匹配失败,则认为当前行是多行事件的组成部分)

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk

date(时间插件)

  • 在Logstash产生了一个Event对象的时候,会给该Event设置一个时间,字段为“@timestamp”,同时,我们的日志内容一般也会有时间,但是这两个时间是不一样的,因为日志内容的时间是该日志打印出来的时间,而“@timestamp”字段的时间是input插件接收到了一条数据并创建Event的时间,所有一般来说的话“@timestamp”的时间要比日志内容的时间晚一点,因为Logstash监控数据变化,数据输入,创建Event导致的时间延迟。这两个时间都可以使用,具体要根据自己的需求来定。
filter {
    date {
        match => ["access_time", "dd/MMM/YYYY:HH:mm:ss Z", "UNIX", "yyyy-MM-dd HH:mm:ss", "dd-MMM-yyyy HH:mm:ss"]
		target => "@timestamp"
		timezone => "Asia/Shanghai"
    } 
}
match:用于配置具体的匹配内容规则,前半部分内容表示匹配实际日志当中的时间戳的名称,后半部分则用于匹配实际日志当中的时间戳格式,
             这个地方是整条配置的核心内容,如果此处规则匹配是无效的,则生成后的日志时间戳将会被input插件读取的时间替代。
            如果时间格式匹配失败,会生成一个tags字段,字段值为 _dateparsefailure,需要重新检查上边的match配置解析是否正确。

target:将匹配的时间戳存储到给定的目标字段中。如果未提供,则默认更新事件的@timestamp字段。

timezone:当需要配置的date里面没有时区信息,而且不是UTC时间,需要设置timezone参数。
###时间戳详解:

年
yyyy  #全年号码。 例如:2015。
yy    #两位数年份。 例如:2015年的15。

月
M     #最小数字月份。 例如:1 for January and 12 for December.。
MM    #两位数月份。 如果需要,填充零。 例如:01 for January  and 12 for Decembe
MMM   #缩短的月份文本。 例如: Jan for January。 注意:使用的语言取决于您的语言环境。 请        参阅区域设置以了解如何更改语言。
MMMM  #全月文本,例如:January。 注意:使用的语言取决于您的语言环境。

日
d   #最少数字的一天。 例如:1月份的第一天1。
dd  #两位数的日子,如果需要的话可以填零.例如:01 for the 1st of the month。

时
H   #最小数字小时。 例如:0表示午夜。
HH  #两位数小时,如果需要填零。 例如:午夜00。

分
m   #最小的数字分钟。 例如:0。
mm  #两位数分钟,如果需要填零。 例如:00。

秒
s    #最小数字秒。 例如:0。
ss   #两位数字,如果需要填零。 例如:00。

毫秒( 秒的小数部分最大精度是毫秒(SSS)。除此之外,零附加。)
S    #十分之一秒。例如:0为亚秒值012
SS   #百分之一秒 例如:01为亚秒值01
SSS  #千分之一秒 例如:012为亚秒值012

时区偏移或身份
Z    #时区偏移,结构为HHmm(Zulu/UTC的小时和分钟偏移量)。例如:-0700。
ZZ   #时区偏移结构为HH:mm(小时偏移和分钟偏移之间的冒号)。 例如:-07:00。
ZZZ  #时区身份。例如:America/Los_Angeles。 
         注意:有效的ID在列表中列出http://joda-time.sourceforge.net/timezones.html

案例

过滤message数据中,只获得时间参数


vim /etc/logstash/conf.d/filebeat.conf



input {
     beats {
       port => "5044"
   }
}


filter{
      grok{
           match => {"message" => ".* -\ -\ \[%{HTTPDATE:timestamp}\]"}
      }
      date{
           match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
           target => "@timestamp"
           timezone => "Asia/Shanghai"
    }
}



output {
    elasticsearch {
        hosts => ["192.168.242.66:9200","192.168.242.67:9200"]
        index => "%{[fields][server_name]}-%{+YYYY.MM.dd}"
    }

}


###在web界面查看两个时间是否一致

5.EFLK(ELK+filebeat)+filter过滤,10.分布式,elk文章来源地址https://www.toymoban.com/news/detail-562188.html

到了这里,关于5.EFLK(ELK+filebeat)+filter过滤的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • docker搭建Elk+Kafka+Filebeat分布式日志收集系统

    目录 一、介绍 二、集群环境 三、ES集群 四、Kibana  五、Logstash 六、Zookeeper 七、Kafka 八、Filebeat 八、Nginx (一)架构图  (二)组件介绍 1.Elasticsearch 是一个基于Lucene的搜索服务器。提供搜集、分析、存储数据三大功能。它提供了一个分布式多用户能力的全文搜索引擎,基于

    2024年02月04日
    浏览(52)
  • 4.日志分布式-ELK

    ELK平台是一套完整的日志集中处理解决方案,将 ElasticSearch、Logstash 和 Kiabana 三个开源工具配合使用, 完成更强大的用户对日志的查询、排序、统计需求。 ElasticSearch:是基于Lucene(一个全文检索引擎的架构)开发的分布式存储检索引擎,用来存储各类日志。 Elasticsearch 是用

    2024年02月13日
    浏览(47)
  • 【Springcloud】elk分布式日志

    (1)什么是分布式日志 在分布式应用中,日志被分散在储存不同的设备上。如果你管理数十上百台服务器,你还在使用依次登录每台机器的传统方法查阅日志。这样是不是感觉很繁琐和效率低下。所以我们使用集中化的日志管理,分布式日志就是对大规模日志数据进行采集

    2024年02月09日
    浏览(45)
  • 【分布式技术】ELK大型日志收集分析系统

    目录 步骤一:完成JAVA环境部署 步骤二:部署ES节点(三台主机) 步骤三:内核参数修改 步骤四:web端查看验证 步骤五:yum安装nginx  步骤六:完成logstash部署 步骤七:部署kibana  步骤八:测试ELK

    2024年01月20日
    浏览(46)
  • 分布式ELK 企业级日志分析系统

    ELK平台是一套完整的日志集中处理解决方案,将 ElasticSearch、Logstash 和 Kiabana 三个开源工具配合使用, 完成更强大的用户对日志的查询、排序、统计需求。 ElasticSearch :是基于Lucene(一个全文检索引擎的架构)开发的分布式存储检索引擎,用来存储各类日志。Elasticsearch 是用

    2024年02月16日
    浏览(40)
  • 【手记】基于SpringCloud搭建ELK分布式日志追踪

    ELK: 是 Elasticsearch、Logstrash 和 Kibana 的缩写,它们代表的是一套成熟的日志管理系统,ELK Stack已经成为目前最流行的集中式日志解决管理方案。 Elasticsearch: 分布式搜索和分析引擎,具有高可伸缩、高可靠和易管理等特点。基于 Apache Lucene 构建,能对大容量的数据进行接近实

    2024年01月16日
    浏览(47)
  • 【分布式应用】ELK企业级日志分析系统

    目录 一、ELK 简介 1.1 ELK各组件介绍 ElasticSearch: Kiabana: Logstash: 1.2 可以添加的其它组件: Filebeat: 缓存/消息队列(redis、kafka、RabbitMQ等): Fluentd: 1.3 ELK、ELFK、EFLKL 二、为什么要使用 ELK 三、完整日志系统基本特征 四、ELK 的工作原理 五、ELK集群部署 实验环境: 实验步

    2024年02月14日
    浏览(45)
  • 分布式应用:ELK企业级日志分析系统

    目录 一、理论 1.ELK  2.ELK场景  3.完整日志系统基本特征 4.ELK 的工作原理          5.ELK集群准备 6.Elasticsearch部署(在Node1、Node2节点上操作) 7.Logstash 部署(在 Apache 节点上操作) 8.Kiabana 部署(在 Node1 节点上操作) 二、实验          1.Elasticsearch部署(在Node1、Node2节点上

    2024年02月14日
    浏览(42)
  • 【Distributed】分布式ELK日志文件分析系统(一)

      日志主要包括系统日志、应用程序日志和安全日志。系统远维和开发人员可以通过日志了解服务器软硬件信息、检查配置过程中的错误及错误发生的原因。经常分析日志可以了解服务器的负荷,性能安全性,从而及时采取措施纠正错误。   往往单台机器的日志我们使用

    2024年02月15日
    浏览(42)
  • RedissonClient妙用-分布式布隆过滤器

    目录 布隆过滤器介绍 布隆过滤器的落地应用场景 高并发处理  多个过滤器平滑切换 分析总结 布隆过滤器(Bloom Filter)是1970年由布隆提出的。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。 它的优点是空间效

    2024年02月21日
    浏览(24)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包