Merge pull request #740 from crosbymichael/oom-event

Emit oom events

linux/runtime.go

@@ -56,7 +56,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
 		cfg.Runtime = defaultRuntime
 	}
 	c, cancel := context.WithCancel(ic.Context)
-	return &Runtime{
+	r := &Runtime{
 		root:          path,
 		remote:        !cfg.NoShim,
 		runtime:       cfg.Runtime,
@@ -64,7 +64,10 @@ func New(ic *plugin.InitContext) (interface{}, error) {
 		eventsContext: c,
 		eventsCancel:  cancel,
 		monitor:       ic.Monitor,
-	}, nil
+	}
+	// set the events output for a monitor if it generates events
+	ic.Monitor.Events(r.events)
+	return r, nil
 }
 
 type Runtime struct {

metrics/cgroups/cgroups.go

@@ -1,6 +1,8 @@
 package cgroups
 
 import (
+	"time"
+
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/plugin"
 	"github.com/crosbymichael/cgroups"
@@ -39,6 +41,7 @@ type cgroupsMonitor struct {
 	collector *prometheus.Collector
 	oom       *prometheus.OOMCollector
 	context   context.Context
+	events    chan<- *containerd.Event
 }
 
 func (m *cgroupsMonitor) Monitor(c containerd.Container) error {
@@ -58,7 +61,7 @@ func (m *cgroupsMonitor) Monitor(c containerd.Container) error {
 	if err := m.collector.Add(id, cg); err != nil {
 		return err
 	}
-	return m.oom.Add(id, cg)
+	return m.oom.Add(id, cg, m.trigger)
 }
 
 func (m *cgroupsMonitor) Stop(c containerd.Container) error {
@@ -68,3 +71,15 @@ func (m *cgroupsMonitor) Stop(c containerd.Container) error {
 	m.collector.Remove(c.Info().ID)
 	return nil
 }
+
+func (m *cgroupsMonitor) Events(events chan<- *containerd.Event) {
+	m.events = events
+}
+
+func (m *cgroupsMonitor) trigger(id string, cg cgroups.Cgroup) {
+	m.events <- &containerd.Event{
+		Timestamp: time.Now(),
+		Type:      containerd.OOMEvent,
+		ID:        id,
+	}
+}

plugin/monitor.go

@@ -8,6 +8,8 @@ type ContainerMonitor interface {
 	Monitor(containerd.Container) error
 	// Stop stops and removes the provided container from the monitor
 	Stop(containerd.Container) error
+	// Events emits events from the monitor
+	Events(chan<- *containerd.Event)
 }
 
 func NewMultiContainerMonitor(monitors ...ContainerMonitor) ContainerMonitor {
@@ -31,6 +33,9 @@ func (mm *noopContainerMonitor) Stop(c containerd.Container) error {
 	return nil
 }
 
+func (mm *noopContainerMonitor) Events(events chan<- *containerd.Event) {
+}
+
 type multiContainerMonitor struct {
 	monitors []ContainerMonitor
 }
@@ -52,3 +57,9 @@ func (mm *multiContainerMonitor) Stop(c containerd.Container) error {
 	}
 	return nil
 }
+
+func (mm *multiContainerMonitor) Events(events chan<- *containerd.Event) {
+	for _, m := range mm.monitors {
+		m.Events(events)
+	}
+}

vendor.conf

@@ -1,6 +1,6 @@
 github.com/crosbymichael/go-runc 65847bfc51952703ca24b564d10de50d3f2db6e7
 github.com/crosbymichael/console f13f890e20a94bdec6c328cdf9410b7158f0cfa4
-github.com/crosbymichael/cgroups a692a19766b072b86d89620c97a7916b2e2de3e7
+github.com/crosbymichael/cgroups e950a27f3faf567abbf995bfbec90eaddc766d25
 github.com/docker/go-metrics 8fd5772bf1584597834c6f7961a530f06cbfbb87
 github.com/prometheus/client_golang v0.8.0
 github.com/prometheus/client_model fa8ad6fec33561be4280a8f0514318c79d7f6cb6

vendor/github.com/crosbymichael/cgroups/prometheus/metrics.go

@@ -16,6 +16,10 @@ var (
 	ErrCgroupNotExists  = errors.New("cgroup does not exist in the collector")
 )
 
+// Trigger will be called when an event happens and provides the cgroup
+// where the event originated from
+type Trigger func(string, cgroups.Cgroup)
+
 // New registers the Collector with the provided namespace and returns it so
 // that cgroups can be added for collection
 func New(ns *metrics.Namespace) *Collector {

vendor/github.com/crosbymichael/cgroups/prometheus/oom.go

@@ -33,11 +33,12 @@ type OOMCollector struct {
 }
 
 type oom struct {
-	id string
+	id       string
-	c  cgroups.Cgroup
+	c        cgroups.Cgroup
+	triggers []Trigger
 }
 
-func (o *OOMCollector) Add(id string, cg cgroups.Cgroup) error {
+func (o *OOMCollector) Add(id string, cg cgroups.Cgroup, triggers ...Trigger) error {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 	fd, err := cg.OOMEventFD()
@@ -45,8 +46,9 @@ func (o *OOMCollector) Add(id string, cg cgroups.Cgroup) error {
 		return err
 	}
 	o.set[fd] = &oom{
-		id: id,
+		id:       id,
-		c:  cg,
+		c:        cg,
+		triggers: triggers,
 	}
 	// set the gauge's default value
 	o.memoryOOM.WithValues(id).Set(0)
@@ -102,6 +104,9 @@ func (o *OOMCollector) process(fd uintptr, event uint32) {
 		return
 	}
 	o.memoryOOM.WithValues(info.id).Inc(1)
+	for _, t := range info.triggers {
+		t(info.id, info.c)
+	}
 }
 
 func flush(fd uintptr) error {