Elastic Stack学习--logstash入门

    logstash是基于实时管道的数据收集引擎。它像一根处理数据的管道，收集分散的数据，进行汇总处理后输出给下游进行数据分析和展现。

    logstash可以配合Beats组件或者其它第三方组件进行数据收集，数据经过处理后存放到elasticsearch中进行检索和分析。

    除了对接beats以外，logstash有着丰富的组件，能够支持各种数据源接入。包括jdbc、kafka、http等。

    logstash是基于pipeline方式进行数据处理的，pipeline可以理解为数据处理流程的抽象。在一条pipeline数据经过上游数据源汇总到消息队列中，然后由多个工作线程进行数据的转换处理，最后输出到下游组件。一个logstash中可以包含多个pipeline。

基本概念

• pipeline：一条数据处理流程的逻辑抽象，类似于一条管道，数据从一端流入，经过处理后，从另一端流出；一个pipeline包括输入、过滤、输出3个部分，其中输入和输出部分是必选组件，过滤是可选组件；
• instance：一个logstash实例，可以包含多条数据处理流程，即多个pipeline；
• inputs：数据输入组件，用于对接各种数据源，接入数据，支持解码器，允许对数据进行编码解码操作；必选组件；
• filters：数据过滤组件，负责对输入数据进行加工处理；可选组件；
• outputs：数据输出组件，用于对接下游组件，发送处理后的数据，支持解码器，允许对数据进行编码解码操作；必选组件；
• event：pipeline中的数据都是基于事件的，一个event可以看作是数据流中的一条数据或者一条消息；

安装logstash

1）依赖java8，且不支持java9：检查java版本并配置JAVA_HOME环境变量。logstash基于jruby开发，logstash 6.x版本要求运行在java8环境，且目前不支持java9;
2）下载并解压：下载logstash，解压文件；注意logstash所在路径中不可以包含冒号；

tar -zxvf logstash-6.2.2.tar.gz
cd logstash-6.2.2

3）启动logstash，发布第一个事件：通过-e指定一个pipeline的处理流程，指定从stdin中读取event，然后在stdout输出；

bin/logstash -e 'input { stdin { } } output { stdout {} }'

可以看到类似如下日志：

Sending Logstash's logs to /home/work/zion_package/elastic/logstash/logstash-6.2.2/logs which is now configured via log4j2.properties
h[2018-03-14T14:48:00,053][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/home/work/zion_package/elastic/logstash/logstash-6.2.2/modules/fb_apache/configuration"}
[2018-03-14T14:48:00,074][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/home/work/zion_package/elastic/logstash/logstash-6.2.2/modules/netflow/configuration"}
[2018-03-14T14:48:00,172][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.queue", :path=>"/home/work/zion_package/elastic/logstash/logstash-6.2.2/data/queue"}
[2018-03-14T14:48:00,178][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.dead_letter_queue", :path=>"/home/work/zion_package/elastic/logstash/logstash-6.2.2/data/dead_letter_queue"}
[2018-03-14T14:48:00,631][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2018-03-14T14:48:00,672][INFO ][logstash.agent           ] No persistent UUID file found. Generating new UUID {:uuid=>"5901ab9f-fdc9-43dc-a88d-c5c636cf8224", :path=>"/home/work/zion_package/elastic/logstash/logstash-6.2.2/data/uuid"}
[2018-03-14T14:48:01,337][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.2.2"}
[2018-03-14T14:48:01,733][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-03-14T14:48:03,264][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>12, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2018-03-14T14:48:03,453][INFO ][logstash.pipeline        ] Pipeline started succesfully {:pipeline_id=>"main", :thread=>"#<Thread:0x128244e5 run>"}
The stdin plugin is now waiting for input:
[2018-03-14T14:48:03,546][INFO ][logstash.agent           ] Pipelines running {:count=>1, :pipelines=>["main"]}

在控制台随便输入字符串按回车后，发现消息会立刻在控制台输出，此时pipeline的处理过程是从stdin接收我们的输入，然后再在stdout输出：

Hello, this is my first event to logstash !
2018-03-14T06:50:26.261Z yf-beidou-dmp00.yf01.baidu.com Hello, this is my first event to logstash !

logstash会给消息添加ip和时间戳，要退出logstash，输入ctrl+d；

通过Filebeat发送日志到logstash

配置Filebeat

    Filebeat用于追踪服务器上的文件数据，它的设计以可靠和低资源占用为初衷，和业务系统部署在一起时，能够避免因为资源消耗影响到业务系统的正常运行。logstash安装后默认包含Beats input组件，用于接收各种beat组件上报事件。Filebeat也可以直接上报事件给elasticsearch，而不用经过logstash。

1）下载文件样例logstash-tutorial.log.gz，上传至Filebeat将要部署的服务器并解压；该样例为官方文档提供的apache的web日志样例。
2）下载Filebeat，上传服务器并解压；此处下载LINUX 64-BIT的安装包；

tar -zxvf filebeat-6.2.2-linux-x86_64.tar.gz
cd filebeat-6.2.2-linux-x86_64

3）配置Filebeat，修改filebeat.yml，设置监控日志文件路径和上报logstash地址；

filebeat.prospectors:
- type: log
  # 监控日志文件路径
  paths:
    - /path/to/file/logstash-tutorial.log 
output.logstash:
  # 上报logstash地址
  hosts: ["localhost:5044"]

注：设置监控日志文件的路径一定是绝对路径，支持通配符；

4）启动Filebeat；其中-e参数指定输出日志到stderr，而非输出到日志文件；
-c参数指定配置文件路径；-d参数debug选择器；

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

配置logstash

1）创建pipeline配置文件；pipeline配置格式如下：

# 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 {
}

创建一个名为first-pipeline.conf的文件，配置如下：

input {
    beats {
        # 设置beats上报端口
        port => "5044"
    }
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
    # 输出到stdout，同时指定日志解码器为rubydebug
    stdout { codec => rubydebug }
}

2）校验配置是否正确,命令如下。--config.test_and_exit选项会校验配置文件，并输出错误；

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

输出类似如下日志，说明配置文件格式校验通过：

Configuration OK
[2018-03-14T16:23:08,919][INFO ][logstash.runner          ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

3）启动logstash，命令如下。--config.reload.automatic选项能够使得配置文件修改后被自动加载，从而避免重新启动logstash；

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

如果配置正确，可以看到logstash命令行有类似如下输出：

{
          "host" => "yf-beidou-dmp00.yf01.baidu.com",
        "offset" => 22310,
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "message" => "218.30.103.62 - - [04/Jan/2015:05:28:43 +0000] \"GET /blog/geekery/xvfb-firefox.html HTTP/1.1\" 200 10975 \"-\" \"Sogou web spider/4.0(+http://www.sogou.com/docs/help/webmasters.htm#07)\"",
    "prospector" => {
        "type" => "log"
    },
          "beat" => {
            "name" => "yf-beidou-dmp00.yf01.baidu.com",
         "version" => "6.2.2",
        "hostname" => "yf-beidou-dmp00.yf01.baidu.com"
    },
    "@timestamp" => 2018-03-14T08:43:52.564Z,
        "source" => "/home/work/zion_package/elastic/filebeat/logstash-tutorial.log",
      "@version" => "1"
}

通过Grok过滤组件解析日志

通过上面的例子，我们可以将filebeat上报的日志经由logstash输出，接下来将添加filter组件，对日志进行处理。grok组件是logstash的filter组件之一，它可以将非结构化的数据按照一定规则整理成结构化数据，从而便于检索。这些规则需要根据日志格式事先设定好，因此需要了解采集日志的格式。因为我们的样例日志是apache的web日志，因而可以直接使用grok提供的%{COMBINEDAPACHELOG}格式进行日志的过滤，过滤后的日志格式如下：

修改first-pipeline.conf文件，增加filter，如下：

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

因为启动logstash时添加了--config.reload.automatic选项，logstash能够自动加载修改后的配置文件，因而不需要重启；修改保存后可以看到重新加载配置，重启pipeline的日志：

[2018-03-14T17:04:57,325][INFO ][logstash.pipelineaction.reload] Reloading pipeline {"pipeline.id"=>:main}
[2018-03-14T17:05:01,758][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"#<Thread:0x58f295b9 run>"}
[2018-03-14T17:05:02,048][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>12, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2018-03-14T17:05:02,362][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"}
[2018-03-14T17:05:02,422][INFO ][logstash.pipeline        ] Pipeline started succesfully {:pipeline_id=>"main", :thread=>"#<Thread:0x5506295b sleep>"}
[2018-03-14T17:05:02,429][INFO ][org.logstash.beats.Server] Starting server on port: 5044
[2018-03-14T17:05:02,453][INFO ][logstash.agent           ] Pipelines running {:count=>1, :pipelines=>["main"]}

为了能够使filebeat重新读取文件，需要停止filebeat，删除读取文件的保存点记录，并重启filebeat：

cd filebeat-6.2.2-linux-x86_64
rm data/registry

添加filter后，输出的日志格式如下，发现不仅输出了日志原文，同时对日志进行解析切割，存放到相应的字段中。

{
           "host" => "yf-beidou-dmp00.yf01.baidu.com",
       "clientip" => "121.107.188.202",
           "verb" => "GET",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
        "message" => "121.107.188.202 - - [04/Jan/2015:05:27:57 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-dashboard3.png HTTP/1.1\" 200 171717 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.107 Safari/537.36\"",
           "beat" => {
            "name" => "yf-beidou-dmp00.yf01.baidu.com",
         "version" => "6.2.2",
        "hostname" => "yf-beidou-dmp00.yf01.baidu.com"
    },
    "httpversion" => "1.1",
           "auth" => "-",
       "response" => "200",
          "bytes" => "171717",
          "ident" => "-",
       "@version" => "1",
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-dashboard3.png",
         "offset" => 21927,
     "prospector" => {
        "type" => "log"
    },
     "@timestamp" => 2018-03-14T09:06:47.927Z,
         "source" => "/home/work/zion_package/elastic/filebeat/logstash-tutorial.log",
      "timestamp" => "04/Jan/2015:05:27:57 +0000",
          "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.107 Safari/537.36\"",
       "referrer" => "\"-\""
}

使用Geoip过滤器组件增强数据处理

geoip也是filter组件的一种，用于从ip地址中解析出位置信息，并添加到输出日志中；geoip需要配置需要指定存放ip地址的字段，在本例中，我们使用通过gork解析后clientip字段；因为过滤器是按顺序过滤，所以需要确保geoip的过滤器在之前配置的gork过滤器之后，配置文件如下：

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        # 指定要进行ip解析的字段
        source => "clientip"
    }
}
output {
    stdout { codec => rubydebug }
}

停止filebeat，删除data/registry文件，重启filebeats后再次查看日志，发现新增了地理位置信息：

{
           "host" => "yf-beidou-dmp00.yf01.baidu.com",
       "clientip" => "218.30.103.62",
           "verb" => "GET",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
        "message" => "218.30.103.62 - - [04/Jan/2015:05:28:43 +0000] \"GET /blog/geekery/xvfb-firefox.html HTTP/1.1\" 200 10975 \"-\" \"Sogou web spider/4.0(+http://www.sogou.com/docs/help/webmasters.htm#07)\"",
           "beat" => {
            "name" => "yf-beidou-dmp00.yf01.baidu.com",
         "version" => "6.2.2",
        "hostname" => "yf-beidou-dmp00.yf01.baidu.com"
    },
    "httpversion" => "1.1",
           "auth" => "-",
       "response" => "200",
          "bytes" => "10975",
          "ident" => "-",
       "@version" => "1",
        "request" => "/blog/geekery/xvfb-firefox.html",
          "geoip" => {
              "location" => {
            "lat" => 39.9289,
            "lon" => 116.3883
        },
              "latitude" => 39.9289,
        "continent_code" => "AS",
           "region_code" => "11",
         "country_code3" => "CN",
         "country_code2" => "CN",
             "longitude" => 116.3883,
             "city_name" => "Beijing",
          "country_name" => "China",
                    "ip" => "218.30.103.62",
           "region_name" => "Beijing",
              "timezone" => "Asia/Shanghai"
    },
         "offset" => 22310,
     "prospector" => {
        "type" => "log"
    },
     "@timestamp" => 2018-03-14T09:22:31.299Z,
         "source" => "/home/work/zion_package/elastic/filebeat/logstash-tutorial.log",
      "timestamp" => "04/Jan/2015:05:28:43 +0000",
          "agent" => "\"Sogou web spider/4.0(+http://www.sogou.com/docs/help/webmasters.htm#07)\"",
       "referrer" => "\"-\""
}

将logstash数据输出到elasticsearch

1）修改output，写入数据到elasticsearch：修改first-pipeline.conf文件，配置elasticsearch访问地址；

input {
    beats {
        port => "5044"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    elasticsearch {
        # 指定elasticsearch地址，指定多个地址，logstash会自动负载均衡
        hosts => [ "ip1:port1", "ip2,port2" ]
        # 如果设置用户名和密码，则需要指定如下两个字段；
        # 用户必须具有对index的CRUD权限；
        user => "username"
        password => "password"
    }
}

2）重新发送消息：停止filebeat，删除data/registry文件，重启filebeats；
3）检索日志：执行如下语句，查询是否有日志写入elasticsearch，ip和端口是elasticsearch实例的ip和端口；因为first-pipeline.conf中未指定创建index的名称格式，默认为：logstash-yyyy.MM.dd（日期部分需要替换）；如果未指定用户名/密码，则不需要-u参数；

curl -XGET 'ip:port/logstash-2018.03.14/_search?pretty&q=response=200' -u username:password

检索结果类似如下json串：

{
  "took" : 20,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 98,
    "max_score" : 2.3988724,
    "hits" : [
      {
        "_index" : "logstash-2018.03.14",
        "_type" : "doc",
        "_id" : "t76VJGIBe9U4s2F_OL_W",
        "_score" : 2.3988724,
        "_source" : {
          "verb" : "GET",
          "@timestamp" : "2018-03-14T12:56:19.779Z",
          "response" : "200",
          "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\"",
          "source" : "/home/work/zion_package/elastic/filebeat/logstash-tutorial.log",
          "host" : "yf-beidou-dmp00.yf01.baidu.com",
          "auth" : "-",
          "clientip" : "83.149.9.216",
          "geoip" : {
            "country_code3" : "RU",
            "continent_code" : "EU",
            "ip" : "83.149.9.216",
           ... ...

4）在kibana上配置index-pattern，通过Discover检索日志：
点击菜单：Management -> Index Pattern；

点击 Create Index Pattern，创建index pattern；

点击菜单：Discover，检索日志；

参考

官方视频：Getting Started with Logstash
官方文档：Getting Started with Logstash