使用Logstash解析日志

时间  2019-11-17
标签 使用 logstash 解析 日志 栏目 日志分析 繁體版
原文   原文链接

本文翻译自：Logstash Reference [6.5] » Getting Started with Logstash » Parsing Logs with Logstash，我的水平有限，不正确之处望指正。html

在存储你的第一个事件一节中，咱们建立了一个简单的Logstash管道示例来测试你已安装运行的Logstash。在生产环境中，Logstash管道会很是复杂：通常包含一个或多个输入，过滤器，输出插件。git

在本章节，你将建立一个Logstash管道：Filebeat采集Apache web日志数据做为输入，解析输入的日志，建立特定的字段，而后将解析后的数据写入Elasticsearch集群中。您将在配置文件中进行Logstash管道的配置而不是在命令行中配置。web

接下来，点击这里下载示例数据，解压文件。ubuntu

配置Filebeat发送日志到Logstash

在建立Logstash管道以前，你须要配置Filebeat以将日志发送到Logstash中。Filebeat是轻量级，资源占用较低的日志收集工具，从各服务器上收集日志并将这些日志发送到Logstash实例中处理。Filebeat专为可靠性和低延时而设计。Filebeat只占用不多的主机资源，从而Beat输入插件有效的下降了Logstash实例的资源需求。ruby

在典型应用中，Filebeat与Logstash是部署在不一样的主机上的。这里出于教程简便的目的，将它们部署在同一台主机上。bash

Logstash默认安装已含有Beat input插件。Beat输入插件使Logstash能够从Elastic Beat框架接收事件，这意味者任何编写为与Beat框架一块儿工做的Beat(如Packetbeat和Metricbeat)均可以将事件数据发送到Logstash。服务器

要在数据源主机上安装Filebeat，能够在Filebeat产品页中下载对应的安装包。也能够参阅Filebeat入门获取其余安装说明。微信

安装好Filebeat后，须要对其进行配置。在Filebeat安装目录中找到并打开filebeat.yml文件，替换以下行中的配置项。确保paths项为先前下载的Apache示例日志文件（logstash-tutorial.log）路径。app

filebeat.prospectors:
- type: log
  paths:
    - /path/to/file/logstash-tutorial.log  # 须填写绝对路径
output.logstash:
  hosts: ["localhost:5044"]
复制代码

保存配置。为了简化配置，无需配置生成环境中经常使用TLS/SSL配置。框架

在数据源主机上，输入如下命令运行Filebeat：

sudo ./filebeat -e -c filebeat.yml -d "publish"
复制代码

若是以root身份运行Filebeat，则须要更改配置文件的全部权（请参阅Config File Ownership and Permissions）。

Filebeat将尝试在端口5044上创建链接。在Logstash启动一个激活状态的Beats插件前，该端口上没有任何响应，因此在输入插件处于未激活状态前你看到的在5044端口上创建链接失败的信息都是正常的。

为Logstash配置Filebeat输入

接下来，你要建立一个以Beats输入插件做为输入接收来自Beats的事件的Logstash管道配置。

如下的一段内容为一个管道配置的框架：

# 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文件的输入部分，将Logstash实例配置为使用Beats做为输入插件：

beats {
        port => "5044"
    }
复制代码

后面你会配置Logstash输出到Elasticsearch中。目前，你能够先在输出部分添加下面几行配置使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}模式，用下表的形式结构化Apache日志行：

Information Field Name
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注册表文件registry。 例如，运行：

sudo rm data/registry
复制代码

因为Filebeat在注册表文件中存储了每一个文件被读取后的状态，删除注册表文件将强制Filebea从头开始读取文件。下一步，用下面的命令重启Filebeat:

sudo ./filebeat -e -c filebeat.yml -d "publish"
复制代码

若是Filebeat在开始处理事件以前须要等待Logstash从新加载配置文件，则可能会有点延时。在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地址，从地址中获取地理位置信息，并将该位置信息添加到日志中。

使用geoip插件配置你的Logstash实例，在first-pipeline.conf配置文件的filter部分添加以下行：

geoip {
        source => "clientip"
    }
复制代码

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

因为过滤器是按序处理的，在配置文件中请确保geoip部分在grok部分以后，而且都在filter内部。

当你完成这步后，first-pipeline.conf内容应该以下：

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

保存更改。如以前所作的那样，为了强制Filebeat从头开始读取日志文件，关闭Filebeat(Ctrl+C)，删除register注册表文件，运行如下命令重启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日志数据解析为各个字段，能够将数据输出到Elasticsearch中了。

你能够在本身的主机上运行Elasticsearch，也能够在Elastic Cloud上使用咱们托管的Elasticsearch Service。 AWS和GCP均提供Elasticsearch服务。 免费试用Elasticsearch服务。

Logstash管道能够将数据索引到Elasticsearch集群中。编辑first-pipeline.conf配置文件，将输出部分替换为以下内容：

output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}
复制代码

在此配置中，Logstash使用http协议与Elasticsearch创建链接。上面的示例假定Logstash与Elasticsearch运行在同一台主机中。你能够经过hosts配置项进行相似hosts => [ "es-machine:9092" ]的配置去链接一个远端运行的Elasticsearch实例。

到这里，first-pipeline.conf配置文件中input、filter、output都有了适当的配置，配置内容以下：

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

保存更改。如以前所作的那样，为了强制Filebeat从头开始读取日志文件，关闭Filebeat(Ctrl+C)，删除register注册表文件，运行如下命令重启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-$DATA是正确的索引名。查看当前可用索引列表，可用此进行查询：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读取的数据：

image
可阅读 Filebeat使用文档获取有关Kibana加载Filebeat索引的信息。

你已经成功的建立了一个管道，使用Filebeat读取Apache web日志信息做为输入，解析日志建立指定字段，并将解析后的日志数据输出到Elasticsearch集群中。下一步，你将学习如何建立一个使用多个输入输出插件的管道。

关注微信公众号，按期推送文章！

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程