前文已經簡單搭建好了elk，現在就入數據這一方面，我們簡單分析。

1現狀：我司java程序全部使用log4j2經行日志寫文件。其基本例子如下：

<?xml version="1.0" encoding="UTF-8"?>
<!-- status:log4j自身日志，monitorInterval:自動檢測配置文件是否改變，單位：s -->
<configuration status="info" monitorInterval="5" shutdownHook="disable">
    <Properties>
        <!-- 配置日志文件輸出目錄 -->
        <Property name="LOG_HOME">/logdata-local/path/to/log/</Property>
    </Properties>
    <appenders>
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout charset="UTF-8" pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} - %msg%n"/>
            <!--控制臺只輸出level及以上級別的信息（onMatch），其他的直接拒絕（onMismatch） -->
            <ThresholdFilter level="trace" onMatch="ACCEPT" onMismatch="DENY"/>
        </Console>

        <!-- 服務端主日志 -->
        <RollingFile name="asyncservice" fileName="${LOG_HOME}/service.log"
                     filePattern="${LOG_HOME}/service_%d{yyyy-MM-dd}_%i.log">
            <Filters>
                <!-- 打印除error日志所有日志 -->
                <ThresholdFilter level="error" onMatch="DENY" onMismatch="NEUTRAL"/>
                <ThresholdFilter level="trace" onMatch="ACCEPT" onMismatch="DENY"/>
            </Filters>
            <PatternLayout charset="UTF-8"
                           pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} [%L]  - %msg%n"/>
            <Policies>
                <!-- 更新時間 -->
                <TimeBasedTriggeringPolicy modulate="true" interval="1"/>
                <SizeBasedTriggeringPolicy size="500MB"/>
            </Policies>
            <!-- 最多8個日志 -->
            <DefaultRolloverStrategy max="10"/>
        </RollingFile>

        <!-- 服務端錯誤日志 -->
        <RollingFile name="asyncerror" fileName="${LOG_HOME}/error.log"
                     filePattern="${LOG_HOME}/error_%d{yyyy-MM-dd}_%i.log">
            <Filters>
                <!-- 打印error日志 -->
                <ThresholdFilter level="error" onMatch="ACCEPT" onMismatch="DENY"/>
                <ThresholdFilter level="error" onMatch="DENY" onMismatch="ACCEPT"/>
            </Filters>
            <PatternLayout charset="UTF-8"
                           pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} [%L] - %msg%n"/>
            <Policies>
                <!-- 更新時間 -->
                <TimeBasedTriggeringPolicy modulate="true"
                                           interval="1"/>
                <SizeBasedTriggeringPolicy size="500MB"/>
            </Policies>
            <!-- 最多8個日志 -->
            <DefaultRolloverStrategy max="8"/>
        </RollingFile>

        <RollingFile name="asyncmonitor" fileName="${LOG_HOME}/monitor.log"
                     filePattern="${LOG_HOME}/report_%d{yyyy-MM-dd}_%i.log">
            <PatternLayout charset="UTF-8" pattern="%d{HH:mm:ss.SSS} - %msg%n"/>
            <Policies>
                <!-- 更新時間 -->
                <TimeBasedTriggeringPolicy modulate="true" interval="1"/>
                <SizeBasedTriggeringPolicy size="500MB"/>
            </Policies>
            <!-- 最多8個日志 -->
            <DefaultRolloverStrategy max="8"/>
        </RollingFile>

        <Async name="service" bufferSize="102400">
            <AppenderRef ref="asyncservice"/>
        </Async>
        <Async name="error" bufferSize="102400">
            <AppenderRef ref="asyncerror"/>
        </Async>
        <Async name="monitor" bufferSize="102400">
            <AppenderRef ref="asyncmonitor"/>
        </Async>
    </appenders>


    <loggers>
        <Logger name="monitor" level="info" additivity="false">
            <AppenderRef ref="monitor"/>
        </Logger>
        <root level="info">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="service"/>
            <AppenderRef ref="error"/>
        </root>
    </loggers>

</configuration>

其中主要使用了log4j2的異步緩沖池，5秒刷新配置，自動分日期打包文件等特性。

2目標

我們主要關注

pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} [%L]  - %msg%n"

打出來的日志基本是這樣：

2018-03-27 21:22:19.048 [pool-4-thread-1] INFO  com.common.monitor.MonitorService []  - 監控redis連接#執行結果：OK，監控狀態：redis狀態：true(0,0)

我決定使用logstash的grok插件，過濾信息

3成果

input{
    file {
        path => ["/logdata-local/*/*.log"]
        exclude => "*_*.log"
        max_open_files => "18600"
        codec => multiline {
                pattern => "^\s"
                what => "previous"
                }
        }
}
filter{
        grok {
             match =>{
                "message" =>"%{TIMESTAMP_ISO8601:logtime}\s\[%{DATA:logthread}\]\s%{LOGLEVEL:loglevel}\s\s%{DATA:logclass}\s\[\].{4}%{GREEDYDATA:logcontent}"
                }
                remove_field => ["message"]
            }
        date {
                match => ["logtime", "yyyy-MM-dd HH:mm:ss.SSS"]
                target => "@timestamp"
                }
        mutate {
            add_field => { "filepath" => "%{path}" }
              }
        mutate{
        split => ["filepath","/"]
            add_field =>   {
                "idx" => "%{[filepath][1]}-%{[filepath][2]}-%{[filepath][3]}"
            }
            add_field =>   {
                "filename" => "%{[filepath][3]}"
            }
        }
        mutate{
                lowercase => [ "idx" ]
        }
}
output {
        elasticsearch {
                hosts => ["192.168.193.47:9200","192.168.193.47:9200","192.168.193.47:9200"]
                index => "logstash-bank-%{idx}-%{+YYYY.MM.dd}"
                user => elastic
                password => elastic
         }
}

過濾器意思將其自動生成的message信息解析，然后移除。將文件路徑的“/”轉化為“-”建立索引。索引不能有大寫，最后都小寫化。

解析出的日志時間，覆蓋原始的@timestamp字段。

其基本將每條日志的時間、線程、日志級別、類和內容字段分析了出來，以后看運維同事有什么查詢需求，將規范化內容再細化解析。

4kibana呈現

這樣日志就像數據庫一樣，可以按照字段查詢了。