2023年7月17日发(作者：)

Logstash插件filter介绍及ELK相关预警配置此篇⽂章主要讲logstash的过滤器的配置并举例Filterfilter { grok { match => { "path" => "/usr/local/test/logs/%{DATA:fileName}.log"} #获取log⽂件名 } if [message] =~ ".*Exception.*" { # 格式化⽇志信息 grok { } } else { grok { mutate { replace => { "type" => "apache_access" } } } }} match => { "message" => ".*%{LOGLEVEL:priority}.*uid=%{NUMBER:uid},traceId=%{DATA:traceId} .*(%{DATA:method}@%{DATA:class}.java:%{NUM match => {" message " => "s*[impl][in]requestId=%{NUMBER:reqId},reqTime=%{NUMBER:reqTime} req=%{GREEDYDATA:req}" } # 会⾃动⽣成相应正则匹配测试1. 预警及发送http提醒input { # 略}filter { grok { match => { "message" => "s*[impl][in] traceId=%{NUMBER:traceId},reqTime=%{NUMBER:reqTime} req=%{GREEDYDATA:req}" }

# remove_field => [ "message" ] } if [priority] == "ERROR" { #是错误 添加标记 grok { tag_on_failure => "error" } } # 计数 metrics { # 每60秒清空计数器 #should be a multiple of 5s clear_interval =>60 # 每隔60秒统计⼀次 #Must be a multiple of 5s flush_interval =>60 # 计数器数据保存的字段名 priority的值默认就有 是⽇志的级别 meter => "events_%{priority}" # 增加"metric"，作为标记 add_tag => "metric" # 3秒内的message数据才统计，避免延迟 # ignore_older_than => 3 } # 如果event中标记为“metric”的 if "metric" in [tags] { # 执⾏ruby代码 ruby { # 如果level为warn的数量⼩于3条，就忽略此事件(即不发送任何消息)。 code => " if event['events_WARN']['count'] < 3" } }}output { # 如果event中标记为“metric”，表⽰只发送计数的消息。 if "metric" in [tags]{ # 通过http请求公众号发送微信消息。(此处简略描述) http { http_method => "post"

url => "116.1.1.112:33333/xxx?count=%{[events_WARN][count]}" } } stdout { codec => rubydebug } # 标准输出# elasticsearch {# action => "index" #The operation on ES# hosts => "127.0.0.1:9344" #ElasticSearch host, can be array.# index => "logstash-%{+}" #The index to write data to.# }}2. 预警 发送email提醒# input 与 filter 参考上⾯的output { # 如果event中标记为“error”，表⽰出错 if "error" in [tags] {

email {

from => "shyn_xxxxx@"

to => "123456@"

via => "smtp"

port => 25 subject => "[%{@timestamp} xxx服务器 ⽇志发现异常!]"

address => ""

domain => ""

body => "u have new bug ! %{message}"

username => "shyn_xxxxx@"

password => "password"

} } stdout { codec => rubydebug }}3. 指定Exeception异常提醒filter { if [message] =~ "s*(Arithmetic|Runtime)Exceptions*" { ruby { code => “” add_tag => "specific_exeception" } } grok { match => { "message" => "s*[impl][in] traceId=%{NUMBER:traceId},reqTime=%{NUMBER:reqTime} req=%{GREEDYDATA:req}" } } if [priority] == "ERROR" { grok { tag_on_failure => "error" } }}Logstash的⼀些配置file {

path => "/data/web/logstash/logFile/*/*" start_position => "beginning" #从⽂件开始处读写}* 有⼀些⽐较有⽤的配置项，可以⽤来指定 FileWatch 库的⾏为：discover_intervallogstash 每隔多久去检查⼀次被监听的 path 下是否有新⽂件。默认值是 15 秒。exclude不想被监听的⽂件可以排除出去，这⾥跟 path ⼀样⽀持 glob 展开。sincedb_path如果你不想⽤默认的 $HOME/.sincedb(Windows 平台上在C:b)，可以通过这个配置定义 sincedb ⽂件到其他位置。sincedb_write_intervallogstash 每隔多久写⼀次 sincedb ⽂件，默认是 15 秒。stat_intervallogstash 每隔多久检查⼀次被监听⽂件状态（是否有更新），默认是 1 秒。start_positionlogstash 从什么位置开始读取⽂件数据，默认是结束位置，也就是说 logstash 进程会以类似 tail -F 的形式运⾏。如果你是要导⼊原有数据，把这个设定改成 "beginning"，logstash 进程就从头开始读取，有点类似cat，但是读到最后⼀⾏不会终⽌，⽽是继续变成 tail -F。这⾥推荐些⽹站供读者查阅

logstash的数据类型和基本语法

logstash官⽹查询插件及配置

logstash较详细中⽂介绍

还有当时查询elk时觉得挺有⽤的⼀些⽹站链接也在这⾥贴⼀贴

分布式系统⽇志跟踪

Logstash ⽗⼦关系配置

分布式系统全链路监控介绍包括架构，原理等

分布式实时⽇志处理平台ELK

分布式系统调⽤链监控