forked from containerd/containerd
/
v2.go
113 lines (100 loc) · 2.77 KB
/
v2.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
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
}