« 将数据发送到 Elastic Cloud（托管的 Elasticsearch 服务） 保护您与 Elasticsearch 的连接 »
Elastic 文档 ›Logstash 参考 [8.14] ›创建 Logstash 管道

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"
}
« 将数据发送到 Elastic Cloud（托管的 Elasticsearch 服务） 保护您与 Elasticsearch 的连接 »