logstash 5.1.1 学习

时间 2019-11-18

原文原文链接

Logstash 5.1.1 安装配置

安装和配置

java 环境：

[root@ ~]# tar xf jdk-8u71-linux-x64.tar.gz 
[root@ ~]# mv jdk1.8.0_71 /usr/local/java_1.8.0
[root@ ~]# ln -s /usr/local/java_1.8.0/bin/java /usr/bin/java
[root@ ~]# vim /etc/profile
# 添加:
export JAVA_HOME=/usr/local/java_1.8.0
export CLASSPATH=$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/jre/lib
export PATH=$PATH:$JAVA_HOME/bin
[root@ ~]# source /etc/profile

安装logstash

直接使用rpm包安装：logstash-5.1.1.rpmhtml

[root@ ~]# rpm -ivh logstash-5.1.1.rpm

初始配置logstash

logstash安装后的所在目录为：/usr/share/logstash/，这里为了方便统一的管理，作规范的配置java

# 统一配置目录，/etc/logstash在安装logstash后就存在了配置文件，启动后又去/usr/share/logstash/config找，why？
[root@ ~]# ln -s /etc/logstash /usr/share/logstash/config
[root@ ~]# ln -s /usr/share/logstash/bin/* /usr/local/bin/
# 调整jvm使用内存
[root@ ~]# vim /etc/logstash/jvm.options 
 -Xms128m
 -Xmx256m
# 修改logstash基本配置
[root@ ~]# vim /etc/logstash/logstash.yml 
pipeline:
  workers: 4
  batch:
    size: 125
    delay: 5
path.config: /etc/logstash/conf.d
path.logs: /data/logs/logstash
http.port: 9600
http.host: "192.168.31.140"

hello world

作个简单的测试node

[root@ ~]# logstash  -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'
Sending Logstash's logs to /data/logs/logstash which is now configured via log4j2.properties
The stdin plugin is now waiting for input:
hello world
{
    "@timestamp" => 2017-06-21T06:55:00.471Z,
      "@version" => "1",
          "host" => "baseos-1",
       "message" => "hello world",
          "tags" => []
}

以上的例子从标准输入中获取数据，再从标准输出中输出。mysql

在输出的结果中，"@timestamp"标记事件发生的时间点，"host"标记事件发生的主机，"tags"标记事件的某个方面的属性（这是一个数组，能够有多个属性），"type"标记事件的惟一类型。linux

Es 统一采用UTC时间存成长整型数据，会比北京时间晚八个小时。nginx

命令行参数：

-e：直接执行后接的配置字符串，能够为空。空字符串''就默认是'input{stdin{}}output{stdout{codec=>rubydebug}}'。
-f：后接配置文件，能够将复杂的配置写到配置文件中。
-t：测试，用来测试后接的配置文件语法是否正确。写法：logstash -t -f conf_file_path
-l：后接日志文件，默认日志为配置文件logstash.yml中指定的日志文件。

Logstash 数据类型

array

数组能够是单个或者多个字符串值。git

path => [ "/var/log/messages", "/var/log/*.log" ]
path => "/data/mysql/mysql.log"

若是指定了屡次，追加数组。此实例path数组包含三个字符串元素。github

boolean

布尔值必须是TRUE或者false。true和false不能有引号。redis

ssl_enable => true

bytes

指定字节单位。支持的单位有SI (k M G T P E Z Y) 和 Binary (Ki Mi Gi Ti Pi Ei Zi Yi)。Binary单位基于1024，SI单位基于1000。不区分大小写和忽略值与单位之间的空格。若是没有指定单位，默认是byte。sql

my_bytes => "1113" # 1113 bytes
my_bytes => "10MiB" # 10485760 bytes
my_bytes => "100kib" # 102400 bytes
my_bytes => "180 mb" # 180000000 bytes

Codec

logstash编码名称用来表示数据编码。用于input和output段。便于数据的处理。若是input和output使用合适的编码，就无需单独的filter对数据进行处理。

codec => "json"

hash

键值对，注意多个键值对用空格分隔，而不是逗号。

match => {
"field1" => "value1"
"field2" => "value2"
... }

number

必须是有效的数值，浮点数或者整数。
port => 33

password

一个单独的字符串。
my_password => "password"

path

一个表明有效的操做系统路径。
my_path => "/tmp/logstash"

string

name => "Hello world"
name => 'It\'s a beautiful day'

Logstash 插件学习

Logstash必需要有input和output。

input插件

这里只介绍 stdin、file 和beats 插件

标准输入：stdin

一个标准输入配置实例：

input {
    stdin {
        add_field => {"key" => "value"}
        codec => "plain"
        tags  => ["add"]
        type => "stdin"
    }
}

输出结果：

The stdin plugin is now waiting for input:
hello
{
    "@timestamp" => 2017-06-21T07:34:57.899Z,
      "@version" => "1",
          "host" => "baseos-1",
       "message" => "hello",
          "type" => "stdin",
           "key" => "value",
          "tags" => [
        [0] "add"
    ]
}

参数:

add_field：用于向Event中添加字段，hash类型，默认值：{} 。
codec：输入数据以后对数据进行解码，常见的就是在output指定输出结果为rubydebug 格式,string类型，默认值："line"。
tags：能够在Event中增长标签，以便于在后续的处理流程中使用，array类型。
type：若是采用elasticsearch作store，在默认状况下将做为elasticsearch的type，string类型。type字段做为事件自己的一部分存储，在kibana中经过type字段进行索引。

读取文件：file

文件读取插件主要用来抓取文件的变化信息，将变化信息封装成Event进程处理或者传递。

Logstash使用一个叫FileWatch的 Ruby Gem 库来监听文件变化。这个库支持glob展开文件路劲(只支持绝对路径，不会自动递归目录)，并且会记录一个隐藏的数据文件来跟踪被监听的日志文件的当前读取位置（默认这个文件叫.sincedb）。该数据库文件记录了每一个被监听文件的inode、major number、minor number 和 pos。

一个读取文件配置实例：

input {
    file {
        path => ["/data/logs/nginx/*.log"]
        exclude => ["/data/logs/nginx/nginx.pid","/data/logs/nginx/error.log"]
        discover_interval => 5
        sincedb_path => "/data/database/logstash/.sincedb_nginx_log"
        sincedb_write_interval => 10
        type => "nginx_log"
        start_position => "beginning"
    }
}

参数：

discover_interval：logstash每隔多久去检查一次被监听的path路径下是否有新文件，默认15秒。
exclude：不想监听的文件路径。
close_older：一个已经被监听的文件，若是超过这个时间没有新内容更新，则关闭这个文件的句柄，默认3600秒。
ignore_older：每次检查文件列表时，若是一个文件的最后修改时间超过这个值，就忽略这个文件，默认86400秒，一天。
sincedb_path：sincedb文件的位置，默认$HOME/.sincedb。
sincedb_write_interval：多久写一次sincedb文件，默认15秒。
star_interval：每隔多久检查一次被监听文件状态是否有更新，默认15秒。
start_position：从什么位置开始读取文件数据，默认是结束位置。“beginning”从头开始读取。该选项仅对第一次被监听的文件起做用，若是sincedb文件已经记录了这个文件了，那么logstash依然会从记录过的pos开始读取，能够删除sincedb文件来让logstash从头开始读取被监听文件。

详细参考：https://www.elastic.co/guide/en/logstash/5.0/plugins-inputs-file.html

获取Nginx日志,并输出到标准输出

[root@ ~]# vim /etc/logstash/conf.d/nginx_access.conf 
input {
    file {
        path => ["/data/logs/nginx/*.log"]
        exclude => ["/data/logs/nginx/nginx.pid","/data/logs/nginx/error.log"]
        discover_interval => 5
        sincedb_path => "/data/database/logstash/.sincedb_nginx_log"
        sincedb_write_interval => 10
        type => "nginx_log"
        start_position => "beginning"
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

测试效果：

[root@baseos-1_192.168.31.140 ~]# logstash -f /etc/logstash/conf.d/nginx_access.conf 
Sending Logstash's logs to /data/logs/logstash which is now configured via log4j2.properties
{
          "path" => "/data/logs/nginx/logstash.wangshenjin.com_access.log",
    "@timestamp" => 2017-06-21T07:10:36.270Z,
      "@version" => "1",
          "host" => "baseos-1",
       "message" => "192.168.31.140 - - [21/Jun/2017:15:10:36 +0800] \"GET / HTTP/1.1\" 200 33 \"-\" \"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.27.1 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\" \"-\"",
          "type" => "nginx_log",
          "tags" => []
}

beats插件

Beats插件用于创建监听服务，接收Filebeat或者其余beat发送的Events

filebeat:

filebeat 是基于原先 logstash-forwarder 的源码改造出来的。换句话说：filebeat 就是新版的 logstash-forwarder，也会是 Elastic Stack 在 shipper 端的第一选择。

filebeat 配置：

filebeat:
  prospectors:
    -
      paths:
        - /usr/local/nginx/logs/*.com.log
      input_type: log
      document_type: nginx-access
      tail_files: true
output:
  logstash:
    hosts: ["192.168.31.140:5044"]
shipper:
  tags: ["nginx_log"]

logstash 配置：

input {
    beats {
        port => 5044
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

效果：

[root@ conf.d]# logstash -f filebeat.conf           
Sending Logstash's logs to /data/logs/logstash which is now configured via log4j2.properties
{
    "@timestamp" => 2017-06-21T10:49:12.596Z,
        "offset" => 605,
      "@version" => "1",
    "input_type" => "log",
          "beat" => {
        "hostname" => "salt-master",
            "name" => "salt-master",
         "version" => "5.1.1"
    },
          "host" => "salt-master",
        "source" => "/usr/local/nginx/logs/access.log",
       "message" => "192.168.31.1 - - [24/Feb/2017:17:00:59 +0800] \"GET / HTTP/1.1\" 301 184 \"-\" \"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36\"",
          "type" => "nginx-access",
          "tags" => [
        [0] "beats_input_codec_plain_applied"
    ]
}

参数：

port：监听的端口，数值类型，必填字段。
host：监听的IP，默认值为"0.0.0.0"，string类型，非必填。
client_inactivity_timeout：多长时间关闭链接，默认值60s，数值类型，非必填。
ssl：是否启动SSL,默认值false,布尔类型，非必填。
ssl_certificate：证书路径，string类型，非必填。
- 若是logstash使用IP访问，证书里的subjectAltName字段须要包含logstash的IP，不然报错：
- x509: cannot validate certificate for 192.168.1.241 because it doesn't contain any IP SANs
ssl_key：SSL秘钥，string类型，非必填。
ssl_key_passphrase：SSL秘钥密码，string类型，非必填。
ssl_verify_mode：是否定证客户端，可选值“none”、”peer”、”force_peer”，默认none，非必填。

详细参考：https://www.elastic.co/guide/en/logstash/5.0/plugins-inputs-beats.html

beats SSL：

这里采用 easyrsa 生成logstash证书。

若是logstash使用IP访问，证书里的subjectAltName字段须要包含logstash的IP。easyrsa签署证书时，能够采用--subject-alt-name来添加：

./easyrsa  gen-req logstash_server nopass  # 生成证书签署文件
./easyrsa --subject-alt-name="IP:192.168.31.140" sign server logstash_server # 签署证书

Filter插件

grok

Logstash使用grok模块对任意文本解析并结构化输出，Logstash默认带有120中匹配模式。

grok的语法格式为 %{SYNTAX:SEMANTIC}，前面是grok-pattrens中定义的变量，后面能够自定义变量的名称，　若是有双引号""或者中括号[]，须要加进行转义。

SYNTAX 是要匹配的模式，例如3.14匹配 NUMBER 模式，127.0.0.1 匹配 IP 模式。
SEMANTIC 是匹配到的文本片断的标识，例如 “3.14” 能够是一个时间的持续时间，因此能够简单地叫作"duration" ，字符串"55.3.244.1"能够被标识为“client”。

因此，grok过滤器表达式能够写成： %{NUMBER:duration} %{IP:client}

默认状况下，全部的SEMANTIC是以字符串的方式保存，若是想要转换一个SEMANTIC的数据类型，例如转换一个字符串为整形，能够写成以下的方式： %{NUMBER:num:int}

一个例子，示例日志以下所示：

55.3.244.1 GET /index.html 15824 0.043

filter 配置以下所示：

filter {
    grok {
        match => {
            "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
        }
    }
}

效果以下所示：

55.3.244.1 GET /index.html 15824 0.043
{
      "duration" => "0.043",
       "request" => "/index.html",
    "@timestamp" => 2017-06-22T01:49:27.773Z,
        "method" => "GET",
         "bytes" => "15824",
      "@version" => "1",
          "host" => "baseos-1",
        "client" => "55.3.244.1",
       "message" => "55.3.244.1 GET /index.html 15824 0.043",
          "tags" => []
}

一个nginx日志例子：

nginx日志格式：

'$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for"';

filter 配置以下所示：

filter {
    if [type] == "nginx-access-log" {
        grok {
            match => {
                "message" => "%{IPORHOST:remote_addr} - %{USERNAME:remote_user} \[%{HTTPDATE:[@metadata][timestamp]}\] \"%{DATA:request}\" %{INT:status} %{INT
:body_bytes_sent} \"%{DATA:http_referer}\" \"%{DATA:http_user_agent}\" \"%{USERNAME:http_x_forwarded_for}\""
            }
        }
    }
}

效果以下所示：

Sending Logstash's logs to /data/logs/logstash which is now configured via log4j2.properties
{
             "remote_addr" => "192.168.31.130",
                 "request" => "GET / HTTP/1.1",
         "body_bytes_sent" => "33",
                 "message" => "192.168.31.130 - - [22/Jun/2017:13:50:33 +0800] \"GET / HTTP/1.1\" 200 33 \"-\" \"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\" \"-\"",
                    "type" => "nginx-access-log",
                    "tags" => [],
         "http_user_agent" => "curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2",
             "remote_user" => "-",
                    "path" => "/data/logs/nginx/logstash.wangshenjin.com_access.log",
              "@timestamp" => 2017-06-22T05:50:34.860Z,
            "http_referer" => "-",
                "@version" => "1",
                    "host" => "baseos-1",
    "http_x_forwarded_for" => "-",
                  "status" => "200"
}

message是每段读进来的日志，IPORHOST、USERNAME、HTTPDATE等都是patterns/grok-patterns中定义好的正则格式名称，对照日志进行编写。

logstash 默认自带120种正则格式，参考：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

自定义正则

不少时候，日志格式都是自定义的，这时候咱们须要根据实际状况自定义正则。

这里以 remote_addr、request为例子，定义三个正则：

IPADDR [0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}
REQUESTPATH (?:/[\\A-Za-z0-9$.+!*'(){},~:;=@#% \[\]_<>^\-&?]*)+
REQUESTPRO ([^"]*)

使用上面的自定义正则：

IPADDR [0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}
REQUESTPRO ([^"]*)
REQUESTPATH (?:/[\\A-Za-z0-9$.+!*'(){},~:;=@#% \[\]_<>^\-&?]*)+

NGXACCESSLOG %{IPADDR:client_ip} - (%{USERNAME:user}|-) \[%{HTTPDATE:log_timestamp}\] \"%{WORD:request_mathod} %{REQUESTPATH:request_path} %{REQUESTPRO:request_protocol}\" %{NUMBER:http_status} %{NUMBER:body_bytes_sent} (%{GREEDYDATA:http_referer}|-) \"%{DATA:http_user_agent}\" \"%{USERNAME:http_x_forwarded_for}\"

logstash 的配置：

filter {
    grok {
        patterns_dir => "/etc/logstash/conf.d/patterns/mypattern"
        match => {
            "message" => "%{NGXACCESSLOG}"
        }
    }
}

output {
    stdout {
        codec => rubydebug
    }
}

效果以下所示：

{
           "log_timestamp" => "11/Oct/2017:19:32:22 +0800",
         "body_bytes_sent" => "13",
                 "message" => "192.168.31.1 - - [11/Oct/2017:19:32:22 +0800] \"GET /test/t/ HTTP/1.1\" 200 13 \"-\" \"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36\" \"-\"",
                    "type" => "logstash",
          "request_mathod" => "GET",
         "http_user_agent" => "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36",
                    "path" => "/data/logs/nginx/logstash.wangshenjin.com_access.log",
              "@timestamp" => 2017-10-30T04:08:08.412Z,
            "http_referer" => "\"-\"",
                "@version" => "1",
                    "host" => "baseos-1",
    "http_x_forwarded_for" => "-",
            "request_path" => "/test/t/",
               "client_ip" => "192.168.31.1",
             "http_status" => "200",
                    "user" => "-",
        "request_protocol" => "HTTP/1.1"
}

正则调试：https://grokdebug.herokuapp.com

统一管理grok正则：

# 新建一个目录，统一存放自定义grok正则
[root@ conf.d]# mkdir patterns
[root@ conf.d]# vim patterns/nginx_access
NGINXACCESS %{IPORHOST:remote_addr} - %{USERNAME:remote_user} \[%{HTTPDATE:log_timestamp}\] \"%{DATA:request}\" %{INT:status} %{INT:body_bytes_sent} \"%{DATA:http_referer}\" \"%{DATA:http_user_agent}\" \"%{USERNAME:http_x_forwarded_for}\"

[root@ conf.d]# vim nginx_access.conf
****
filter {
    if [type] == "nginx-access-log" {
        grok {
            patterns_dir => "/etc/logstash/conf.d/patterns"   //设置自定义正则路径     
            match => {
                "message" => "%{NGINXACCESS}"
            }
        }
    }
}
****

date

格式化日期

date {
        match => [ "log_timestamp", "YYYY-MM-dd HH:mm:ss"]
    }

geoip

借助GeoIP数据库来实现显示请求来源的地理位置，GeoIP 库能够根据 IP 地址提供对应的地域信息，包括国别，省市，经纬度等。

获取GeoIP数据库

wget http://geolite.maxmind.com/download/geoip/database/GeoLite2-City.mmdb.gz  
gzip -d GeoLite2-City.mmdb.gz

配置：

filter {
***
    geoip {
        source => "client_ip"
        fields => ["city_name" , "country_name" ,"continent_code" , "continent_name" ]
        database => "/etc/logstash/GeoLite2-City.mmdb"
   }
}

参数：

source：设置解析IP地址的字段
fields：geoip数据库重须要的字段，主要有：city_name, continent_code, country_code2, country_code3, country_name, dma_code, ip, latitude, longitude, postal_code, region_name，timezone。
target：将geoip数据保存到一个字段内
database：IP地址数据库

output插件

redis

output配置以下：

output {
    redis {
        host => "127.0.0.1"
        port => 6000
        password => "8a6715"
        data_type => "channel"
        key => "logstash-%{+yyyy.MM.dd}"
    }
}

redis效果：

127.0.0.1:6000> subscribe logstash-2017.06.22
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "logstash-2017.06.22"
3) (integer) 1
1) "message"
2) "logstash-2017.06.22"
3) "{\"remote_addr\":\"192.168.31.130\",\"request\":\"GET / HTTP/1.1\",\"body_bytes_sent\":\"33\",\"message\":\"192.168.31.130 - - [22/Jun/2017:15:51:22 +0800] \\\"GET / HTTP/1.1\\\" 200 33 \\\"-\\\" \\\"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\\\" \\\"-\\\"\",\"type\":\"nginx-access-log\",\"tags\":[],\"http_user_agent\":\"curl/7.19.7 (x86_64-redhat-linux-gnu) libcurl/7.19.7 NSS/3.14.0.0 zlib/1.2.3 libidn/1.18 libssh2/1.4.2\",\"remote_user\":\"-\",\"path\":\"/data/logs/nginx/logstash.wangshenjin.com_access.log\",\"@timestamp\":\"2017-06-22T07:51:23.484Z\",\"http_referer\":\"-\",\"@version\":\"1\",\"host\":\"baseos-1\",\"http_x_forwarded_for\":\"-\",\"status\":\"200\"}"

参数：

batch：设为true，经过发送一条rpush命令，存储一批的数据
- 默认为false：1条rpush命令，存储1条数据
- 设为true以后，1条rpush会发送batch_events条数据或发送batch_timeout秒（取决于哪个先到达）
batch_events：一次rpush多少条
- 默认50条
batch_timeout：一次rpush最多消耗多少s
- 默认5s
codec：对输出数据进行codec，避免使用logstash的separate filter
congestion_interval：每多长时间进行一次拥塞检查(拥塞保护（仅用于data_type为list）
- 默认1s，设为0，表示对每rpush一个，都进行检测
congestion_threshold：list中最多能够存在多少个item数据，默认是0：表示禁用拥塞检测
- 当list中的数据量达到congestion_threshold，会阻塞直到有其余消费者消费list中的数据
- 做用：防止OOM
data_type
- list：使用rpush
- channel：使用publish
db：使用redis的数据库，默认使用0号
host：数组
- eg.["127.0.0.1:6380", "127.0.0.1"]
- 能够指定port，会覆盖全局port
port：全局port，默认6379
key：list或channel的名字
- 支持动态key，例如：logstash-%{type}
password：redis密码，默认不使用密码
reconnect_interval：失败重连的间隔，默认为1s
timeout：链接超时，默认5s

elasticsearch

output配置以下：

output {
        elasticsearch {
            hosts => ["192.168.1.147:9200","192.168.1.151:9200"]
            index  => "nginx-access-log-%{+YYYY.MM.dd}" #定义index pattern
        }
    }

参数：

hosts：elasticsearch服务列表
- Default value is ["127.0.0.1"]
- Value type is string
action：指定es的行为，index, delete, create, update
- Default value is "index"：index a document（该document就是一个来自于logstash的event）
- delete：经过id删除一个document（须要指定document_id）
- create：index a document（若是该document已经在index中存在，则失败）
- update：经过id更新一个document
index：指定文档索引，支持字符扩展%{foo}
- Value type is string
- Default value is "logstash-%{+YYYY.MM.dd}"
- 便于删除老数据
- 在语法解析的时候，看到+号开头的，会自动认为后面是时间格式，尝试用时间格式来解析后续字符串。因此，以前处理过程当中不要给自定义的字段起一个+号开头的名字
- 索引名中不能有大写字母
- 有时也会自定义为：logstash-%{servicename}-%{+YYYY.MM.dd}
cacert：验证server合法性的.cer或.pem文件路径
- Value type is path
- There is no default value for this setting.
codec：用于输出数据以前对数据进行解码的的编解码器，而不须要一个单独的LogStash过滤器。
- Value type is codec
- 默认值："plain"
document_id：文档索引ID，用来覆盖es中现有相同ID文档
- Value type is string
- There is no default value for this setting.
document_type：文档索引类型，支持字符扩展%{foo}，不设置文档类型，文档类型会被设置为logs
- Value type is string
- There is no default value for this setting.
user：进入es cluster的用户
password：进入es cluster的密码
timeout：Set the timeout for network operations and requests sent Elasticsearch. If a timeout occurs,the request will be retried.
flush_size：默认500，logstash攒够500条数据再一次性向es发送。从5.0开始，flush_size值不能超过batch_size,否者以batch_size为批量发送的大小
- Value type is number
- Default value is 500
idle_flush_time：默认1s，若是1s内没攒够500条仍是会一次性将攒的数据发出去给es
ssl:开启SSL与ES集群通讯，未指定该参数，是否启动SSL将由ES的URL决定，若是显式禁用SSL，在主机中给出一个HTTPS URL，该插件也将拒绝启动。
- Value type is boolean
- There is no default value for this setting.

详细参考：https://www.elastic.co/guide/en/logstash/5.0/plugins-outputs-elasticsearch.html