Revert "Wait for client side copy goroutines to start"

This reverts commit 06dc87ae59.

Revert "Change oom metric to const"

This reverts commit e800f08f9f.

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby 2017-08-18 16:09:18 -04:00
parent e3696cfb4e
commit 4950c26757
5 changed files with 11 additions and 47 deletions

View File

@ -36,7 +36,6 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
var ( var (
f io.ReadWriteCloser f io.ReadWriteCloser
set []io.Closer set []io.Closer
cwg sync.WaitGroup
ctx, cancel = context.WithCancel(context.Background()) ctx, cancel = context.WithCancel(context.Background())
wg = &sync.WaitGroup{} wg = &sync.WaitGroup{}
) )
@ -53,9 +52,7 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
return nil, err return nil, err
} }
set = append(set, f) set = append(set, f)
cwg.Add(1)
go func(w io.WriteCloser) { go func(w io.WriteCloser) {
cwg.Done()
io.Copy(w, ioset.in) io.Copy(w, ioset.in)
w.Close() w.Close()
}(f) }(f)
@ -65,9 +62,7 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
} }
set = append(set, f) set = append(set, f)
wg.Add(1) wg.Add(1)
cwg.Add(1)
go func(r io.ReadCloser) { go func(r io.ReadCloser) {
cwg.Done()
io.Copy(ioset.out, r) io.Copy(ioset.out, r)
r.Close() r.Close()
wg.Done() wg.Done()
@ -80,15 +75,12 @@ func copyIO(fifos *FIFOSet, ioset *ioSet, tty bool) (_ *wgCloser, err error) {
if !tty { if !tty {
wg.Add(1) wg.Add(1)
cwg.Add(1)
go func(r io.ReadCloser) { go func(r io.ReadCloser) {
cwg.Done()
io.Copy(ioset.err, r) io.Copy(ioset.err, r)
r.Close() r.Close()
wg.Done() wg.Done()
}(f) }(f)
} }
cwg.Wait()
return &wgCloser{ return &wgCloser{
wg: wg, wg: wg,
dir: fifos.Dir, dir: fifos.Dir,

View File

@ -49,7 +49,6 @@ func WithStart(binary, address string, debug bool) ClientOpt {
if err != nil { if err != nil {
terminate(cmd) terminate(cmd)
} }
reaper.Default.Delete(cmd.Process.Pid)
}() }()
log.G(ctx).WithFields(logrus.Fields{ log.G(ctx).WithFields(logrus.Fields{
"pid": cmd.Process.Pid, "pid": cmd.Process.Pid,

View File

@ -65,7 +65,6 @@ func (m *cgroupsMonitor) Monitor(c runtime.Task) error {
func (m *cgroupsMonitor) Stop(c runtime.Task) error { func (m *cgroupsMonitor) Stop(c runtime.Task) error {
info := c.Info() info := c.Info()
m.collector.Remove(info.ID, info.Namespace) m.collector.Remove(info.ID, info.Namespace)
m.oom.Remove(info.ID, info.Namespace)
return nil return nil
} }

View File

@ -9,7 +9,6 @@ import (
"github.com/containerd/cgroups" "github.com/containerd/cgroups"
metrics "github.com/docker/go-metrics" metrics "github.com/docker/go-metrics"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
@ -19,21 +18,20 @@ func NewOOMCollector(ns *metrics.Namespace) (*OOMCollector, error) {
return nil, err return nil, err
} }
c := &OOMCollector{ c := &OOMCollector{
fd: fd, fd: fd,
set: make(map[uintptr]*oom), memoryOOM: ns.NewLabeledGauge("memory_oom", "The number of times a container received an oom event", metrics.Total, "container_id", "namespace"),
desc: ns.NewDesc("memory_oom", "The number of times a container has received an oom event", metrics.Total, "container_id", "namespace"), set: make(map[uintptr]*oom),
} }
go c.start() go c.start()
ns.Add(c)
return c, nil return c, nil
} }
type OOMCollector struct { type OOMCollector struct {
mu sync.Mutex mu sync.Mutex
fd int memoryOOM metrics.LabeledGauge
set map[uintptr]*oom fd int
desc *prometheus.Desc set map[uintptr]*oom
} }
type oom struct { type oom struct {
@ -41,7 +39,6 @@ type oom struct {
namespace string namespace string
c cgroups.Cgroup c cgroups.Cgroup
triggers []Trigger triggers []Trigger
count int
} }
func (o *OOMCollector) Add(id, namespace string, cg cgroups.Cgroup, triggers ...Trigger) error { func (o *OOMCollector) Add(id, namespace string, cg cgroups.Cgroup, triggers ...Trigger) error {
@ -53,10 +50,12 @@ func (o *OOMCollector) Add(id, namespace string, cg cgroups.Cgroup, triggers ...
} }
o.set[fd] = &oom{ o.set[fd] = &oom{
id: id, id: id,
namespace: namespace,
c: cg, c: cg,
triggers: triggers, triggers: triggers,
namespace: namespace,
} }
// set the gauge's default value
o.memoryOOM.WithValues(id, namespace).Set(0)
event := unix.EpollEvent{ event := unix.EpollEvent{
Fd: int32(fd), Fd: int32(fd),
Events: unix.EPOLLHUP | unix.EPOLLIN | unix.EPOLLERR, Events: unix.EPOLLHUP | unix.EPOLLIN | unix.EPOLLERR,
@ -67,37 +66,11 @@ func (o *OOMCollector) Add(id, namespace string, cg cgroups.Cgroup, triggers ...
return nil return nil
} }
func (o *OOMCollector) Remove(id, namespace string) {
o.mu.Lock()
defer o.mu.Unlock()
for fd, t := range o.set {
if t.id == id && t.namespace == namespace {
unix.Close(int(fd))
delete(o.set, fd)
return
}
}
}
// Close closes the epoll fd // Close closes the epoll fd
func (o *OOMCollector) Close() error { func (o *OOMCollector) Close() error {
return unix.Close(int(o.fd)) return unix.Close(int(o.fd))
} }
func (o *OOMCollector) Describe(ch chan<- *prometheus.Desc) {
o.mu.Lock()
defer o.mu.Unlock()
ch <- o.desc
}
func (o *OOMCollector) Collect(ch chan<- prometheus.Metric) {
o.mu.Lock()
defer o.mu.Unlock()
for _, t := range o.set {
ch <- prometheus.MustNewConstMetric(o.desc, prometheus.GaugeValue, float64(t.count), t.id, t.namespace)
}
}
func (o *OOMCollector) start() { func (o *OOMCollector) start() {
var events [128]unix.EpollEvent var events [128]unix.EpollEvent
for { for {
@ -134,7 +107,7 @@ func (o *OOMCollector) process(fd uintptr, event uint32) {
unix.Close(int(fd)) unix.Close(int(fd))
return return
} }
info.count++ o.memoryOOM.WithValues(info.id, info.namespace).Inc(1)
for _, t := range info.triggers { for _, t := range info.triggers {
t(info.id, info.c) t(info.id, info.c)
} }

View File

@ -9,6 +9,7 @@ import (
type TaskInfo struct { type TaskInfo struct {
ID string ID string
Runtime string Runtime string
Spec []byte
Namespace string Namespace string
} }