Merge pull request #531 from dqminh/cgroup_id

Feature: synchronize cgroup id with ebpf program via shared map

Author: bobrik

.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,6 +198,10 @@ func (m *fanotifyMonitor) Resolve(id int) string {
 	return m.observer.lookup(id)
 }
 
+func (m *fanotifyMonitor) SubscribeCgroupChange(ch chan<- ChangeNotification) error {
+	return m.observer.subscribeCgroupChange(ch)
+}
+
 // The following kernel patch is required to take advantage of this (included in v6.6-rc1):
 // * https://git.kernel.org/torvalds/c/0ce7c12e88cf ("kernfs: attach uuid for every kernfs and report it in fsid")
 func attachFanotify(path string) (io.Reader, error) {

cgroup/monitor.go

@@ -1,11 +1,24 @@
 package cgroup
 
 import (
+	"errors"
 	"log"
 )
 
+// ErrCgroupIDMapUnsupported is returned when cgroup id map is not available
+var ErrCgroupIDMapUnsupported = errors.New("cgroup change subscription failed (fanotify not available)")
+
+// 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(ch chan<- ChangeNotification) error
 }
 
 // Monitor resolves cgroup ids into their respective paths
@@ -34,3 +47,9 @@ func NewMonitor(path string) (*Monitor, error) {
 func (m *Monitor) Resolve(id int) string {
 	return m.inner.Resolve(id)
 }
+
+// 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,13 +19,15 @@ type observer struct {
 	lock        sync.Mutex
 	inodeToPath map[int]*resolved
 	pathToInode map[string]int
+	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<- ChangeNotification{},
 	}
 
 	for inode, name := range initial {
@@ -83,6 +85,12 @@ func (o *observer) add(inode int, path string) {
 
 	o.inodeToPath[inode] = r
 	o.pathToInode[path] = inode
+	for _, ch := range o.cgroupChans {
+		ch <- ChangeNotification{
+			ID:   inode,
+			Path: path,
+		}
+	}
 }
 
 func (o *observer) remove(path string) {
@@ -96,6 +104,13 @@ func (o *observer) remove(path string) {
 
 	r := o.inodeToPath[inode]
 	r.dead = time.Now()
+	for _, ch := range o.cgroupChans {
+		ch <- ChangeNotification{
+			ID:     inode,
+			Path:   path,
+			Remove: true,
+		}
+	}
 }
 
 func (o *observer) lookup(inode int) string {
@@ -113,3 +128,17 @@ func (o *observer) lookup(inode int) string {
 
 	return r.path
 }
+
+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 <- ChangeNotification{
+				ID:   inode,
+				Path: path,
+			}
+		}
+	}()
+	return nil
+}

cgroup/walker.go

@@ -45,6 +45,10 @@ func (m *walkerMonitor) Resolve(id int) string {
 	return m.mapping[id]
 }
 
+func (m *walkerMonitor) SubscribeCgroupChange(_ chan<- ChangeNotification) error {
+	return ErrCgroupIDMapUnsupported
+}
+
 func walk(dir string) (map[int]string, error) {
 	mapping := map[int]string{}
 

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"`

decoder/cgroup.go

@@ -13,16 +13,6 @@ type CGroup struct {
 	monitor *cgroup.Monitor
 }
 
-// NewCgroupDecoder creates a new cgroup decoder
-func NewCgroupDecoder() (*CGroup, error) {
-	monitor, err := cgroup.NewMonitor("/sys/fs/cgroup")
-	if err != nil {
-		return nil, fmt.Errorf("error creating cgroup monitor: %w", err)
-	}
-
-	return &CGroup{monitor}, nil
-}
-
 // Decode transforms cgroup id to path in cgroupfs
 func (c *CGroup) Decode(in []byte, _ config.Decoder) ([]byte, error) {
 	cgroupID, err := strconv.Atoi(string(in))

decoder/decoder.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/cloudflare/ebpf_exporter/v2/cgroup"
 	"github.com/cloudflare/ebpf_exporter/v2/config"
 	"github.com/cloudflare/ebpf_exporter/v2/kallsyms"
 	lru "github.com/hashicorp/golang-lru/v2"
@@ -29,20 +30,15 @@ type Set struct {
 }
 
 // NewSet creates a Set with all known decoders
-func NewSet(skipCacheSize int) (*Set, error) {
-	cgroup, err := NewCgroupDecoder()
-	if err != nil {
-		return nil, fmt.Errorf("error creating cgroup decoder: %w", err)
-	}
-
+func NewSet(skipCacheSize int, monitor *cgroup.Monitor) (*Set, error) {
 	ksym, err := kallsyms.NewDecoder("/proc/kallsyms")
 	if err != nil {
 		return nil, fmt.Errorf("error creating ksym decoder: %w", err)
 	}
 
 	s := &Set{
 		decoders: map[string]Decoder{
-			"cgroup":       cgroup,
+			"cgroup":       &CGroup{monitor},
 			"dname":        &Dname{},
 			"errno":        &Errno{},
 			"hex":          &Hex{},

decoder/decoder_test.go

@@ -149,7 +149,7 @@ func TestDecodeLabels(t *testing.T) {
 	}
 
 	for i, c := range cases {
-		s, err := NewSet(0)
+		s, err := NewSet(0, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -255,7 +255,7 @@ func TestDecodeSkipLabels(t *testing.T) {
 	}
 
 	for i, c := range cases {
-		s, err := NewSet(100)
+		s, err := NewSet(100, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -322,7 +322,7 @@ func TestDecoderSetConcurrency(t *testing.T) {
 		},
 	}
 
-	s, err := NewSet(0)
+	s, err := NewSet(0, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -387,7 +387,7 @@ func TestDecoderSetCache(t *testing.T) {
 		},
 	}
 
-	s, err := NewSet(0)
+	s, err := NewSet(0, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -458,7 +458,7 @@ func BenchmarkCache(b *testing.B) {
 		},
 	}
 
-	s, err := NewSet(0)
+	s, err := NewSet(0, nil)
 	if err != nil {
 		b.Fatal(err)
 	}

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";