ELK之Logstash日志处理模块

当我们采集日志数据源，然后想进行进一步处理，我们可以使用Logstash来帮我们进一步处理。

1

$ wget https://artifacts.elastic.co/downloads/logstash/logstash-7.14.1-linux-x86_64.tar.gz

1.2 部署

1
2


$ tar zxf logstash-7.14.1-linux-x86_64.tar.gz
$ cd logstash-7.14.1

1.3 配置文件配置

logstash.yml

path.data: /opt/testerzhang/elk/datadir/logstashdir

新建规则文件rule.conf

input {
    stdin { }
}


output {
    stdout {}
}

1.4 启动 Logstash

使用-t参数验证规则文件是否配置有问题

1

$ bin/logstash -t -f config/rule.conf

启动

1

$ bin/logstash -f config/rule.conf

如果正常启动，日志不会报错，同时可以查看进程是否存活。

调试

$ bin/logstash -f config/rule.conf --verbose --debug

这样就可以调试你的规则了。

后台启动，并动态加载配置文件

1

$ nohup ./bin/logstash -f ./config/rule.conf --config.reload.automatic  &

如果你在一台机器上启动两个不同的logstash，可能会有如下的报错：

[2021-09-15T10:44:59,670][FATAL][logstash.runner          ] Logstash could not be started because there is already another instance using the configured data directory.  If you wish to run multiple instances, you must change the "path.data" setting.

查看是否进程还在：

如果进程在就kill掉进程，再运行一次。
如果没有进程，之前运行的instance有缓冲，保存在path.data里面有.lock文件，删除掉就可以。可以在你的path.data目录操作:

$ find . -name *.lock
./.lock
$ rm .lock

2.安装Logstash插件(可选)

2.1查看可用插件

1

$ bin/logstash-plugin list

2.2安装新插件

从 https://github.com/logstash-plugins/ 安装新插件

1

$ bin/logstash-plugin install logstash-filter-json

2.3安装成功

Validating logstash-filter-json
Installing logstash-filter-json
Installation successful

3.配置日志规则

在logstash中，包括了三个阶段:

输入input --> 处理filter（不是必须的） --> 输出output

3.1 input 配置节

input {
  beats {
    host => "10.10.10.6"
    port => 9600
  }
}

配置启动端口为9600，启动IP:10.10.10.6

3.2 output 配置节

output {

  elasticsearch {
    hosts => ["http://10.10.10.10:9201"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }

  stdout {
    codec => rubydebug
  }
}

上面是以输出数据到搜索引擎elasticsearch为例

elasticsearch配置参数解析：

hosts: es的访问地址，建议使用非master节点。
user: 访问es的用户名。
password：访问es的密码。
index：在es中的索引名称。
template：设置自己的es模板路径。
template_name：使用es中的索引模板名称。
上方的es的密码是明文的，可能存在泄漏，可以使用 logstash keystore来解决。

codec => rubydebug 是配合调试使用，在启动命令后面增加--verbose --debug，这样直接在前台启动,便可以看到日志输出到终端

3.3 filter配置节

3.3.1 input进来的数据匹配规则进行丢弃

丢弃这条日志

drop {}

过滤删除指定字段RetryJob

drop {
  remove_field => [ "RetryJob" ]
}

3.3.2 grok过滤说明

Grok 是 Logstash 最重要的插件。你可以在 grok 里预定义好命名正则表达式，在稍后(grok参数或者其他正则表达式里)引用它。

grok表达式的打印复制格式的完整语法是下面这样的：

%{PATTERN_NAME:capture_name:data_type}

grok-patterns内置的正则表达式，实在不太多了，看不过来了。

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
EMAILLOCALPART [a-zA-Z][a-zA-Z0-9_.+-=:]+
EMAILADDRESS %{EMAILLOCALPART}@%{HOSTNAME}
HTTPDUSER %{EMAILADDRESS}|%{USER}
INT (?:[+-]?(?:[0-9]+))
BASE10NUM (?<![0-9.+-])(?>[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+)))
NUMBER (?:%{BASE10NUM})
BASE16NUM (?<![0-9A-Fa-f])(?:[+-]?(?:0x)?(?:[0-9A-Fa-f]+))
BASE16FLOAT \b(?<![0-9A-Fa-f.])(?:[+-]?(?:0x)?(?:(?:[0-9A-Fa-f]+(?:\.[0-9A-Fa-f]*)?)|(?:\.[0-9A-Fa-f]+)))\b
POSINT \b(?:[1-9][0-9]*)\b
NONNEGINT \b(?:[0-9]+)\b
WORD \b\w+\b
NOTSPACE \S+
SPACE \s*
DATA .*?
GREEDYDATA .*
QUOTEDSTRING (?>(?<!\\)(?>"(?>\\.|[^\\"]+)+"|""|(?>'(?>\\.|[^\\']+)+')|''|(?>`(?>\\.|[^\\`]+)+`)|``))
UUID [A-Fa-f0-9]{8}-(?:[A-Fa-f0-9]{4}-){3}[A-Fa-f0-9]{12}
# Networking
MAC (?:%{CISCOMAC}|%{WINDOWSMAC}|%{COMMONMAC})
CISCOMAC (?:(?:[A-Fa-f0-9]{4}\.){2}[A-Fa-f0-9]{4})
WINDOWSMAC (?:(?:[A-Fa-f0-9]{2}-){5}[A-Fa-f0-9]{2})
COMMONMAC (?:(?:[A-Fa-f0-9]{2}:){5}[A-Fa-f0-9]{2})
IPV6 ((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(((:[0-9A-Fa-f]{1,4}){1,7})|((:[0-9A-Fa-f]{1,4}){0,5}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:)))(%.+)?
IPV4 (?<![0-9])(?:(?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5])[.](?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5])[.](?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5])[.](?:[0-1]?[0-9]{1,2}|2[0-4][0-9]|25[0-5]))(?![0-9])
IP (?:%{IPV6}|%{IPV4})
HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b)
IPORHOST (?:%{IP}|%{HOSTNAME})
HOSTPORT %{IPORHOST}:%{POSINT}
# paths
PATH (?:%{UNIXPATH}|%{WINPATH})
UNIXPATH (/([\w_%!$@:.,~-]+|\\.)*)+
TTY (?:/dev/(pts|tty([pq])?)(\w+)?/?(?:[0-9]+))
WINPATH (?>[A-Za-z]+:|\\)(?:\\[^\\?*]*)+
URIPROTO [A-Za-z]+(\+[A-Za-z+]+)?
URIHOST %{IPORHOST}(?::%{POSINT:port})?
# uripath comes loosely from RFC1738, but mostly from what Firefox
# doesn't turn into %XX
URIPATH (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%_\-]*)+
#URIPARAM \?(?:[A-Za-z0-9]+(?:=(?:[^&]*))?(?:&(?:[A-Za-z0-9]+(?:=(?:[^&]*))?)?)*)?
URIPARAM \?[A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]<>]*
URIPATHPARAM %{URIPATH}(?:%{URIPARAM})?
URI %{URIPROTO}://(?:%{USER}(?::[^@]*)?@)?(?:%{URIHOST})?(?:%{URIPATHPARAM})?
# Months: January, Feb, 3, 03, 12, December
MONTH \b(?:Jan(?:uary|uar)?|Feb(?:ruary|ruar)?|M(?:a|)?r(?:ch|z)?|Apr(?:il)?|Ma(?:y|i)?|Jun(?:e|i)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|O(?:c|k)?t(?:ober)?|Nov(?:ember)?|De(?:c|z)(?:ember)?)\b
MONTHNUM (?:0?[1-9]|1[0-2])
MONTHNUM2 (?:0[1-9]|1[0-2])
MONTHDAY (?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])
# Days: Monday, Tue, Thu, etc...
DAY (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?)
# Years?
YEAR (?>\d\d){1,2}
HOUR (?:2[0123]|[01]?[0-9])
MINUTE (?:[0-5][0-9])
# '60' is a leap second in most time standards and thus is valid.
SECOND (?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)
TIME (?!<[0-9])%{HOUR}:%{MINUTE}(?::%{SECOND})(?![0-9])
# datestamp is YYYY/MM/DD-HH:MM:SS.UUUU (or something like it)
DATE_US %{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR}
DATE_EU %{MONTHDAY}[./-]%{MONTHNUM}[./-]%{YEAR}
ISO8601_TIMEZONE (?:Z|[+-]%{HOUR}(?::?%{MINUTE}))
ISO8601_SECOND (?:%{SECOND}|60)
TIMESTAMP_ISO8601 %{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:?%{MINUTE}(?::?%{SECOND})?%{ISO8601_TIMEZONE}?
DATE %{DATE_US}|%{DATE_EU}
DATESTAMP %{DATE}[- ]%{TIME}
TZ (?:[PMCE][SD]T|UTC)
DATESTAMP_RFC822 %{DAY} %{MONTH} %{MONTHDAY} %{YEAR} %{TIME} %{TZ}
DATESTAMP_RFC2822 %{DAY}, %{MONTHDAY} %{MONTH} %{YEAR} %{TIME} %{ISO8601_TIMEZONE}
DATESTAMP_OTHER %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{TZ} %{YEAR}
DATESTAMP_EVENTLOG %{YEAR}%{MONTHNUM2}%{MONTHDAY}%{HOUR}%{MINUTE}%{SECOND}
HTTPDERROR_DATE %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}
# Syslog Dates: Month Day HH:MM:SS
SYSLOGTIMESTAMP %{MONTH} +%{MONTHDAY} %{TIME}
PROG [\x21-\x5a\x5c\x5e-\x7e]+
SYSLOGPROG %{PROG:program}(?:\[%{POSINT:pid}\])?
SYSLOGHOST %{IPORHOST}
SYSLOGFACILITY <%{NONNEGINT:facility}.%{NONNEGINT:priority}>
HTTPDATE %{MONTHDAY}/%{MONTH}/%{YEAR}:%{TIME} %{INT}
# Shortcuts
QS %{QUOTEDSTRING}
# Log formats
SYSLOGBASE %{SYSLOGTIMESTAMP:timestamp} (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logsource} %{SYSLOGPROG}:
COMMONAPACHELOG %{IPORHOST:clientip} %{HTTPDUSER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-)
COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}
HTTPD20_ERRORLOG \[%{HTTPDERROR_DATE:timestamp}\] \[%{LOGLEVEL:loglevel}\] (?:\[client %{IPORHOST:clientip}\] ){0,1}%{GREEDYDATA:errormsg}
HTTPD24_ERRORLOG \[%{HTTPDERROR_DATE:timestamp}\] \[%{WORD:module}:%{LOGLEVEL:loglevel}\] \[pid %{POSINT:pid}:tid %{NUMBER:tid}\]( \(%{POSINT:proxy_errorcode}\)%{DATA:proxy_errormessage}:)?( \[client %{IPORHOST:client}:%{POSINT:clientport}\])? %{DATA:errorcode}: %{GREEDYDATA:message}
HTTPD_ERRORLOG %{HTTPD20_ERRORLOG}|%{HTTPD24_ERRORLOG}
# Log Levels
LOGLEVEL ([Aa]lert|ALERT|[Tt]race|TRACE|[Dd]ebug|DEBUG|[Nn]otice|NOTICE|[Ii]nfo|INFO|[Ww]arn?(?:ing)?|WARN?(?:ING)?|[Ee]rr?(?:or)?|ERR?(?:OR)?|[Cc]rit?(?:ical)?|CRIT?(?:ICAL)?|[Ff]atal|FATAL|[Ss]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)

3.3.3 grok正则表达式调试

那问题来了，写了正则表达式，如何调试呢？

可以在kibana网页中的dev tools里grok Debugger进行调试

如下例子，在页面中需填写内容：

页面文本框Sample Data（写入需要解析的日志）

2019-12-11 11:04:45 032 orderCallbackJob_Worker-1 INFO com.testerzhang.openability.manager.product.SalesOrderManager.listUnprocessedSalesOrderItem:80 - {"bodyMap":{"params":"{\"triggerTime\":1576033485026,\"batchSize\":10}","size":0},"endTime":"2019-12-11 11:04:45.032","method":"listUnprocessedSalesOrderItem","module":"com.testerzhang.openability.manager.product.SalesOrderManager","result":"fail","startTime":"2019-12-11 11:04:45.031"}

页面文本框Grok Pattern（写入正则表达式）

%{TIMESTAMP_ISO8601:timestamp} %{WORD} %{NOTSPACE} %{LOGLEVEL:loglevel} %{NOTSPACE} - %{GREEDYDATA:content}

页面文本框Structured Data（调试后得出的结果）

{
  "loglevel": "INFO",
  "content": "{\"bodyMap\":{\"params\":\"{\\\"triggerTime\\\":1576033485026,\\\"batchSize\\\":10}\",\"size\":0},\"endTime\":\"2019-12-11 11:04:45.032\",\"method\":\"listUnprocessedSalesOrderItem\",\"module\":\"com.testerzhang.openability.manager.product.SalesOrderManager\",\"result\":\"fail\",\"startTime\":\"2019-12-11 11:04:45.031\"}",
  "timestamp": "2019-12-11 11:04:45"
}

3.3.4 grok格式问题

match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{WORD} %{NOTSPACE} %{LOGLEVEL:loglevel} %{NOTSPACE} - %{GREEDYDATA:temMsg}"}
match => ["message","%{TIMESTAMP_ISO8601:timestamp} %{WORD} %{NOTSPACE} %{LOGLEVEL:loglevel} %{NOTSPACE} - %{GREEDYDATA:temMsg}"]

注意：可以这两种格式，{} 和 => 搭配，[] 和 , 搭配。

3.3.5 Mutate 过滤器

一般grok处理日志格式数据，然后会接着使用Mutate来进一步处理数据。

配置选项：

配置选项	用途
add_field	向事件添加新字段
remove_field	从事件中删除任意字段
add_tag	向事件添加任意标签
remove_tag	从事件中删除标签（如果存在）
convert	将字段值转换为另一种数据类型
id	向现场事件添加唯一的ID
lowercase	将字符串字段转换为其小写形式
replace	用新值替换字段
strip	删除开头和结尾的空格
uppercase	将字符串字段转换为等效的大写字母
update	用新值更新现有字段
rename	重命名事件中的字段
gsub	用于查找和替换字符串中的替换
merge	合并数组或 hash 事件

3.4 例子

$ cat config/test.conf
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    host => "10.10.10.6"
    port => 9600
  }
}

filter {

  grok{
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{WORD} %{NOTSPACE} %{LOGLEVEL:loglevel} %{NOTSPACE} - %{GREEDYDATA:content}"}
      }

  mutate {
    rename => {"content" => "message"}
   }


}

output {

  elasticsearch {
    hosts => ["http://10.10.10.10:9201"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }

  stdout {
    codec => rubydebug
  }
}

4.更多例子

4.1普通过滤

采集日志:

2021-09-13 00:40:21,937 [Quartz_Scheduler_Worker-1] INFO  com.testerzhang.nc.api.checker.StartTaskChecker.exec(StartTaskChecker.java:35) StartTaskChecker|exec|success|"taskSize"="0"

配置规则：

$ cat rule.conf
input {
  beats {
   port => 9800
  }
}

filter {

  grok{
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{WORD} %{NOTSPACE} %{LOGLEVEL:loglevel} %{NOTSPACE} - %{GREEDYDATA:temMsg}"}
      }

  mutate {
    rename => {"temMsg" => "message"}
   }


}

output {

  elasticsearch {
    hosts => ["http://localhost:9201"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }

  stdout {
    codec => rubydebug
  }
}

生成了一个索引名字:filebeat-7.14.1-2021.09.13

4.2 过滤一些不要的关键字

filter {

  if [message] =~ "refreshConnections" or [message] =~ "RetryFailRNChecker" {

    drop { }
  } else if [message] =~ "refreshConnections" {
    # do something
  } else {
    # do something
  }


  ...


}

4.3 设置索引名

根据采集日志的字段fields.log_type来作为动态索引名

output {

  elasticsearch {
    hosts => ["http://localhost:9201"]

    index => "%{[fields][log_type]}-%{+YYYY.MM.dd}"

  }

}

4.4 当符合某个条件，可以单独在插入另外一个index

output {

  elasticsearch {
    hosts => ["http://localhost:9201"]

    index => "%{[fields][log_type]}-%{+YYYY.MM.dd}"

  }

  if [fields][appname] =~ "testerzhang" {
     elasticsearch {
         hosts => ["http://10.10.10.10:9201","http://10.10.10.11:9201","http://10.10.10.12:9201"]
         index => "event-%{[fields][appname]}-%{+YYYY.MM.dd}"
     }
  }

}

本文没有授权给任何组织、企业和个人转载，未经作者允许禁止转载！

欢迎关注我的公众号testerzhang，原创技术文章第一时间推送。

公众号二维码

ELK之Logstash日志处理模块

1.下载部署

1.1下载

1.2 部署

1.3 配置文件配置

1.4 启动 Logstash

2.安装Logstash插件(可选)

2.1查看可用插件

2.2安装新插件

2.3安装成功

3.配置日志规则

3.1 input 配置节

3.2 output 配置节

3.3 filter配置节

3.3.1 input进来的数据匹配规则进行丢弃

3.3.2 grok过滤说明

3.3.3 grok正则表达式调试

3.3.4 grok格式问题

3.3.5 Mutate 过滤器

配置选项：

3.4 例子

4.更多例子

4.1普通过滤

4.2 过滤一些不要的关键字

4.3 设置索引名

4.4 当符合某个条件，可以单独在插入另外一个index

ELK之Logstash日志处理模块

相关文章：