add ability to subscribe to cgroup change

In order to prepare for synchronisation of interested cgroup id with bpf
programs, this adds a way to subscribe to cgroup change via a channel
share with cgroup monitor.

Author: dqminh

cgroup/fanotify.go

@@ -199,6 +199,10 @@ func (m *fanotifyMonitor) Resolve(id int) string {
 	return m.observer.lookup(id)
 }
 
+func (m *fanotifyMonitor) SubscribeCgroupChange(ch chan<- CgroupChange) 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,21 @@
 package cgroup
 
 import (
+	"errors"
 	"log"
 )
 
+var ErrCgroupIdMapUnsupported = errors.New("cgroup change subscription failed (fanotify not available)")
+
+type CgroupChange struct {
+	Id     int
+	Path   string
+	Remove bool
+}
+
 type monitor interface {
 	Resolve(id int) string
+	SubscribeCgroupChange(chan<- CgroupChange) error
 }
 
 // Monitor resolves cgroup ids into their respective paths
@@ -34,3 +44,7 @@ func NewMonitor(path string) (*Monitor, error) {
 func (m *Monitor) Resolve(id int) string {
 	return m.inner.Resolve(id)
 }
+
+func (m *Monitor) SubscribeCgroupChange(ch chan<- CgroupChange) 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<- CgroupChange
 }
 
 func newObserver(initial map[int]string) *observer {
 	observer := observer{
 		lock:        sync.Mutex{},
 		inodeToPath: map[int]*resolved{},
 		pathToInode: map[string]int{},
+		cgroupChans: []chan<- CgroupChange{},
 	}
 
 	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 <- CgroupChange{
+			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 <- CgroupChange{
+			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<- CgroupChange) error {
+	o.cgroupChans = append(o.cgroupChans, ch)
+	// send the initial cgroup mapping
+	go func() {
+		for path, inode := range o.pathToInode {
+			ch <- CgroupChange{
+				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<- CgroupChange) error {
+	return ErrCgroupIDMapUnsupported
+}
+
 func walk(dir string) (map[int]string, error) {
 	mapping := map[int]string{}