大数据日志分析系统-logstash

logstash简介

Logstash 是一个开源的数据收集引擎，它具有备实时数据传输能力。它可以统一过滤来自不同源的数据，并按照开发者的制定的规范输出到目的地。

logstash-2.2.2的配置：

从logstash-forward 到kafka的配置

ubuntu@sp1:~/logstashBeforeChangeConf$ cat /home/ubuntu/logstash-2.2.2/config/before-kafka-access.conf

input {

lumberjack {

port => "5044"

ssl_certificate => "/home/ubuntu/logstash-2.2.2/config/lumberjack.crt"

ssl_key => "/home/ubuntu/logstash-2.2.2/config/lumberjack.key"

type => "fc_access"

}

output {

if "_grokparsefailure" not in [tags] {

# stdout { codec => rubydebug }

kafka {

topic_id => "kafka_es"

bootstrap_servers => "sp1:9092,sp2:9092,sp3:9092,sp4:9092,sp5:9092,sp6:9092,sp7:9092"

compression_type => "snappy"

acks => ["1"]

value_serializer => "org.apache.kafka.common.serialization.StringSerializer"

timeout_ms => 10000

retries => 5

retry_backoff_ms => 100

send_buffer_bytes => 102400

workers => 2

}

从kafka到es配置

其中包括了对日志各个字段的解析,以及对异常日志过滤（同时注意其中过滤了不属于当前时间前后5天的时间的日志，为了防止异常日志创建索引过多导致es报红）

ubuntu@sp1:~/logstashAfterChangeConf$ cat /home/ubuntu/logstash-2.2.2/config/after-kafa-access.conf

input {

kafka {

topic_id => "kafka_es"

group_id => "kafka_es"

zk_connect => "sp1:2181,sp2:2181,sp3:2181,sp4:2181,sp5:2181,sp6:2181,sp7:2181"

consumer_threads => 1

consumer_restart_on_error => true

consumer_restart_sleep_ms => 5000

decorate_events => true

consumer_timeout_ms => 1000

queue_size => 100

auto_offset_reset => "smallest"

rebalance_max_retries => 50

}

filter {

mutate {

add_field => [ "messageClone", "%{message}" ]

}

mutate {

split => { "messageClone" => '"' }

add_field => {"agent" => "%{[messageClone][3]}"}

}

useragent {

source => "agent"

}

mutate {

split => { "message" => " " }

add_field => {"timestamp" => "%{[message][0]}"}

add_field => {"reqtime" => "%{[message][1]}"}

add_field => {"clientIP" => "%{[message][2]}"}

add_field => {"squidCache" => "%{[message][3]}"}

add_field => {"repsize" => "%{[message][4]}"}

add_field => {"reqMethod" => "%{[message][5]}"}

add_field => {"requestURL" => "%{[message][6]}"}

add_field => {"username" => "%{[message][7]}"}

add_field => {"requestOriginSite" => "%{[message][8]}"}

add_field => {"mime" => "%{[message][9]}"}

add_field => {"referer" => "%{[message][10]}"}

add_field => {"agentCheck" => "%{[message][11]}"}

add_field => {"dnsGroup" => "%{[message][-1]}"}

remove_field => ["offset", "kafka", "@version", "file", "message", "messageClone"]

}

if [agentCheck] =~ "ChinaCache" {

grok { match => { "agentCheck" => "OOPS" } }

}

mutate {

convert => {

"timestamp" => "float"

"reqtime" => "integer"

"repsize" => "integer"

}

remove_field => ["agentCheck"]

}

ruby {

code => "event['timestamp_str'] = Time.at(event['timestamp']).strftime('%Y-%m-%dT%H:%M:%S.%LZ')"

}

date { match => [ "timestamp_str", "ISO8601" ]

}

mutate {

split => { "requestURL" => '/' }

add_field => {"uriHost" => "%{[requestURL][2]}"}

remove_field => ["timestamp_str"]

}

mutate {

join => { "requestURL" => '/' }

}

ruby {

code => "event.cancel if 5 * 24 * 3600 < (event['@timestamp']-::Time.now).abs"

}

output {

if "ChinaCache" not in [agent] {

# stdout { codec => "rubydebug" }

elasticsearch {

index => "logstash-%{+YYYY.MM.dd.HH}"

workers => 1

flush_size => 5000

idle_flush_time => 1

hosts => ["es-ip-1:9200","es-ip-2:9200","es-ip-3:9200","es-ip-4:9200","es-ip-5:9200","es-ip-6:9200","es-ip-7:9200"]

}

启动命令：

nohup /home/ubuntu/logstash-2.2.2/bin/logstash -f /home/ubuntu/logstash-2.2.2/config/after-kafa-access.conf 2>&1 > /home/ubuntu/logstash-2.2.2/logs/logstash-after-kafka-access.log &

nohup /home/ubuntu/logstash-2.2.2/bin/logstash -f /home/ubuntu/logstash-2.2.2/config/before-kafka-access.conf 2>&1 > /home/ubuntu/logstash-2.2.2/logs/logstash-before-kafka.log &

logstash-6.1.1配置

从filbeat到kafka的配置

ubuntu@sp26:~/apps/logstash-6.1.1$ cat filebeat5055-kafkasp26-3.conf

input {

beats {

port => "5055"

type => "log"

}

output {

# stdout { codec => rubydebug }

kafka {

codec => "json"

bootstrap_servers => "37:9092,38:9092,39:9092,40:9092,41:9092"

topic_id => "test"

compression_type => "snappy"

value_serializer => "org.apache.kafka.common.serialization.StringSerializer"

}

检测
/home/ubuntu/apps/logstash-6.1.1/bin/logstash -f /home/ubuntu/apps/logstash-6.1.1/filebeat5055-kafkasp26-3.conf --config.test_and_exit
启动

nohup /home/ubuntu/apps/logstash-6.1.1/bin/logstash -f /home/ubuntu/apps/logstash-6.1.1/filebeat5055-kafkasp26-3.conf --config.reload.automatic 2>&1 > /home/ubuntu/apps/logstash-6.1.1/logs/filebeat5055-kafkasp26-3.log &