Katyusha's blog

1
// Package detector 对采样指标做时序异常判定。
2
//
3
// 当前实现采用"持续高占用"规则（阈值 + 连续窗口数），保证零幻觉、可解释。
4
// 后续可在此扩展 EWMA / 3-sigma / Spectral Residual 等检测算法。
5
package detector
6

7
import (
8
  "time"
9

10
  "github.com/os2026/ebpf-rca/internal/collector"
11
)
12

13
// Signal 表示一次已确认的异常信号。
14
type Signal struct {
15
  Sample      collector.Sample
16
  WindowStart time.Time
17
  WindowEnd   time.Time
18
}
19

20
// CPUDetector 检测持续的高 CPU 占用线程。
21
type CPUDetector struct {
22
  Threshold    float64 // CPU 占用阈值（单核占比）
23
  SustainTicks int     // 连续超过阈值多少个窗口才触发
24
  counters     map[uint32]int
25
  firstSeen    map[uint32]time.Time
26
  fired        map[uint32]bool
27
}
28

29
// NewCPUDetector 构造检测器。
30
func NewCPUDetector(threshold float64, sustain int) *CPUDetector {
31
  if sustain < 1 {
32
    sustain = 1
33
  }
34
  return &CPUDetector{
35
    Threshold:    threshold,
36
    SustainTicks: sustain,
37
    counters:     make(map[uint32]int),
38
    firstSeen:    make(map[uint32]time.Time),
39
    fired:        make(map[uint32]bool),
40
  }
41
}
42

43
// Detect 处理一个窗口的样本，返回本窗口新触发的异常信号。
44
func (d *CPUDetector) Detect(samples []collector.Sample, now time.Time) []Signal {
45
  active := make(map[uint32]bool, len(samples))
46
  var signals []Signal
47

48
  for _, s := range samples {
49
    if s.CPUUtil < d.Threshold {
50
      continue
51
    }
52
    active[s.Pid] = true
53
    if d.counters[s.Pid] == 0 {
54
      d.firstSeen[s.Pid] = now
55
    }
56
    d.counters[s.Pid]++
57
    if d.counters[s.Pid] >= d.SustainTicks && !d.fired[s.Pid] {
58
      d.fired[s.Pid] = true
59
      signals = append(signals, Signal{
60
        Sample:      s,
61
        WindowStart: d.firstSeen[s.Pid],
62
        WindowEnd:   now,
63
      })
64
    }
65
  }
66

67
  // 本窗口回落到阈值以下的线程，重置其状态（允许下次重新触发）。
68
  for pid := range d.counters {
69
    if !active[pid] {
70
      delete(d.counters, pid)
71
      delete(d.firstSeen, pid)
72
      delete(d.fired, pid)
73
    }
74
  }
75
  return signals
76
}