Katyusha's blog

1
package detector
2

3
import (
4
  "time"
5

6
  "github.com/os2026/ebpf-rca/internal/collector"
7
)
8

9
// IOSignal 表示一次已确认的 I/O 延迟抖动异常。
10
type IOSignal struct {
11
  Sample      collector.IOSample
12
  WindowStart time.Time
13
  WindowEnd   time.Time
14
}
15

16
// IODetector 检测 P99 时延持续超阈值的块设备。
17
type IODetector struct {
18
  P99ThresholdMs float64
19
  SustainTicks   int
20
  counters       map[uint32]int
21
  firstSeen      map[uint32]time.Time
22
  fired          map[uint32]bool
23
}
24

25
// NewIODetector 构造检测器（阈值单位为毫秒）。
26
func NewIODetector(p99ThresholdMs float64, sustain int) *IODetector {
27
  if sustain < 1 {
28
    sustain = 1
29
  }
30
  return &IODetector{
31
    P99ThresholdMs: p99ThresholdMs,
32
    SustainTicks:   sustain,
33
    counters:       make(map[uint32]int),
34
    firstSeen:      make(map[uint32]time.Time),
35
    fired:          make(map[uint32]bool),
36
  }
37
}
38

39
// Detect 处理一个窗口的样本，返回本窗口新触发的异常信号。
40
func (d *IODetector) Detect(samples []collector.IOSample, now time.Time) []IOSignal {
41
  active := make(map[uint32]bool, len(samples))
42
  var signals []IOSignal
43

44
  for _, s := range samples {
45
    if s.P99LatMs < d.P99ThresholdMs {
46
      continue
47
    }
48
    active[s.Dev] = true
49
    if d.counters[s.Dev] == 0 {
50
      d.firstSeen[s.Dev] = now
51
    }
52
    d.counters[s.Dev]++
53
    if d.counters[s.Dev] >= d.SustainTicks && !d.fired[s.Dev] {
54
      d.fired[s.Dev] = true
55
      signals = append(signals, IOSignal{
56
        Sample:      s,
57
        WindowStart: d.firstSeen[s.Dev],
58
        WindowEnd:   now,
59
      })
60
    }
61
  }
62

63
  for dev := range d.counters {
64
    if !active[dev] {
65
      delete(d.counters, dev)
66
      delete(d.firstSeen, dev)
67
      delete(d.fired, dev)
68
    }
69
  }
70
  return signals
71
}