creeks日志重要级规则配置流程

现在我们改为使用ES + Kibana 进行日志的查询,为了减少过多的日志存储,为日志进行重要级划分然后根据不同的重要级别进行删除

当前级别划分:

• DEBUG : LOW
• INFO : MIDDLE
• WARN/ERROR : HIGH

在error-collect服务中我们在进行日志存储到ES的时候会进行一次重要级改变,来将一些我们任务重要的业务日志进行重要级提升,存储更长的时间

规则配置流程

规则的配置分为两种

• LOGGER配置 (具体到某个类)
• PACKAGE配置 (配置包下的所有类)

当我们需要配置某个服务下的规则时我们可以再K8S的ERROR-COLLECT服务的feishu.json进行配置

示例

通过配置methodNames进行方法的匹配 (支持正则表达式)

目前常用的匹配规则

# 排除常用查询关键字开头的方法
^((?!^query|^search|^find|^select|^get|^list|^page|^export).)*$

如果发现需要的日志优先级不符合预期,可以去调整规则

流程采集的过程

在业务服务中的logback-spring.xml引用自定义的采集Appender

里面实现了日志消息发送到队列的逻辑

collector中会启动一个后台线程,将其内部队列的消息进行消费执行error-collect的接口

package com.codo.tech.error.collect.client;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/** 将日志收集到当前k8s集群的elasticsearch中 */
public class LocalLogCollector implements Runnable {

  // 任务队列，避免阻塞；同时限定了容量，防止内存泄漏
  private final BlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>(200);

  // 清理周期，单位：毫秒
  private static final long clearPeriod = 600_000;
  // 未能入队的记录数
  private final AtomicInteger missedCount = new AtomicInteger(0);
  // 上次清理时间戳
  private long lastClearTime = System.currentTimeMillis();

  // 标记后台线程是否已启动
  private final AtomicBoolean started = new AtomicBoolean(false);

  // 默认的服务地址，使用当前 k8s 集群的 error-collect-service
  private String serviceUrl = "http://error-collect-service:8080";

  /**
   * 是否运行在 k8s 集群中.
   *
   * <p>当运行在开发人员的 PC 上时，避免打印报警信息，因为无法访问 k8s 集群的 error-collect-service
   */
  private boolean isInKubernetes = false;

  public void collect(LogRecord record) {
    startSenderThread();

    final boolean success = queue.offer(record);
    if (!success) {
      final int newCount = missedCount.incrementAndGet();

      final long now = System.currentTimeMillis();
      if (lastClearTime + clearPeriod > now) {
        // 通过 compareAndSet 避免并发时重复执行清理操作
        if (missedCount.compareAndSet(newCount, 0)) {
          if (isInKubernetes) {
            System.err.println("[ErrorCollectClient] " + newCount + " records failed to enqueue");
          }
          lastClearTime = now;
        }
      }
    }
  }

  @Override
  public void run() {
    while (true) {
      try {
        LogRecord record = queue.take();
        HttpUtil.post(serviceUrl + "/remote/logCollectRemoteApiV1/collectAndNotify", record);
      } catch (InterruptedException e) {
        if (isInKubernetes) {
          System.err.println("LocalLogCollector is interrupted");
        }
        break;
      } catch (Exception ignore) {
      }
    }
  }

  private void startSenderThread() {
    if (started.get()) {
      return;
    }

    // 启动线程，保证只初始化一次
    if (started.compareAndSet(false, true)) {
      detectServiceUrl();
      detectKubernetesEnv();

      final Thread thread = new Thread(this);
      thread.setDaemon(true);
      thread.setName("LocalLogCollect");
      thread.start();
    }
  }

  /** 从环境变量中获取服务地址，替换默认值 */
  private void detectServiceUrl() {
    String url = System.getenv(LogConstants.ENV_SERVICE_URL);
    if (url != null && url.startsWith("http")) {
      serviceUrl = url;
    }
  }

  /** 判断是否在 codo kubernetes 集群中 */
  private void detectKubernetesEnv() {
    String env = System.getenv(LogConstants.ENV_CODO_ENV);
    if (env != null && !env.isEmpty()) {
      isInKubernetes = true;
    }
  }
}