SOFABoot 健康检查能力分析

开源中国有个年度开源软件的活动，里面有两个 SOFA 相关的项目(SOFABoot & SOFARPC)，大家帮忙点两下一起投个票: www.oschina.net/project/top… 。同时也欢迎大家关注 SOFAStack

Liveness Check & Readiness Check

Spring Boot 提供了一个基础的健康检查的能力，中间件和应用都可以扩展来实现自己的健康检查逻辑。但是 Spring Boot 的健康检查只有 Liveness Check 的能力，缺少 Readiness Check 的能力，这样会有比较致命的问题。当一个微服务应用启动的时候，必须要先保证启动后应用是健康的，才可以将上游的流量放进来（来自于 RPC，网关，定时任务等等流量），否则就可能会导致一定时间内大量的错误发生。

针对 Spring Boot 缺少 Readiness Check 能力的情况， SOFABoot 增加了 Spring Boot 现有的健康检查的能力，提供了 Readiness Check 的能力。利用 Readiness Check 的能力， SOFA 中间件中的各个组件只有在 Readiness Check 通过之后，才将流量引入到应用的实例中，比如 RPC ，只有在 Readiness Check 通过之后，才会向服务注册中心注册，后面来自上游应用的流量才会进入。

除了中间件可以利用 Readiness Check 的事件来控制流量的进入之外， PAAS 系统也可以通过访问 http://localhost:8080/actuator/readiness 来获取应用的 Readiness Check 的状况，用来控制例如负载均衡设备等等流量的进入。

使用方式

SOFABoot 的健康检查能力需要引入：

<dependency>
    <groupId>com.alipay.sofa</groupId>
    <artifactId>healthcheck-sofa-boot-starter</artifactId>
</dependency>
复制代码

区别于 SpringBoot 的：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
复制代码

详细工程科参考： sofa-boot

健康检查启动日志

代码分析

既然是个Starter，那么就先从 spring.factories 文件来看：

org.springframework.context.ApplicationContextInitializer=\
com.alipay.sofa.healthcheck.initializer.SofaBootHealthCheckInitializer

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.alipay.sofa.healthcheck.configuration.SofaBootHealthCheckAutoConfiguration
复制代码

SofaBootHealthCheckInitializer

SofaBootHealthCheckInitializer 实现了 ApplicationContextInitializer 接口。

ApplicationContextInitializer 是 Spring 框架原有的概念，这个类的主要目的就是在 ConfigurableApplicationContext 类型（或者子类型）的 ApplicationContext 做 refresh 之前，允许我们 对 ConfigurableApplicationContext 的实例做进一步的设置或者处理。

public class SofaBootHealthCheckInitializer implements ApplicationContextInitializer<ConfigurableApplicationContext> {
    @Override
    public void initialize(ConfigurableApplicationContext applicationContext) {
        Environment environment = applicationContext.getEnvironment();
        if (SOFABootEnvUtils.isSpringCloudBootstrapEnvironment(environment)) {
            return;
        }
        // init logging.level.com.alipay.sofa.runtime argument
        String healthCheckLogLevelKey = Constants.LOG_LEVEL_PREFIX
                                        + HealthCheckConstants.SOFABOOT_HEALTH_LOG_SPACE;
        SofaBootLogSpaceIsolationInit.initSofaBootLogger(environment, healthCheckLogLevelKey);
      SofaBootHealthCheckLoggerFactory.getLogger(SofaBootHealthCheckInitializer.class).info(
            "SOFABoot HealthCheck Starting!");
    }
}
复制代码

SofaBootHealthCheckInitializer 在 initialize 方法中主要做了两件事：

• 验证当前 environment 是否是 SpringCloud 的（3.0.0 开始支持 springCloud ，之前版本无此 check ）
• 初始化 logging.level

这两件事和健康检查没有什么关系，但是既然放在这个模块里面还是来看下。

1、springCloud 环境验证

首先就是为什么会有这个验证。 SOFABoot 在支持 SpringcLoud 时遇到一个问题，就是当在 classpath 中添加 spring-cloud-context 依赖关系时, org.springframework.context.ApplicationContextInitializer 会被调用两次。具体背景可参考 # issue1151 && # issue 232

private final static String SPRING_CLOUD_MARK_NAME = "org.springframework.cloud.bootstrap.BootstrapConfiguration";

public static boolean isSpringCloudBootstrapEnvironment(Environment environment) {
    if (environment instanceof ConfigurableEnvironment) {
        return !((ConfigurableEnvironment) environment).getPropertySources().contains(
            SofaBootInfraConstants.SOFA_BOOTSTRAP)
               && isSpringCloud();
    }
    return false;
}

public static boolean isSpringCloud() {
    return ClassUtils.isPresent(SPRING_CLOUD_MARK_NAME, null);
}
复制代码

上面这段代码是 SOFABoot 提供的一个用于区分 引导上下文 和 应用上下文 的方法：

• 检验是否有 "org.springframework.cloud.bootstrap.BootstrapConfiguration" 这个类来判断当前是否引入了 spingCloud 的引导配置类
• 从 environment 中获取 MutablePropertySources 实例，验证 MutablePropertySources 中是否包括 sofaBootstrap （ 如果当前环境是 SOFA bootstrap environment ，则包含 sofaBootstrap ；这个是在 SofaBootstrapRunListener 回调方法中设置进行的 ）

2、初始化 logging.level

这里是处理 SOFABoot 日志空间隔离的。

public static void initSofaBootLogger(Environment environment, String runtimeLogLevelKey) {
    // 初始化 logging.path 参数
    String loggingPath = environment.getProperty(Constants.LOG_PATH);
    if (!StringUtils.isEmpty(loggingPath)) {
        System.setProperty(Constants.LOG_PATH, environment.getProperty(Constants.LOG_PATH));
        ReportUtil.report("Actual " + Constants.LOG_PATH + " is [ " + loggingPath + " ]");
    }

    //for example : init logging.level.com.alipay.sofa.runtime argument
    String runtimeLogLevelValue = environment.getProperty(runtimeLogLevelKey);
    if (runtimeLogLevelValue != null) {
        System.setProperty(runtimeLogLevelKey, runtimeLogLevelValue);
    }

    // init file.encoding
    String fileEncoding = environment.getProperty(Constants.LOG_ENCODING_PROP_KEY);
    if (!StringUtils.isEmpty(fileEncoding)) {
        System.setProperty(Constants.LOG_ENCODING_PROP_KEY, fileEncoding);
    }
}
复制代码

SofaBootHealthCheckAutoConfiguration

这个类是 SOFABoot 健康检查机制的自动化配置实现。

@Configuration
public class SofaBootHealthCheckAutoConfiguration {
    /** ReadinessCheckListener: 容器刷新之后回调 */
    @Bean
    public ReadinessCheckListener readinessCheckListener() {
        return new ReadinessCheckListener();
    }
    /** HealthCheckerProcessor: HealthChecker处理器 */
    @Bean
    public HealthCheckerProcessor healthCheckerProcessor() {
        return new HealthCheckerProcessor();
    }
    /** HealthCheckerProcessor: HealthIndicator处理器 */
    @Bean
    public HealthIndicatorProcessor healthIndicatorProcessor() {
        return new HealthIndicatorProcessor();
    }
    /** AfterReadinessCheckCallbackProcessor: ReadinessCheck之后的回调处理器 */
    @Bean
    public AfterReadinessCheckCallbackProcessor afterReadinessCheckCallbackProcessor() {
        return new AfterReadinessCheckCallbackProcessor();
    }
    /** 返回 SofaBoot健康检查指标类 实例*/
    @Bean
    public SofaBootHealthIndicator sofaBootHealthIndicator() {
        return new SofaBootHealthIndicator();
    }

    @ConditionalOnClass(Endpoint.class)
    public static class ConditionReadinessEndpointConfiguration {
        @Bean
        @ConditionalOnEnabledEndpoint
        public SofaBootReadinessCheckEndpoint sofaBootReadinessCheckEndpoint() {
            return new SofaBootReadinessCheckEndpoint();
        }
    }

    @ConditionalOnClass(Endpoint.class)
    public static class ReadinessCheckExtensionConfiguration {
        @Bean
        @ConditionalOnMissingBean
        @ConditionalOnEnabledEndpoint
        public ReadinessEndpointWebExtension readinessEndpointWebExtension() {
            return new ReadinessEndpointWebExtension();
        }
    }
}
复制代码

ReadinessCheckListener

public class ReadinessCheckListener implements PriorityOrdered,
                                   ApplicationListener<ContextRefreshedEvent> 
复制代码

从代码来看， ReadinessCheckListener 实现了 ApplicationListener 监听器接口，其所监听的事件对象是 ContextRefreshedEvent ，即当容器上下文刷新完成之后回调。 SOFABoot 中通过这个监听器来完成 readniess check 的处理。

onApplicationEvent 回调方法：

public void onApplicationEvent(ContextRefreshedEvent event) {
    // healthCheckerProcessor init
    healthCheckerProcessor.init();
    // healthIndicatorProcessor init
    healthIndicatorProcessor.init();
    // afterReadinessCheckCallbackProcessor init
    afterReadinessCheckCallbackProcessor.init();
    // readiness health check execute
    readinessHealthCheck();
}
复制代码

• 初始化 healthCheckerProcessor ，这个里面就是将当前所有的 HealthChecker 类型的 bean 找出来，然后放在一个 map 中，等待后面的 readiness check 。

public void init() {
    // 是否已经初始化了
    if (isInitiated.compareAndSet(false, true)) {
        // applicationContext 应用上下文不能为null
        Assert.notNull(applicationContext, () -> "Application must not be null");
        // 获取所有类型是 HealthChecker 的bean
        Map<String, HealthChecker> beansOfType = applicationContext
                .getBeansOfType(HealthChecker.class);
        // 排序
        healthCheckers = HealthCheckUtils.sortMapAccordingToValue(beansOfType,
                applicationContext.getAutowireCapableBeanFactory());
        // 构建日志信息，对应在健康检查日志里面打印出来的是：
        // ./logs/health-check/common-default.log:Found 0 HealthChecker implementation
        StringBuilder healthCheckInfo = new StringBuilder(512).append("Found ")
                .append(healthCheckers.size()).append(" HealthChecker implementation:")
                .append(String.join(",", healthCheckers.keySet()));
        logger.info(healthCheckInfo.toString());
    }
}
复制代码

• 初始化 healthIndicatorProcessor ，将所有的 healthIndicator 类型的 bean 找出来，然后放在一个 map 中等待 readiness check 。如果想要在 SOFABoot 的 Readiness Check 里面增加一个检查项，那么可以直接扩展 Spring Boot 的 HealthIndicator 这个接口。

public void init() {
    // 是否已经初始化
    if (isInitiated.compareAndSet(false, true)) {
        // applicationContext 验证
        Assert.notNull(applicationContext, () -> "Application must not be null");
        // 获取所有HealthIndicator类型的bean
        Map<String, HealthIndicator> beansOfType = applicationContext
                .getBeansOfType(HealthIndicator.class);
        // 支持 Reactive 方式
        if (ClassUtils.isPresent(REACTOR_CLASS, null)) {
            applicationContext.getBeansOfType(ReactiveHealthIndicator.class).forEach(
                    (name, indicator) -> beansOfType.put(name, () -> indicator.health().block()));
        }
        // 排序
        healthIndicators = HealthCheckUtils.sortMapAccordingToValue(beansOfType,
                applicationContext.getAutowireCapableBeanFactory());
        // 构建日志信息
        // Found 2 HealthIndicator implementation:
        // sofaBootHealthIndicator, diskSpaceHealthIndicator
        StringBuilder healthIndicatorInfo = new StringBuilder(512).append("Found ")
                .append(healthIndicators.size()).append(" HealthIndicator implementation:")
                .append(String.join(",", healthIndicators.keySet()));
        logger.info(healthIndicatorInfo.toString());
    }
}
复制代码

• 初始化 afterReadinessCheckCallbackProcessor 。如果想要在 Readiness Check 之后做一些事情，那么可以扩展 SOFABoot 的这个接口

public void init() {
    //  是否已经初始化
    if (isInitiated.compareAndSet(false, true)) {
        // applicationContext 验证
        Assert.notNull(applicationContext, () -> "Application must not be null");
        // 找到所有 ReadinessCheckCallback 类型的 bean 
        Map<String, ReadinessCheckCallback> beansOfType = applicationContext
                .getBeansOfType(ReadinessCheckCallback.class);
        // 排序
        readinessCheckCallbacks = HealthCheckUtils.sortMapAccordingToValue(beansOfType,
                applicationContext.getAutowireCapableBeanFactory());
        // 构建日志
        StringBuilder applicationCallbackInfo = new StringBuilder(512).append("Found ")
                .append(readinessCheckCallbacks.size())
                .append(" ReadinessCheckCallback implementation: ")
                .append(String.join(",", beansOfType.keySet()));
        logger.info(applicationCallbackInfo.toString());
    }
}
复制代码

• readinessHealthCheck ，前面的几个 init 方法中均是为 readinessHealthCheck 做准备的，到这里 SOFABoot 已经拿到了当前多有的 HealthChecker 、 HealthIndicator 和 ReadinessCheckCallback 类型的 bean 信息。

  // readiness health check
  public void readinessHealthCheck() {
      // 是否跳过所有check,可以通过 com.alipay.sofa.healthcheck.skip.all 配置项配置决定
      if (skipAllCheck()) {
          logger.warn("Skip all readiness health check.");
      } else {
          // 是否跳过所有 HealthChecker 类型bean的 readinessHealthCheck,
          // 可以通过com.alipay.sofa.healthcheck.skip.component配置项配置
          if (skipComponent()) {
              logger.warn("Skip HealthChecker health check.");
          } else {
              //HealthChecker 的 readiness check
              healthCheckerStatus = healthCheckerProcessor
                  .readinessHealthCheck(healthCheckerDetails);
          }
          // 是否跳过所有HealthIndicator 类型bean的readinessHealthCheck
          // 可以通过 com.alipay.sofa.healthcheck.skip.indicator配置项配置
          if (skipIndicator()) {
              logger.warn("Skip HealthIndicator health check.");
          } else {
              //HealthIndicator 的 readiness check
              healthIndicatorStatus = healthIndicatorProcessor
                  .readinessHealthCheck(healthIndicatorDetails);
          }
      }
      // ReadinessCheck 之后的回调函数，做一些后置处理
      healthCallbackStatus = afterReadinessCheckCallbackProcessor
          .afterReadinessCheckCallback(healthCallbackDetails);
      if (healthCheckerStatus && healthIndicatorStatus && healthCallbackStatus) {
          logger.info("Readiness check result: success");
      } else {
          logger.error("Readiness check result: fail");
      }
  }
  复制代码

Readiness Check 做了什么

前面是 SOFABoot 健康检查组件处理健康检查逻辑的一个大体流程，了解到了 Readiness 包括检查 HealthChecker 类型的 bean 和 HealthIndicator 类型的 bean 。其中 HealthIndicator 是 SpringBoot 自己的接口 ，而 HealthChecker 是 SOFABoot 提供的接口。下面继续通过 XXXProcess 来看下 Readiness Check 到底做了什么？

HealthCheckerProcessor

HealthChecker 的健康检查处理器， readinessHealthCheck 方法

public boolean readinessHealthCheck(Map<String, Health> healthMap) {
    Assert.notNull(healthCheckers, "HealthCheckers must not be null.");
    logger.info("Begin SOFABoot HealthChecker readiness check.");
    boolean result = healthCheckers.entrySet().stream()
            .map(entry -> doHealthCheck(entry.getKey(), entry.getValue(), true, healthMap, true))
            .reduce(true, BinaryOperators.andBoolean());
    if (result) {
        logger.info("SOFABoot HealthChecker readiness check result: success.");
    } else {
        logger.error("SOFABoot HealthChecker readiness check result: failed.");
    }
    return result;
}
复制代码

这里每个 HealthChecker 又委托给 doHealthCheck 来检查

private boolean doHealthCheck(String beanId, HealthChecker healthChecker, boolean isRetry,
                              Map<String, Health> healthMap, boolean isReadiness) {
    Assert.notNull(healthMap, "HealthMap must not be null");
    Health health;
    boolean result;
    int retryCount = 0;
    // check 类型  readiness ？ liveness
    String checkType = isReadiness ? "readiness" : "liveness";
    do {
        // 获取 Health 对象
        health = healthChecker.isHealthy();
        // 获取 健康检查状态结果
        result = health.getStatus().equals(Status.UP);
        if (result) {
            logger.info("HealthChecker[{}] {} check success with {} retry.", beanId, checkType,retryCount);
            break;
        } else {
            logger.info("HealthChecker[{}] {} check fail with {} retry.", beanId, checkType,retryCount);
        }
        // 重试 && 等待
        if (isRetry && retryCount < healthChecker.getRetryCount()) {
            try {
                retryCount += 1;
                TimeUnit.MILLISECONDS.sleep(healthChecker.getRetryTimeInterval());
            } catch (InterruptedException e) {
                logger
                    .error(
                        String
                            .format(
                                "Exception occurred while sleeping of %d retry HealthChecker[%s] %s check.",
                                retryCount, beanId, checkType), e);
            }
        }
    } while (isRetry && retryCount < healthChecker.getRetryCount());
    // 将当前 实例 bean 的健康检查结果存到结果集healthMap中
    healthMap.put(beanId, health);
    try {
        if (!result) {
            logger
                .error(
                    "HealthChecker[{}] {} check fail with {} retry; fail details:{}; strict mode:{}",
                    beanId, checkType, retryCount,
                    objectMapper.writeValueAsString(health.getDetails()),
                    healthChecker.isStrictCheck());
        }
    } catch (JsonProcessingException ex) {
        logger.error(
            String.format("Error occurred while doing HealthChecker %s check.", checkType), ex);
    }
    // 返回健康检查结果
    return !healthChecker.isStrictCheck() || result;
}
复制代码

这里的 doHealthCheck 结果需要依赖具体 HealthChecker 实现类的处理。通过这样一种方式可以 SOFABoot 可以很友好的实现对所以 HealthChecker 的健康检查。 HealthIndicatorProcessor 的 readinessHealthCheck 和 HealthChecker 的基本差不多；有兴趣的可以自行阅读源码 Alipay-SOFABoot 。

AfterReadinessCheckCallbackProcessor

这个接口是 SOFABoot 提供的一个扩展接口， 用于在 Readiness Check 之后做一些事情。其实现思路和前面的 XXXXProcessor 是一样的，对之前初始化时得到的所有的 ReadinessCheckCallbacks 实例 bean 逐一进行回调处理。

public boolean afterReadinessCheckCallback(Map<String, Health> healthMap) {
    logger.info("Begin ReadinessCheckCallback readiness check");
    Assert.notNull(readinessCheckCallbacks, "ReadinessCheckCallbacks must not be null.");

    boolean result = readinessCheckCallbacks.entrySet().stream()
            .map(entry -> doHealthCheckCallback(entry.getKey(), entry.getValue(), healthMap))
            .reduce(true, BinaryOperators.andBoolean());

    if (result) {
        logger.info("ReadinessCheckCallback readiness check result: success.");
    } else {
        logger.error("ReadinessCheckCallback readiness check result: failed.");
    }
    return result;
}
复制代码

同样也是委托给了 doHealthCheckCallback 来处理

private boolean doHealthCheckCallback(String beanId,
                                      ReadinessCheckCallback readinessCheckCallback,
                                      Map<String, Health> healthMap) {
    Assert.notNull(healthMap, () -> "HealthMap must not be null");
    boolean result = false;
    Health health = null;
    try {
        health = readinessCheckCallback.onHealthy(applicationContext);
        result = health.getStatus().equals(Status.UP);
        // print log 省略
    } catch (Throwable t) {
        // 异常处理
    } finally {
        // 存入 healthMap
        healthMap.put(beanId, health);
    }
    return result;
}
复制代码

扩展 Readiness Check 能力

按照上面的分析，我们可以自己来实现下这几个扩展。

实现 HealthChecker 接口

@Component
public class GlmapperHealthChecker implements HealthChecker {

    @Override
    public Health isHealthy() {
        // 可以检测数据库连接是否成功
        // 可以检测zookeeper是否启动成功
        // 可以检测 redis 客户端是否启动成功
        // everything you want ...
        if(OK){
            return Health.up().build();
        }
        return Health.down().build();
    }

    @Override
    public String getComponentName() {
        // 组件名
        return "GlmapperComponent";
    }
    
    @Override
    public int getRetryCount() {
        // 重试次数
        return 1;
    }

    @Override
    public long getRetryTimeInterval() {
        // 重试间隔
        return 0;
    }

    @Override
    public boolean isStrictCheck() {
        return false;
    }
}
复制代码

实现 ReadinessCheckCallback 接口

@Component
public class GlmapperReadinessCheckCallback implements ReadinessCheckCallback {

    @Override
    public Health onHealthy(ApplicationContext applicationContext) {
        Object glmapperHealthChecker = applicationContext.getBean("glmapperHealthChecker");
        if (glmapperHealthChecker instanceof GlmapperHealthChecker){
            return Health.up().build();
        }
        return Health.down().build();
    }
}
复制代码

再来看下健康检查日志：

可以看到我们自己定义的检查类型 ready 了。

从日志看到有一个 sofaBootHealthIndicator ，实现了 HealthIndicator 接口。

public class SofaBootHealthIndicator implements HealthIndicator {
    private static final String    CHECK_RESULT_PREFIX = "Middleware";
    @Autowired
    private HealthCheckerProcessor healthCheckerProcessor;

    @Override
    public Health health() {
        Map<String, Health> healths = new HashMap<>();
        // 调用了 healthCheckerProcessor 的 livenessHealthCheck
        boolean checkSuccessful = healthCheckerProcessor.livenessHealthCheck(healths);
        if (checkSuccessful) {
            return Health.up().withDetail(CHECK_RESULT_PREFIX, healths).build();
        } else {
            return Health.down().withDetail(CHECK_RESULT_PREFIX, healths).build();
        }
    }
}
复制代码

livenessHealthCheck 和 readinessHealthCheck 两个方法都是交给 doHealthCheck 来处理的，没有看出来有什么区别。

小结

本文基于 SOFABoot 3.0.0 版本，与之前版本有一些区别。详细变更见：SOFABoot upgrade_3_x。本篇文章简单介绍了 SOFABoot 对 SpringBoot 健康检查能力扩展的具体实现细节。

最后再来补充下 liveness 和 readiness ，从字面意思来理解， liveness 就是是否是活的， readiness 就是意思是否可访问的。

• readiness ：应用即便已经正在运行了，它仍然需要一定时间才能 提供 服务，这段时间可能用来加载数据，可能用来构建缓存，可能用来注册服务，可能用来选举 Leader 等等。总之 Readiness 检查通过前是不会有流量发给应用的。目前 SOFARPC 就是在 readiness check 之后才会将所有的服务注册到注册中心去。
• liveness ：检测应用程序是否正在运行