flume+elasticsearch日誌收集分析

因爲不一樣版存在不兼容問題也試了好屢次，網上找別人寫的博客版本最後終於成功了。

flume：1.6.0

elasticsearch：1.7.6

kibana：4.1.9

從elasticsearch的lib目錄拷貝elasticsearch-1.7.6.jar，lucene-core-4.10.4.jar到flume的lib目錄,咱們服務器使用的是logcxx和spdlog，使用syslog方式收集日誌，（因爲spdlog的syslog不能寫到遠程服務器，本身要改下源碼用udp發送出去就好了）。

日誌內容：<15>(Balance)[2016-12-15 14:53:28,610][DEBUG] ServerName[Entrance] statistics_total: total_UserCount = [cur_total=11,max_total=0],

紅色：syslog用日誌級別和日誌類型生成的一個數字（本身修改spdlog的syslog須要在一條日誌前加上尖括號和這個數字）

綠色：下面正則表達式要過濾出來的服務器名字或者模塊名字

藍色：時間，下面也會轉換成時間戳

褐色：級別

黑色：內容

logcxx配置：

log4j.appender.RootSyslog=org.apache.log4j.net.SyslogAppender
log4j.appender.RootSyslog.SyslogHost=127.0.0.1 #flume地址
log4j.appender.RootSyslog.Facility=USER
log4j.appender.RootSyslog.Append=true
log4j.appender.RootSyslog.layout=org.apache.log4j.PatternLayout
log4j.appender.RootSyslog.layout.ConversionPattern=(服務器名字或者模塊)[%d][%p] %m%n

flume配置：

agent.sources = sysSource
agent.channels = memoryChannel
agent.sinks = k1

#syslog source
agent.sources.sysSource.type = syslogudp
agent.sources.sysSource.bind = 0.0.0.0
agent.sources.sysSource.port = 514
agent.sources.sysSource.channels = memoryChannel
agent.sources.sysSource.interceptors = i1 i2 i3
#時間
agent.sources.sysSource.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
#服務器IP
agent.sources.sysSource.interceptors.i2.type = org.apache.flume.interceptor.HostInterceptor$Builder
agent.sources.sysSource.interceptors.i2.hostHeader = host
#正則表達式過濾出自定義圓括號中的字符串做爲服務器類型名
agent.sources.sysSource.interceptors.i3.type=org.apache.flume.interceptor.RegexExtractorInterceptor$Builder
agent.sources.sysSource.interceptors.i3.regex = ((?<=\\()[^\\)]+)
agent.sources.sysSource.interceptors.i3.serializers = s1 
agent.sources.sysSource.interceptors.i3.serializers.s1.name = server_type

#ElasticSearchSink
agent.sinks.k1.type = org.apache.flume.sink.elasticsearch.ElasticSearchSink
agent.sinks.k1.channel = memoryChannel
#elasticsearch地址
agent.sinks.k1.hostNames = 10.105.92.225:9300 
#elasticsearch索引名字
agent.sinks.k1.indexName = server_log
agent.sinks.k1.batchSize = 100
agent.sinks.k1.indexType = log
#這個要和elasticsearch名字同樣
agent.sinks.k1.clusterName = log-es
agent.sinks.k1.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer

#memory channel
agent.channels.memoryChannel.type = memory
agent.channels.memoryChannel.capacity = 10000
agent.channels.memoryChannel.transacionCapacity = 100
agent.channels.memoryChannel.byteCapacityBufferPercentage = 20
agent.channels.memoryChannel.byteCapacity  = 10240000
 

elasticsearch配置:

cluster.name: log-es
node.name: "log-es01"
node.master: true
node.data: true
 

能夠給elasticsearch配置kibana和head插件