cgroup2: implement containerd.events.TaskOOM event

How to test (from https://github.com/opencontainers/runc/pull/2352#issuecomment-620834524):
  (host)$ sudo swapoff -a
  (host)$ sudo ctr run -t --rm --memory-limit $((1024*1024*32)) docker.io/library/alpine:latest foo
  (container)$ sh -c 'VAR=$(seq 1 100000000)'

An event `/tasks/oom {"container_id":"foo"}` will be displayed in `ctr events`.

Signed-off-by: Akihiro Suda <akihiro.suda.cz@hco.ntt.co.jp>

pkg/oom/oom.go

@@ -0,0 +1,30 @@
+// +build linux
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package oom
+
+import (
+	"context"
+)
+
+// Watcher watches OOM events
+type Watcher interface {
+	Close() error
+	Run(ctx context.Context)
+	Add(id string, cg interface{}) error
+}

pkg/oom/v1/v1.go

@@ -16,7 +16,7 @@
    limitations under the License.
 */
 
-package oom
+package v1
 
 import (
 	"context"
@@ -24,28 +24,30 @@ import (
 
 	"github.com/containerd/cgroups"
 	eventstypes "github.com/containerd/containerd/api/events"
+	"github.com/containerd/containerd/pkg/oom"
 	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/containerd/runtime/v2/shim"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
 )
 
 // New returns an epoll implementation that listens to OOM events
 // from a container's cgroups.
-func New(publisher shim.Publisher) (*Epoller, error) {
+func New(publisher shim.Publisher) (oom.Watcher, error) {
 	fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
 	if err != nil {
 		return nil, err
 	}
-	return &Epoller{
+	return &epoller{
 		fd:        fd,
 		publisher: publisher,
 		set:       make(map[uintptr]*item),
 	}, nil
 }
 
-// Epoller implementation for handling OOM events from a container's cgroup
-type Epoller struct {
+// epoller implementation for handling OOM events from a container's cgroup
+type epoller struct {
 	mu sync.Mutex
 
 	fd        int
@@ -59,12 +61,12 @@ type item struct {
 }
 
 // Close the epoll fd
-func (e *Epoller) Close() error {
+func (e *epoller) Close() error {
 	return unix.Close(e.fd)
 }
 
 // Run the epoll loop
-func (e *Epoller) Run(ctx context.Context) {
+func (e *epoller) Run(ctx context.Context) {
 	var events [128]unix.EpollEvent
 	for {
 		select {
@@ -86,8 +88,12 @@ func (e *Epoller) Run(ctx context.Context) {
 	}
 }
 
-// Add the cgroup to the epoll monitor
-func (e *Epoller) Add(id string, cg cgroups.Cgroup) error {
+// Add cgroups.Cgroup to the epoll monitor
+func (e *epoller) Add(id string, cgx interface{}) error {
+	cg, ok := cgx.(cgroups.Cgroup)
+	if !ok {
+		return errors.Errorf("expected cgroups.Cgroup, got: %T", cgx)
+	}
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	fd, err := cg.OOMEventFD()
@@ -105,7 +111,7 @@ func (e *Epoller) Add(id string, cg cgroups.Cgroup) error {
 	return unix.EpollCtl(e.fd, unix.EPOLL_CTL_ADD, int(fd), &event)
 }
 
-func (e *Epoller) process(ctx context.Context, fd uintptr) {
+func (e *epoller) process(ctx context.Context, fd uintptr) {
 	flush(fd)
 	e.mu.Lock()
 	i, ok := e.set[fd]

pkg/oom/v2/v2.go

@@ -0,0 +1,113 @@
+// +build linux
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package v2
+
+import (
+	"context"
+
+	cgroupsv2 "github.com/containerd/cgroups/v2"
+	eventstypes "github.com/containerd/containerd/api/events"
+	"github.com/containerd/containerd/pkg/oom"
+	"github.com/containerd/containerd/runtime"
+	"github.com/containerd/containerd/runtime/v2/shim"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+// New returns an implementation that listens to OOM events
+// from a container's cgroups.
+func New(publisher shim.Publisher) (oom.Watcher, error) {
+	return &watcher{
+		itemCh:    make(chan item),
+		publisher: publisher,
+	}, nil
+}
+
+// watcher implementation for handling OOM events from a container's cgroup
+type watcher struct {
+	itemCh    chan item
+	publisher shim.Publisher
+}
+
+type item struct {
+	id  string
+	ev  cgroupsv2.Event
+	err error
+}
+
+// Close closes the watcher
+func (w *watcher) Close() error {
+	return nil
+}
+
+// Run the loop
+func (w *watcher) Run(ctx context.Context) {
+	lastOOMMap := make(map[string]uint64) // key: id, value: ev.OOM
+	for {
+		select {
+		case <-ctx.Done():
+			w.Close()
+			return
+		case i := <-w.itemCh:
+			if i.err != nil {
+				delete(lastOOMMap, i.id)
+				continue
+			}
+			lastOOM := lastOOMMap[i.id]
+			if i.ev.OOM > lastOOM {
+				if err := w.publisher.Publish(ctx, runtime.TaskOOMEventTopic, &eventstypes.TaskOOM{
+					ContainerID: i.id,
+				}); err != nil {
+					logrus.WithError(err).Error("publish OOM event")
+				}
+			}
+			if i.ev.OOM > 0 {
+				lastOOMMap[i.id] = i.ev.OOM
+			}
+		}
+	}
+}
+
+// Add cgroups.Cgroup to the epoll monitor
+func (w *watcher) Add(id string, cgx interface{}) error {
+	cg, ok := cgx.(*cgroupsv2.Manager)
+	if !ok {
+		return errors.Errorf("expected *cgroupsv2.Manager, got: %T", cgx)
+	}
+	// FIXME: cgroupsv2.Manager does not support closing eventCh routine currently.
+	// The routine shuts down when an error happens, mostly when the cgroup is deleted.
+	eventCh, errCh := cg.EventChan()
+	go func() {
+		for {
+			i := item{id: id}
+			select {
+			case ev := <-eventCh:
+				i.ev = ev
+				w.itemCh <- i
+			case err := <-errCh:
+				i.err = err
+				w.itemCh <- i
+				// we no longer get any event/err when we got an err
+				logrus.WithError(err).Warn("error from *cgroupsv2.Manager.EventChan")
+				return
+			}
+		}
+	}()
+	return nil
+}