|
| 1 | +//go:build linux |
| 2 | + |
| 3 | +package probes |
| 4 | + |
| 5 | +import ( |
| 6 | + "context" |
| 7 | + "sync" |
| 8 | + |
| 9 | + "github.com/cilium/ebpf" |
| 10 | + "github.com/cilium/ebpf/link" |
| 11 | + log "github.com/sirupsen/logrus" |
| 12 | + "go.opentelemetry.io/ebpf-profiler/libpf" |
| 13 | +) |
| 14 | + |
| 15 | +// attachReq is enqueued by OnExecutable for the worker goroutine to process. |
| 16 | +type attachReq struct { |
| 17 | + filePath string |
| 18 | + fileID libpf.FileID |
| 19 | + specs []ProbeSpec |
| 20 | +} |
| 21 | + |
| 22 | +// attacher owns the state for matching new binaries against the spec list and |
| 23 | +// attaching paired entry/exit uprobes to them on a worker goroutine. The |
| 24 | +// hot-path callback only holds the mutex long enough to dedupe by FileID and |
| 25 | +// run regexes. |
| 26 | +type attacher struct { |
| 27 | + progEntry *ebpf.Program |
| 28 | + progExit *ebpf.Program |
| 29 | + specs []ProbeSpec |
| 30 | + |
| 31 | + mu sync.Mutex |
| 32 | + attached map[libpf.FileID]struct{} |
| 33 | + links map[libpf.FileID][]link.Link |
| 34 | + |
| 35 | + queue chan attachReq |
| 36 | +} |
| 37 | + |
| 38 | +func newAttacher(progEntry, progExit *ebpf.Program, specs []ProbeSpec, queueDepth int) *attacher { |
| 39 | + return &attacher{ |
| 40 | + progEntry: progEntry, |
| 41 | + progExit: progExit, |
| 42 | + specs: specs, |
| 43 | + attached: make(map[libpf.FileID]struct{}), |
| 44 | + links: make(map[libpf.FileID][]link.Link), |
| 45 | + queue: make(chan attachReq, queueDepth), |
| 46 | + } |
| 47 | +} |
| 48 | + |
| 49 | +// OnExecutable is the cheap-path callback invoked from the otel ebpf-profiler |
| 50 | +// reporter goroutine. It must not block: dedupe, regex-match, and enqueue. |
| 51 | +// All disk I/O happens in attachWorker. |
| 52 | +func (a *attacher) OnExecutable(filePath string, fileID libpf.FileID) { |
| 53 | + a.mu.Lock() |
| 54 | + if _, seen := a.attached[fileID]; seen { |
| 55 | + a.mu.Unlock() |
| 56 | + return |
| 57 | + } |
| 58 | + a.attached[fileID] = struct{}{} |
| 59 | + a.mu.Unlock() |
| 60 | + |
| 61 | + var matched []ProbeSpec |
| 62 | + for _, s := range a.specs { |
| 63 | + if s.FileMatchRE.MatchString(filePath) { |
| 64 | + matched = append(matched, s) |
| 65 | + } |
| 66 | + } |
| 67 | + if len(matched) == 0 { |
| 68 | + log.Debugf("probes: received %s (fileID=%s) — no spec matched", filePath, fileID.StringNoQuotes()) |
| 69 | + return |
| 70 | + } |
| 71 | + log.Debugf("probes: received %s (fileID=%s) — %d spec(s) matched", filePath, fileID.StringNoQuotes(), len(matched)) |
| 72 | + |
| 73 | + select { |
| 74 | + case a.queue <- attachReq{filePath: filePath, fileID: fileID, specs: matched}: |
| 75 | + default: |
| 76 | + // Queue full: log and forget. We've already marked this fileID as |
| 77 | + // "seen" so we won't try again, which is fine for v1 — the user |
| 78 | + // can restart with a smaller probe-config or a deeper queue. |
| 79 | + log.Warnf("probes: attach queue full, dropping %s (fileID=%s)", filePath, fileID.StringNoQuotes()) |
| 80 | + } |
| 81 | +} |
| 82 | + |
| 83 | +// run is the attachWorker goroutine. Returns when the queue is drained after |
| 84 | +// ctx is cancelled. |
| 85 | +func (a *attacher) run(ctx context.Context) { |
| 86 | + for { |
| 87 | + select { |
| 88 | + case <-ctx.Done(): |
| 89 | + return |
| 90 | + case req := <-a.queue: |
| 91 | + a.handle(req) |
| 92 | + } |
| 93 | + } |
| 94 | +} |
| 95 | + |
| 96 | +func (a *attacher) handle(req attachReq) { |
| 97 | + ex, err := link.OpenExecutable(req.filePath) |
| 98 | + if err != nil { |
| 99 | + log.Warnf("probes: open executable %s: %v", req.filePath, err) |
| 100 | + return |
| 101 | + } |
| 102 | + |
| 103 | + var newLinks []link.Link |
| 104 | + for _, s := range req.specs { |
| 105 | + cookie := s.Cookie() |
| 106 | + |
| 107 | + entryLink, err := ex.Uprobe(s.EntrySymbol, a.progEntry, &link.UprobeOptions{ |
| 108 | + Cookie: cookie, |
| 109 | + PID: 0, |
| 110 | + }) |
| 111 | + if err != nil { |
| 112 | + log.Warnf("probes: attach entry %s @ %s: %v", s.EntrySymbol, req.filePath, err) |
| 113 | + continue |
| 114 | + } |
| 115 | + exitLink, err := ex.Uprobe(s.ExitSymbol, a.progExit, &link.UprobeOptions{ |
| 116 | + Cookie: cookie, |
| 117 | + PID: 0, |
| 118 | + }) |
| 119 | + if err != nil { |
| 120 | + log.Warnf("probes: attach exit %s @ %s: %v", s.ExitSymbol, req.filePath, err) |
| 121 | + // Roll back the entry uprobe so we don't have orphan stack pushes |
| 122 | + // with no exit to drain them. |
| 123 | + if cerr := entryLink.Close(); cerr != nil { |
| 124 | + log.Warnf("probes: rollback entry %s: %v", s.EntrySymbol, cerr) |
| 125 | + } |
| 126 | + continue |
| 127 | + } |
| 128 | + newLinks = append(newLinks, entryLink, exitLink) |
| 129 | + log.Debugf("probes: attached pair %s/%s @ %s (spec_id=%d, id=%s)", |
| 130 | + s.EntrySymbol, s.ExitSymbol, req.filePath, s.SpecID, s.ID) |
| 131 | + } |
| 132 | + if len(newLinks) == 0 { |
| 133 | + return |
| 134 | + } |
| 135 | + |
| 136 | + a.mu.Lock() |
| 137 | + a.links[req.fileID] = append(a.links[req.fileID], newLinks...) |
| 138 | + a.mu.Unlock() |
| 139 | +} |
| 140 | + |
| 141 | +// closeAllLinks tears down every attached uprobe link. Called from |
| 142 | +// service.Close. |
| 143 | +func (a *attacher) closeAllLinks() { |
| 144 | + a.mu.Lock() |
| 145 | + defer a.mu.Unlock() |
| 146 | + for fid, links := range a.links { |
| 147 | + for _, l := range links { |
| 148 | + if err := l.Close(); err != nil { |
| 149 | + log.Warnf("probes: close link for fileID=%s: %v", fid.StringNoQuotes(), err) |
| 150 | + } |
| 151 | + } |
| 152 | + } |
| 153 | + a.links = nil |
| 154 | +} |
0 commit comments