如何提高分布式系统的可观察性：Insight Tool的引入

文章目录

• 前言
• 系统的三大Insight视角
• 三大Insight视角信息的收集
• Insight Tool的设计实现
• Log/Metric Insight命令实现
• Log采集部分的实现
• Metric采集部分的实现
• 引用

前言

---

前一篇文章刚刚讨论过在分布式系统中如何构建一套metric框架来帮助我们做系统性能指标的收集，以此帮助我们能够更好地了解系统运行的情况。在这里，metric信息是一种能够让我们看到系统内在具体运行情况的一个信息。这里还会有其它信息，比如常见的系统log，这个也是广泛被用来查看定位系统问题的一个主要信息源。总结地来说，不管metric也好，log也罢，这些都是为了能够提升系统的Observability，即可观察性。分布式系统的运行是极其复杂的，我们需要一定的措施和手段来提高其可观察性。本文我们就来聊聊这个话题。

系统的三大Insight视角

---

关于如何提升系统的可观察性，我们首先要了解的一点是我们需要去看哪些内部的Insight视角，这些视角信息是有用的，对我们有帮助的。其实在上文部分，已经提到了几点Insight视角：

• Metric信息，metric信息用来查看系统某些关键指标信息的。
• Log日志，查看系统具体运行逻辑信息。
• Conf配置信息，系统运行的使用配置信息情况。

三大Insight视角信息的收集

---

接下来的问题就是：以上三大视角信息我们如何来进行收集呢？

首先是Metric信息，系统的Metric信息是保留在Metric框架内部的，因此我们需要将这些metric信息的值从框架中拉取出来，就是get value of all一次。

Log日志信息的获取，日志信息的获取我们可以先简化地考虑为是实时日志的获取，而不是获取历史日志。实时流的获取就不用走读取系统本地文件的方式了，我们可以直接获取Log打印对象，进行输出流数据的实时采集。

最后是Conf配置信息，系统当前使用的Conf配置信息是3个Insight视角信息里最容易获取的。Conf系统在被系统服务初始加载完毕后，或被赋值到其内部Configuration实例内，我们只需通过获取这个实例信息来打印即可。

Insight Tool的设计实现

---

以上三大Insight信息如果在不做任何改进的情况下，我们的观察办法是比较低效且各自独立的：

1）到JMX页面观察当前metric指标
2）登录系统服务节点，查看本地日志
3）登录系统服务节点，查看本地Conf配置文件（此方式并不完全可靠，存在系统加载完配置后，存在配置被更改但系统又还没加载新的配置文件的情况）

在改进系统Observability的情况下，我们将要引入Insight Tool的工具来整合上述略显低效的方式，收集的方式如前文小节所述，此收集展示过程如下所示“

• 1)系统服务构建出Metric/Log/GetConf三大Endpoint，在这些endpoint里面执行收集逻辑
• 2)用户通过Insight Tool工具命令触发某个行为命令，获取某metric值，获取某服务log
• 3)Insight Tool发送http请求到目标地址中，随后请求抵达，触发了Serverlet的处理。
• 4)Serverlet返回客户端结果。

此过程简图如下所示：

Log/Metric Insight命令实现

---

以下笔者将重点介绍Log以及Metric的Insight代码功能实现，基于的原理在上文中已经提到过了，但是在此小节部分将会展示更多的细节内容。此部分代码实现参考自HDDS相关issue：HDDS-1935:Improve the visibility with Ozone Insight tool。

Log采集部分的实现

---

在HDDS-1935对此的实现中，采用的是获取Root Log对象进行实时log的写出，然后客户端将日志进行过滤，此过程流程图如下所示：

在Client Side端的细节步骤如下：

• 1)获取目标要查看的insight point(具体服务，类等等)
• 2)获取此insight point的log源信息
• 3)遍历log源信息，进行log level的重置操作
• 4)提取log源对应所属的component
• 5)根据log源信息进行log的实时获取
  • 5.1)根据log源信息，开启独立线程进行日志信息流的获取
    5.1.1)构造实时日志获取http的get请求
    5.1.2)得到执行返回的回复结果
    5.1.3)从得到的日志结果内容中过滤出目标期待的log类的日志内容
  • 5.2)启动日志获取线程
  • 5.3)等待日志获取线程的执行结束

上述关键Log源的定义如下：

/**
 * Log Source的定义.
 */
public class LoggerSource {

  /**
   * 此日志所属的模块
   */
  private Component component;

  /**
   * log类具体名称
   */
  private String loggerName;

  /**
   * 日志级别
   */
  private Level level;

  public LoggerSource(Component component, String loggerName, Level level) {
    this.component = component;
    this.loggerName = loggerName;
    this.level = level;
  }
  ...

Client端相关代码如下：

  @Override
  public Void call() throws Exception {
    OzoneConfiguration conf =
        getInsightCommand().createOzoneConfiguration();
    // 1)获取目标要查看的insight point(具体服务，类等等)
    InsightPoint insight =
        getInsight(conf, insightName);

    // 2)获取此insight point的log源信息
    List<LoggerSource> loggers = insight.getRelatedLoggers(verbose);

    // 3)遍历log源信息，进行log level的重置操作
    for (LoggerSource logger : loggers) {
      setLogLevel(conf, logger.getLoggerName(), logger.getComponent(),
          logger.getLevel());
    }

    // 4)提取log源对应所属的component
    Set<Component> sources = loggers.stream().map(LoggerSource::getComponent)
        .collect(Collectors.toSet());
    try {
      // 5)根据log源信息进行log的实时获取
      streamLog(conf, sources, loggers);
    } finally {
      for (LoggerSource logger : loggers) {
        setLogLevel(conf, logger.getLoggerName(), logger.getComponent(),
            Level.INFO);
      }
    }
    return null;
  }

...

  // 实时日志流数据的获取
  private void streamLog(OzoneConfiguration conf, Set<Component> sources,
      List<LoggerSource> relatedLoggers) {
    List<Thread> loggers = new ArrayList<>();
    // 1)根据log源信息，开启独立线程进行日志信息流的获取
    for (Component sourceComponent : sources) {
      loggers.add(new Thread(
          () -> streamLog(conf, sourceComponent, relatedLoggers)));
    }

    // 2)启动日志获取线程
    for (Thread thread : loggers) {
      thread.start();
    }

    // 3)等待日志获取线程的执行结束
    for (Thread thread : loggers) {
      try {
        thread.join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  // 实时日志获取方法
  private void streamLog(OzoneConfiguration conf, Component logComponent,
      List<LoggerSource> loggers) {
    HttpClient client = HttpClientBuilder.create().build();

    // 1)构造实时日志获取http的get请求
    HttpGet get = new HttpGet(getHost(conf, logComponent) + "/logstream");
    try {
      // 2)得到执行返回的回复结果
      HttpResponse execute = client.execute(get);
      try (BufferedReader bufferedReader = new BufferedReader(
          new InputStreamReader(execute.getEntity().getContent(),
              StandardCharsets.UTF_8))) {
        bufferedReader.lines()
            .filter(line -> {
              // 3)从得到的日志结果内容中过滤出目标期待的log类的日志内容
              for (LoggerSource logger : loggers) {
                if (line.contains(logger.getLoggerName())) {
                  return true;
                }
              }
              return false;
            })
            // 4)处理并输出日志行内容
            .map(this::processLogLine)
            .map(l -> "[" + logComponent.prefix() + "] " + l)
            .forEach(System.out::println);
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

...
  /**
   * 根据所属的不同component模块，构造不同的地址的请求地址
   */
  public String getHost(OzoneConfiguration conf, Component component) {
    if (component.getHostname() != null) {
      return "http://" + component.getHostname() + ":" + component.getPort();
    } else if (component.getName() == Type.SCM) {
      Optional<String> scmHost =
          HddsUtils.getHostNameFromConfigKeys(conf,
              ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY,
              ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY);

      return "http://" + scmHost.get() + ":9876";
    } else if (component.getName() == Type.OM) {
      Optional<String> omHost =
          HddsUtils.getHostNameFromConfigKeys(conf,
              OMConfigKeys.OZONE_OM_ADDRESS_KEY);
      return "http://" + omHost.get() + ":9874";
    } else {
      throw new IllegalArgumentException(
          "Component type is not supported: " + component.getName());
    }
  }

Server端的处理逻辑相对来讲，简单很多，

/**
 * Servlet to stream the current logs to the response.
 */
public class LogStreamServlet extends HttpServlet {

  private static final String PATTERN = "%d [%p|%c|%C{1}] %m%n";

  @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {

    WriterAppender appender =
        new WriterAppender(new PatternLayout(PATTERN), resp.getWriter());
    appender.setThreshold(Level.TRACE);

    try {
      // 将当期的所有实时log输出到回复对象中
      Logger.getRootLogger().addAppender(appender);
      try {
    	// 进行长时间睡眠直到此操作被中断
        Thread.sleep(Integer.MAX_VALUE);
      } catch (InterruptedException e) {
        //interrupted
      }
    } finally {
      Logger.getRootLogger().removeAppender(appender);
    }
  }

}

Metric采集部分的实现

---

下面我们来看另一部分metric insight的实现部分。HDDS-1935对此是实现了一个基于内存计数的metric sink，然后通过servlet的请求处理，触发此metric sink的内存计数结果写出，流程图如下：

Client端代码如下：

  public Void call() throws Exception {
    OzoneConfiguration conf =
        getInsightCommand().createOzoneConfiguration();
    // 1)获取目标要查看的insight point(具体服务)
    InsightPoint insight =
        getInsight(conf, insightName);
    // 2)提取metric源对应所属的component
    Set<Component> sources =
        insight.getMetrics().stream().map(MetricGroupDisplay::getComponent)
            .collect(Collectors.toSet());
    // 3)根据metric源信息获取metric值
    Map<Component, List<String>> metrics = getMetrics(conf, sources);
    System.out.println(
        "Metrics for `" + insightName + "` (" + insight.getDescription() + ")");
    System.out.println();
    // 4)遍历待展示的metric,从上步结果中取出值
    for (MetricGroupDisplay group : insight.getMetrics()) {
      System.out.println(group.getDescription());
      System.out.println();
      for (MetricDisplay display : group.getMetrics()) {
        System.out.println("  " + display.getDescription() + ": " + selectValue(
            metrics.get(group.getComponent()), display));
      }
      System.out.println();
      System.out.println();

    }
    return null;
  }
...

  /**
   * 构造HTTP请求，获取metric信息数据
   */
  private List<String> getMetrics(OzoneConfiguration conf,
      Component component) {
    HttpClient client = HttpClientBuilder.create().build();
    HttpGet get = new HttpGet(getHost(conf, component) + "/prom");
    try {
      HttpResponse execute = client.execute(get);
      if (execute.getStatusLine().getStatusCode() != 200) {
        throw new RuntimeException(
            "Can't read prometheus metrics endpoint" + execute.getStatusLine()
                .getStatusCode());
      }
      try (BufferedReader bufferedReader = new BufferedReader(
          new InputStreamReader(execute.getEntity().getContent(),
              StandardCharsets.UTF_8))) {
        return bufferedReader.lines().collect(Collectors.toList());
      }
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

...
  /**
   * 根据所属的不同component模块，构造不同的地址的请求地址
   */
  public String getHost(OzoneConfiguration conf, Component component) {
     // 此部分操作同上
  }

Server端代码如下：

/**
 * Servlet to publish hadoop metrics in prometheus format.
 */
public class PrometheusServlet extends HttpServlet {

  public PrometheusMetricsSink getPrometheusSink() {
    return
        (PrometheusMetricsSink) getServletContext().getAttribute(
            BaseHttpServer.PROMETHEUS_SINK);
  }

  @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {
	// 1)触发一次全量metric publish到metric sink
    DefaultMetricsSystem.instance().publishMetricsNow();
    // 2)metric sink将自身维护的metric写出到回复信息内
    getPrometheusSink().writeMetrics(resp.getWriter());
    resp.getWriter().flush();
  }
}

以上关键的PrometheusSink即上文阐述的基于内存维护结果的metric sink，定义如下：

/**
 * Metrics sink for prometheus exporter.
 * <p>
 * Stores the metric data in-memory and return with it on request.
 */
public class PrometheusMetricsSink implements MetricsSink {

  /**
   * 每个metric的输出结果cache
   */
  private final Map<String, String> metricLines = new ConcurrentHashMap<>();
  ...

  /**
   * 写出metric结果到指定writer中
   */
  public void writeMetrics(Writer writer) throws IOException {
	// PrometheusMetricsSink将当前维护的metric信息进行写出
    for (String line : metricLines.values()) {
      writer.write(line);
    }
  }

通过上述两部分Insight功能的实现，逻辑改动其实并不复杂，也没有对核心代码做主要的修改。但是可以让我们能够通过简单命令行快速地了解系统运行的内部情况，在一定程度上增强了系统的可观察性。我们在实际的系统开发中也可以多多运用这样的改造来监控我们的系统运行情况。

引用

---

[1].https://issues.apache.org/jira/browse/HDDS-1935 . Improve the visibility with Ozone Insight tool