logstash 安装 配置

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

#　输入
input {
  ...
}
# 过滤器
filter {
  ...
}
# 输出
output {
  ...
}

input {
    # file为常用文件插件，插件内选项很多，可根据需求自行判断
    file {
        path => "/var/lib/mysql/slow.log"
        # 要导入的文件的位置，可以使用*，例如/var/log/nginx/*.log
        Excude =>”*.gz”
        # 要排除的文件
        start_position => "beginning"
        # 从文件开始的位置开始读,end表示从结尾开始读
        ignore_older => 0  
        # 多久之内没修改过的文件不读取，0为无限制，单位为秒
        sincedb_path => "/dev/null"
        # 记录文件上次读取位置，输出到null表示每次都从文件首行开始解析
        type => "mysql-slow"
        # type字段，可表明导入的日志类型
    }   
}

input {
    # 从文件读取日志信息
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }
}
# filter {
#
# }
output {
    # 标准输出
    stdout { codec => rubydebug }
}

logstash -f logstash.conf

       "message" => "Apr 10 00:53:29 cacti systemd: logstash.service: main process exited, code=exited, status=1/FAILURE",
    "@timestamp" => 2018-04-12T02:33:17.825Z,
          "host" => "cacti",
      "@version" => "1",
          "type" => "system",
          "path" => "/var/log/messages"
}
{
       "message" => "Apr 10 00:53:29 cacti systemd: Unit logstash.service entered failed state.",
    "@timestamp" => 2018-04-12T02:33:17.825Z,
          "host" => "cacti",
      "@version" => "1",
          "type" => "system",
          "path" => "/var/log/messages"

input {
    # redis插件为常用插件，插件内选项很多，可根据需求自行判断
    redis {
        batch_count => 1 
        # EVAL命令返回的事件数目，设置为5表示一次请求返回5条日志信息
        data_type => "list" 
        # logstash redis插件工作方式
        key => "logstash-test-list" 
        # 监听的键值
        host => "127.0.0.1" 
        # redis地址
        port => 6379 
        # redis端口号
        password => "123qwe" 
        # 如果有安全认证，此项为认证密码
        db => 0 
        # 如果应用使用了不同的数据库，此为redis数据库的编号，默认为0。
        threads => 1 
        # 启用线程数量
      }
}

output {
    # tdout { codec => "rubydebug" }
    # 筛选过滤后的内容输出到终端显示

    elasticsearch {  # 导出到es，最常用的插件
        codec => "json"
        # 导出格式为json
        hosts => ["127.0.0.1:9200"]
        # ES地址+端口
        index => "logstash-slow-%{+YYYY.MM.dd}"
        # 导出到index内，可以使用时间变量
        user => "admin"
        password => "xxxxxx"
        # ES如果有安全认证就使用账号密码验证，无安全认证就不需要
        flush_size => 500
        # 默认500，logstash一次性攒够500条的数据在向es发送
        idle_flush_time => 1
        # 默认1s，如果1s内没攒够500，还是会一次性把数据发给ES
    }   
}

output {
     redis{  # 输出到redis的插件，下面选项根据需求使用
         batch => true
         # 设为false，一次rpush，发一条数据，true为发送一批
         batch_events => 50
         # 一次rpush发送多少数据
         batch_timeout => 5
         # 一次rpush消耗多少时间
         codec => plain
         # 对输出数据进行codec，避免使用logstash的separate filter
         congestion_interval => 1
         # 多长时间进项一次拥塞检查
         congestion_threshold => 5
         # 限制一个list中可以存在多少个item，当数量足够时，就会阻塞直到有其他消费者消费list中的数据
         data_type => list
         # 使用list还是publish
         db => 0
         # 使用redis的那个数据库，默认为0号
         host => ["127.0.0.1:6379"]
         # redis 的地址和端口，会覆盖全局端口
         key => xxx
         # list或channel的名字
         password => xxx
         # redis的密码，默认不使用
         port => 6379
         # 全局端口，默认6379，如果host已指定，本条失效
         reconnect_interval => 1
         # 失败重连的间隔，默认为1s
         timeout => 5
         # 连接超时的时间
         workers => 1
         # 工作进程
     }
}

filter {  # 插件很多，这里选取我使用过的插件做讲述
    if ([message] =~ "正则表达式")  {  drop {}  }
    # 正则匹配=~,!~,包含判断in,not in ，字符串匹配==,!=,等等，匹配之后可以做任何操作，这里过滤掉匹配行，除了做过滤操作，if后面可以作任意操作，甚至可以为匹配到的任意行做单独的正则分割操作

    multiline {
        pattern => "正则表达式"
        negate => true
        what => "previous"
        # 多行合并，由于一些日志存在一条多行的情况，这个模块可以进行指定多行合并，通过正则匹配，匹配到的内容上面的多行合并为一条日志。
    }   

    grok {
        match => { "message" => "正则表达式"
         # 正则匹配日志，可以筛选分割出需要记录的字段和值
        }   
        remove_field => ["message"]
        # 删除不需要记录的字段
   }   

    date {
        match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"] 
       # 记录@timestamp时间，可以设置日志中自定的时间字段，如果日志中没有时间字段，也可以自己生成
        target=>“@timestamp”
        # 将匹配的timestamp字段放在指定的字段 默认是@timestamp
    }

    ruby {
        code => "event.timestamp.time.localtime"
        # timestamp时区锁定
    }   
}