add cgroup_id_map config to sync cgroup id to ebpf programs

For some usecases, it's useful for ebpf programs to filter their metrics
based on cgroups. It's generally hard to ebpf program to be able to
filter cgroup at runtime. This make it easier by allowing ebpf exporter
to update known interesting cgroup id at runtime via a shared BPF map.

Author: dqminh

.vscode/config-schema.yaml

@@ -71,6 +71,16 @@ properties:
                 type: number
             labels:
               $ref: "#/definitions/labels"
+  cgroup_id_map:
+    type: object
+    additionalProperties: false
+    properties:
+      name:
+        type: string
+      regexps:
+        type: array
+        items:
+          type: string
   tracing:
     type: object
     additionalProperties: false

cgroup/fanotify.go

@@ -63,9 +63,8 @@ func newFanotifyMonitor(path string) (*fanotifyMonitor, error) {
 	}
 
 	go func() {
-		if err := m.readFanotifyLoop(); err != nil {
-			log.Fatalf("Error running fanotify loop: %v", err)
-		}
+		err := m.readFanotifyLoop()
+		log.Fatalf("Fanotify loop terminated with err:%v", err)
 	}()
 
 	return m, nil
@@ -199,7 +198,7 @@ func (m *fanotifyMonitor) Resolve(id int) string {
 	return m.observer.lookup(id)
 }
 
-func (m *fanotifyMonitor) SubscribeCgroupChange(ch chan<- CgroupChange) error {
+func (m *fanotifyMonitor) SubscribeCgroupChange(ch chan<- ChangeNotification) error {
 	return m.observer.subscribeCgroupChange(ch)
 }
 

cgroup/monitor.go

@@ -5,17 +5,20 @@ import (
 	"log"
 )
 
-var ErrCgroupIdMapUnsupported = errors.New("cgroup change subscription failed (fanotify not available)")
+// ErrCgroupIDMapUnsupported is returned when cgroup id map is not available
+var ErrCgroupIDMapUnsupported = errors.New("cgroup change subscription failed (fanotify not available)")
 
-type CgroupChange struct {
-	Id     int
+// ChangeNotification is the notification returned by cgroup monitor when a subscribed
+// cgroup has been added or removed
+type ChangeNotification struct {
+	ID     int
 	Path   string
 	Remove bool
 }
 
 type monitor interface {
 	Resolve(id int) string
-	SubscribeCgroupChange(chan<- CgroupChange) error
+	SubscribeCgroupChange(ch chan<- ChangeNotification) error
 }
 
 // Monitor resolves cgroup ids into their respective paths
@@ -45,6 +48,8 @@ func (m *Monitor) Resolve(id int) string {
 	return m.inner.Resolve(id)
 }
 
-func (m *Monitor) SubscribeCgroupChange(ch chan<- CgroupChange) error {
+// SubscribeCgroupChange receives cgroup change notifications. This requires
+// kernel with fanotify support for cgroup
+func (m *Monitor) SubscribeCgroupChange(ch chan<- ChangeNotification) error {
 	return m.inner.SubscribeCgroupChange(ch)
 }

cgroup/observer.go

@@ -19,15 +19,15 @@ type observer struct {
 	lock        sync.Mutex
 	inodeToPath map[int]*resolved
 	pathToInode map[string]int
-	cgroupChans []chan<- CgroupChange
+	cgroupChans []chan<- ChangeNotification
 }
 
 func newObserver(initial map[int]string) *observer {
 	observer := observer{
 		lock:        sync.Mutex{},
 		inodeToPath: map[int]*resolved{},
 		pathToInode: map[string]int{},
-		cgroupChans: []chan<- CgroupChange{},
+		cgroupChans: []chan<- ChangeNotification{},
 	}
 
 	for inode, name := range initial {
@@ -86,7 +86,7 @@ func (o *observer) add(inode int, path string) {
 	o.inodeToPath[inode] = r
 	o.pathToInode[path] = inode
 	for _, ch := range o.cgroupChans {
-		ch <- CgroupChange{
+		ch <- ChangeNotification{
 			ID:   inode,
 			Path: path,
 		}
@@ -105,7 +105,7 @@ func (o *observer) remove(path string) {
 	r := o.inodeToPath[inode]
 	r.dead = time.Now()
 	for _, ch := range o.cgroupChans {
-		ch <- CgroupChange{
+		ch <- ChangeNotification{
 			ID:     inode,
 			Path:   path,
 			Remove: true,
@@ -129,12 +129,12 @@ func (o *observer) lookup(inode int) string {
 	return r.path
 }
 
-func (o *observer) subscribeCgroupChange(ch chan<- CgroupChange) error {
+func (o *observer) subscribeCgroupChange(ch chan<- ChangeNotification) error {
 	o.cgroupChans = append(o.cgroupChans, ch)
 	// send the initial cgroup mapping
 	go func() {
 		for path, inode := range o.pathToInode {
-			ch <- CgroupChange{
+			ch <- ChangeNotification{
 				ID:   inode,
 				Path: path,
 			}

cgroup/walker.go

@@ -45,7 +45,7 @@ func (m *walkerMonitor) Resolve(id int) string {
 	return m.mapping[id]
 }
 
-func (m *walkerMonitor) SubscribeCgroupChange(_ chan<- CgroupChange) error {
+func (m *walkerMonitor) SubscribeCgroupChange(_ chan<- ChangeNotification) error {
 	return ErrCgroupIDMapUnsupported
 }
 

config/config.go

@@ -11,11 +11,12 @@ import (
 
 // Config describes how to configure and extract metrics
 type Config struct {
-	Name    string   `yaml:"name"`
-	Metrics Metrics  `yaml:"metrics"`
-	Tracing Tracing  `yaml:"tracing"`
-	Kaddrs  []string `yaml:"kaddrs"`
-	BPFPath string
+	Name        string      `yaml:"name"`
+	Metrics     Metrics     `yaml:"metrics"`
+	Tracing     Tracing     `yaml:"tracing"`
+	Kaddrs      []string    `yaml:"kaddrs"`
+	CgroupIDMap CgroupIDMap `yaml:"cgroup_id_map"`
+	BPFPath     string
 }
 
 // Metrics is a collection of metrics attached to a program
@@ -45,6 +46,14 @@ type Histogram struct {
 	Labels           []Label             `yaml:"labels"`
 }
 
+// CgroupIDMap describes the cgroup that the bpf programs are interested in.
+// The cgroups that match the provided regexps will be available to the bpf program
+// as a shared map with provided name.
+type CgroupIDMap struct {
+	Name    string   `yaml:"name"`
+	Regexps []string `yaml:"regexps"`
+}
+
 // Tracing is a collection of spans attached to a program
 type Tracing struct {
 	Spans []Span `yaml:"spans"`

examples/cgroup_id_map.bpf.c

@@ -0,0 +1,40 @@
+#include <vmlinux.h>
+#include <bpf/bpf_tracing.h>
+#include "maps.bpf.h"
+
+struct {
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __uint(max_entries, 1024);
+    __type(key, u64);
+    __type(value, u64);
+} cgroup_sched_migrations_total SEC(".maps");
+
+struct {
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __uint(max_entries, 1024);
+    __type(key, u64);
+    __type(value, u64);
+} cgroup_sched_migrations_not_match_total SEC(".maps");
+
+struct {
+    __uint(type, BPF_MAP_TYPE_LRU_HASH);
+    __uint(max_entries, 1024);
+    __type(key, u64);
+    __type(value, u64);
+} cgroup_id_map SEC(".maps");
+
+SEC("tp_btf/sched_migrate_task")
+int BPF_PROG(sched_migrate_task)
+{
+    u64 *ok;
+    u64 cgroup_id = bpf_get_current_cgroup_id();
+    ok = bpf_map_lookup_elem(&cgroup_id_map, &cgroup_id);
+    if (ok) {
+        increment_map(&cgroup_sched_migrations_total, &cgroup_id, 1);
+    } else {
+        increment_map(&cgroup_sched_migrations_not_match_total, &cgroup_id, 1);
+    }
+    return 0;
+}
+
+char LICENSE[] SEC("license") = "GPL";

examples/cgroup_id_map.yaml

@@ -0,0 +1,24 @@
+metrics:
+  counters:
+    - name: cgroup_sched_migrations_total
+      help: Number of sched:sched_migrate_task events per cgroup
+      labels:
+        - name: cgroup
+          size: 8
+          decoders:
+            - name: uint
+            - name: cgroup
+
+    - name: cgroup_sched_migrations_not_match_total
+      help: Number of sched:sched_migrate_task events per cgroup not match cgroup id map
+      labels:
+        - name: cgroup
+          size: 8
+          decoders:
+            - name: uint
+            - name: cgroup
+
+cgroup_id_map:
+  name: cgroup_id_map
+  regexps:
+    - ^.*(system.slice/.*)$

exporter/cgroup_id_map.go

@@ -0,0 +1,83 @@
+package exporter
+
+import (
+	"fmt"
+	"log"
+	"regexp"
+	"unsafe"
+
+	"github.com/aquasecurity/libbpfgo"
+	"github.com/cloudflare/ebpf_exporter/v2/cgroup"
+	"github.com/cloudflare/ebpf_exporter/v2/config"
+)
+
+// CgroupIDMap synchronises cgroup changes with the shared bpf map.
+type CgroupIDMap struct {
+	bpfMap *libbpfgo.BPFMap
+	ch     chan cgroup.ChangeNotification
+	cache  map[string]*regexp.Regexp
+}
+
+func newCgroupIDMap(module *libbpfgo.Module, cfg config.Config) (*CgroupIDMap, error) {
+	m, err := module.GetMap(cfg.CgroupIDMap.Name)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get map %q: %w", cfg.CgroupIDMap.Name, err)
+	}
+
+	keySize := m.KeySize()
+	if keySize != 8 {
+		return nil, fmt.Errorf("key size for map %q is not expected 8 bytes (u64), it is %d bytes", cfg.CgroupIDMap.Name, keySize)
+	}
+	valueSize := m.ValueSize()
+	if valueSize != 8 {
+		return nil, fmt.Errorf("value size for map %q is not expected 8 bytes (u64), it is %d bytes", cfg.CgroupIDMap.Name, valueSize)
+	}
+
+	c := &CgroupIDMap{
+		bpfMap: m,
+		ch:     make(chan cgroup.ChangeNotification, 10),
+		cache:  map[string]*regexp.Regexp{},
+	}
+
+	for _, expr := range cfg.CgroupIDMap.Regexps {
+		if _, ok := c.cache[expr]; !ok {
+			compiled, err := regexp.Compile(expr)
+			if err != nil {
+				return nil, fmt.Errorf("error compiling regexp %q: %w", expr, err)
+			}
+			c.cache[expr] = compiled
+		}
+	}
+
+	return c, nil
+}
+
+func (c *CgroupIDMap) subscribe(m *cgroup.Monitor) error {
+	return m.SubscribeCgroupChange(c.ch)
+}
+
+func (c *CgroupIDMap) runLoop() {
+	for update := range c.ch {
+		if update.Remove {
+			key := uint64(update.ID)
+			err := c.bpfMap.DeleteKey(unsafe.Pointer(&key))
+			log.Printf("Error deleting key from CgroupIDMap: %v", err)
+		} else {
+			key := uint64(update.ID)
+			value := uint64(1)
+			if c.checkMatch(update.Path) {
+				err := c.bpfMap.Update(unsafe.Pointer(&key), unsafe.Pointer(&value))
+				log.Printf("Error updating CgroupIDMap: %v", err)
+			}
+		}
+	}
+}
+
+func (c *CgroupIDMap) checkMatch(path string) bool {
+	for _, compiled := range c.cache {
+		if compiled.MatchString(path) {
+			return true
+		}
+	}
+	return false
+}

exporter/exporter.go

@@ -15,6 +15,7 @@ import (
 	"unsafe"
 
 	"github.com/aquasecurity/libbpfgo"
+	"github.com/cloudflare/ebpf_exporter/v2/cgroup"
 	"github.com/cloudflare/ebpf_exporter/v2/config"
 	"github.com/cloudflare/ebpf_exporter/v2/decoder"
 	"github.com/cloudflare/ebpf_exporter/v2/tracing"
@@ -36,8 +37,9 @@ var percpuMapTypes = map[libbpfgo.MapType]struct{}{
 
 // Exporter is a ebpf_exporter instance implementing prometheus.Collector
 type Exporter struct {
-	configs                  []config.Config
-	modules                  map[string]*libbpfgo.Module
+	configs []config.Config
+	modules map[string]*libbpfgo.Module
+
 	perfEventArrayCollectors []*perfEventArraySink
 	kaddrs                   map[string]uint64
 	enabledConfigsDesc       *prometheus.Desc
@@ -238,12 +240,31 @@ func (e *Exporter) attachConfig(ctx context.Context, cfg config.Config) error {
 		return fmt.Errorf("error validating maps for config %q: %w", cfg.Name, err)
 	}
 
+	// attach cgroup id map if exists
+	if len(cfg.CgroupIDMap.Name) > 0 {
+		if err := e.attachCgroupIDMap(module, cfg); err != nil {
+			return err
+		}
+	}
+
 	e.attachedProgs[cfg.Name] = attachments
 	e.modules[cfg.Name] = module
 
 	return nil
 }
 
+func (e *Exporter) attachCgroupIDMap(module *libbpfgo.Module, cfg config.Config) error {
+	cgMap, err := newCgroupIDMap(module, cfg)
+	if err != nil {
+		return err
+	}
+	if err := cgMap.subscribe(e.cgroupMonitor); err != nil {
+		return err
+	}
+	go cgMap.runLoop()
+	return nil
+}
+
 // Detach detaches bpf programs and maps for exiting
 func (e *Exporter) Detach() {
 	e.activeMutex.Lock()