【SpringCloud&Alibaba开源全栈项目实战】：SpringBoot整合ELK实现分布式日志收集

一. 前言

其实早前就想计划出这篇文章，但是最近主要精力在完善微服务、系统权限设计、微信小程序和管理前端的功能，不过好在有群里小伙伴的一起帮忙反馈问题，基础版的功能已经差不多，也在此谢过，希望今后大家还是能够相互学习，一起进步~

ELK是Elasticsearch、Logstash、Kibana三个开源软件的组合，相信很多童鞋使用ELK有去做过分布式日志收集。流程概括为：微服务应用把Logback输出的日志通过HTTP传输至LogStash，然后经过分析过滤，转发至ES，再由Kibana提供检索和统计可视化界面。

在本实战案例中，使用Spring AOP、Logback横切认证接口来记录用户登录日志，收集到ELK，通过SpringBoot整合RestHighLevelClient实现对ElasticSearch数据检索和统计。从日志搜集到数据统计，一次性的走个完整，快速入门ElasticSearch。

本篇涉及的前后端全部源码已上传gitee和github，熟悉有来项目的童鞋快速过一下步骤即可。

二. 需求

基于ELK的日志搜集的功能，本篇实现的需求如下：

1. 记录系统用户登录日志，信息包括用户IP、登录耗时、认证令牌JWT
2. 统计十天内用户登录次数、今日访问IP和总访问IP
3. 充分利用记录的JWT信息，通过黑名单的方式让JWT失效实现强制下线

实现效果：

访问地址：http://www.youlai.tech

• Kibana日志可视化统计

• 登录次数统计、今日访问IP统计、总访问IP统计

• 登录信息，强制用户下线，演示的是自己强制自己下线的效果

三. Docker快速搭建ELK环境

1. 拉取镜像

docker pull elasticsearch:7.10.1
docker pull kibana:7.10.1
docker pull logstash:7.10.1

2. elasticsearch部署

1. 环境准备

# 创建文件
mkdir -p /opt/elasticsearch/{plugins,data}  /etc/elasticsearch
touch /etc/elasticsearch/elasticsearch.yml 
chmod -R 777 /opt/elasticsearch/data/  
vim /etc/elasticsearch/elasticsearch.yml
# 写入
cluster.name: elasticsearch
http.cors.enabled: true                               
http.cors.allow-origin: "*"                     
http.host: 0.0.0.0
node.max_local_storage_nodes: 100

2. 启动容器

docker run -d --name=elasticsearch --restart=always \
-e discovery.type=single-node \
-e ES_JAVA_OPTS="-Xms256m -Xmx256m" \
-p 9200:9200 \
-p 9300:9300 \
-v /etc/elasticsearch/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml \
-v /opt/elasticsearch/data:/usr/share/elasticsearch/data \
-v /opt/elasticsearch/plugins:/usr/share/elasticsearch/plugins \
elasticsearch:7.10.1

3. 验证和查看ElasticSearch版本

curl -XGET localhost:9200

2. kibana部署

1. 环境准备

# 创建文件
mkdir -p /etc/kibana
vim /etc/kibana/kibana.yml

# 写入
server.name: kibana
server.host: "0"
elasticsearch.hosts: [ "http://elasticsearch:9200" ]
i18n.locale: "zh-CN"

2. 启动容器

docker run -d --restart always -p 5601:5601 --name kibana --link elasticsearch \
-e ELASTICSEARCH_URL=http://elasticsearch:9200 \
-v /etc/kibana/kibana.yml:/usr/share/kibana/config/kibana.yml \
kibana:7.10.1

3. logstash部署

1. 环境准备

• 配置 logstash.yml

# 创建文件
mkdir -p /etc/logstash/config
vim /etc/logstash/config/logstash.yml

# 写入
http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]
xpack.management.pipeline.id: ["main"]

• 配置pipeline.yml

# 创建文件
vim  /etc/logstash/config/pipeline.yml 

# 写入(注意空格)
- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline/logstash.config"

• 配置logstash.conf

# 创建文件
mkdir -p /etc/logstash/pipeline
vim /etc/logstash/pipeline/logstash.conf 

# 写入
input {
    tcp {
      port => 5044
      mode => "server"
      host => "0.0.0.0"
      codec => json_lines
    }
}
filter{

}
output {
    elasticsearch {
        hosts => ["elasticsearch:9200"]
        # 索引名称，没有会自动创建
        index => "%{[project]}-%{[action]}-%{+YYYY-MM-dd}"
    }
}

2. 启动容器

docker run -d --restart always -p 5044:5044 -p 9600:9600 --name logstash --link elasticsearch \
-v /etc/logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml \
-v /etc/logstash/config/pipeline.yml:/usr/share/logstash/config/pipeline.yml \
-v /etc/logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \
logstash:7.10.1

4. 测试

访问： http://localhost:5601/

四. Spring AOP + Logback 横切打印登录日志

1. Spring AOP横切认证接口添加日志

代码坐标： common-web#LoginLogAspect

@Aspect
@Component
@AllArgsConstructor
@Slf4j
@ConditionalOnProperty(value = "spring.application.name", havingValue = "youlai-auth")
public class LoginLogAspect {

    @Pointcut("execution(public * com.youlai.auth.controller.AuthController.postAccessToken(..))")
    public void Log() {
    }

    @Around("Log()")
    public Object doAround(ProceedingJoinPoint joinPoint) throws Throwable {

        LocalDateTime startTime = LocalDateTime.now();
        Object result = joinPoint.proceed();

        // 获取请求信息
        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        HttpServletRequest request = attributes.getRequest();

        // 刷新token不记录
        String grantType=request.getParameter(AuthConstants.GRANT_TYPE_KEY);
        if(grantType.equals(AuthConstants.REFRESH_TOKEN)){
            return result;
        }

        // 时间统计
        LocalDateTime endTime = LocalDateTime.now();
        long elapsedTime = Duration.between(startTime, endTime).toMillis(); // 请求耗时（毫秒）

        // 获取接口描述信息
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        String description = signature.getMethod().getAnnotation(ApiOperation.class).value();// 方法描述

        String username = request.getParameter(AuthConstants.USER_NAME_KEY); // 登录用户名
        String date = startTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd")); // 索引名需要，因为默认生成索引的date时区不一致

        // 获取token
        String token = Strings.EMPTY;
        if (request != null) {
            JSONObject jsOnObject= JSONUtil.parseObj(result);
            token = jsonObject.getStr("value");
        }

        String clientIP = IPUtils.getIpAddr(request);  // 客户端请求IP（注意：如果使用Nginx代理需配置）
        String region = IPUtils.getCityInfo(clientIP); // IP对应的城市信息

        // MDC 扩展logback字段，具体请看logback-spring.xml的自定义日志输出格式
        MDC.put("elapsedTime", StrUtil.toString(elapsedTime));
        MDC.put("description", description);
        MDC.put("region", region);
        MDC.put("username", username);
        MDC.put("date", date);
        MDC.put("token", token);
        MDC.put("clientIP", clientIP);

        log.info("{} 登录，耗费时间 {} 毫秒", username, elapsedTime); // 收集日志这里必须打印一条日志，内容随便吧，记录在message字段，具体看logback-spring.xml文件
        return result;
    }
}

2. Logback日志上传至LogStash

代码坐标：common-web#logback-spring.xml

    localhost:5044
    
        
            
                Asia/Shanghai
            
            
            
                
                    {
                    "project": "${APP_NAME}",
                    "date": "%X{date}", 
                    "action":"login",
                    "pid": "${PID:-}",
                    "thread": "%thread",
                    "message": "%message",
                    "elapsedTime": "%X{elapsedTime}",
                    "username":"%X{username}",
                    "clientIP": "%X{clientIP}",
                    "region":"%X{region}",
                    "token":"%X{token}",
                    "loginTime": "%date{\"yyyy-MM-dd HH:mm:ss\"}",
                    "description":"%X{description}"
                    }
                
            
        
    
    5 minutes




    

• localhost:5044 Logstash配置的input收集数据的监听
• %X{username} 输出MDC添加的username的值

五. SpringBoot整合ElasticSearch客户端RestHighLevelClient

1. pom依赖

代码坐标： common-elasticsearch#pom.xml

客户端的版本需和服务器的版本对应，这里也就是7.10.1

    org.elasticsearch.client
    elasticsearch-rest-high-level-client
    
        
            org.elasticsearch.client
            elasticsearch-rest-client
        

        
            elasticsearch
            org.elasticsearch
        
    



    org.elasticsearch
    elasticsearch
    7.10.1



    org.elasticsearch.client
    elasticsearch-rest-client
    7.10.1



    org.springframework.boot
    spring-boot-starter-data-elasticsearch

2. yml 配置

spring:
  elasticsearch:
    rest:
      uris: ["http://localhost:9200"]
      cluster-nodes:
        - localhost:9200

3. RestHighLevelClientConfig 配置类

代码坐标： common-elasticsearch#RestHighLevelClientConfig

@ConfigurationProperties(prefix = "spring.elasticsearch.rest")
@Configuration
@AllArgsConstructor
public class RestHighLevelClientConfig {

    @Setter
    private List clusterNodes;

    @Bean
    public RestHighLevelClient restHighLevelClient() {

        HttpHost[] hosts = clusterNodes.stream()
                .map(this::buildHttpHost) // eg: new HttpHost("127.0.0.1", 9200, "http")
                .toArray(HttpHost[]::new);
        return new RestHighLevelClient(RestClient.builder(hosts));
    }

    private HttpHost buildHttpHost(String node) {
        String[] nodeInfo = node.split(":");
        return new HttpHost(nodeInfo[0].trim(), Integer.parseInt(nodeInfo[1].trim()), "http");
    }
}

4. RestHighLevelClient API封装

代码坐标： common-elasticsearch#ElasticSearchService

• 暂只简单封装实现需求里需要的几个方法，计数、去重计数、日期聚合统计、列表查询、分页查询、删除，后续可扩展...

@Service
@AllArgsConstructor
public class ElasticSearchService {

    private RestHighLevelClient client;

    /**
     * 计数
     */
    @SneakyThrows
    public long count(QueryBuilder queryBuilder, String... indices) {
        // 构造请求
        CountRequest countRequest = new CountRequest(indices);
        countRequest.query(queryBuilder);

        // 执行请求
        CountResponse countRespOnse= client.count(countRequest, RequestOptions.DEFAULT);
        long count = countResponse.getCount();
        return count;
    }

    /**
     * 去重计数
     */
    @SneakyThrows
    public long countDistinct(QueryBuilder queryBuilder, String field, String... indices) {
        String distinctKey = "distinctKey"; // 自定义计数去重key，保证上下文一致

        // 构造计数聚合 cardinality:集合中元素的个数
        CardinalityAggregationBuilder aggregatiOnBuilder= AggregationBuilders
                .cardinality(distinctKey).field(field);

        // 构造搜索源
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder).aggregation(aggregationBuilder);

        // 构造请求
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchSourceBuilder);

        // 执行请求
        SearchResponse searchRespOnse= client.search(searchRequest, RequestOptions.DEFAULT);
        ParsedCardinality result = searchResponse.getAggregations().get(distinctKey);
        return result.getValue();
    }

    /**
     * 日期聚合统计
     *
     * @param queryBuilder 查询条件
     * @param field        聚合字段，如：登录日志的 date 字段
     * @param interval     统计时间间隔，如：1天、1周
     * @param indices      索引名称
     * @return
     */
    @SneakyThrows
    public Map dateHistogram(QueryBuilder queryBuilder, String field, DateHistogramInterval interval, String... indices) {

        String dateHistogramKey = "dateHistogramKey"; // 自定义日期聚合key，保证上下文一致

        // 构造聚合
        AggregationBuilder aggregatiOnBuilder= AggregationBuilders
                .dateHistogram(dateHistogramKey) //自定义统计名，和下文获取需一致
                .field(field) // 日期字段名
                .format("yyyy-MM-dd") // 时间格式
                .calendarInterval(interval) // 日历间隔，例： 1s->1秒 1d->1天 1w->1周 1M->1月 1y->1年 ...
                .minDocCount(0); // 最小文档数，比该值小就忽略

        // 构造搜索源
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder
                .query(queryBuilder)
                .aggregation(aggregationBuilder)
                .size(0);

        // 构造SearchRequest
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchSourceBuilder);

        searchRequest.indicesOptions(
                IndicesOptions.fromOptions(
                        true, // 是否忽略不可用索引
                        true, // 是否允许索引不存在
                        true, // 通配符表达式将扩展为打开的索引
                        false // 通配符表达式将扩展为关闭的索引
                ));

        // 执行请求
        SearchResponse searchRespOnse= client.search(searchRequest, RequestOptions.DEFAULT);

        // 处理结果
        ParsedDateHistogram dateHistogram = searchResponse.getAggregations().get(dateHistogramKey);

        Iterator iterator = dateHistogram.getBuckets().iterator();

        Map map = new HashMap<>();
        while (iterator.hasNext()) {
            Histogram.Bucket bucket = iterator.next();
            map.put(bucket.getKeyAsString(), bucket.getDocCount());
        }
        return map;
    }

    /**
     * 列表查询
     */
    @SneakyThrows
    public  List search(QueryBuilder queryBuilder, Class clazz, String... indices) {
        List list = this.search(queryBuilder, null, 1, ESConstants.DEFAULT_PAGE_SIZE, clazz, indices);
        return list;
    }

    /**
     * 分页列表查询
     */
    @SneakyThrows
    public  List search(QueryBuilder queryBuilder, SortBuilder sortBuilder, Integer page, Integer size, Class clazz, String... indices) {
        // 构造SearchSourceBuilder
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder);
        searchSourceBuilder.sort(sortBuilder);
        searchSourceBuilder.from((page - 1) * size);
        searchSourceBuilder.size(size);
        // 构造SearchRequest
        SearchRequest searchRequest = new SearchRequest(indices);
        searchRequest.source(searchSourceBuilder);
        // 执行请求
        SearchResponse searchRespOnse= client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();

        List list = CollectionUtil.newArrayList();
        for (SearchHit hit : searchHits) {
            T t = JSONUtil.toBean(hit.getSourceAsString(), clazz);
            t.setId(hit.getId()); // 数据的唯一标识
            t.setIndex(hit.getIndex());// 索引
            list.add(t);
        }
        return list;
    }

    /**
     * 删除
     */
    @SneakyThrows
    public boolean deleteById(String id, String index) {
        DeleteRequest deleteRequest = new DeleteRequest(index,id);
        DeleteResponse deleteRespOnse= client.delete(deleteRequest, RequestOptions.DEFAULT);
        return true;
    }
}

六. 后台接口

在SpringBoot整合了ElasticSearch的高级客户端RestHighLevelClient，以及简单了封装方法之后，接下来就准备为前端提供统计数据、分页列表查询记录、根据ID删除记录接口了。

1. 首页控制台

首页控制台需要今日IP访问数，历史总IP访问数、近十天每天的登录次数统计，具体代码如下：

代码坐标： youlai-admin#DashboardController

@Api(tags = "首页控制台")
@RestController
@RequestMapping("/api.admin/v1/dashboard")
@Slf4j
@AllArgsConstructor
public class DashboardController {

    ElasticSearchService elasticSearchService;

    @ApiOperation(value = "控制台数据")
    @GetMapping
    public Result data() {
        Map data = new HashMap<>();

        // 今日IP数
        long todayIpCount = getTodayIpCount();
        data.put("todayIpCount", todayIpCount);

        // 总IP数
        long totalIpCount = getTotalIpCount();
        data.put("totalIpCount", totalIpCount);

        // 登录统计
        int days = 10; // 统计天数
        Map loginCount = getLoginCount(days);
        data.put("loginCount", loginCount);

        return Result.success(data);
    }

    
    private long getTodayIpCount() {
        String date = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery("date", date);
        String indexName = ESConstants.LOGIN_INDEX_PATTERN + date; //索引名称
        
        // 这里使用clientIP聚合计数，为什么加.keyword后缀呢？下文给出截图
        long todayIpCount = elasticSearchService.countDistinct(termQueryBuilder, "clientIP.keyword", indexName);
        return todayIpCount;
    }

    private long getTotalIpCount() {
        long totalIpCount = elasticSearchService.countDistinct(null, "clientIP.keyword", ESConstants.LOGIN_INDEX_PATTERN);
        return totalIpCount;
    }

    private Map getLoginCount(int days) {

        LocalDateTime now = LocalDateTime.now();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

        String startDate = now.plusDays(-days).format(formatter);
        String endDate = now.format(formatter);

        String[] indices = new String[days]; // 查询ES索引数组
        String[] xData = new String[days]; // 柱状图x轴数据
        for (int i = 0; i  totalCountMap = elasticSearchService.dateHistogram(
                boolQueryBuilder,
                "date", // 根据date字段聚合统计登录数 logback-spring.xml 中的自定义扩展字段 date
                DateHistogramInterval.days(1),
                indices);

        // 当前用户统计
        HttpServletRequest request = RequestUtils.getRequest();
        String clientIP = IPUtils.getIpAddr(request);

        boolQueryBuilder.must(QueryBuilders.termQuery("clientIP", clientIP));
        Map myCountMap = elasticSearchService.dateHistogram(boolQueryBuilder, "date", DateHistogramInterval.days(1), indices);


        // 组装echarts数据
        Long[] totalCount = new Long[days];
        Long[] myCount = new Long[days];

        Arrays.sort(xData);// 默认升序
        for (int i = 0; i  map = new HashMap<>(4);

        map.put("xData", xData); // x轴坐标
        map.put("totalCount", totalCount); // 总数
        map.put("myCount", myCount); // 我的

        return map;
    }
}

• 聚合字段clientIP为什么添加.keyword后缀?

2. 登录记录分页查询接口

代码坐标： youlai-admin # LoginRecordController

@Api(tags = "登录记录")
@RestController
@RequestMapping("/api.admin/v1/login_records")
@Slf4j
@AllArgsConstructor
public class LoginRecordController {

    ElasticSearchService elasticSearchService;

    ITokenService tokenService;

    @ApiOperation(value = "列表分页")
    @ApiImplicitParams({
            @ApiImplicitParam(name = "page", value = "页码", defaultValue = "1", paramType = "query", dataType = "Long"),
            @ApiImplicitParam(name = "limit", value = "每页数量", defaultValue = "10", paramType = "query", dataType = "Long"),
            @ApiImplicitParam(name = "startDate", value = "开始日期", paramType = "query", dataType = "String"),
            @ApiImplicitParam(name = "endDate", value = "结束日期", paramType = "query", dataType = "String"),
            @ApiImplicitParam(name = "clientIP", value = "客户端IP", paramType = "query", dataType = "String")
    })
    @GetMapping
    public Result list(
            Integer page,
            Integer limit,
            String startDate,
            String endDate,
            String clientIP
    ) {

        // 日期范围
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("date");

        if (StrUtil.isNotBlank(startDate)) {
            rangeQueryBuilder.from(startDate);
        }
        if (StrUtil.isNotBlank(endDate)) {
            rangeQueryBuilder.to(endDate);
        }

        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(rangeQueryBuilder);

        if (StrUtil.isNotBlank(clientIP)) {
            queryBuilder.must(QueryBuilders.wildcardQuery("clientIP", "*" + clientIP + "*"));
        }
        // 总记录数
        long count = elasticSearchService.count(queryBuilder, ESConstants.LOGIN_INDEX_PATTERN);

        // 排序
        FieldSortBuilder sortBuilder = new FieldSortBuilder("@timestamp").order(SortOrder.DESC);

        // 分页查询
        List list = elasticSearchService.search(queryBuilder, sortBuilder, page, limit, LoginRecord.class, ESConstants.LOGIN_INDEX_PATTERN);

        // 遍历获取会话状态
        list.forEach(item -> {
            String token = item.getToken();
            int tokenStatus = 0;
            if (StrUtil.isNotBlank(token)) {
                tokenStatus = tokenService.getTokenStatus(item.getToken());
            }
            item.setStatus(tokenStatus);
        });

        return Result.success(list, count);
    }


    @ApiOperation(value = "删除登录记录")
    @ApiImplicitParam(name = "ids", value = "id集合", required = true, paramType = "query", dataType = "String")
    @DeleteMapping
    public Result delete(@RequestBody List documents) {
        documents.forEach(document -> elasticSearchService.deleteById(document.getId(), document.getIndex()));
        return Result.success();
    }

}

3. 强制下线接口

代码坐标： youlai-admin#TokenController

• 这里还是将JWT添加至黑名单，然后在网关限制被加入黑名单的JWT登录

@Api(tags = "令牌接口")
@RestController
@RequestMapping("/api.admin/v1/tokens")
@Slf4j
@AllArgsConstructor
public class TokenController {

    ITokenService tokenService;

    @ApiOperation(value = "强制下线")
    @ApiImplicitParam(name = "token", value = "访问令牌", required = true, paramType = "query", dataType = "String")
    @PostMapping("/{token}/_invalidate")
    @SneakyThrows
    public Result invalidateToken(@PathVariable String token) {
        boolean status = tokenService.invalidateToken(token);
        return Result.judge(status);
    }

}

代码坐标： youlai-admin#TokenServiceImpl

@Override
@SneakyThrows
public boolean invalidateToken(String token) {

    JWTPayload payload = JWTUtils.getJWTPayload(token);

    // 计算是否过期
    long currentTimeSecOnds= System.currentTimeMillis() / 1000;
    Long exp = payload.getExp();
    if (exp 

docker exec -it youlai-auth /bin/sh
echo "Asia/Shanghai" > /etc/timezone
docker restart youlai-auth

proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;