标签：Logstash

由于是首次部署，第一次想着是单独部署logstash、elasticsearch、kibana，然后通过配置实现日志的监控，以下为部署步骤，但是最终失败，只能采取docker-compose来部署，以下内容可以略过，仅作为参考。

一、每个单独部署

先部署elasticsearch，因为logstash要设置日志输出位置，而输出位置正是elasticsearch，所以需要先部署启动elasticsearch，logstash才能部署并启动成功。

1、elasticsearch部署

1.制作elasticsearch镜像

docker pull docker.elastic.co/elasticsearch/elasticsearch:6.5.4

2.创建并启动elasticsearch容器

docker run -p 9200:9200 -p 9300:9300 -e “discovery.type=single-node” -d docker.elastic.co/elasticsearch/elasticsearch:6.5.4

通过本地浏览器访问localhost:9200，返回如下内容，则证明elasticsearch部署成功

官方参考：https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html

2、logstash部署

1.制作logstash镜像

docker pull docker.elastic.co/logstash/logstash:6.5.4

2.创建并启动logstash容器

docker run --rm -it -p 4560:4560 -v /home/xijie/app/mylogstash/pipeline/:/usr/share/logstash/pipeline/ -d docker.elastic.co/logstash/logstash:6.5.4

注意文件pipeline的映射，在pipeline中需要创建logstash的配置文件logstash-springboot.conf，该文件内容如下：

input {
    tcp {
        port => 4560
        codec => json_lines
    }
}
output{
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

注意output中设置了elasticsearch的连接地址：localhost:9200

官方参考文章：https://www.elastic.co/guide/en/logstash/current/docker-config.html

3、kibana部署

1.制作镜像

docker pull docker.elastic.co/kibana/kibana:6.5.4

2.创建并启动容器

docker run -p 5601:5601 --mount type=bind,src=/home/xijie/app/mykibana/kibana.yml,dst=/usr/share/kibana/config/kibana.yml -d mysql/mysql-server:5.7

二、docker-compose一起部署

参考文章：https://www.jianshu.com/p/c2f6e80b2756

1、第一步在docker上安装ELK

1.创建目录

mkdir /home/xijie/app/myelk

2.从github上拉取部署elk所需资料

$ git clone https://github.com/deviantony/docker-elk.git

下载完毕的资料目录如下：

3.进入刚下载的文件夹内

$ cd docker-elk

4.通过docker-compose创建并启动容器

$ docker-compose up -d

5.这个时候通过docker ps可以看到logstash、elasticsearch、kibana容器已经创建并且启动。

可以看该elk容器的默认端口为：

• 5000: Logstash TCP input.
• 9200: Elasticsearch HTTP
• 9300: Elasticsearch TCP transport
• 5601: Kibana

Kibana的web入口：http://localhost:5601

6.接下来进行参数配置，来实现springboot通过ELK查看日志信息。

修改logstash的配置，

进入logstash配置文件所在目录：

cd /home/xijie/app/myelk/dokcer-elk/logstash/pipeline

打开配置文件：

vim logstash.conf

修改内容如下：

input{
        tcp {
                mode => "server"
                port => 5000
                codec => json_lines
                tags => ["data-http"]
        }
}
filter{
    json{
        source => "message"
        remove_field => ["message"]
    }
}
output{
    if "data-http" in [tags]{
        elasticsearch{
                hosts=> ["elasticsearch:9200"]
                index => "data-http-%{+YYYY.MM.dd}"
                }
        stdout{codec => rubydebug}
    }
}

注：

input标签为logstash进数据接口，filter标签为数据过滤器，output为数据出去接口。

input标签使用的是tcp，说明springboot客户端需要将日志传递到该接口，该接口正是logstash服务器接口。

filter将message字段去掉，只是为了当展示springboot的http请求接口的数据更加规整，而不是全部展示在message字段中。

output标签将数据传递给了elasticsearch，这里使用了if，当判断所出数据为所指定tag，才进行下面的配置。特别要注意index的配置，该值在kibana中需要使用，这里指定的index值为：data-http，要注意该值与tag是没有关系的，要注意区分。

参考文章：https://www.elastic.co/guide/en/logstash/current/index.html

这个时候重启elk即可。

进入docker-elk目录

/home/xijie/app/myelk/docker-elk

然后重启

docker-compose restart

关于elasticsearch、logstash、kibana的配置都在对应目录下的config文件夹中的.yml文件中，只需要修改该文件即可。

2、springboot日志系统配置logstash

1.pom中配置logstash

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

2.application.properties文件中配置如下：

#logstash日志收集地址，即logstash服务器地址
logstash.ip_port=172.168.0.165:5000
#日志保存级别
logging.all.level=info
#日志保存地址，该值与logstash没关系，是当日志存在本地File文件内的文件夹地址
logging.levelfile=/home/logs/data-center-service

3.logback.xml文件内容如下，该文件在resources目录

<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
    <property resource="application.properties"></property>
    <!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径-->
    <property name="LOG_HOME" value="${logging.levelfile}" />
    <property name="LOG_LEVEL" value="${logging.all.level}" />
    <!-- 控制台输出 -->
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50}:%L - %msg  %n</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>
    <!-- 按照每天生成日志文件 -->
    <appender name="FILE"  class="ch.qos.logback.core.rolling.RollingFileAppender">
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!--日志文件输出的文件名-->
            <FileNamePattern>${LOG_HOME}/data-center-service.log.%d{yyyy-MM-dd}.log</FileNamePattern>
            <!--日志文件保留天数-->
            <MaxHistory>30</MaxHistory>
        </rollingPolicy>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
            <charset>UTF-8</charset>
        </encoder>
        <!--日志文件最大的大小-->
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <MaxFileSize>100MB</MaxFileSize>
        </triggeringPolicy>
    </appender>
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>${logstash.ip_port}</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
        <queueSize>1048576</queueSize>
        <keepAliveDuration>5 minutes</keepAliveDuration>
        <!--<customFields>{"application-name":"data-repo-interface"}</customFields>-->
        <!--<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
            <evaluator> <!– 默认为 ch.qos.logback.classic.boolex.JaninoEventEvaluator –>
                <expression>return message.contains("billing");</expression>
            </evaluator>
            <OnMatch>ACCEPT</OnMatch>
            <OnMismatch>DENY</OnMismatch>
        </filter>-->
    </appender>

    <logger name="elk_logger" level="INFO" additivity="false">
        <appender-ref ref="logstash"/>
    </logger>

    <!--<logger name="com.zaxxer" level="${LOG_LEVEL}"/>-->
    <!--<logger name="org.apache.ibatis" level="${LOG_LEVEL}"/>-->
    <!--<logger name="org.mybatis.spring" level="${LOG_LEVEL}"/>-->
    <!--<logger name="org.springframework" level="${LOG_LEVEL}"/>-->
    <!--<logger name="java.sql.Connection" level="${LOG_LEVEL}"/>-->
    <!--<logger name="java.sql.Statement" level="${LOG_LEVEL}"/>-->
    <!--<logger name="java.sql.PreparedStatement" level="${LOG_LEVEL}"/>-->

    <!-- 日志输出级别 -->
    <root level="${LOG_LEVEL}">
        <appender-ref ref="STDOUT" />
        <appender-ref ref="FILE" />
        <!--<appender-ref ref="logstash" />-->
    </root>
</configuration>

在上述配置文件中

<appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>${logstash.ip_port}</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
        <queueSize>1048576</queueSize>
        <keepAliveDuration>5 minutes</keepAliveDuration>
        <!--<customFields>{"application-name":"data-repo-interface"}</customFields>-->
        <!--<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
            <evaluator> <!– 默认为 ch.qos.logback.classic.boolex.JaninoEventEvaluator –>
                <expression>return message.contains("billing");</expression>
            </evaluator>
            <OnMatch>ACCEPT</OnMatch>
            <OnMismatch>DENY</OnMismatch>
        </filter>-->
    </appender>

    <logger name="elk_logger" level="INFO" additivity="false">
        <appender-ref ref="logstash"/>
    </logger>

是logstash的主要配置。注意该配置中name=”elk_logger”，这样的指定，即

Logger elkLogger = LoggerFactory.getLogger("elk_logger");

当使用slf4j生成Logger时，只要指定Tag为“elk_logger”的日志输出或打印，都会上传到logstash服务器，并存储到elasticsearch，用kibana查看。

3、配置kibana，现在只要服务器通过指定的Tag打印日志，日志信息将会上传logstash解析，并且存储到elasticsearch，然后只需要kibana配置对应的elasticsearch的index即可看到所需的日志信息。

通过浏览器访问kibana，http://localhost:5601

由于本人是通过虚拟机部署的服务，而且虚拟机的ip为172.168.0.165,所以在宿主机中通过访问http://172.168.0.165:5601即可。

然后点击左侧的management模块。

接下来点击Index Patterns，创建index pattern

接下来填入信息，在index pattern框中填入上述创建ElK时logstash配置文件中的index信息“data-http-*”

填写的同时，下面会显示出elasticsearch中已经有该index得数据，（注意：在配置kibana时，应该先运行服务器，让服务区打印出对应的index日志，elasticsearch中也保存了该日志，这时才能配置kibana成功）。

然后点击右侧的next step。

在Time Filter field name列表中选择@timestamp，然后按下Create index pattern按钮及创建成功。

这时点击左侧的Discover模块，选中data-http-*标签即可。在右侧就显示出了日志信息。如下图：

这样整个ELK与日志系统就搭建完毕了。

4、接下讲一下我们工程中关于http接口日志的配置

由于我们提供的服务器接口是上游，是给其他部门服务的，这里会牵扯到大量的专业数据，而为了避免数据问题的纠纷与接口错误问题的排查，所以需要将特定的接口请求数据保存，这样可以通过resquest与response来排查到底是哪个部门的问题。

1.使用了AOP对每次请求进行日志拦截。

定义SysLogAspect类，该类内容如下：

package com;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.SpringContextUtil;
import com.LoggerEntity;
import com.MailService;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.reflect.Field;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Aspect
@Component
@Slf4j
public class SysLogAspect {

    Logger elkLogger = LoggerFactory.getLogger("elk_logger");


    /**
     * 开始时间
     */
    private long startTime = 0L;
    /**
     * 结束时间
     */
    private long endTime = 0L;

//  @Autowired
//  private ILogService logService;

    public static Map<String, Object> getKeyAndValue(Object obj) {
        Map<String, Object> map = new HashMap<>();
        // 得到类对象
        Class userCla = (Class) obj.getClass();
        /* 得到类中的所有属性集合 */
        Field[] fs = userCla.getFields();
        for (int i = 0; i < fs.length; i++) {
            Field f = fs[i];
            f.setAccessible(true); // 设置些属性是可以访问的
            Object val = new Object();
            try {
                val = f.get(obj);
                // 得到此属性的值
                map.put(f.getName(), val);// 设置键值
            } catch (IllegalArgumentException e) {
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }

        }
        return map;
    }

    //通过该注解来判断该接口请求是否进行cut
    @Pointcut("@annotation(com.hongdaoai.datastore.common.annotation.SysLog)")
    public void cutController() {
    }

    @Before("cutController()")
    public void doBeforeInServiceLayer(JoinPoint joinPoint) {
        log.debug("doBeforeInServiceLayer");
        startTime = System.currentTimeMillis();
    }

    @After("cutController()")
    public void doAfterInServiceLayer(JoinPoint joinPoint) {
        log.debug("doAfterInServiceLayer");
    }

    @Around("cutController()")
    public Object recordSysLog(ProceedingJoinPoint joinPoint) throws Throwable {

        RequestAttributes ra = RequestContextHolder.getRequestAttributes();

        ServletRequestAttributes sra = (ServletRequestAttributes) ra;

        HttpServletRequest request = sra.getRequest();

        HttpServletResponse response = sra.getResponse();

        //ELK日志实体类
        LoggerEntity elkLog = new LoggerEntity();

        //应用程序名称
        elkLog.setApplicationName(SpringContextUtil.getApplicationName());

        //profile.active
        elkLog.setProfileActive(SpringContextUtil.getActiveProfile());

        // 请求的类名
        elkLog.setClassName(joinPoint.getTarget().getClass().getName());

        // 请求的方法名
        elkLog.setMethodName(joinPoint.getSignature().getName());

        //请求完整地址
        elkLog.setUrl(request.getRequestURL().toString());

        //请求URI
        elkLog.setUri(request.getRequestURI());

        //请求类型
        elkLog.setRequestMethod(request.getMethod());

        String queryString = request.getQueryString();

        Object[] args = joinPoint.getArgs();

        String params = "";

        //获取请求参数集合并进行遍历拼接
        if (args.length > 0) {

            if ("POST".equals(request.getMethod())) {

                //param
                Object object = args[0];

                Map<String, Object> map = new HashMap<>();

                Map paramMap = getKeyAndValue(object);

                map.put("param", paramMap);

                if (args.length > 1) {

                    object = args[1];

                    Map bodyMap = getKeyAndValue(object);

                    map.put("body", bodyMap);
                }

                params = JSON.toJSONStringWithDateFormat(map, "yyyy-MM-dd HH:mm:ss", SerializerFeature.UseSingleQuotes);

            } else if ("GET".equals(request.getMethod())) {

                params = queryString;

            }
        }

        //请求参数内容(param)
        elkLog.setRequestParamData(params);

        //客户端IP
        elkLog.setClientIp(request.getRemoteAddr());

        //终端请求方式,普通请求,ajax请求
        elkLog.setRequestType(request.getHeader("X-Requested-With"));

        //sessionId
        elkLog.setSessionId(request.getRequestedSessionId());

        //请求时间
//        elkLog.setRequestDateTime(new Date(startTime));
        elkLog.setRequestDateTime(new Date());

        Object result = null;

        try {

            // 环绕通知 ProceedingJoinPoint执行proceed方法的作用是让目标方法执行，这也是环绕通知和前置、后置通知方法的一个最大区别。
            result = joinPoint.proceed();

        } catch (Exception e) {

            endTime = System.currentTimeMillis();

            //请求耗时(单位:毫秒)
            elkLog.setSpentTime(endTime - startTime);

            //接口返回时间
            elkLog.setResponseDateTime(new Date(endTime));

            //请求时httpStatusCode代码
            elkLog.setHttpStatusCode(String.valueOf(response.getStatus()));

            String elkLogData = JSON.toJSONStringWithDateFormat(elkLog, "yyyy-MM-dd HH:mm:ss.SSS");

            elkLogger.error(elkLogData);

//            log.error(e.getMessage(), e);

            throw e;

//            return result;

        }

        endTime = System.currentTimeMillis();

        //请求耗时(单位:毫秒)
        elkLog.setSpentTime(endTime - startTime);

        if (!elkLog.getMethodName().equals("getBookOriginalContent")) {

            //接口返回数据
            elkLog.setResponseData(JSON.toJSONStringWithDateFormat(result, "yyyy-MM-dd HH:mm:ss", SerializerFeature.UseSingleQuotes));

        }

        //接口返回时间
        elkLog.setResponseDateTime(new Date(endTime));

        //请求时httpStatusCode代码
        elkLog.setHttpStatusCode(String.valueOf(response.getStatus()));

//        if (SpringContextUtil.getActiveProfile().equals("prod")) {
        String s = JSON.toJSONStringWithDateFormat(elkLog, "yyyy-MM-dd HH:mm:ss.SSS");
        elkLogger.info(s);

        return result;
    }


}

SpringContextUtil类内容如下：

package com;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.Locale;

@Component
public class SpringContextUtil implements ApplicationContextAware {

    private static ApplicationContext context = null;

    // 传入线程中
    public static <T> T getBean(String beanName) {
        return (T) context.getBean(beanName);
    }

    // 国际化使用
    public static String getMessage(String key) {
        return context.getMessage(key, null, Locale.getDefault());
    }

    /// 获取应用程序名称
    public static String getApplicationName() {
        return context.getEnvironment().getProperty("spring.application.name");
    }

    /// 获取当前环境
    public static String getActiveProfile() {
        return context.getEnvironment().getActiveProfiles()[0];
    }

    /* (non Javadoc)
     * @Title: setApplicationContext
     * @Description: spring获取bean工具类
     * @param applicationContext
     * @throws BeansException
     * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext)
            throws BeansException {
        this.context = applicationContext;
    }
}

LoggerEntity类内容如下：

package com;

import lombok.Data;

import java.io.Serializable;
import java.util.Date;

//@Entity
//@Table(name = "t_logger_infos")
@Data
public class LoggerEntity implements Serializable {

    //应用程序名
    private String applicationName;

    //spring.profiles.active
    private String profileActive;

    //客户端请求ip
    private String clientIp;

    //客户端请求路径
    private String uri;

    //客户端请求完整路径
    private String url;

    //请求方法名
    private String methodName;

    //请求类名
    private String className;

    //终端请求方式,普通请求,ajax请求
    private String requestType;

    //请求方式method,post,get等
    private String requestMethod;

    //请求参数内容,json
    private String requestParamData;

    //请求body参数内容,json
    private String requestBodyData;

    //请求接口唯一session标识
    private String sessionId;

    //请求时间
    private Date requestDateTime;

    //接口返回时间
    private Date responseDateTime;

    //接口返回数据json
    private String responseData;

    //请求时httpStatusCode代码，如：200,400,404等
    private String httpStatusCode;

    //请求耗时秒单位
    private long spentTime;

}

SysLog注解类如下：

package com;

import java.lang.annotation.*;

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface SysLog {

    boolean isLog() default true;

}

在SysLogAspect切面类中使用@SysLog注解来判断该接口是否进行日志传递。

所以在我们写的接口中只要在方法上加入该注解，即可控制日志是否上传。

如该Controller中所定义接口：

@SysLog(isLog = true)
@RequestMapping(value = "/getDataList", method = RequestMethod.POST)
public ApiResult getInputDataList(@RequestBody VersionVo vo) {
    return success();
}