Npd代码结构了解

1 pkg目录结构

这里是各个monitor的实现。

.
├── custompluginmonitor             -> 用于实现自定义监视插件。node-problem-detector 支持通过插件监视特定状态或事件，此目录中包含与用户自定义插件相关的代码。
├── exporters                       -> 用于包含导出器（exporters）的实现，导出器负责将监视到的数据发送到外部系统，比如监控系统（例如 Prometheus）。导出器在数据收集和监控集成中起到关键作用。
├── healthchecker                   -> 实现健康检查逻辑，确保 node-problem-detector 本身和其他依赖的组件处于健康状态。这是监控和维护软件稳定性的重要部分。
├── logcounter                      -> 实现了 logcounter 的功能，通常用于计数和分析系统日志，以检测异常行为或问题。它可以与系统日志监控结合使用。
├── problemdaemon                   -> 实现问题守护程序。这个守护程序负责监测和识别节点中的问题，通过问题发现机制收集指标并生成告警。
├── problemdetector                 -> 包含问题检测的核心逻辑。这是 node-problem-detector 的主要功能部分，负责从不同的来源收集信息，并根据这些信息识别和报告节点问题。
├── problemmetrics                  -> 包含问题指标的实现，用于收集、处理和导出有关节点状态和问题的数据。这些指标可以展示在监控界面上，帮助用户理解系统的健康状况。
├── systemlogmonitor                -> 实现了系统日志监控的功能。它处理系统日志，分析日志内容，以检测可能导致问题的事件和错误。
├── systemstatsmonitor              -> 用于监控系统的各种统计数据，例如 CPU 使用率、内存使用情况等。它负责收集和报告这些统计信息，以帮助验证节点的健康状态。
├── types                           -> 用于定义项目中使用的各种类型结构体和常量。这些类型通常用于数据传输和处理，为核心逻辑提供数据模型。
├── util                            -> 包含一些公用工具函数和库，这些函数可以在其他模块中复用，提高代码的可维护性和可读性。
└── version                         -> 保存与版本相关的信息和结构。它用于管理版本号并提供版本信息的功能

2 problemdaemon

2.1 Register

pkg/problemdaemon/problem_daemon.go:32

var (
        handlers = make(map[types.ProblemDaemonType]types.ProblemDaemonHandler)
)

// Register registers a problem daemon factory method, which will be used to create the problem daemon.
func Register(problemDaemonType types.ProblemDaemonType, handler types.ProblemDaemonHandler) {
        handlers[problemDaemonType] = handler
}

注册一个问题守护进程工厂方法。
参数：

• problemDaemonType: 需要注册的守护进程类型。
• handler: 对应的处理器，其中包含创建守护进程的方法。
  作用：将指定类型的处理器添加到 handlers 映射中，以便后续根据类型获取对应的处理器。

2.2 GetProblemDaemonNames

pkg/problemdaemon/problem_daemon.go:37

// GetProblemDaemonNames retrieves all available problem daemon types.
func GetProblemDaemonNames() []types.ProblemDaemonType {
        problemDaemonTypes := []types.ProblemDaemonType{}
        for problemDaemonType := range handlers {
                problemDaemonTypes = append(problemDaemonTypes, problemDaemonType)
        }
        return problemDaemonTypes
}

检索所有可用的问题守护进程类型。

返回值：返回一个包含所有注册的守护进程类型的切片。

实现细节：遍历 handlers 映射，将每种问题守护进程类型添加到返回的切片中。

2.3 GetProblemDaemonHandlerOrDie

pkg/problemdaemon/problem_daemon.go:46

// GetProblemDaemonHandlerOrDie retrieves the ProblemDaemonHandler for a specific type of problem daemon, panic if error occurs..
func GetProblemDaemonHandlerOrDie(problemDaemonType types.ProblemDaemonType) types.ProblemDaemonHandler {
        handler, ok := handlers[problemDaemonType]
        if !ok {
                panic(fmt.Sprintf("Problem daemon handler for %v does not exist", problemDaemonType))
        }
        return handler
}

获取特定类型问题守护进程的处理器。

参数：问题守护进程类型。

返回值：返回对应的处理器。

实现细节：如果指定类型没有找到对应的处理器，则触发恐慌（panic），这通常用于在初始化时确保配置的有效性。

2.4 NewProblemDaemons

pkg/problemdaemon/problem_daemon.go:55

// NewProblemDaemons creates all problem daemons based on the configurations provided.
func NewProblemDaemons(monitorConfigPaths types.ProblemDaemonConfigPathMap) []types.Monitor {
        problemDaemonMap := make(map[string]types.Monitor)
        for problemDaemonType, configs := range monitorConfigPaths {
                for _, config := range *configs {
                        if _, ok := problemDaemonMap[config]; ok {
                                // Skip the config if it's duplicated.
                                klog.Warningf("Duplicated problem daemon configuration %q", config)
                                continue
                        }
                        problemDaemonMap[config] = handlers[problemDaemonType].CreateProblemDaemonOrDie(config)
                }
        }

        problemDaemons := []types.Monitor{}
        for _, problemDaemon := range problemDaemonMap {
                problemDaemons = append(problemDaemons, problemDaemon)
        }
        return problemDaemons
}

根据提供的配置创建所有问题守护进程。

参数：

monitorConfigPaths: 包含每种问题守护进程类型及其对应配置路径的映射。
返回值：返回创建的所有 Monitor 实例的切片。
实现细节

• 创建空映射：使用 problemDaemonMap 存储已创建的守护进程，以避免重复。
• 遍历配置：
  • 对于配置中的每种问题守护进程类型及其相应的配置路径，检查是否已经创建。
  • 如果配置已存在，则记录警告并跳过。
  • 使用注册的处理器创建新的问题守护进程并将其添加到映射中。
• 返回值构建：将地图中的所有守护进程转换为切片并返回。

3 problem_detector

3.1 problemDetector定义

pkg/problemdetector/problem_detector.go:33

// ProblemDetector collects statuses from all problem daemons and update the node condition and send node event.
type ProblemDetector interface {
        Run(context.Context) error
}

type problemDetector struct {
        monitors  []types.Monitor
        exporters []types.Exporter
}

字段：

monitors: 存储监控器的切片，用于收集问题状态。

exporters: 存储导出器的切片，用于将收集到的状态发送到外部系统或用户。

方法说明： Run(context.Context) error 启动问题检测器，并在上下文被取消或出现错误时返回错误。

3.2 NewProblemDetector

pkg/problemdetector/problem_detector.go:40

// NewProblemDetector creates the problem detector. Currently we just directly passed in the problem daemons, but
// in the future we may want to let the problem daemons register themselves.
func NewProblemDetector(monitors []types.Monitor, exporters []types.Exporter) ProblemDetector {
        return &problemDetector{
                monitors:  monitors,
                exporters: exporters,
        }
}

创建一个新的问题检测器实例。

参数：

monitors: 监控器数组，用于检测不同问题。

exporters: 导出器数组，用于将状态发送给外部系统。

返回值：返回实现 ProblemDetector 接口的新的 problemDetector 实例。

3.3 Run

pkg/problemdetector/problem_detector.go:48

// Run starts the problem detector.
func (p *problemDetector) Run(ctx context.Context) error {
        // Start the log monitors one by one.
        var chans []<-chan *types.Status
        failureCount := 0
        for _, m := range p.monitors {
                ch, err := m.Start()
                if err != nil {
                        // Do not return error and keep on trying the following config files.
                        klog.Errorf("Failed to start problem daemon %v: %v", m, err)
                        failureCount++
                        continue
                }
                if ch != nil {
                        chans = append(chans, ch)
                }
        }
        allMonitors := p.monitors

        if len(allMonitors) == failureCount {
                return fmt.Errorf("no problem daemon is successfully setup")
        }

        defer func() {
                for _, m := range allMonitors {
                        m.Stop()
                }
        }()

        ch := groupChannel(chans)
        klog.Info("Problem detector started")

        for {
                select {
                case <-ctx.Done():
                        return nil
                case status := <-ch:
                        for _, exporter := range p.exporters {
                                exporter.ExportProblems(status)
                        }
                }
        }
}

执行问题检测器的主要运行逻辑。

实现细节：

• 开始监控器：遍历并启动每个监控器。
• 如果启动失败，记录错误并增加失败计数。
• 如果成功，则将返回的通道添加到 chans 切片。
• 检查失败情况：如果所有监控器都失败，则返回错误。
• 延迟关闭：在函数结束时停止所有监控器。
• 通道组合：调用 groupChannel 函数，用于将所有监控器的状态通道组合成单一通道。
• 监听状态：使用 select 不断监听上下文和监控器的状态。
• 如果收到状态，从所有导出器导出问题。

3.4 groupChannel

pkg/problemdetector/problem_detector.go:91

func groupChannel(chans []<-chan *types.Status) <-chan *types.Status {
        statuses := make(chan *types.Status)
        for _, ch := range chans {
                go func(c <-chan *types.Status) {
                        for status := range c {
                                statuses <- status
                        }
                }(ch)
        }
        return statuses
}

将多个状态通道组合为一个通道。

参数：切片 chans，包含多个状态通道。

返回值：返回一个新的通道 statuses，用于接收来自所有监控器的状态。

实现细节

启动多个 goroutine，每个 goroutine 从一个监控器的通道读取状态，并将其转发到新的 statuses 通道中。

4 problemmetrics

4.1 ProblemMetricsManager定义

pkg/problemmetrics/problem_metrics.go:40

// ProblemMetricsManager manages problem-converted metrics.
// ProblemMetricsManager is thread-safe.
type ProblemMetricsManager struct {
        problemCounter           metrics.Int64MetricInterface
        problemGauge             metrics.Int64MetricInterface
        problemTypeToReason      map[string]string
        problemTypeToReasonMutex sync.Mutex
}

字段：

• problemCounter: 一个整型指标接口，用于计数特定问题发生的次数。
• problemGauge: 一个整型指标接口，用于表示特定问题是否对节点产生影响。
• problemTypeToReason: 一个映射，保存问题类型与原因的关系。
• problemTypeToReasonMutex: 一个互斥锁，用于保护 problemTypeToReason 的并发访问。

4.2 init

pkg/problemmetrics/problem_metrics.go:31

// GlobalProblemMetricsManager is a singleton of ProblemMetricsManager,
// which should be used to manage all problem-converted metrics across all
// problem daemons.
var GlobalProblemMetricsManager *ProblemMetricsManager

func init() {
        GlobalProblemMetricsManager = NewProblemMetricsManagerOrDie()
}

GlobalProblemMetricsManager: 全局唯一的 ProblemMetricsManager 实例，用于管理所有问题相关的指标。

init 函数: 在程序启动时初始化 GlobalProblemMetricsManager，确保其可用。

4.3 NewProblemMetricsManagerOrDie

pkg/problemmetrics/problem_metrics.go:47

func NewProblemMetricsManagerOrDie() *ProblemMetricsManager {
        pmm := ProblemMetricsManager{}

        var err error
        pmm.problemCounter, err = metrics.NewInt64Metric(
                metrics.ProblemCounterID,
                string(metrics.ProblemCounterID),
                "Number of times a specific type of problem have occurred.",
                "1",
                metrics.Sum,
                []string{"reason"})
        if err != nil {
                klog.Fatalf("Failed to create problem_counter metric: %v", err)
        }

        pmm.problemGauge, err = metrics.NewInt64Metric(
                metrics.ProblemGaugeID,
                string(metrics.ProblemGaugeID),
                "Whether a specific type of problem is affecting the node or not.",
                "1",
                metrics.LastValue,
                []string{"type", "reason"})
        if err != nil {
                klog.Fatalf("Failed to create problem_gauge metric: %v", err)
        }

        pmm.problemTypeToReason = make(map[string]string)

        return &pmm
}

创建并初始化 ProblemMetricsManager 实例。

实现细节：

创建 problemCounter 指标，用于记录特定问题的发生次数，接收标签 reason。

创建 problemGauge 指标，用于指示特定问题是否正在影响节点，接收标签 type 和 reason。

如果创建指标失败，则程序会崩溃（Fatalf）。

初始化问题类型到原因的映射。

4.4 IncrementProblemCounter

pkg/problemmetrics/problem_metrics.go:79

// IncrementProblemCounter increments the value of a problem counter.
func (pmm *ProblemMetricsManager) IncrementProblemCounter(reason string, count int64) error {
        if pmm.problemCounter == nil {
                return errors.New("problem counter is being incremented before initialized.")
        }

        return pmm.problemCounter.Record(map[string]string{"reason": reason}, count)
}

增加问题计数器的值，根据 reason 标签记录统计数据。

参数：

reason: 问题发生的原因。

count: 要增加的数量。

返回值：返回可能的错误。

4.5 SetProblemGauge

pkg/problemmetrics/problem_metrics.go:88

// SetProblemGauge sets the value of a problem gauge.
func (pmm *ProblemMetricsManager) SetProblemGauge(problemType string, reason string, value bool) error {
        if pmm.problemGauge == nil {
                return errors.New("problem gauge is being set before initialized.")
        }

        pmm.problemTypeToReasonMutex.Lock()
        defer pmm.problemTypeToReasonMutex.Unlock()

        // We clear the last reason, because the expected behavior is that at any point of time,
        // for each type of permanent problem, there should be at most one reason got set to 1.
        // This behavior is consistent with the behavior of node condition in Kubernetes.
        // However, problemGauges with different "type" and "reason" are considered as different
        // metrics in Prometheus. So we need to clear the previous metrics explicitly.
        if lastReason, ok := pmm.problemTypeToReason[problemType]; ok {
                err := pmm.problemGauge.Record(map[string]string{"type": problemType, "reason": lastReason}, 0)
                if err != nil {
                        return fmt.Errorf("failed to clear previous reason %q for type %q: %v",
                                problemType, lastReason, err)
                }
        }

        pmm.problemTypeToReason[problemType] = reason

        var valueInt int64
        if value {
                valueInt = 1
        }
        return pmm.problemGauge.Record(map[string]string{"type": problemType, "reason": reason}, valueInt)
}

设置问题量规的值，指示特定问题对节点的影响。

参数：

• problemType: 问题类型（例如，节点不可用）。
• reason: 具体原因。
• value: 表示问题的状态，true表示问题存在 (1)，false表示问题消失 (0)。

实现细节：

• 使用互斥锁保护对 problemTypeToReason 的并发访问。
• 清除之前的原因（如果存在），保证每个问题类型在任何时刻都最多只有一个原因被设置为 1。
• 记录新的量规值。

5 exporters

目录结构：

.
├── k8sexporter
│   ├── condition
│   ├── k8s_exporter.go
│   └── problemclient
├── prometheusexporter
│   └── prometheus_exporter.go
├── register.go
├── register_test.go
└── stackdriver
    ├── config
    ├── gce
    ├── stackdriver_exporter.go
    └── stackdriver_exporter_test.go

这里就涉及到了三个exporter，分别是k8sexporter、prometheusexporter、stackdriver_exporter

这三个是比较独立的，因为能力是完全不一样的，

• exporter：专注于将types.Status的event和conditation上报，并且会有一些http服务接口开放(healthz、conditions、/debug/pprof)，
• prometheusexporter：直接用的prometheuse的库，简单理解就是为了让prometheuse可以采集到数据
• stackdriver_exporter：处理与 Google Stackdriver 的集成，Stackdriver 是 Google Cloud 提供的监控和管理平台，该exporter负责将指标数据发送到
  Stackdriver，以便在 Google Cloud 环境中监控应用程序和集群状态。

这里来说说我的理解吧，prometheusexporter和stackdriver_exporter这两个exporter是为了将metric数据发送到对应的监控平台，而k8sexporter则是为了将types.Status发送到k8s，目前已经开始向register继承方向延展，如stackdriver_exporter已经使用该模式并且已经统一了struct的实现方法，但是prometheusexporter和k8sexporter还没完成演变，目前还是通过package直接调用的方式使用的。

这个模块就不进行代码级别解析了，整体是非常简单的。

6 healthchecker

目录结构：

├── health_checker.go
├── health_checker_darwin.go
├── health_checker_linux.go
├── health_checker_test.go
├── health_checker_windows.go
└── types
├── types.go
├── types_test.go
├── types_unix.go
└── types_windows.go

6.1 healthChecker定义
pkg/healthchecker/health_checker.go:31

type healthChecker struct {
  component       string
  service         string
  enableRepair    bool
  healthCheckFunc func () (bool, error)
  // The repair is "best-effort" and ignores the error from the underlying actions.
  // The bash commands to kill the process will fail if the service is down and hence ignore.
  repairFunc         func ()
  uptimeFunc         func () (time.Duration, error)
  crictlPath         string
  healthCheckTimeout time.Duration
  coolDownTime       time.Duration
  loopBackTime       time.Duration
  logPatternsToCheck map[string]int
}

字段含义：

• component：被检查的 Kubernetes 组件名称（如 Kubelet、Kube Proxy 等）。
• service：服务名称或标识符。
• enableRepair：布尔值，指示是否在发现服务不健康时尝试修复它。
• healthCheckFunc：一个函数，用于执行健康检查，返回健康状态和可能的错误。
• repairFunc：一个函数，用于尝试修复不健康的服务。
• uptimeFunc：一个函数，用于获取服务的运行时间。
• crictlPath：指向 CRI 工具的路径。
• healthCheckTimeout：健康检查的超时设置。
• coolDownTime：在尝试修复之前要求服务必须正常运行的时间。
• loopBackTime：用于回溯检查日志的时间段。
• logPatternsToCheck：一个映射，包含要检查的日志模式及其出现次数阈值。

6.2 NewHealthChecker

pkg/healthchecker/health_checker.go:48

// NewHealthChecker returns a new health checker configured with the given options.
func NewHealthChecker(hco *options.HealthCheckerOptions) (types.HealthChecker, error) {
  hc := &healthChecker{
    component:          hco.Component,
    enableRepair:       hco.EnableRepair,
    crictlPath:         hco.CriCtlPath,
    healthCheckTimeout: hco.HealthCheckTimeout,
    coolDownTime:       hco.CoolDownTime,
    service:            hco.Service,
    loopBackTime:       hco.LoopBackTime,
    logPatternsToCheck: hco.LogPatterns.GetLogPatternCountMap(),
  }
  hc.healthCheckFunc = getHealthCheckFunc(hco)
  hc.repairFunc = getRepairFunc(hco)
  hc.uptimeFunc = getUptimeFunc(hco.Service)
  return hc, nil
}

NewHealthChecker会根据输入的options.HealthCheckerOptions来进行types.HealthChecker的构造。

6.3 CheckHealth

pkg/healthchecker/health_checker.go:67

// CheckHealth checks for the health of the component and tries to repair if enabled.
// Returns true if healthy, false otherwise.
func (hc *healthChecker) CheckHealth() (bool, error) {
        healthy, err := hc.healthCheckFunc()
        if err != nil {
                return healthy, err
        }
        logPatternHealthy, err := logPatternHealthCheck(hc.service, hc.loopBackTime, hc.logPatternsToCheck)
        if err != nil {
                return logPatternHealthy, err
        }
        if healthy && logPatternHealthy {
                return true, nil
        }

        // The service is unhealthy.
        // Attempt repair based on flag.
        if hc.enableRepair {
                // repair if the service has been up for the cool down period.
                uptime, err := hc.uptimeFunc()
                if err != nil {
                        klog.Infof("error in getting uptime for %v: %v\n", hc.component, err)
                        return false, nil
                }
                klog.Infof("%v is unhealthy, component uptime: %v\n", hc.component, uptime)
                if uptime > hc.coolDownTime {
                        klog.Infof("%v cooldown period of %v exceeded, repairing", hc.component, hc.coolDownTime)
                        hc.repairFunc()
                }
        }
        return false, nil
}

检查组件的健康状态，返回是否健康，并在不健康的情况下尝试修复。

逻辑：

• 调用 healthCheckFunc 执行健康检查。
• 使用 logPatternHealthCheck 检查日志模式。
• 如果组件不健康且启用了修复，检查组件的运行时间并调用修复函数。

6.4 logPatternHealthCheck

pkg/healthchecker/health_checker.go:100

// logPatternHealthCheck checks for the provided logPattern occurrences in the service logs.
// Returns true if the pattern is empty or does not exist logThresholdCount times since start of service, false otherwise.
func logPatternHealthCheck(service string, loopBackTime time.Duration, logPatternsToCheck map[string]int) (bool, error) {
        if len(logPatternsToCheck) == 0 {
                return true, nil
        }
        uptimeFunc := getUptimeFunc(service)
        klog.Infof("Getting uptime for service: %v\n", service)
        uptime, err := uptimeFunc()
        if err != nil {
                klog.Warningf("Failed to get the uptime: %+v", err)
                return true, err
        }

        logStartTime := time.Now().Add(-uptime).Format(types.LogParsingTimeLayout)
        if loopBackTime > 0 && uptime > loopBackTime {
                logStartTime = time.Now().Add(-loopBackTime).Format(types.LogParsingTimeLayout)
        }
        for pattern, count := range logPatternsToCheck {
                healthy, err := checkForPattern(service, logStartTime, pattern, count)
                if err != nil || !healthy {
                        return healthy, err
                }
        }
        return true, nil
}

检查服务日志中指定模式的出现次数。

参数：

• service：被监控的服务名称。
• loopBackTime：用于确定要检查的日志时间范围。
• logPatternsToCheck：要检查的日志模式及其出现的阈值。
  返回值：布尔值表示健康状态和可能的错误。

6.5 healthCheckEndpointOKFunc

pkg/healthchecker/health_checker.go:126

// healthCheckEndpointOKFunc returns a function to check the status of an http endpoint
func healthCheckEndpointOKFunc(endpoint string, timeout time.Duration) func() (bool, error) {
        return func() (bool, error) {
                httpClient := http.Client{Timeout: timeout}
                response, err := httpClient.Get(endpoint)
                if err != nil || response.StatusCode != http.StatusOK {
                        return false, nil
                }
                return true, nil
        }
}

返回一个函数，该函数检查给定 HTTP 端点的健康状态。

参数：端点 URI 和超时时间。

返回值：健康检查函数。

6.6 getHealthCheckFunc

pkg/healthchecker/health_checker.go:138

// getHealthCheckFunc returns the health check function based on the component.
func getHealthCheckFunc(hco *options.HealthCheckerOptions) func() (bool, error) {
        switch hco.Component {
        case types.KubeletComponent:
                return healthCheckEndpointOKFunc(types.KubeletHealthCheckEndpoint(), hco.HealthCheckTimeout)
        case types.KubeProxyComponent:
                return healthCheckEndpointOKFunc(types.KubeProxyHealthCheckEndpoint(), hco.HealthCheckTimeout)
        case types.DockerComponent:
                return func() (bool, error) {
                        if _, err := execCommand(hco.HealthCheckTimeout, getDockerPath(), "ps"); err != nil {
                                return false, nil
                        }
                        return true, nil
                }
        case types.CRIComponent:
                return func() (bool, error) {
                        _, err := execCommand(
                                hco.HealthCheckTimeout,
                                hco.CriCtlPath,
                                "--timeout="+hco.CriTimeout.String(),
                                "--runtime-endpoint="+hco.CriSocketPath,
                                "pods",
                                "--latest",
                        )
                        if err != nil {
                                return false, nil
                        }
                        return true, nil
                }
        default:
                klog.Warningf("Unsupported component: %v", hco.Component)
        }

        return nil
}

根据组件类型返回相应的健康检查方法。

参数：健康检查选项。

返回值：执行健康检查的函数。

6.7 execCommand

pkg/healthchecker/health_checker.go:174

// execCommand executes the bash command and returns the (output, error) from command, error if timeout occurs.
func execCommand(timeout time.Duration, command string, args ...string) (string, error) {
        ctx, cancel := context.WithTimeout(context.Background(), timeout)
        defer cancel()
        cmd := exec.CommandContext(ctx, command, args...)
        out, err := cmd.CombinedOutput()
        if err != nil {
                klog.Infof("command %v failed: %v, %s\n", cmd, err, string(out))
                return "", err
        }

        return strings.TrimSuffix(string(out), "\n"), nil
}

执行给定的命令，并在指定的超时时间内返回输出和错误。

参数：超时时间、命令和参数。

返回值：命令的输出和错误信息。

6.8 getUptimeFunc

pkg/healthchecker/health_checker_linux.go:32

// getUptimeFunc returns the time for which the given service has been running.
func getUptimeFunc(service string) func() (time.Duration, error) {
        return func() (time.Duration, error) {
                // Using InactiveExitTimestamp to capture the exact time when systemd tried starting the service. The service will
                // transition from inactive -> activating and the timestamp is captured.
                // Source : https://www.freedesktop.org/wiki/Software/systemd/dbus/
                // Using ActiveEnterTimestamp resulted in race condition where the service was repeatedly killed by plugin when
                // RestartSec of systemd and invoke interval of plugin got in sync. The service was repeatedly killed in
                // activating state and hence ActiveEnterTimestamp was never updated.
                out, err := execCommand(types.CmdTimeout, "systemctl", "show", service, "--property=InactiveExitTimestamp")

                if err != nil {
                        return time.Duration(0), err
                }
                val := strings.Split(out, "=")
                if len(val) < 2 {
                        return time.Duration(0), errors.New("could not parse the service uptime time correctly")
                }
                t, err := time.Parse(types.UptimeTimeLayout, val[1])
                if err != nil {
                        return time.Duration(0), err
                }
                return time.Since(t), nil
        }
}

返回一个函数，计算指定服务的运行时间。

参数：

• service string - 被监控服务的名称。

• 返回值：返回一个函数，该函数返回服务运行的持续时间或错误。

• 实现细节

• 使用 systemctl show 命令获取 InactiveExitTimestamp，这是服务从 inactive 状态转换时的时间戳。

• 解析命令的输出，将时间戳转换为 time.Time 类型，并计算自该时间以来的持续时间。

• 此方法可用于获取服务的当前运行时间，以便评估是否在修复之前达到冷却时间。

6.9 getRepairFunc

pkg/healthchecker/health_checker_linux.go:58

// getRepairFunc returns the repair function based on the component.
func getRepairFunc(hco *options.HealthCheckerOptions) func() {
        // Use `systemctl kill` instead of `systemctl restart` for the repair function.
        // We start to rely on the kernel message difference for the two commands to
        // indicate if the component restart is due to an administrative plan (restart)
        // or a system issue that needs repair (kill).
        // See https://github.com/kubernetes/node-problem-detector/issues/847.
        switch hco.Component {
        case types.DockerComponent:
                // Use "docker ps" for docker health check. Not using crictl for docker to remove
                // dependency on the kubelet.
                return func() {
                        execCommand(types.CmdTimeout, "pkill", "-SIGUSR1", "dockerd")
                        execCommand(types.CmdTimeout, "systemctl", "kill", "--kill-who=main", hco.Service)
                }
        default:
                // Just kill the service for all other components
                return func() {
                        execCommand(types.CmdTimeout, "systemctl", "kill", "--kill-who=main", hco.Service)
                }
        }
}

根据服务的组件类型返回一个演示修复功能的方法。

参数：

• hco *options.HealthCheckerOptions - 健康检查的选项。
• 返回值：返回一个函数，该函数执行相应修复操作。

实现细节

• 对于 Docker 组件，通过 pkill 发送 SIGUSR1 信号以更优雅地停止 Docker 守护进程。
• 对于其他组件，调用 systemctl kill 命令终止主要进程。
• 这种设计使得修复功能能够根据不同的组件采取合适的处理方式。

6.10 checkForPattern

pkg/healthchecker/health_checker_linux.go:82

// checkForPattern returns (true, nil) if logPattern occurs less than logCountThreshold number of times since last
// service restart. (false, nil) otherwise.
func checkForPattern(service, logStartTime, logPattern string, logCountThreshold int) (bool, error) {
        out, err := execCommand(types.CmdTimeout, "/bin/sh", "-c",
                // Query service logs since the logStartTime
                `journalctl --unit "`+service+`" --since "`+logStartTime+
                        // Grep the pattern
                        `" | grep -i "`+logPattern+
                        // Get the count of occurrences
                        `" | wc -l`)
        if err != nil {
                return true, err
        }
        occurrences, err := strconv.Atoi(out)
        if err != nil {
                return true, err
        }
        if occurrences >= logCountThreshold {
                klog.Infof("%s failed log pattern check, %s occurrences: %v", service, logPattern, occurrences)
                return false, nil
        }
        return true, nil
}

检查指定服务的日志中是否包含特定模式，并返回其出现次数是否达到阈值。

参数：

• service：被监控服务的名称。
• logStartTime：起始时间，用于过滤日志。
• logPattern：需要检查的日志模式。
• logCountThreshold：模式出现的最大次数阈值。
• 返回值：返回一个布尔值，指示是否健康，以及可能的错误。

实现细节

• 使用 journalctl 命令获取从指定时间开始的服务日志，并通过 grep 过滤符合条件的日志模式。
• 使用 wc -l 计算匹配的行数。
• 如果日志模式出现的次数大于阈值，记录相关信息并返回健康检查失败的结果。

7 logcounter

7.1 logCounter定义

pkg/logcounter/log_counter.go:42

type logCounter struct {
        logCh         <-chan *systemtypes.Log
        buffer        systemlogmonitor.LogBuffer
        pattern       string
        revertPattern string
        clock         clock.Clock
}

字段解释：

• logCh：只读通道，用于接收系统日志。
• buffer：存储日志的缓冲区，以便分析和计数。
• pattern：处理日志时使用的匹配模式。
• revertPattern：可用来反向计数的模式。
• clock：用于获取当前时间的时钟，支持模拟时钟功能。

7.2 NewJournaldLogCounter

pkg/logcounter/log_counter.go:50

func NewJournaldLogCounter(options *options.LogCounterOptions) (types.LogCounter, error) {
        watcher := journald.NewJournaldWatcher(watchertypes.WatcherConfig{
                Plugin:       "journald",
                PluginConfig: map[string]string{journaldSourceKey: options.JournaldSource},
                LogPath:      options.LogPath,
                Lookback:     options.Lookback,
                Delay:        options.Delay,
        })
        logCh, err := watcher.Watch()
        if err != nil {
                return nil, fmt.Errorf("error watching journald: %v", err)
        }
        return &logCounter{
                logCh:         logCh,
                buffer:        systemlogmonitor.NewLogBuffer(bufferSize),
                pattern:       options.Pattern,
                revertPattern: options.RevertPattern,
                clock:         clock.RealClock{},
        }, nil
}

初始化并返回一个新的 logCounter 实例。

参数：

• options *options.LogCounterOptions - 包含配置信息（例如日志源、路径、模式等）。
• 返回值：返回一个实现了 types.LogCounter 接口的 logCounter 实例。

实现细节

• 创建一个 journald 日志观察者实例来监控指定的日志路径。
• 如果监视操作成功，建立日志通道并创建 logCounter 的实例，设置缓冲区、匹配模式等。

7.3 Count

pkg/logcounter/log_counter.go:71

func (e *logCounter) Count() (count int, err error) {
        start := e.clock.Now()
        for {
                select {
                case log, ok := <-e.logCh:
                        if !ok {
                                err = fmt.Errorf("log channel closed unexpectedly")
                                return
                        }
                        if start.Before(log.Timestamp) {
                                return
                        }
                        e.buffer.Push(log)
                        if len(e.buffer.Match(e.pattern)) != 0 {
                                count++
                        }
                        if e.revertPattern != "" && len(e.buffer.Match(e.revertPattern)) != 0 {
                                count--
                        }
                case <-e.clock.After(timeout):
                        return
                }
        }
}

统计在运行期间匹配的日志条目数量。

返回值：返回符合条件的日志计数及任何潜在的错误。

实现细节

• 使用当前时间 (start) 记录开始时刻。
• 在无限循环中，使用 select 语句监听日志通道和超时事件。
• 如果从 logCh 读取到日志：
  • 检查通道是否已经关闭。
  • 如果读取到的日志时间戳晚于开始时间，则停止计数。
  • 将日志条目压入缓冲区。
  • 检查当前日志是否与模式匹配，如果匹配则计数加一；如果匹配反向模式，则计数减一。
  • 如果在超时时间内没有收到新日志，结束循环并返回计数。