码迷,mamicode.com
首页 > 其他好文 > 详细

统一日志ELK部署配置(3)——logstash

时间:2018-06-13 18:07:43      阅读:299      评论:0      收藏:0      [点我收藏+]

标签:日志收集

一、下载
从官网下载:https://www.elastic.co/downloads/logstash
二、配置
1、修改config下的jvm.options:
1??根据需要修改最大堆和最小堆
2??我这里使用的jdk1.8,gc使用G1,所以需要重新配置;
-XX:+UnlockDiagnosticVMOptions
-XX:+UseCompressedOops
-XX:+UseG1GC
-XX:MaxGCPauseMillis=200
-XX:G1ReservePercent=20
-XX:+G1SummarizeConcMark
-XX:InitiatingHeapOccupancyPercent=40
-XX:+AlwaysPreTouch
-XX:+DisableExplicitGC

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

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!