统一日志ELK部署配置（3）——logstash

标签：日志收集

2、新建配置文件service.conf

3、在service.conf配置数据收集、解析及输出：
#==数据源配置，从kafka
input{
kafka{
    bootstrap_servers => ["10.31.140.96:9092,10.31.140.99:9092,10.31.140.93:9092"]
    topics => "elk-service-log"
    group_id => "service-log-group"
    codec => "json"
    auto_offset_reset => "earliest"
            max_partition_fetch_bytes => "52428700"
    max_poll_records => "300"
    consumer_threads => "16"
}

}
#==数据解析——这里的日志内容是json格式，所以用以下解析
filter{
mutate{
gsub => ["message", "[\n|\r|\f|\t]", " "]
gsub => ["message", "[\]", "/"]
}
json{
source => "message" #==filebeat收集的日志内容在message字段中，JSON格式
#target => "doc"
}

if [type] == "nginx-log" {     #==匹配字段内容，去除filebeat中不需要的内容
    grok{
        match => {
            "@version" => "1"
        }
        remove_field => ["@version"]
        remove_field => ["offset"]
        remove_field => ["beat"]
        remove_field => ["@version"]
        remove_field => ["source"]
        remove_field => ["input_type"]
        remove_field => ["message"]
        #remove_field => ["request_body"]
    }
} else {
    grok{
        match => {
            "@version" => "1"
        }
        remove_field => ["@version"]
        remove_field => ["offset"]
        remove_field => ["beat"]
        remove_field => ["@version"]
        remove_field => ["source"]
        remove_field => ["input_type"]
    }
}

}
#== 数据输出到elasticsearch集群
output{
if [type] == "service-log" { #==type字段是在filebeat中添加的自定义字段，用于日志内容区分
elasticsearch{
hosts => ["es1:9200","es2:9200","es3:9200"]
index => "service-log-%{+yyyy.MM.dd}" #==每天一个日志索引
codec => "json"
manage_template => true
template => "/mnt/logstash-5.6.4/templates/service-log.json" #日志索引模板
template_name=>"service-log"
template_overwrite=>true
}
}

if [type] == "nginx-log" {
  elasticsearch{
    hosts => ["es1:9200","es2:9200","es3:9200"]
    index => "nginx-log-%{+yyyy.MM.dd}"
    codec => "json"
    manage_template => true
    template => "/mnt/logstash-5.6.4/templates/nginx-log.json"
    template_name=>"nginx-log"
    template_overwrite=>true
  }
}

}
三、日志索引模板：
日志索引模板是用来创建索引时规定字段类型、设置索引配置的，设置得当可以提高Elasticsearch性能，减少Elasticsearch对资源的消耗。
日志模板如下：
{
"template": "service-log*",
"settings": {
"index.number_of_shards": 12,
"index.number_of_replicas": 0,
"index.refresh_interval":"10s"
},
"mappings": {
"java": {
"_all": {
"analyzer": "ik_max_word", #==ik分词
"enabled": false #==禁用all
},
"properties": {
"@timestamp": {
"format": "dateOptionalTime",
"type": "date"
},
"date": {
"type": "keyword"
},
"tranceId": {
"type": "keyword"
},
"sequenceId": {
"type": "keyword"
},
"level": {
"type": "keyword"
},
"appName": {
"type": "text",
"analyzer":"ik_max_word"
},
"serverName": {
"type": "text",
"analyzer":"ik_max_word"
},
"port": {
"type": "integer"
},
"class": {
"analyzer": "ik_max_word",
"type":"text"
},
"method": {
"type": "text",
"analyzer":"ik_max_word"
},
"line": {
"type": "integer"
},
"message": {
"analyzer": "ik_max_word",
"type": "text"
}
}
}
}
}
四、启动：
bin/logstash -f service.conf &

五、维护：
如果不同的日志内容格式需要接入到ELK，那么在filebeat端增加一个type标识内容，在logstash中根据type进行处理，然后写入Elasticsearch时新建一个索引模板写入即可。

注意事项：
template的命名一定要和logstash中输出到Elasticsearch时的配置匹配，否则会导致template不生效，但是还会写入到Elasticsearch；
如："template": "service-log*"，那么配置为 index => "service-log-%{+yyyy.MM.dd}"，配置为index => "nginx-log-%{+yyyy.MM.dd}"就会提示找不到template

统一日志ELK部署配置（3）——logstash

标签：日志收集

原文地址：http://blog.51cto.com/jfzhang/2129020