MLK分布式日志处理

目录

场景以及优点

Java代码集合logstash，es进行检索

将信息通过logstash输入到es中

然后我们将es中的日志信息输出（做一个日志的接口）

---

 

场景以及优点

当系统发现故障时，那么我们需要登录服务器grep 等脚本工具去日志查找原因，在没有日志系统的情况下，如果服务器部署了多个实例，我们需要进每个实例的容器种去寻找日志文件（如果量毕比较大的话，比如每天产生一个新的logfile，那么多节点时间一长就会导致文件过多）

(50条消息) tail 命令详解_Coder杨公子的博客-CSDN博客_tail命令

优化：

我们可以将日志集中管理，并且提供集中检索的方法（ELK=elasticsearch+logstash+kibana），

logstash：负责数据的收集整理，相当于一个通道

Elasticsearch：数据的存储和分析

 logstash的工作原理：

logstash事件处理管道有三个阶段：1.输入->过滤器->输出，输入生成事件，然后过滤器去过滤修改他们将他们输出到其他地方

 1.首先docker pull一共logstash的tar包，进行解压得到镜像

 2.然后修改配置文件

 3.最后重启容器即可

Java代码集合logstash，es进行检索

将信息通过logstash输入到es中

1.logstash日志配置文件

"1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml" />
    <springProperty scope="context" name="springAppName" source="spring.application.name" />
    
    <property name="LOG_FILE" value="${BUILD_FOLDER:-build}/${springAppName}" />
    
    <property name="CONSOLE_LOG_PATTERN" value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />
    
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFOlevel>
        filter>
        
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}pattern>
            <charset>utf8charset>
        encoder>
    appender>
 
    
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>82.157.198.247:4560destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
    appender>
 
    
    <root level="DEBUG">
        <appender-ref ref="console" />
        <appender-ref ref="logstash" />
    root>
configuration>

2.启动类上利用LoggerFactory打印日志 

package com.wyh.logstash;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class LogStashApplication {
 
    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(LogStashApplication.class);
        logger.error("自己写的bug");
        SpringApplication.run(LogStashApplication.class, args);
    }
 
}

3.logstash依赖

<dependency>
            <groupId>net.logstash.logbackgroupId>
            <artifactId>logstash-logback-encoderartifactId>
            <version>6.3version>
        dependency>

此时会发现日志通过logstash这通道输出到es中

然后我们将es中的日志信息输出（做一个日志的接口）

0.实体类

package com.wyh.logstashsys.pojo;
 
import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.DateFormat;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;
 
import java.util.Date;
 
@Data
@Document(indexName = "test_log")
public class LogPojo {
 
    @Id
    private String id;
    @Field(type = FieldType.Integer)
    private Integer port;
    @Field(type = FieldType.Text)
    private String message;
    @Field(name = "@version",type = FieldType.Keyword)
    private String version;
    //注意：如果不配置date_time
    @Field(name = "@timestamp",type = FieldType.Date,format = DateFormat.date_time)
    private Date timestamp;
    @Field(type = FieldType.Keyword)
    private String host;
 
}

1.接口

public interface LogService {
    /**
     * 1.查询es中的日志信息
     * @return
     */
    List selectPage(Integer page,Integer rows);
}

2.业务实现类（指定时间范围得到hits，然后放到集合中进行返回）

 
@Service
public class LogServiceImpl implements LogService{
 
    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
 
    /**
     * 1.分页查询es中的分布式日志
     * @param page
     * @param rows
     * @return
     */
    @Override
    public List selectPage(Integer page, Integer rows) {
        //1.设置查询参数：小于当前时间15分钟的时间对象
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,-15);
        //2.查询最近15分钟以内的日志
        Query query = new NativeSearchQuery(QueryBuilders.rangeQuery("@timestamp").gte(calendar.getTime()));
        query.setPageable(PageRequest.of(page-1,rows));
        //3.es客户端进行查询
        SearchHits searchHits = elasticsearchRestTemplate.search(query, LogPojo.class);
 
        //4.对结果进行循环遍历输出
        Vector vector = new Vector<>();
        for (SearchHit searchHit : searchHits) {
            vector.add(searchHit.getContent());
        }
 
        return vector;
    }
}

 3.接口层

 /**
     * 1.得到分布式事务的分页结果
     * @param page
     * @param rows
     * @return
     */
    @RequestMapping("/")
    public List showLogs(Integer page,Integer rows){
        return logService.selectPage(page,rows);
    }