Logstash 配置示例
编辑Logstash 配置示例编辑
这些示例说明了如何配置 Logstash 来过滤事件、处理 Apache 日志和 syslog 消息,以及使用条件语句来控制哪些事件由过滤器或输出处理。
如果您需要帮助构建 grok 模式,请尝试使用 Grok 调试器。
配置过滤器编辑
过滤器是一种内联处理机制,它提供了灵活的方式来切片和切块您的数据,以满足您的需求。让我们看看一些过滤器在实际中的应用。以下配置文件设置了 grok
和 date
过滤器。
input { stdin { } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }
使用此配置运行 Logstash
bin/logstash -f logstash-filter.conf
现在,将以下行粘贴到您的终端中,然后按 Enter 键,以便它由 stdin 输入处理
127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"
您应该看到类似于以下内容的输出返回到 stdout
{ "message" => "127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] \"GET /xampp/status.php HTTP/1.1\" 200 3891 \"http://cadenza/xampp/navi.php\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"", "@timestamp" => "2013-12-11T08:01:45.000Z", "@version" => "1", "host" => "cadenza", "clientip" => "127.0.0.1", "ident" => "-", "auth" => "-", "timestamp" => "11/Dec/2013:00:01:45 -0800", "verb" => "GET", "request" => "/xampp/status.php", "httpversion" => "1.1", "response" => "200", "bytes" => "3891", "referrer" => "\"http://cadenza/xampp/navi.php\"", "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"" }
如您所见,Logstash(在 grok
过滤器的帮助下)能够解析日志行(它恰好是 Apache 的“组合日志”格式),并将其分解成许多不同的离散信息片段。这在您开始查询和分析我们的日志数据时非常有用。例如,您将能够轻松地对 HTTP 响应代码、IP 地址、引用者等运行报告。Logstash 自带了许多 grok 模式,因此如果您需要解析常见的日志格式,很可能有人已经为您完成了这项工作。有关更多信息,请参阅 GitHub 上的 Logstash grok 模式 列表。
此示例中使用的另一个过滤器是 date
过滤器。此过滤器解析时间戳,并将其用作事件的时间戳(无论您何时摄取日志数据)。您会注意到,此示例中的 @timestamp
字段设置为 2013 年 12 月 11 日,即使 Logstash 在之后某个时间点摄取了该事件。这在回填日志时非常方便。它使您能够告诉 Logstash“将此值用作此事件的时间戳”。
处理 Apache 日志编辑
让我们做一些真正 有用的事情:处理 apache2 访问日志文件!我们将从本地主机上的文件读取输入,并使用 条件语句 根据我们的需要处理事件。首先,创建一个名为 logstash-apache.conf 的文件,其中包含以下内容(您可以根据需要更改日志的文件路径)
input { file { path => "/tmp/access_log" start_position => "beginning" } } filter { if [path] =~ "access" { mutate { replace => { "type" => "apache_access" } } grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } } date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }
然后,创建您在上面配置的输入文件(在本例中为“/tmp/access_log”,其中包含以下日志条目(或使用您自己的 Web 服务器中的某些日志条目)
71.141.244.242 - kurt [18/May/2011:01:48:10 -0700] "GET /admin HTTP/1.1" 301 566 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3" 134.39.72.245 - - [18/May/2011:12:40:18 -0700] "GET /favicon.ico HTTP/1.1" 200 1189 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; InfoPath.2; .NET4.0C; .NET4.0E)" 98.83.179.51 - - [18/May/2011:19:35:08 -0700] "GET /css/main.css HTTP/1.1" 200 1837 "http://www.safesand.com/information.htm" "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1"
现在,使用 -f 标志运行 Logstash 以传入配置文件
bin/logstash -f logstash-apache.conf
现在您应该在 Elasticsearch 中看到您的 apache 日志数据!Logstash 打开并读取了指定的输入文件,处理它遇到的每个事件。添加到此文件的任何其他行也将被捕获,由 Logstash 作为事件处理,并存储在 Elasticsearch 中。作为额外的好处,它们被存储在带有设置为“apache_access”的“type”字段中(这是由输入配置中的 type ⇒ “apache_access”行完成的)。
在此配置中,Logstash 仅监视 apache access_log,但通过更改上述配置中的一行,可以轻松地监视 access_log 和 error_log(实际上,任何以“log”结尾的文件),
input { file { path => "/tmp/*_log" ...
当您重新启动 Logstash 时,它将处理 error 和 access 日志。但是,如果您检查您的数据(使用 elasticsearch-kopf,也许),您会看到 access_log 被分解成离散的字段,但 error_log 没有。这是因为我们使用了 grok
过滤器来匹配标准的组合 apache 日志格式,并自动将数据拆分成单独的字段。如果我们可以 根据其格式控制行的解析方式,那不是很好吗?好吧,我们可以……
请注意,Logstash 没有重新处理已经在 access_log 文件中看到的事件。从文件读取时,Logstash 会保存其位置,并且仅处理添加的新行。很酷吧!
使用条件语句编辑
您可以使用条件语句来控制哪些事件由过滤器或输出处理。例如,您可以根据事件出现在哪个文件(access_log、error_log 和其他以“log”结尾的随机文件)中对每个事件进行标记。
input { file { path => "/tmp/*_log" } } filter { if [path] =~ "access" { mutate { replace => { type => "apache_access" } } grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } date { match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ] } } else if [path] =~ "error" { mutate { replace => { type => "apache_error" } } } else { mutate { replace => { type => "random_logs" } } } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }
此示例使用 type
字段标记所有事件,但实际上并没有解析 error
或 random
文件。错误日志的类型很多,因此如何标记它们实际上取决于您正在处理的日志。
类似地,您可以使用条件语句将事件定向到特定的输出。例如,您可以
- 向 nagios 发送任何状态为 5xx 的 apache 事件的警报
- 将任何 4xx 状态记录到 Elasticsearch
- 通过 statsd 记录所有状态代码命中
要告诉 nagios 任何状态代码为 5xx 的 http 事件,您首先需要检查 type
字段的值。如果它是 apache,那么您可以检查 status
字段是否包含 5xx 错误。如果是,则将其发送到 nagios。如果不是 5xx 错误,则检查 status
字段是否包含 4xx 错误。如果是,则将其发送到 Elasticsearch。最后,无论 status
字段包含什么,都将所有 apache 状态代码发送到 statsd
output { if [type] == "apache" { if [status] =~ /^5\d\d/ { nagios { ... } } else if [status] =~ /^4\d\d/ { elasticsearch { ... } } statsd { increment => "apache.%{status}" } } }
处理 Syslog 消息编辑
Syslog 是 Logstash 最常见的用例之一,也是它处理得非常好的用例(只要日志行大致符合 RFC3164)。Syslog 是事实上的 UNIX 网络日志记录标准,它将消息从客户端机器发送到本地文件,或通过 rsyslog 发送到集中式日志服务器。对于此示例,您不需要运行的 syslog 实例;我们将从命令行模拟它,以便您可以了解发生了什么。
首先,让我们为 Logstash + syslog 创建一个简单的配置文件,名为 logstash-syslog.conf。
input { tcp { port => 5000 type => syslog } udp { port => 5000 type => syslog } } filter { if [type] == "syslog" { grok { match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" } add_field => [ "received_at", "%{@timestamp}" ] add_field => [ "received_from", "%{host}" ] } date { match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ] } } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }
使用此新配置运行 Logstash
bin/logstash -f logstash-syslog.conf
通常,客户端机器会连接到 Logstash 实例的 5000 端口并发送其消息。对于此示例,我们只需 telnet 到 Logstash 并输入日志行(类似于我们之前将日志行输入 STDIN 的方式)。打开另一个 shell 窗口以与 Logstash syslog 输入进行交互,并输入以下命令
telnet localhost 5000
复制并粘贴以下行作为示例。(随意尝试一些您自己的行,但请记住,如果 grok
过滤器不适合您的数据,它们可能无法解析)。
Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154] Dec 23 14:42:56 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log) Dec 22 18:28:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'.
现在您应该在原始 shell 中看到 Logstash 的输出,因为它处理和解析消息!
{ "message" => "Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)", "@timestamp" => "2013-12-23T22:30:01.000Z", "@version" => "1", "type" => "syslog", "host" => "0:0:0:0:0:0:0:1:52617", "syslog_timestamp" => "Dec 23 14:30:01", "syslog_hostname" => "louis", "syslog_program" => "CRON", "syslog_pid" => "619", "syslog_message" => "(www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)", "received_at" => "2013-12-23 22:49:22 UTC", "received_from" => "0:0:0:0:0:0:0:1:52617", "syslog_severity_code" => 5, "syslog_facility_code" => 1, "syslog_facility" => "user-level", "syslog_severity" => "notice" }