日志优化实践经验分享

作者: c80bbe47f715 | 来源:发表于2020-01-13 17:26 被阅读0次

背景

最近公司对原有单体应用进行业务拆分，将每个有自己特定功能的模块作为一个微服务，每个微服务单独部署，开发过程中遇到了一些问题，一是调用接口报错，虽有打印日志，但不知道是哪个服务出了问题，是前端传参有问题，还是系统A或者系统B提供的接口导致；二是接口出现性能问题时只能通过打印日志定位，三是log日志打印太随意，应该打error级别日志，使用info、warn级别等问题，总之，问题排查非常麻烦。下面我们来导论一下这个问题。

解决思路

定义日志打印规范，给每个请求生成一个唯一traceId，当请求调用其他服务时，我们传递这个traceId。在输出日志时，将这个traceId打印到日志文件中，这样从日志文件中，根据traceId就可以分析一个请求完整的调用链路。

一、日志打印优化实践

1.1 日志打印规范

针对项目中日志级别乱打，大量地输出无效日志带来的问题，我们在项目中定义了如下规范：

1.【必须】排查问题用的数据量较大详细日志打debug，关键业务流程节点、第三方接口日志打info，不需要监控的错误（如后台表单参数校验失败等）打warn，需要监控并处理的错误（系统不应该出现的错误）打error。

**2. **【必须】第三方接口根据情况尽可能打印完整入参出参，如果数据量较大的接口，至少用info打印关键字段，debug打印完整信息。

**3. **【必须】所有fallback打印error日志，包括调用参数，触发fallback的原因。

**4. **【建议】后台管理的增删改操作尽可能多记录日志，以便快速定位问题。

全局异常捕获类日志打印规范代码实现：

@RestControllerAdvice
public class BaseGlobalExceptionHandler {
    private static final Logger log = LoggerFactory.getLogger(BaseGlobalExceptionHandler.class);

    @ExceptionHandler({BaseBizException.class})
    @ResponseStatus(HttpStatus.BAD_REQUEST)
    public BaseResult<?> baseBusinessExceptionHandler(BaseBizException exception) {
        if (log.isWarnEnabled()) {
            log.warn("[BizException]业务异常信息 ex={}", exception.getMessage(), exception);
        }
        return BaseResult.error(exception.getCode(), exception.getMessage());
    }

    @ExceptionHandler(MethodArgumentNotValidException.class)
    @ResponseStatus(HttpStatus.BAD_REQUEST)
    public BaseResult methodArgumentNotValidException(MethodArgumentNotValidException exception) {
        return bindingResultHandler(exception.getBindingResult());
    }

    @ExceptionHandler(BindException.class)
    @ResponseStatus(HttpStatus.BAD_REQUEST)
    public BaseResult bindException(BindException exception) {
        return bindingResultHandler(exception.getBindingResult());
    }

    public BaseResult bindingResultHandler(BindingResult bindingResult){
        Map<String,String> errorMap = new HashMap<>(bindingResult.getFieldErrorCount());
        if (bindingResult.hasErrors()) {
            for (FieldError fieldError : bindingResult.getFieldErrors()) {
                errorMap.put(fieldError.getField(), fieldError.getDefaultMessage());
            }
        }
        if (log.isWarnEnabled()) {
            log.warn("[bindingResult]参数校验异常 ex={}", bindingResult);
        }
        return BaseResult.error(ErrorEnum.PARAM_ERROR.getCode(),          ErrorEnum.PARAM_ERROR.getMessage(), errorMap);
    }
    @ExceptionHandler({Exception.class})
    @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
    public BaseResult<?> exceptionHandler(Exception e) {
        if (log.isErrorEnabled()) {
            log.error("[InternalException]内部异常信息 ex={}", e.getMessage(), e);
        }
        return BaseResult.error(BaseErrorEnum.UNKNOW_SYSTEM_ERROR);
    }
}

业务代码非error级别异常处理，此处的异常因为是记录日志导致的，所以异常的时候，一是不影响主业务逻辑，二是不进行异常告警。

@Transactional(rollbackFor = Exception.class)
public void updateWorkOrderStatus(WorkOrderStatusDto workOrderStatus) {
    //......工单业务处理逻辑代码省略
    try {
        // 记录日志
        List<WorkOrderFile> workOrderFiles = workOrderFileService.selectList(new QueryWrapper<WorkOrderFile>()
                .lambda().eq(WorkOrderFile::getWorkOrderId, workOrderStatus.getWorkOrderId()));
        List<Integer> fileIds = workOrderFiles.stream().map(WorkOrderFile::getFileId).collect(Collectors.toList());
        saveWorkerOrderLog(workOrderStatus.getWorkOrderId(), workOrder.getDescription(), fileIds);
    } catch (Exception e) { 
        // 工单状态变更通知或记录日志异常不影响业务正常执行
        log.warn("工单状态变更通知或记录日志异常", e);
    }
}

业务代码处理error级别异常。此处库存异动处理失败，需要及时通知相关人员进行处理。

@Override
    public void dealChangeFromBoh(InventoryChangeFromBohDto dto, String batchNo) {
      //......业务处理代码省略
      // 库存异动
      try {
        handleChangeFromBoh(dto);
      } catch (Exception e) {
        log.error("库存异动，处理失败,bohNo:{}", batchNo, e);
      }
    }

1.2 日志加入traceId

业务筛分后，一个接口请求可能会涉及到多个服务，比如获取门店商品接口，需要调用会员、订单、商品等服务完成数据构建，当获取门店商品接口异常时，很难找出哪个服务异常导致。因此在不大改代码的前提下使用一个traceId实现分布式链路追踪。

APM组件目前比较流行的有Zipkin、Pinpoint、SkyWalking、CAT，大家可以根据实际的业务需求和场景进行选型，项目中我们使用Zipkin，主要是结合我们项目中使用的技术和业务需求。

需在要生成traceId的所有工程中，pom.xml引入以下依赖

<dependency>
   <groupId>org.springframework.cloud</groupId>
   <artifactId>spring-cloud-starter-zipkin</artifactId>
</dependency>

由于需要给前端返回traceId，网关请求返回值需要加上如下代码:

private void addTraceIdToHeader(ServerHttpResponse response) {
    String traceId = Optional.ofNullable(tracer.currentSpan())
            .map(Span::context)
            .map(TraceContext::traceIdString)
            .orElse(null);

    if (StringUtils.isEmpty(traceId)) {
        if (log.isInfoEnabled()) {
            log.info("trace id empty");
        }
        return;
    }

    String header = HEADER_NAME_TRACE_ID;
    if (log.isDebugEnabled()) {
        log.debug("add trace id [{}] to header [{}]", traceId, header);
    }
    response.getHeaders().add(header, traceId);
}

加入traceId后服务端日志打印如下：

{"@timestamp":"2020-01-04T18:15:31.275+08:00","hostname":"service-order","severity":"INFO","service":"service-sale","trace":"6c9a14a717a52819","span":"6fef879e39f264f5","parent":"6c9a14a717a52819","exportable":"false","pid":"1","thread":"http-nio-30130-exec-3","class":"c.h.s.s.c.shop.FeeController","message":"获取门店包装费状态"}
{"@timestamp":"2020-01-04T18:15:31.607+08:00","hostname":"service-sale","severity":"INFO","service":"service-sale","trace":"473d9eeea42d75fe","span":"310ae5240722958c","parent":"473d9eeea42d75fe","exportable":"false","pid":"1","thread":"http-nio-30130-exec-4","class":"c.h.s.s.s.u.s.i.UserSubscribeServiceImpl","message":"订阅提醒 "}

前端请求返回结果新增X-trace-id参数：

Response Headers    
    Connection:keep-alive
    Content-Type:application/json;charset=UTF-8
    Date:Sat, 04 Jan 2020 08:48:34 GMT
    Transfer-Encoding:chunked
    X-trace-id:2a5a9283dfe955d0

1.3 traceId线程穿透及解决项目中遇到的问题

1.3.1 虽然使用Spring Cloud Sleuth为我们生成traceId，极大的减少了开发工作量，但是在调用链路中如果使用了Thread或ThreadPoolExecutor，就会导致traceId无法穿透，如果不解决线程穿透问题，此类接口打印traceId有点鸡肋。

1.3.2 通过网关发起的请求，接口如果超过10秒没有执行完连接就会断开并返回接口请求超时提示信息，举个例子，当运营操作菜单批量绑定商品功能时(此功能有使用多线程)，只要执行此功能，前端基本上会报接口请求超时，实际菜单批量绑定商品接口最后执成功或失败都有可能，但发起请求一端拿不到执行结果信息，执行进度和结果完全靠猜或长期积累的经验进行判断，这类功能用户使用体验极差，问题也不好定位。

为解决以上问题，我们封装了一个工具类：

/**
 * 线程池需要使用的时候该service时，需要线程池使用sleuth的线程池定义，否则线程池无法向下传递
 * 
 * @author zhangbowen@heytea.com
 * @since 2019-12-27 09:31
 */
public interface LogManager {
    /**
     * 操作记录begin 操作记录开始,当有userId的时候则为人工操作；没有userId则为定时任务
     * 
     * @param type
     *            业务类型 批量发布、批量上架等
     */
    void begin(Integer type);

    /**
     * 操作记录end 操作执行完成后，更新操作记录的状态
     *
     */
    void end();

    /**
     * 操作日志log 记录操作日志,正常和错误日志都可以记录
     *
     * @param type
     *            1:正常 0：错误
     * @param description
     *            操作描述
     */
    void log(Integer type, String description);

    /**
     * 操作日志success 记录正常的操作日志
     *
     * @param description
     *            操作描述
     */
    void success(String description);

    /**
     * 操作日志error 记录错误的操作日志
     *
     * @param description
     *            操作描述
     */
    void error(String description);

    /**
     * 批量新增操作日志
     *
     * @param executionLogDtoList
     *            批量的操作日志
     */
    void batchLog(List<ExecutionLogDto> executionLogDtoList);

    /**
     * 查询操作日志
     *
     * @param traceId
     *            链路id
     * @return
     */
    OperateLogDto findOperateLog(String traceId);

    /**
     * 生成新的traceId
     *
     * @return
     */
    String newTraceId();
}

/**
 * 线程池需要使用的时候该service时，需要线程池使用sleuth的线程池定义，否则traceId无法从线程池向下传递
 * 
 * @author zhangbowen@heytea.com
 * @since 2019-12-27 09:32
 */
@Component
@Slf4j
public class LogManagerImpl implements LogManager {

    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private Tracing tracing;
    @Autowired
    private LogManager logManager;

    /**
     * 操作记录begin 操作记录开始,当有userId的时候则为人工操作；没有userId则为定时任务
     * 
     * @param type
     *            业务类型 批量发布、批量上架等
     */
    @Override
    public void begin(Integer type) {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            String traceId = traceContext != null ? traceContext.traceIdString() : logManager.newTraceId();

            //@formatter:off
            OperateRecord operateRecord = new OperateRecord().setCreateTime(LocalDateTime.now())
                    .setStatus(OperateStatusEnum.BEGIN.getCode())
                    .setType(type).setTraceId(traceId);
            //@formatter:on

            BaseUser baseUser = RequestContextUtil.getBaseUser();
            if (baseUser.getUserId() == null) {
                // 定时任务
                Integer userId = CommonConstant.DEFAULT_USER_ID;
                log.info("【操作记录begin】定时任务调用traceId:{},用户id:{}", traceId, userId);
                //@formatter:off
                operateRecord.setOperateType(OperateTypeEnum.JOB.getCode())
                        .setUserId(userId);
                //@formatter:on
            } else {
                // 人工调用
                log.info("【操作记录begin】人工调用调用traceId:{},用户id:{}", traceId, baseUser.getUserId());
                //@formatter:off
                operateRecord.setOperateType(OperateTypeEnum.ARTIFICIAL.getCode())
                        .setUserId(baseUser.getUserId().intValue());
                //@formatter:on
            }

            log.info("【操作记录begin】插入操作记录:{}", JSON.toJSONString(operateRecord));
            mongoTemplate.insert(operateRecord);
        } catch (Exception e) {
            log.error("【操作记录begin】操作失败", e);
        }
    }

    /**
     * 操作记录end 操作执行完成后，更新操作记录的状态
     *
     */
    @Override
    public void end() {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            if (traceContext == null) {
                log.error("【操作记录end】没有traceId");
                return;
            }

            String traceId = traceContext.traceIdString();
            log.info("【操作记录end】更新操作记录 traceId:{}", JSON.toJSONString(traceId));
            mongoTemplate.updateFirst(BasicQuery.query(Criteria.where(OperateRecordConstant.TRACE_ID).is(traceId)),
                BasicUpdate.update(OperateRecordConstant.STATUS, OperateStatusEnum.END.getCode()), OperateRecord.class);
        } catch (Exception e) {
            log.error("【操作记录end】操作失败", e);
        }
    }

    /**
     * 操作日志log 记录操作日志,正常和错误日志都可以记录
     * 
     * @param type
     *            1:正常 0：错误
     * @param description
     *            操作描述
     */
    @Override
    public void log(Integer type, String description) {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            if (traceContext == null) {
                log.error("【操作日志log】没有traceId");
                return;
            }
            //@formatter:off
            ExecutionLog executionLog = new ExecutionLog()
                    .setTraceId(traceContext.traceIdString())
                    .setDescription(description)
                    .setType(type)
                    .setCreateTime(LocalDateTime.now());
            //@formatter:on
            log.info("【操作日志log】插入执行日志:{}", JSON.toJSONString(executionLog));
            mongoTemplate.insert(executionLog);
        } catch (Exception e) {
            log.error("【操作日志log】操作失败", e);
        }
    }

    /**
     * 操作日志success 记录正常的操作日志
     * 
     * @param description
     *            操作描述
     */
    @Override
    public void success(String description) {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            if (traceContext == null) {
                log.error("【操作日志success】没有traceId");
                return;
            }
            //@formatter:off
            ExecutionLog executionLog = new ExecutionLog()
                    .setTraceId(traceContext.traceIdString())
                    .setDescription(description)
                    .setType(ResultTypeEnum.SUCCESS.getCode())
                    .setCreateTime(LocalDateTime.now());
            //@formatter:on
            log.info("【操作日志success】插入执行日志:{}", JSON.toJSONString(executionLog));
            mongoTemplate.insert(executionLog);
        } catch (Exception e) {
            log.error("【操作日志success】操作失败", e);
        }
    }

    /**
     * 操作日志error 记录错误的操作日志
     *
     * @param description
     *            操作描述
     */
    @Override
    public void error(String description) {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            if (traceContext == null) {
                log.error("【操作日志error】没有traceId");
                return;
            }
            //@formatter:off
            ExecutionLog executionLog = new ExecutionLog()
                    .setTraceId(traceContext.traceIdString())
                    .setDescription(description)
                    .setType(ResultTypeEnum.FAIL.getCode())
                    .setCreateTime(LocalDateTime.now());
            //@formatter:on
            log.info("【操作日志error】插入执行日志:{}", JSON.toJSONString(executionLog));
            mongoTemplate.insert(executionLog);
        } catch (Exception e) {
            log.error("【操作日志error】操作失败", e);
        }
    }

    /**
     * 批量新增操作日志
     *
     * @param executionLogDtoList
     *            批量的操作日志
     */
    @Override
    public void batchLog(List<ExecutionLogDto> executionLogDtoList) {
        try {
            TraceContext traceContext = tracing.currentTraceContext().get();
            if (traceContext == null) {
                log.error("【批量新增操作日志】没有traceId");
                return;
            }
            //@formatter:off
            List<ExecutionLog> executionLogList = executionLogDtoList.stream().map(dto -> new ExecutionLog().setTraceId(traceContext.traceIdString())
                    .setType(dto.getType())
                    .setDescription(dto.getDescription())
                    .setCreateTime(LocalDateTime.now())).collect(Collectors.toList());
            //@formatter:on
            if (log.isDebugEnabled()) {
                log.debug("【批量新增操作日志】批量插入执行日志:{}", JSON.toJSONString(executionLogList));
            }
            mongoTemplate.insertAll(executionLogList);
        } catch (Exception e) {
            log.error("【批量新增操作日志】操作失败", e);
        }
    }

    /**
     * 查询操作日志
     * 
     * @param traceId
     *            链路id
     * @return
     */
    @Override
    public OperateLogDto findOperateLog(String traceId) {
        // 查询操作记录
        OperateRecord operateRecord = mongoTemplate
            .findOne(new Query(Criteria.where(OperateRecordConstant.TRACE_ID).is(traceId)), OperateRecord.class);
        if (operateRecord == null) {
            return null;
        }
        // 操作日志
        List<ExecutionLog> executionLogList = mongoTemplate
            .find(new Query(Criteria.where(ExecutionLogConstant.TRACE_ID).is(traceId)), ExecutionLog.class);

        // 构建操作记录返回值
        OperateLogDto operateLogDto = new OperateLogDto().buildOperateLogDto(operateRecord).setExecutionLogs(
            executionLogList.stream().map(OperateLogDto.ExecutionLogDto::new).collect(Collectors.toList()));
        log.info("查看操作记录日志:{},traceId:{}", JSON.toJSONString(operateLogDto), traceId);
        return operateLogDto;
    }

    /**
     * 生成新的traceId
     *
     * @return
     */
    @Override
    @NewSpan
    public String newTraceId() {
        TraceContext traceContext = tracing.currentTraceContext().get();
        String traceId = traceContext.traceIdString();
        log.info("生成新的traceId:{}", traceId);
        return traceId;
    }
}

具体业务代码实现：

菜单批量绑定商品接口请求入口使用线程池是为解决接口运行超过网关所设超时问题

public class ProductPolicyController {

    @Autowired
    private ProductPolicyService productPolicyService;
    @Autowired
    @Qualifier("commonTraceableExecutor")
    private ExecutorService commonExecutorService;
    @Autowired
    private LogManager logManager;

     @PostMapping("/batch-binding")
    @ApiOperation("菜单批量绑定商品")
    public BaseResult batchBindingProducts(@Validated @RequestBody BatchBindingProductForm productForm) {
        log.info("菜单批量绑定商品入参：productForm:[{}]", JSON.toJSON(productForm));
        //记录一条操作日志后，直接异步调用业务方法，给前端返回一个接口请求成功，目的是防止接口执行时间过长导致网关断开连接
        logManager.begin(ExecuteTypeEnum.BATCH_INSERT.getCode());
        commonExecutorService.execute(()->{
            productPolicyService.batchBindingProducts(productForm);
        });
        return BaseResult.success();
    }

}

sleuth线程池配置

@Configuration
public class ExecutorConfig {

    @Autowired
    private ApplicationContext applicationContext;

    private ThreadPoolExecutor policyPoolExecutor =
            new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(100000));

    private ThreadPoolExecutor mtPoolExecutor =
            new ThreadPoolExecutor(30, 30, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(100000));

    /**
     *供后台操作异步使用！！！
     */
    private ThreadPoolExecutor commonPoolExecutor =
            new ThreadPoolExecutor(10, 10, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(100000));

    /**
     *供后台操作异步使用！！！
     */
    @Bean(value = "commonTraceableExecutor")
    public ExecutorService commonExecutorService(){
        return new TraceableExecutorService(applicationContext, commonPoolExecutor);
    }

    /**
     * 后台菜单线程池
     */
    @Bean(value = "policyTraceableExecutor")
    public ExecutorService policyExecutorService(){
        return new TraceableExecutorService(applicationContext, policyPoolExecutor);
    }

}


public void batchBindingProducts(BatchBindingProductForm productForm) {
   //......
    final CountDownLatch latch = new CountDownLatch(policyIds.size());
    for (Policy policy : policies) {
        traceableExecutorService.execute(() -> {
            try {
                //......
                
              logManager.success(policy.getId() + "-【" + policy.getName() + "】批量添加商品成功");
                log.info("批量添加商品成功 policy:[{}], products:[{}]", policy.getId(), newProductIds);

            } catch (Exception e) {
                log.error("批量添加商品成功 policy:[{}], productForm:[{}]", policy.getId(), JSON.toJSON(productForm));
                logManager.error(policy.getId() + "-【" + policy.getName() + "】批量添加商品失败");
            }finally {
                latch.countDown();
            }
        });
    }
    try {
        latch.await();
    } catch (Exception e) {
        log.error("批量添加商品失败  productForm:[{}]", JSON.toJSON(productForm));
        logManager.error("批量添加商品失败");
    }
    log.info("批量添加商品成功 productForm:[{}]", JSON.toJSON(productForm));
    logManager.end();
}

由于记录了每次请求执行结果以及执行过程详细日志，前端就可以实现长轮询来给用户提供请求执行进度，或提供一个专门的页面让运营来查询进度

二、后续优化方向

新增异常告警系统，为什么新增异常监控系统，举个例子：前端-->A服务-->B服务-->C服务，我们现在的处理方式是把traceId返回给前端，异常时前端或运维会直接找到A服务的负责人，这时A服务负责人会很快的拿traceId查找问题，发现不是自己负责服务问题，接着找下游服务的负责人，这样一个问题可能需要按服务调用先后依次全部排查，定位问题效率就非常低。

异常告警系统功能：

约定收集信息模板，比如规定必填字段：异常服务名，服务报异常事件，生成一个全局唯一ID，非必填字段：traceId，用户id等
定位告警策略：定义每个服务的告警策略，比如告警次数、告警方式（短信、或邮件提醒）等

三、总结

各种原因，日志优化穿插在不同版本业务开发迭代中，以上优化使日志级别分类更清晰，告警日志更明确，给请求链路加上traceId，在解决traceId线程穿透问题的同时，把耗时超过10秒的各种批量发布问题导致前端报请求超时问题解决了（有更优的方案，欢迎一起导论），欢迎各位大佬分享自己的干货。

日志优化实践经验分享

背景

解决思路

一、日志打印优化实践

1.1 日志打印规范

1.2 日志加入traceId

1.3 traceId线程穿透及解决项目中遇到的问题

二、后续优化方向

三、总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

喜茶技术分享