logstash5.4 安装配置

原创
2017/05/25 17:09
阅读数 299

一、下载用yum 安装

二、插件

查看可安装的插件
/usr/share/logstash/bin/logstash-plugin list

安装插件
/usr/share/logstash/bin/logstash-plugin install logstash-filter-date
/usr/share/logstash/bin/logstash-plugin install logstash-patterns-core #定义好的patterns
/usr/share/logstash/bin/logstash-plugin install logstash-filter-grok
/usr/share/logstash/bin/logstash-plugin install logstash-filter-mutate

#input kafka,output elasticsearch 需要的话自己安装

三、编写配置

input {
  udp{
    type => "tcpdump"
    port => 514
  }

  kafka {
    bootstrap_servers => "master200:9092"
    codec => "json"
    topics => ["file_dt"]
    id => "kafka_id"
    group_id => "kafka_gid"
    client_id => "kafka_cli"
    auto_offset_reset => "earliest"
    session_timeout_ms => "80000"
    request_timeout_ms => "810000"
    heartbeat_interval_ms => "1000"
    consumer_threads => 50
    decorate_events => true
  }
}


filter {
  if [type] == "tcpdump" {
    grok {

      match => ["message", "\<%{NUMBER:no}\> \*(?<log_time>%{MONTH:month} %{MONTHDAY:day} %{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second}): \%(?<mid>[a-zA-Z0-9_\-]+): %{WORD:client}\(%{CISCOMAC:mac}\) (?<status>[a-zA-Z0-9_\-\s]+)([,.]{1} reason: (?<reason>[a-zA-Z0-9_\-\s]+))?(: ([a-zA-Z0-9_\-\s]+\((?<ap>\w+\d+)\)))?",
                "message", "\<%{NUMBER:no}\> \*(?<log_time>%{MONTH:month} %{MONTHDAY:day} %{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second}): \%(?<mid>[a-zA-Z0-9_\-]+): (?<status>[a-zA-Z0-9_\-\s]+)[,:]? MAC Address:? %{CISCOMAC:mac}, AP Name (?<ap>\w+\d+), base radio MAC(: )?%{CISCOMAC:radiomac}, (IP address: %{IPV4:ip})?(User Name (?<username>[a-zA-Z0-9_\-]+))?, SSID:? (?<ssid>\w+-\w+)"
               ]

    }
    if [mid] == "APMG-6-STA_UPDT" {
      mutate{
        add_field => ["动作", "UPDT"]
      }


    }


    date{
      match => [
        "log_time","MMM dd HH:mm:ss"
               ]
      target => ["log_time110"]
      locale => "en"
      timezone => "Asia/Shanghai"


         }

  }
}



output {

    stdout {
        codec => rubydebug
    }

    if [type] == "file_dt" {
      elasticsearch {
        hosts => ["master200:9200","slave95:9200"]
        index => "kafka-%{type}-%{+YYYYMMdd}"
        document_type => "%{type}"
      }
    }

    else if [type] == "tcpdump" {
      elasticsearch {
        hosts => ["master200:9200","slave95:9200"]
        index => "logstash-%{type}-%{+YYYY.MM.dd}"
        document_type => "%{type}"
        template => "/etc/logstash/conf.d/tcp_template.json"
        template_overwrite => true
        flush_size => 5000
        idle_flush_time => 10
     }
    }


}

四、启动

/usr/share/logstash/bin/logstash --path.settings /etc/logstash -f /etc/logstash/conf.d/logstash.conf -w 10

五、其他注意

#测试地址:http://grokdebug.herokuapp.com/

#测试配置语法的正确性,使用下面的命令
logstash -f first-pipeline.conf --configtest

 

展开阅读全文
打赏
0
1 收藏
分享
加载中
更多评论
打赏
0 评论
1 收藏
0
分享
返回顶部
顶部