ebpf_exporter

1
package exporter
2

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

7
	"github.com/aquasecurity/libbpfgo"
8
	"github.com/cloudflare/ebpf_exporter/v2/config"
9
	"github.com/cloudflare/ebpf_exporter/v2/decoder"
10
	"github.com/prometheus/client_golang/prometheus"
11
)
12

13
type perfEventArraySink struct {
14
	counterConfig config.Counter
15
	counterVec    *prometheus.CounterVec
16
	dropCounter   prometheus.Counter
17
}
18

19
func newPerfEventArraySink(decoders *decoder.Set, module *libbpfgo.Module, counterConfig config.Counter) *perfEventArraySink {
20
	var (
21
		receiveCh = make(chan []byte)
22
		lostCh    = make(chan uint64)
23
	)
24

25
	sink := &perfEventArraySink{
26
		counterConfig: counterConfig,
27
		dropCounter:   createDropCounterForPerfMap(counterConfig),
28
	}
29
	sink.resetCounterVec()
30

31
	perfEventBuf, err := module.InitPerfBuf(counterConfig.Name, receiveCh, lostCh, 1024)
32

33
	if err != nil {
34
		log.Fatalf("Can't init PerfBuf: %s", err)
35
	}
36

37
	go func(sink *perfEventArraySink, receiveCh <-chan []byte) {
38
		for rawBytes := range receiveCh {
39
			// https://github.com/cilium/ebpf/pull/94#discussion_r425823371
40
			// https://lore.kernel.org/patchwork/patch/1244339/
41
			var validDataSize uint
42
			for _, labelConfig := range sink.counterConfig.Labels {
43
				validDataSize += labelConfig.Size
44
			}
45

46
			labelValues, err := decoders.DecodeLabels(rawBytes[:validDataSize], counterConfig.Name, sink.counterConfig.Labels)
47
			if err != nil {
48
				if err == decoder.ErrSkipLabelSet {
49
					continue
50
				}
51

52
				log.Printf("Failed to decode labels: %s", err)
53
			}
54

55
			sink.counterVec.WithLabelValues(labelValues...).Inc()
56
		}
57
	}(sink, receiveCh)
58

59
	go func(sink *perfEventArraySink, lostCh <-chan uint64) {
60
		for droppedEvents := range lostCh {
61
			sink.dropCounter.Add(float64(droppedEvents))
62
		}
63
	}(sink, lostCh)
64

65
	go func(sink *perfEventArraySink) {
66
		flushDuration := time.Hour
67
		if sink.counterConfig.FlushInterval > 0 {
68
			flushDuration = sink.counterConfig.FlushInterval
69
		}
70

71
		ticker := time.NewTicker(flushDuration)
72

73
		for {
74
			<-ticker.C
75
			sink.resetCounterVec()
76
		}
77
	}(sink)
78

79
	perfEventBuf.Poll(300)
80

81
	return sink
82
}
83

84
func (s *perfEventArraySink) Collect(ch chan<- prometheus.Metric) {
85
	s.counterVec.Collect(ch)
86
}
87

88
func (s *perfEventArraySink) Describe(ch chan<- *prometheus.Desc) {
89
	s.counterVec.Describe(ch)
90
}
91

92
func (s *perfEventArraySink) resetCounterVec() {
93
	s.counterVec = createCounterVecForPerfMap(s.counterConfig, labelNamesFromCounterConfig(s.counterConfig))
94
}
95

96
func createCounterVecForPerfMap(counterConfig config.Counter, labelNames []string) *prometheus.CounterVec {
97
	return prometheus.NewCounterVec(prometheus.CounterOpts{
98
		Namespace: prometheusNamespace,
99
		Name:      counterConfig.Name,
100
		Help:      counterConfig.Help,
101
	}, labelNames)
102
}
103

104
func createDropCounterForPerfMap(counterConfig config.Counter) prometheus.Counter {
105
	return prometheus.NewCounterVec(prometheus.CounterOpts{
106
		Namespace: "dropped_perf_event_map_events",
107
		Name:      counterConfig.Name,
108
		Help:      "Dropped perf map events",
109
	}, []string{}).WithLabelValues()
110
}
111

112
func labelNamesFromCounterConfig(counterConfig config.Counter) (labelNames []string) {
113
	for _, label := range counterConfig.Labels {
114
		labelNames = append(labelNames, label.Name)
115
	}
116

117
	return
118
}
119