1、server.conf文件

[elk@logserver bin]$ cat server.conf
input {  
        redis {  
                host => "10.10.45.200"  
                data_type => "list"  
                key => "elk_frontend_access:redis"  
                port =>"5379"  
        }  
}  
output { 
    if "_grokparsefailure" in [tags] {
    }else{
        if [type] == "www1_access"{
                elasticsearch {
                   hosts => "10.10.45.200:8200"
                  index => "logstash-www1-frontend-%{+YYYY.MM.dd}"
        }
        }
       if [type] == "flight1_access"{
                 elasticsearch {
                        hosts => "10.10.45.200:8200"
                index => "logstash-flight1-frontend-%{+YYYY.MM.dd}"
         }
        }
    }   
}

2、agent.conf文件

[elk@www1-n02 bin]$ cat agent.conf
input {  
        file {  
                type => "www1_access"  
                path => ["/data/logs/www1.mangocity.com-access_log"]  
        }
        file {
                type => "flight1_access"
                path => ["/data/logs/flight1-access_log"]
        }
}  
filter {
ruby {
init => "@kname = [‘http_clientip‘,‘http_x_forwarded_for‘,‘time_local‘,‘request‘,‘status‘,‘body_bytes_sent‘,‘request_body‘,‘content_length‘,‘http_referer‘,‘http_user_agent‘,‘http_cookie‘,‘remote_addr‘,‘hostname‘,‘upstream_addr‘,‘upstream_response_time‘,‘request_time‘]"
code => "new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘message‘).split(‘|‘))])
new_event.remove(‘@timestamp‘)
event.append(new_event)"
}
if [request] {
ruby {
init => "@kname = [‘method‘,‘uri‘,‘verb‘]"
code => "new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘request‘).split(‘ ‘))])
new_event.remove(‘@timestamp‘)
event.append(new_event)
"
}
if [uri] {
ruby {
init => "@kname = [‘url_path‘,‘url_args‘]"
code => "new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘uri‘).split(‘?‘))])
new_event.remove(‘@timestamp‘)
event.append(new_event)
"
}
kv {
prefix => "url_"
source => "url_args"
field_split => "& "
remove_field => [ "url_args","uri","request" ]
}
}
}
mutate {
convert => ["body_bytes_sent" , "integer", "content_length", "integer", "upstream_response_time", "float","request_time", "float"]
}
date {
match => [ "time_local", "dd/MMM/yyyy:hh:mm:ss Z" ]
locale => "en"
}
        grok {
        match => { "message" => "%{IP:clientip}" }
  }
        geoip 
{
        source => "clientip"
        }
}
output {
        redis {  
                host => "10.10.45.200"  
                data_type => "list"  
                key => "elk_frontend_access:redis"  
                port=>"5379"  
        }  
}

参考博文：http://blog.csdn.net/wangyangzhizhou/article/details/53314022

logstash 创建多个索引