github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Allow blocking and non-blocking metrics collection

Browse Source 
Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby 2017-10-06 12:04:05 -04:00
parent ed519bb5ce
commit d92f6eea1f
3 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
metrics/cgroups/cgroups.go
View File

@ -71,7 +71,7 @@ func (m *cgroupsMonitor) Monitor(c runtime.Task) error {
func (m *cgroupsMonitor) Stop(c runtime.Task) error {
info := c.Info()
t := c.(*linux.Task)
m.collector.collect(info.ID, info.Namespace, t.Cgroup(), m.collector.storedMetrics, nil)
m.collector.collect(info.ID, info.Namespace, t.Cgroup(), m.collector.storedMetrics, false, nil)
m.collector.Remove(info.ID, info.Namespace)
return nil
}

9
metrics/cgroups/metric.go
View File

@ -28,13 +28,18 @@ func (m *metric) desc(ns *metrics.Namespace) *prometheus.Desc {
return ns.NewDesc(m.name, m.help, m.unit, append([]string{"container_id", "namespace"}, m.labels...)...)
}
func (m *metric) collect(id, namespace string, stats *cgroups.Metrics, ns *metrics.Namespace, ch chan<- prometheus.Metric) {
func (m *metric) collect(id, namespace string, stats *cgroups.Metrics, ns *metrics.Namespace, ch chan<- prometheus.Metric, block bool) {
values := m.getValues(stats)
for _, v := range values {
// block signals to block on the sending the metrics so none are missed
if block {
ch <- prometheus.MustNewConstMetric(m.desc(ns), m.vt, v.v, append([]string{id, namespace}, v.l...)...)
continue
}
// non-blocking metrics can be dropped if the chan is full
select {
case ch <- prometheus.MustNewConstMetric(m.desc(ns), m.vt, v.v, append([]string{id, namespace}, v.l...)...):
default:
break
}
}
}

15
metrics/cgroups/metrics.go
View File

@ -77,22 +77,23 @@ func (c *collector) Collect(ch chan<- prometheus.Metric) {
wg := &sync.WaitGroup{}
for _, t := range c.cgroups {
wg.Add(1)
go c.collect(t.id, t.namespace, t.cgroup, ch, wg)
go c.collect(t.id, t.namespace, t.cgroup, ch, true, wg)
}
SelectLoop:
storedLoop:
for {
// read stored metrics until the channel is flushed
select {
case value := <-c.storedMetrics:
ch <- value
case m := <-c.storedMetrics:
ch <- m
default:
break SelectLoop
break storedLoop
}
}
c.mu.RUnlock()
wg.Wait()
}
func (c *collector) collect(id, namespace string, cg cgroups.Cgroup, ch chan<- prometheus.Metric, wg *sync.WaitGroup) {
func (c *collector) collect(id, namespace string, cg cgroups.Cgroup, ch chan<- prometheus.Metric, block bool, wg *sync.WaitGroup) {
if wg != nil {
defer wg.Done()
}
@ -102,7 +103,7 @@ func (c *collector) collect(id, namespace string, cg cgroups.Cgroup, ch chan<- p
return
}
for _, m := range c.metrics {
m.collect(id, namespace, stats, c.ns, ch)
m.collect(id, namespace, stats, c.ns, ch, block)
}
}