超哥 docker部署ELK(logstash、elasticsearch、kibana)，监控日志
文章目录
[隐藏]
- 一、每个单独部署
- 1、elasticsearch部署
- 2、logstash部署
- 3、kibana部署
- 二、docker-compose一起部署
- 1、第一步在docker上安装ELK
- 2、springboot日志系统配置logstash
- 3、配置kibana，现在只要服务器通过指定的Tag打印日志，日志信息将会上传logstash解析，并且存储到elasticsearch，然后只需要kibana配置对应的elasticsearch的index即可看到所需的日志信息。
- 4、接下讲一下我们工程中关于http接口日志的配置
由于是首次部署，第一次想着是单独部署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"> roperty resource="application.properties">roperty> <!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径--> roperty name="LOG_HOME" value="${logging.levelfile}" /> roperty name="LOG_LEVEL" value="${logging.all.level}" /> <!-- 控制台输出 --> <appender name="STDOUT" > <encoder > <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符--> attern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50}:%L - %msg %nattern> <charset>UTF-8</charset> </encoder> </appender> <!-- 按照每天生成日志文件 --> <appender name="FILE" > <rollingPolicy > <!--日志文件输出的文件名--> <FileNamePattern>${LOG_HOME}/data-center-service.log.%d{yyyy-MM-dd}.log</FileNamePattern> <!--日志文件保留天数--> <MaxHistory>30</MaxHistory> </rollingPolicy> <encoder > <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符--> attern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%nattern> <charset>UTF-8</charset> </encoder> <!--日志文件最大的大小--> <triggeringPolicy > <MaxFileSize>100MB</MaxFileSize> </triggeringPolicy> </appender> <appender name="logstash" > <destination>${logstash.ip_port}</destination> <encoder charset="UTF-8" /> <queueSize>1048576</queueSize> <keepAliveDuration>5 minutes</keepAliveDuration> <!--<customFields>{"application-name":"data-repo-interface"}</customFields>--> <!--<filter > <level>INFO</level> </filter> <filter > <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" > <destination>${logstash.ip_port}</destination> <encoder charset="UTF-8" /> <queueSize>1048576</queueSize> <keepAliveDuration>5 minutes</keepAliveDuration> <!--<customFields>{"application-name":"data-repo-interface"}</customFields>--> <!--<filter > <level>INFO</level> </filter> <filter > <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(); } 原文出处：csdn -> https://blog.csdn.net/zhizhuodewo6/article/details/86630712
本站所发布的一切资源仅限用于学习和研究目的；不得将上述内容用于商业或者非法用途，否则，一切后果请用户自负。本站信息来自网络，版权争议与本站无关。您必须在下载后的24个小时之内，从您的电脑中彻底删除上述内容。如果您喜欢该程序，请支持正版软件，购买注册，得到更好的正版服务。如果侵犯你的利益，请发送邮箱到 [email protected]，我们会很快的为您处理。
