使用Logstash解析日志

在储存你的第一个事件中，你创建了一个基本的Logstash管道来测试你的Logstash设置，在现实情况中，Logstash管道要复杂一些：它通常有一个或多个输入、过滤器和输出插件。

在本节中，你将创建一个Logstash管道，该管道使用Filebeat将Apache web日志作为输入，解析这些日志以从日志中创建特定的、命名的字段，并将解析后的数据写入到Elasticsearch集群中，与其在命令行中定义管道配置，不如在配置文件中定义管道。

首先，到这里下载本示例中使用的示例数据集，然后解压文件。

配置Filebeat，将日志行发送到Logstash

在创建Logstash管道之前，你将配置Filebeat将日志行发送到Logstash，Filebeat客户端是一个轻量级的、资源友好的工具，它从服务器上的文件收集日志，并将这些日志转发给你的Logstash实例进行处理。Filebeat是为可靠性和低延迟而设计的，Filebeat在主机上占用的资源较少，而Beats input插件将Logstash实例上的资源需求最小化。

在一个典型的用例中，Filebeat运行在一台单独的机器上，而不是运行Logstash实例的机器上，为了本教程的目的，Logstash和Filebeat在同一台机器上运行。

默认的Logstash安装包含Beats input插件，Beats input插件允许Logstash从Elastic Beats框架接收事件，也就是说任何Beat written使用Beats框架工作，例如Packetbeat和Metricbeat，也可以将事件数据发送到Logstash。

要在数据源机器上安装Filebeat，请从Filebeat产品页面下载适当的包。你还可以参考Beats文档中的Filebeat入门来获得更多的安装说明。

安装Filebeat后，需要对其进行配置，打开位于Filebeat安装目录中的filebeat.yml文件，并将内容替换为以下几行，确保paths指向Apache日志示例文件logstash-tutorial.log，那是你先前下载的日志：

filebeat.prospectors:
- type: log
  paths:
    - /path/to/file/logstash-tutorial.log 
output.logstash:
  hosts: ["localhost:5044"]

paths => Filebeat处理的文件的绝对路径

保存你的更改。

为了简化配置，你不会像在现实场景中那样指定TLS/SSL设置。

在数据源机器上，使用以下命令运行Filebeat：

sudo ./filebeat -e -c filebeat.yml -d "publish"

如果以root身份运行Filebeat，则需要更改配置文件的所有权（请参阅Beats平台参考中的配置文件所有权和权限）。

Filebeat将尝试在端口5044上连接，在使用活动的Beats插件开始之前，该端口上不会有任何响应，因此你看到的任何关于在该端口上连接失败的消息目前都是正常的。

为Filebeat输入配置Logstash

接下来，你将创建一个Logstash配置管道，使用Beats input插件从Beats接收事件。

下面的文本表示配置管道的结构：

# The # character at the beginning of a line indicates a comment. Use
# comments to describe your configuration.
input {
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
}

这个结构是非功能性的，因为输入和输出部分没有定义任何有效的选项。

为了开始，在你的Logstash主目录中，复制并粘贴配置管道结构到一个文件中，并命名为first-pipeline.conf。

接下来，通过添加以下几行到first-pipeline.conf文件中的input部分来配置你的Logstash实例来使用Beats输入插件：

beats {
   port => "5044"
}

稍后你将配置Logstash以写入到Elasticsearch，现在，你可以向output部分添加以下一行，以便在运行Logstash时将输出打印到stdout：

stdout { codec => rubydebug }

当你完成后，first-pipeline.conf文件内容应该看起来向这样：

input {
    beats {
        port => "5044"
    }
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
    stdout { codec => rubydebug }
}

要验证你的配置，请运行以下命令：

bin/logstash -f first-pipeline.conf --config.test_and_exit

--config.test_and_exit选项解析配置文件并报告任何错误。

如果配置文件通过了配置测试，则使用以下命令启动Logstash：

bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automatic选项允许自动重新加载配置文件，这样你不必在每次修改配置文件时停止并重新启动Logstash。

当Logstash启动时，你可能会看到一个或多个关于Logstash忽略pipelines.yml文件的警告消息，你可以忽略这个警告，pipelines.yml文件用于在一个Logstash实例中运行多个管道，对于这里显示的示例，你正在运行一个管道。

如果你的管道工作正常，你应该会看到一系列事件，如下面写到控制台的：

{
    "@timestamp" => 2017-11-09T01:44:20.071Z,
        "offset" => 325,
      "@version" => "1",
          "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
          "host" => "My-MacBook-Pro.local",
    "prospector" => {
        "type" => "log"
    },
        "source" => "/path/to/file/logstash-tutorial.log",
       "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ]
}
...

使用Grok过滤器插件解析Web日志

现在你有了一个可以从Filebeat读取日志行的工作管道，但是你会注意到日志消息的格式并不理想，你希望解析日志消息，以便从日志中创建特定的、命名的字段，为此你将使用grok过滤器插件。

grok过滤器插件是在Logstash中默认可用的几个插件之一，有关如何管理Logstash插件的详细信息，请参阅插件管理器的参考文档。

grok过滤器插件允许你将非结构化日志数据解析为结构化和可查询的数据。

因为grok过滤器插件在传入的日志数据中寻找模式，配置插件需要你决定关于如何定义对你的用例有意义的模式，web服务器日志示例代表性的行如下：

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

行开头的IP地址很容易识别，方括号中的时间戳也很容易识别，要解析数据，可以使用%{COMBINEDAPACHELOG} grok模式，该模式使用以下模式来构造来自Apache日志的行：

信息	字段名
IP Address	`clientip`
User ID	`ident`
User Authentication	`auth`
timestamp	`timestamp`
HTTP Verb	`verb`
Request body	`request`
HTTP Version	`httpversion`
HTTP Status Code	`response`
Bytes served	`bytes`
Referrer URL	`referrer`
User agent	`agent`

如果你需要帮助构建grok模式，请尝试使用grok调试器，Grok调试器是基本许可证下的X-Pack特性，因此可以免费使用。

编辑first-pipeline.conf文件，用以下文本替换整个filter部分：

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

完成后，first-pipeline.conf的内容应该看起来像这样：

input {
    beats {
        port => "5044"
    }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    stdout { codec => rubydebug }
}

保存你的更改，因为你已经启用了自动配置重载，你不必重新启动Logstash来获取更改，但是，你需要强制Filebeat来从头读取日志文件。要做到这一点，请转到正在运行Filebeat的终端窗口并按Ctrl+C关闭Filebeat，然后删除Filebeat注册表文件，例如，运行：

sudo rm data/registry

由于Filebeat存储它在注册表中获取的每个文件的状态，因此删除注册表文件将迫使Filebeat从头读取它正在获取的所有文件。

接下来，使用以下命令重新启动Filebeat：

sudo ./filebeat -e -c filebeat.yml -d "publish"

如果需要等待Logstash重新加载配置文件，那么在Filebeat开始处理事件之前可能会有一点延迟。

Logstash应用grok模式后，事件将会有以下JSON表示：

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
         "offset" => 325,
           "auth" => "-",
          "ident" => "-",
           "verb" => "GET",
     "prospector" => {
        "type" => "log"
    },
         "source" => "/path/to/file/logstash-tutorial.log",
        "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
     "@timestamp" => 2017-11-09T02:51:12.416Z,
       "response" => "200",
          "bytes" => "203023",
       "clientip" => "83.149.9.216",
       "@version" => "1",
           "beat" => {
            "name" => "My-MacBook-Pro.local",
        "hostname" => "My-MacBook-Pro.local",
         "version" => "6.0.0"
    },
           "host" => "My-MacBook-Pro.local",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000"
}

注意，该事件包含原始消息，但是日志消息也被分解为特定的字段。

使用Geoip过滤器插件增强数据

除了解析日志数据以获得更好的搜索之外，过滤器插件可以从现有数据中获得补充信息，作为一个例子，geoip插件查找IP地址，从地址中获取地理位置信息，并将该位置信息添加到日志中。

通过向first-pipeline.conf文件的filter部分添加以下代码行，配置你的Logstash实例以使用geoip过滤器插件：

geoip {
  source => "clientip"
}

geoip插件配置要求你指定包含要查找的IP地址的源字段的名称来查找，在本例中，clientip字段包含IP地址。

因为过滤器是按顺序计算的，确保geoip部分位于配置文件的grok部分之后，并且grok和geoip部分都嵌套在filter部分中。

当你完成后，first-pipeline.conf的内容看起来想这样：

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    stdout { codec => rubydebug }
}

保存你的更改，如前所述，要强制Filebeat从头读取日志文件，请关闭Filebeat（按Ctrl+C），删除注册表文件，然后使用以下命令重新启动Filebeat：

sudo ./filebeat -e -c filebeat.yml -d "publish"

注意，事件现在包含地理位置信息：

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "geoip" => {
              "timezone" => "Europe/Moscow",
                    "ip" => "83.149.9.216",
              "latitude" => 55.7485,
        "continent_code" => "EU",
             "city_name" => "Moscow",
          "country_name" => "Russia",
         "country_code2" => "RU",
         "country_code3" => "RU",
           "region_name" => "Moscow",
              "location" => {
            "lon" => 37.6184,
            "lat" => 55.7485
        },
           "postal_code" => "101194",
           "region_code" => "MOW",
             "longitude" => 37.6184
    },
    ...

将数据索引到Elasticsearch中

现在web日志被分解成特定的字段，Logstash管道可以将数据索引到一个Elasticsearch集群中。

你可以在自己的硬件上运行Elasticsearch，或者在Elastic Cloud上使用我们托管的Elasticsearch服务，Elasticsearch服务在AWS和GCP上都可用，免费试用Elasticsearch服务。

Logstash管道可以将数据索引到Elasticsearch集群，编辑first-pipeline.conf文件并将整个output部分替换为以下文本：

output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

使用这个配置，Logstash使用http协议连接到Elasticsearch，上面的例子假设在相同的实例上运行了Logstash和Elasticsearch。你可以使用hosts配置指定一个远程的Elasticsearch实例：hosts => [ "es-machine:9092" ]。

此时，你的first-pipeline.conf文件已经正确配置了输入、过滤器和输出部分，并且看起来像这样：

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

保存你的更改，如前所述，要强制Filebeat从头读取日志文件，请关闭Filebeat（按Ctrl+C），删除注册表文件，然后使用以下命令重新启动Filebeat：

sudo ./filebeat -e -c filebeat.yml -d "publish"

测试你的管道

现在已经配置了Logstash管道将数据索引到一个Elasticsearch集群中，你可以查询Elasticsearch。

尝试一个基于grok过滤器插件创建的字段的Elasticsearch测试查询，将$DATE替换为当前日期，使用YYYY.MM.DD格式：

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200'

索引名中使用的日期基于UTC，而不是Logstash运行所在的时区，如果查询返回index_not_found_exception，请确保logstash-$DATE反映了索引的实际名称，要查看可用索引的列表，可以使用这个查询：curl 'localhost:9200/_cat/indices?v'。

你应该得到多份命中返回，例如：

{
  "took": 50,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 98,
    "max_score": 2.793642,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "3IzDnl8BW52sR0fx5wdV",
        "_score": 2.793642,
        "_source": {
          "request": "/presentations/logstash-monitorama-2013/images/frontend-response-codes.png",
          "agent": """"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "geoip": {
            "timezone": "Europe/Moscow",
            "ip": "83.149.9.216",
            "latitude": 55.7485,
            "continent_code": "EU",
            "city_name": "Moscow",
            "country_name": "Russia",
            "country_code2": "RU",
            "country_code3": "RU",
            "region_name": "Moscow",
            "location": {
              "lon": 37.6184,
              "lat": 55.7485
            },
            "postal_code": "101194",
            "region_code": "MOW",
            "longitude": 37.6184
          },
          "offset": 2932,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] "GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1" 200 52878 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"http://semicomplete.com/presentations/logstash-monitorama-2013/"""",
          "@timestamp": "2017-11-09T03:11:35.304Z",
          "response": "200",
          "bytes": "52878",
          "clientip": "83.149.9.216",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:13:45 +0000"
        }
      },
    ...

尝试另一个搜索从IP地址派生的地理信息，将$DATE替换为当前日期，使用YYYY.MM.DD格式：

curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=geoip.city_name=Buffalo'

一些日志条目来自Buffalo，因此查询产生以下响应：

{
  "took": 9,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 2,
    "max_score": 2.6390574,
    "hits": [
      {
        "_index": "logstash-2017.11.09",
        "_type": "doc",
        "_id": "L4zDnl8BW52sR0fx5whY",
        "_score": 2.6390574,
        "_source": {
          "request": "/blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29",
          "agent": """"Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "geoip": {
            "timezone": "America/New_York",
            "ip": "198.46.149.143",
            "latitude": 42.8864,
            "continent_code": "NA",
            "city_name": "Buffalo",
            "country_name": "United States",
            "country_code2": "US",
            "dma_code": 514,
            "country_code3": "US",
            "region_name": "New York",
            "location": {
              "lon": -78.8781,
              "lat": 42.8864
            },
            "postal_code": "14202",
            "region_code": "NY",
            "longitude": -78.8781
          },
          "offset": 22795,
          "auth": "-",
          "ident": "-",
          "verb": "GET",
          "prospector": {
            "type": "log"
          },
          "source": "/path/to/file/logstash-tutorial.log",
          "message": """198.46.149.143 - - [04/Jan/2015:05:29:13 +0000] "GET /blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29 HTTP/1.1" 200 9316 "-" "Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""",
          "tags": [
            "beats_input_codec_plain_applied"
          ],
          "referrer": """"-"""",
          "@timestamp": "2017-11-09T03:11:35.321Z",
          "response": "200",
          "bytes": "9316",
          "clientip": "198.46.149.143",
          "@version": "1",
          "beat": {
            "name": "My-MacBook-Pro.local",
            "hostname": "My-MacBook-Pro.local",
            "version": "6.0.0"
          },
          "host": "My-MacBook-Pro.local",
          "httpversion": "1.1",
          "timestamp": "04/Jan/2015:05:29:13 +0000"
        }
      },
     ...

如果你正在使用Kibana来可视化你的数据，你还可以在Kibana中探索Filebeat数据：

Logstash 参考指南（使用Logstash解析日志）

有关如何为Filebeat加载Kibana索引模式的信息，请参见Filebeat入门文档。

你已经成功地创建了一个管道，它使用Filebeat将Apache web日志作为输入，解析这些日志以创建特定的日志，从日志中命名字段，并将解析后的数据写入到Elasticsearch集群。接下来，你将了解如何创建使用多个输入和输出插件的管道。