Katyusha's blog

1
package collector
2

3
import (
4
  "fmt"
5
  "time"
6

7
  "github.com/cilium/ebpf/link"
8
  "github.com/cilium/ebpf/rlimit"
9

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

13
const maxStackDepth = 32
14

15
// lockStat 与 lock.bpf.c 中的 struct lock_stat 二进制布局一致。
16
type lockStat struct {
17
  OffcpuNs    uint64
18
  OffcpuCount uint64
19
  MaxOffcpuNs uint64
20
  LastWaker   uint32
21
  StackID     int32
22
  Comm        [16]byte
23
}
24

25
// LockSample 是单个线程在一个窗口内的 off-CPU 阻塞派生指标。
26
type LockSample struct {
27
  Pid         uint32
28
  Comm        string
29
  OffcpuRatio float64 // 阻塞型 off-CPU 时间占墙钟比例(0..1)
30
  BlockCount  uint64  // 窗口内阻塞次数
31
  MaxOffcpuMs float64 // 单次最长阻塞(毫秒,累计最大值)
32
  LastWaker   uint32  // 最近唤醒者 tid
33
  StackID     int32   // 阻塞内核栈 id（用于符号化）
34
}
35

36
// LockCollector 加载锁竞争场景的 eBPF 程序并读取 off-CPU 阻塞数据。
37
type LockCollector struct {
38
  objs  lockObjects
39
  links []link.Link
40
  ksyms *ksym.Table
41
  prev  map[uint32]lockStat
42
}
43

44
// NewLockCollector 加载字节码、挂载 tracepoint、载入内核符号表。
45
func NewLockCollector() (*LockCollector, error) {
46
  if err := rlimit.RemoveMemlock(); err != nil {
47
    return nil, fmt.Errorf("remove memlock: %w", err)
48
  }
49
  c := &LockCollector{prev: make(map[uint32]lockStat)}
50
  if err := loadLockObjects(&c.objs, nil); err != nil {
51
    return nil, fmt.Errorf("load bpf objects: %w", err)
52
  }
53
  sw, err := link.Tracepoint("sched", "sched_switch", c.objs.HandleSwitch, nil)
54
  if err != nil {
55
    c.Close()
56
    return nil, fmt.Errorf("attach sched_switch: %w", err)
57
  }
58
  c.links = append(c.links, sw)
59

60
  wk, err := link.Tracepoint("sched", "sched_wakeup", c.objs.HandleWakeup, nil)
61
  if err != nil {
62
    c.Close()
63
    return nil, fmt.Errorf("attach sched_wakeup: %w", err)
64
  }
65
  c.links = append(c.links, wk)
66

67
  // 符号表载入失败不致命：仍可输出地址。
68
  if t, err := ksym.Load(); err == nil {
69
    c.ksyms = t
70
  }
71
  return c, nil
72
}
73

74
// Close 卸载探针并释放资源。
75
func (c *LockCollector) Close() {
76
  for _, l := range c.links {
77
    _ = l.Close()
78
  }
79
  c.objs.Close()
80
}
81

82
// Poll 读取 lock_stats，计算自上次调用以来的差分。
83
func (c *LockCollector) Poll(interval time.Duration) ([]LockSample, error) {
84
  cur := make(map[uint32]lockStat)
85
  var key uint32
86
  var val lockStat
87
  it := c.objs.LockStats.Iterate()
88
  for it.Next(&key, &val) {
89
    cur[key] = val
90
  }
91
  if err := it.Err(); err != nil {
92
    return nil, fmt.Errorf("iterate lock_stats: %w", err)
93
  }
94

95
  intervalNs := float64(interval.Nanoseconds())
96
  var samples []LockSample
97
  for tid, v := range cur {
98
    var dOff, dCount uint64
99
    if p, ok := c.prev[tid]; ok {
100
      dOff = v.OffcpuNs - p.OffcpuNs
101
      dCount = v.OffcpuCount - p.OffcpuCount
102
    } else {
103
      dOff, dCount = v.OffcpuNs, v.OffcpuCount
104
    }
105
    if dOff == 0 && dCount == 0 {
106
      continue
107
    }
108
    samples = append(samples, LockSample{
109
      Pid:         tid,
110
      Comm:        commToString(v.Comm),
111
      OffcpuRatio: float64(dOff) / intervalNs,
112
      BlockCount:  dCount,
113
      MaxOffcpuMs: float64(v.MaxOffcpuNs) / 1e6,
114
      LastWaker:   v.LastWaker,
115
      StackID:     v.StackID,
116
    })
117
  }
118
  c.prev = cur
119
  return samples, nil
120
}
121

122
// ResolveStack 将阻塞栈 id 符号化为最多 max 个栈帧函数名。
123
func (c *LockCollector) ResolveStack(id int32, max int) []string {
124
  if id < 0 {
125
    return nil
126
  }
127
  var frames [maxStackDepth]uint64
128
  if err := c.objs.Stackmap.Lookup(uint32(id), &frames); err != nil {
129
    return nil
130
  }
131
  var out []string
132
  for _, a := range frames {
133
    if a == 0 {
134
      break
135
    }
136
    out = append(out, c.ksyms.Resolve(a))
137
    if len(out) >= max {
138
      break
139
    }
140
  }
141
  return out
142
}