搭建完成elk一套，瞬间感觉高大上了很多有木有，后台直接可以查看所有的服务端日志，当然使用一段时间后，由于之前的使用很初级，渐渐的满足不了真正的需求，所有下面总结一下关于logstash在配置上的要点

划分细度日志、以及多行异常合并一行

在将logstash上传到elk后，默认一行数据都会存在于一个_source字段中，这当然不能满足我们的使用，比如我们需要知道日志的等级，时间，用户，公司等信息。要实现这样的功能我们使用logstash的filter来完成

配置文件

某些敏感信息已经以...替换，在使用时请根据具体环境配置

input {
    file {  
        path => ["..."]  
        type => "app"
        codec => multiline {
            pattern => "^201"
            negate => true
            what => "previous"
        }
    } 
 
}
filter{
    grok{
        match => ['message','%{NOTSPACE:date} %{NOTSPACE:time} \[%{NOTSPACE:grade}\] \[(?<companyName>.*)\:(?<companyId>.*),(?<userName>.*)\:(?<userId>.*)\] \[%{NOTSPACE:thread}\]\s+%{NOTSPACE:package}\s+[-]\s+(?<content>.*)']
    }
}
output {
    stdout {
      codec => rubydebug
    }
    if [type] == "app" {  
       elasticsearch { 
            hosts => "..."  
            index =>  "app-%{+YYYY.MM.dd}"
            document_type => "log4j_type"
            user => elastic
            password => ...
        }  
    }  
}

多行配置

 codec => multiline {
            pattern => "^201"
            negate => true
            what => "previous"
    }

这段配置意思是正则匹配^201以201开头的行，在elk中的行开头，直到匹配下一个，当前行才结束，为啥是201开头？因为我们的日志是以日期开头，2018年。所以就通过这种方式判断一个日志的开头与结束

分割字段

filter{
    grok{
        match => ['message','%{NOTSPACE:date} %{NOTSPACE:time} \[%{NOTSPACE:grade}\] \[(?<companyName>.*)\:(?<companyId>.*),(?<userName>.*)\:(?<userId>.*)\] \[%{NOTSPACE:thread}\]\s+%{NOTSPACE:package}\s+[-]\s+(?<content>.*)']
    }
}

在kibana后台中有测试这个grok语法的devTool->grok dubugger 这里我的原始日志是
2018-07-18 10:13:046 [INFO] [xxx:214,xxx:437] [http-nio-8080-exec-5:61040139] FsController:143 - sip-login action:message-count domain:xx.xx.cn userId:434 phone:xxx使用grok的语法过滤后再elk后台就会有对应划分的字段了。

使用调试工具转换后得到

{
  "date": "2018-07-18",
  "package": "FsController:143",
  "companyName": "xxx",
  "thread": "http-nio-8080-exec-5:61040139",
  "userName": "xxx",
  "userId": "437",
  "content": "sip-login action:message-count domain:xx.xx.cn userId:434 phone:xxx",
  "companyId": "214",
  "grade": "INFO",
  "time": "10:13:046"
}

注意其实这样对日志的格式要求很高，要特别注意不要让一个裸异常直接打印在日志里面，否则elk将会可能会解析成一个正常的日志。比如像下面那样

2018-07-18 11:37:54 [Thread-15] INFO  com.xxx.service.biz.impl.taobao.AlimamaServiceImpl -请求地址:https://xxx.com?startTime=2018-07-18 00:00:00&endTime=2018-07-18 23:59:59&payStatus=&queryType=2&t=1531885074032&_input_charset=utf-8&DownloadID=DOWNLOAD_REPORT_TK3_PUB
2018-07-18 11:37:54 [Thread-15] INFO  com.xxx.service.biz.impl.taobao.AlimamaServiceImpl -获取的cookie:v=0; cookie2=xxx; t=88f13241312e4f14d69d3d8c6d4b4dba; _tb_token_=58381eb1e5e4e; cna=f6dCE2MQZwwCAbaWG7lTWi27; cookie32=xxx; cookie31=MTE4MzMyNTk3LGh1YXFpd2ViY29tLGFkbWluQGh1YXFpd2ViLmNvbSxUQg%3D%3D; account-path-guide-s1=true; 118332597_yxjh-filter-1=true; taokeisb2c=; login=W5iHLLyFOGW7aA%3D%3D; alimamapwag=TW96aWxsYS81LjAgKExpbnV4OyBBbmRyb2lkIDcuMS4xOyBNaSBOb3RlIDMgQnVpbGQvTk1GMjZYOyB3dikgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgVmVyc2lvbi80LjAgQ2hyb21lLzY3LjAuMzM5Ni44NyBNb2JpbGUgU2FmYXJpLzUzNy4zNg%3D%3D; alimamapw=UEMARV5OB1IHClwxAVcABQ9QDwFRVgIOWwIHAwFbAlVXAlsADwcHAAZdB1I%3D; rurl=xxx%3D%3D; isg=BLm5VhvkxuaWXZuH8h72PC7kwydTbsh8qzhgitvuNeBfYtn0Ixa9SCeQ4SAJYUWw
java.io.IOException: Invalid header signature; read 0x74636F64213C0A0D, expected 0xE11AB1A1E011CFD0
        at org.apache.poi.poifs.storage.HeaderBlock.<init>(HeaderBlock.java:140)
        at org.apache.poi.poifs.storage.HeaderBlock.<init>(HeaderBlock.java:104)
        at org.apache.poi.poifs.filesystem.POIFSFileSystem.<init>(POIFSFileSystem.java:138)
        at org.apache.poi.hssf.usermodel.HSSFWorkbook.<init>(HSSFWorkbook.java:322)
        at org.apache.poi.hssf.usermodel.HSSFWorkbook.<init>(HSSFWorkbook.java:303)
        at com.xxx.util.excel.ImportExcel.<init>(ImportExcel.java:127)
        at com.xxx.util.excel.ImportExcel.<init>(ImportExcel.java:98)
        at com.xxx.service.biz.impl.taobao.AlimamaServiceImpl.getTbkThirdPaymentDetails(AlimamaServiceImpl.java:350)
        at com.xxx.service.biz.impl.taobao.AlimamaServiceImpl$$FastClassBySpringCGLIB$$781652c5.invoke(<generated>)
        at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
        at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:720)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
        at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:92)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
        at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:85)
        at com.xxx.service.aop.ExceptionHandleAspect.handleException(ExceptionHandleAspect.java:42)
        at sun.reflect.GeneratedMethodAccessor79.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:620)
        at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:609)
        at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:68)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
        at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:655)
        at com.xxx.service.biz.impl.taobao.AlimamaServiceImpl$$EnhancerBySpringCGLIB$$87a5c229.getTbkThirdPaymentDetails(<generated>)
        at com.xxx.service.jobhandler.SyncTbkPaymentDetailsJobHandler.execute(SyncTbkPaymentDetailsJobHandler.java:138)
        at com.xxl.job.core.thread.JobThread.run(JobThread.java:111)