ELK统一日志管理

ELK

为什么用到ELK

        一般我们需要进行日志分析场景：直接在日志文件中 grep、awk 就可以获得自己想要的信息。但在规模较大的场景中，此方法 效率低下，面临问题包括日志量太大如何归档、文本搜索太慢怎么办、如何多维度查询。需要集中化的日志管理，所有服务器 上的日志收集汇总。常见解决思路是建立集中式日志收集系统，将所有节点上的日志统一收集，管理，访问。

        一般大型系统是一个分布式部署的架构，不同的服务模块部署在不同的服务器上，问题出现时，大部分情况需要根据问题暴露 的关键信息，定位到具体的服务器和服务模块，构建一套集中式日志系统，可以提高定位问题的效率。

        一个完整的集中式日志系统，需要包含以下几个主要特点：

• 收集－能够采集多种来源的日志数据
• 传输－能够稳定的把日志数据传输到中央系统
• 存储－如何存储日志数据
• 分析－可以支持 UI 分析
• 警告－能够提供错误报告，监控机制

        ELK提供了一整套解决方案，并且都是开源软件，之间互相配合使用，完美衔接，高效的满足了很多场合的应用。目前主流的 一种日志系统。

ELK简介

        ELK是三个开源软件的缩写，分别表示：Elasticsearch , Logstash, Kibana , 它们都是开源软件。

        Elasticsearch是个开源分布式搜索引擎，提供搜集、分析、存储数据三大功能。它的特点有：分布式，零配置，自动发现，索 引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等。

        Logstash 主要是用来日志的搜集、分析、过滤日志的工具，支持大量的数据获取方式。一般工作方式为c/s架构，client端安装 在需要收集日志的主机上，server端负责将收到的各节点日志进行过滤、修改等操作在一并发往elasticsearch上去。

        Kibana 也是一个开源和免费的工具，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助 汇总、分析和搜索重要数据日志

ELK架构图

       此架构由Logstash收集各个节点上相关日志、数据，并经过分析、过滤后发送给远端服务器上的Elasticsearch进行存储。 Elasticsearch将数据以分片的形式压缩存储并提供多种API供用户查询，操作。用户亦可以更直观的通过配置Kibana Web方便 的对日志查询，并根据数据生成报表。

      但是如果远端Logstash server因故障停止运行，数据便会丢失，所以后续可能会引入redis或者kafka作为消息中间件，将队列 中消息或数据间接传递给Logstash，Logstash过滤、分析后将数据传递给Elasticsearch存储，最后由Kibana将日志和数据呈现 给用户。这样即使远端logstash宕机，消息也会存在消息中间件，等待logstash恢复，继续消费消息，从而避免数据丢失。

Logstash工作原理

      Logstash事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志， webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。

 Input：输入数据到logstash。

一些常用的输入为：

• file：从文件系统的文件中读取，类似于tail -f命令
• syslog：在514端口上监听系统日志消息，并根据RFC3164标准进行解析
• redis：从redis service中读取
• beats：从filebeat中读取 

Filters：数据中间处理，对数据进行操作。

一些常用的过滤器为：

• grok：解析任意文本数据，Grok 是 Logstash 最重要的插件。它的主要作用就是将文本格式的字符串，转换成为具体的结构化 的数据，配合正则表达式使用。内置120多个解析语法。

          官方提供的grok表达式：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

          grok在线调试： https://grokdebug.herokuapp.com/

• mutate：对字段进行转换。例如对字段进行删除、替换、修改、重命名等。
• drop：丢弃一部分events不进行处理。
• clone：拷贝 event，这个过程中也可以添加或移除字段。
• geoip：添加地理信息(为前台kibana图形化展示使用)

Outputs：outputs是logstash处理管道的最末端组件。 

一个event可以在处理过程中经过多重输出，但是一旦所有的outputs都执行结束，这个event也就完成生命周期。 一些常见的outputs为：

• elasticsearch：可以高效的保存数据，并且能够方便和简单的进行查询。
• file：将event数据保存到文件中。
• graphite：将event数据发送到图形化组件中，一个很流行的开源存储图形化展示的组件。
• Codecs：codecs 是基于数据流的过滤器，它可以作为input，output的一部分配置。Codecs可以帮助你轻松的分割发送过来已经被序列化的数据。 

一些常见的codecs：

• json：使用json格式对数据进行编码/解码。
• multiline：将汇多个事件中数据汇总为一个单一的行。比如：java异常信息和堆栈信息。

安装ElasticSearch

下载安装ElasticSearch 6.3.0

mkdir es
cd es
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.3.0.tar.gz
tar -zxvf elasticsearch-6.3.0.tar.gz

修改es配置文件

vim /config/elasticsearch.yml

#本机ip
network.host: xxx.xxx.xx.xxx
#默认监听端口
http.port: 9200

运行ElasticSearch 6.3.0

配置完成，es不允许root运行，新建个帐号来运行

useradd elk
passwd elk

输入俩次密码

回到上级目录并更改elasticsearch的拥有者

chown -R elk elasticsearch-6.3.0

切换到elk用户

su elk

后台运行它，并将启动日志记录到es.log(自己建log文件夹，并给elk用户赋log文件夹操作权限，当然直接后台启动默认log也可 以)

nohup ./bin/elasticsearch >/usr/local/es/log/es.log

安装Logstash

下载安装logstash

mkdir logstash
cd logstash
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.3.0.tar.gz
tar -zxvf logstash-6.3.0.tar.gz
cd logstash-6.3.0/

创建一个配置文件(logstash-es.conf)

vim config/logstash-es.conf

input {
  tcp {
    port => 4560
    codec => json_lines
  }
}
output {
  elasticsearch {
    hosts => "3x.1xx.7x.1xx"
    index => "log_%{+YYYY.MM.dd}_%{[appname]}"
  }
  stdout {
    codec => rubydebug
  }
}

注意：

1.input.tcp : 中配置的是本机地址，ip和端口必须和springboot的logback.xml中的配置完全一样，不能一个配ip一个localhost

2.output.elasticsearch : 配置elasticsearch服务器的ip

3.%{[appname]} : 引用springboot的logback.xml中配置的变量

4.output.stdout : 在终端显示输出信息（可以不配置）

5.conf文件一定要严格按照tab缩进，否则启动logstash时会异常

运行logstash

nohup ./bin/logstash -f ./config/logstash-es.conf >/usr/local/logstash/log/logstash.log &

安装Kibana

mkdir kibana
cd kibana
wget https://artifacts.elastic.co/downloads/kibana/kibana-6.3.0-linux-x86_64.tar.gz
tar -zxvf kibana-6.3.0-linux-x86_64.tar.gz
cd kibana-6.3.0-linux-x86_64

修改配置文件

vim config/kibana.y

server.host: "0.0.0.0"
elasticsearch.url: "http://your es ip:port"
elasticsearch.username: "elastic"
elasticsearch.password: "changeme

运行kibana

nohup ./bin/kibana >/usr/local/kibana/log/kibana.log &

SpringBoot整合Logback，并打印log到logstash

项目pom.xml文件引入logback依赖

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>4.11</version>
</dependency>

 resource里添加logback.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<property resource="application.properties"/>
<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<destination>3x.1xx.7x.1xx:4560</destination>
<!-- <filter class="ch.qos.logback.classic.filter.LevelFilter">-->
<!-- &lt;!&ndash;过滤 INFO&ndash;&gt;-->
<!-- <level>INFO</level>-->
<!-- &lt;!&ndash;匹配到就禁止&ndash;&gt;-->
<!-- <onMatch>DENY</onMatch>-->
<!-- &lt;!&ndash;没有匹配到就允许&ndash;&gt;-->
<!-- <onMismatch>ACCEPT</onMismatch>-->
<!-- </filter>-->
<encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder">
<customFields>{"appname":"app-server1"}</customFields>
</encoder>
</appender>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder charset="UTF-8"> <!--encoder 可以指定字符集，对于中文输出有意义-->
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<logger name="com.ntnikka" level="Error" additivity="false">
<appender-ref ref="LOGSTASH"/>
</logger>
<logger name="com.ntnikka" level="INFO" additivity="false">
<appender-ref ref="STDOUT" />
</logger>
<!-- <root level="Error">-->
<!-- <appender-ref ref="LOGSTASH" />-->
<!-- <appender-ref ref="STDOUT" />-->
<!-- </root>-->
</configuration>

配置logstash IP 和 PORT

<destination>3x.1xx.7x.1xx:4560</destination>

多台IP或者多个端口logstash配置

<connectionStrategy>
    <roundRobin>
        <connectionTTL>5 minutes</connectionTTL>
    </roundRobin>
</connectionStrategy>

向logstash输出日志如果有多个logstash IP或端口可以轮询负载各端口

不同的应用模块可以配置不同的模块名，便于log查找

<customFields>{"appname":"app-server1"}</customFields>

{"appname":"app-server1","xxx":"xxx",..}如果有需要也可以添加其他自定义参数

测试

编写demo示例

@RestController
@RequestMapping("/test")
public class demoController {
  private static final Logger logger = LoggerFactory.getLogger(demoController.class);
  @Autowired
  demoService demoService;
  @sysLog
  @RequestMapping("/sysLog")
  public String testSysLog(){
    // demoService.demoMethod();
    divide();
    return "test demo";
  }
  public static void divide(){
    // int i = 10 /0;
    throw new MaynException("MaynException", "test MaynException",40000);
  }
}

项目中异常统一处理，编写handle类,也可以程序中自己trycatch异常，在catch代码块中自己通过logback打印异常日志

/**
* 异常处理器
*/
@RestControllerAdvice
public class MaynExceptionHandler {
  private Logger logger = LoggerFactory.getLogger(this.getClass());

  /**
  * 处理自定义异常
  */
  @ExceptionHandler(MaynException.class)
  public R handleMaynException(MaynException e) { R r = new R();
    r.put("code", e.getCode());
    r.put("msg", e.getMessage()); logger.error(e.getMessage(),e); return r;
  }

  @ExceptionHandler(DuplicateKeyException.class)
  public R handleDuplicateKeyException(DuplicateKeyException e) { logger.error(e.getMessage(), e);
    return R.error("数据库中已存在该记录");
  }


  @ExceptionHandler(Exception.class) 
  public R handleException(Exception e) {
    System.out.println("===================== enter exceptionHandle ================="); logger.error(e.getMessage(), e);
    return R.error();
  }
}

在统一处理中记录异常日志

logger.error(e.getMessage(), e);

启动springboot项目，调用接口测试，控制台输出

08:48:23.369 [http-nio-9090-exec-5] ERROR com.ntnikka.common.RRExceptionHandler - MaynException
com.ntnikka.exception.MaynException: MaynException
at com.ntnikka.rhlogsystem.controller.demoController.divide(demoController.java:36)
at com.ntnikka.rhlogsystem.controller.demoController.testSysLog(demoController.java:30)
at
com.ntnikka.rhlogsystem.controller.demoController$$FastClassBySpringCGLIB$$dc504d05.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at
org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:7
49)
at
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at
org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoin
Point.java:88)
at com.ntnikka.aspect.SysLogAspect.around(SysLogAspect.java:52)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvic
e.java:644)
at
org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:633)
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:70)
at
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at
org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:93
)
at
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at
org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688
)
at
com.ntnikka.rhlogsystem.controller.demoController$$EnhancerBySpringCGLIB$$213fe19f.testSysLog(<generated>)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:189)
at
org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java
:138)
at
org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(Servle
tInvocableHandlerMethod.java:102)
at
org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(Req
uestMappingHandlerAdapter.java:892)
at
org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestM
appingHandlerAdapter.java:797)
at
org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapte
r.java:87)
at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1038)
at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:942)
at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1005)
at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:897)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:634)
at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:882)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:741)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at com.alibaba.druid.support.http.WebStatFilter.doFilter(WebStatFilter.java:123)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:92)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at
org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:93)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at
org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200)
at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)
at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)
at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:200)
at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)
at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:490)
at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:139)
at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)
at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74)
at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343)
at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:408)
at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)
at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:834)
at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1415)
at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
at java.lang.Thread.run(Thread.java:748)

查看kibana

 

 异常日志成功输出到kibana，项目整合logback成功