+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: AGPL-3.0
+
// Package crunchstat reports resource usage (CPU, memory, disk,
// network) for a cgroup.
package crunchstat
"os"
"strconv"
"strings"
+ "syscall"
"time"
)
-// This magically allows us to look up userHz via _SC_CLK_TCK:
-
-/*
-#include <unistd.h>
-#include <sys/types.h>
-#include <pwd.h>
-#include <stdlib.h>
-*/
-import "C"
-
// A Reporter gathers statistics for a cgroup and writes them to a
// log.Logger.
type Reporter struct {
// Where cgroup accounting files live on this system, e.g.,
// "/sys/fs/cgroup".
- CgroupRoot string
+ CgroupRoot string
// Parent cgroup, e.g., "docker".
CgroupParent string
// Interval between samples. Must be positive.
- Poll time.Duration
+ PollPeriod time.Duration
+
+ // Temporary directory, will be monitored for available, used & total space.
+ TempDir string
// Where to write statistics. Must not be nil.
Logger *log.Logger
- reportedStatFile map[string]string
- lastNetSample map[string]ioSample
- lastDiskSample map[string]ioSample
- lastCPUSample cpuSample
+ reportedStatFile map[string]string
+ lastNetSample map[string]ioSample
+ lastDiskIOSample map[string]ioSample
+ lastCPUSample cpuSample
+ lastDiskSpaceSample diskSpaceSample
- done chan struct{}
+ done chan struct{} // closed when we should stop reporting
+ flushed chan struct{} // closed when we have made our last report
}
// Start starts monitoring in a new goroutine, and returns
// Callers should not modify public data fields after calling Start.
func (r *Reporter) Start() {
r.done = make(chan struct{})
+ r.flushed = make(chan struct{})
go r.run()
}
// Nothing will be logged after Stop returns.
func (r *Reporter) Stop() {
close(r.done)
+ <-r.flushed
}
func (r *Reporter) readAllOrWarn(in io.Reader) ([]byte, error) {
content, err := ioutil.ReadAll(in)
if err != nil {
- r.Logger.Print(err)
+ r.Logger.Printf("warning: %v", err)
}
return content, err
}
statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid)
stats, err := ioutil.ReadFile(statsFilename)
if err != nil {
- r.Logger.Print(err)
+ r.Logger.Printf("notice: %v", err)
continue
}
return strings.NewReader(string(stats)), nil
continue
}
delta := ""
- if prev, ok := r.lastDiskSample[dev]; ok {
+ if prev, ok := r.lastDiskIOSample[dev]; ok {
delta = fmt.Sprintf(" -- interval %.4f seconds %d write %d read",
sample.sampleTime.Sub(prev.sampleTime).Seconds(),
sample.txBytes-prev.txBytes,
sample.rxBytes-prev.rxBytes)
}
r.Logger.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta)
- r.lastDiskSample[dev] = sample
+ r.lastDiskIOSample[dev] = sample
}
}
}
var outstat bytes.Buffer
for _, key := range wantStats {
- if val, ok := thisSample.memStat[key]; ok {
- outstat.WriteString(fmt.Sprintf(" %d %s", val, key))
+ // Use "total_X" stats (entire hierarchy) if enabled,
+ // otherwise just the single cgroup -- see
+ // https://www.kernel.org/doc/Documentation/cgroup-v1/memory.txt
+ if val, ok := thisSample.memStat["total_"+key]; ok {
+ fmt.Fprintf(&outstat, " %d %s", val, key)
+ } else if val, ok := thisSample.memStat[key]; ok {
+ fmt.Fprintf(&outstat, " %d %s", val, key)
}
}
r.Logger.Printf("mem%s\n", outstat.String())
}
}
+type diskSpaceSample struct {
+ hasData bool
+ sampleTime time.Time
+ total uint64
+ used uint64
+ available uint64
+}
+
+func (r *Reporter) doDiskSpaceStats() {
+ s := syscall.Statfs_t{}
+ err := syscall.Statfs(r.TempDir, &s)
+ if err != nil {
+ return
+ }
+ bs := uint64(s.Bsize)
+ nextSample := diskSpaceSample{
+ hasData: true,
+ sampleTime: time.Now(),
+ total: s.Blocks * bs,
+ used: (s.Blocks - s.Bfree) * bs,
+ available: s.Bavail * bs,
+ }
+
+ var delta string
+ if r.lastDiskSpaceSample.hasData {
+ prev := r.lastDiskSpaceSample
+ interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
+ delta = fmt.Sprintf(" -- interval %.4f seconds %d used",
+ interval,
+ int64(nextSample.used-prev.used))
+ }
+ r.Logger.Printf("statfs %d available %d used %d total%s\n",
+ nextSample.available, nextSample.used, nextSample.total, delta)
+ r.lastDiskSpaceSample = nextSample
+}
+
type cpuSample struct {
hasData bool // to distinguish the zero value from real data
sampleTime time.Time
}
defer cpusetFile.Close()
b, err := r.readAllOrWarn(cpusetFile)
+ if err != nil {
+ return 0
+ }
sp := strings.Split(string(b), ",")
cpus := int64(0)
for _, v := range sp {
return
}
- nextSample := cpuSample{true, time.Now(), 0, 0, r.getCPUCount()}
var userTicks, sysTicks int64
fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks)
- userHz := float64(C.sysconf(C._SC_CLK_TCK))
- nextSample.user = float64(userTicks) / userHz
- nextSample.sys = float64(sysTicks) / userHz
+ userHz := float64(100)
+ nextSample := cpuSample{
+ hasData: true,
+ sampleTime: time.Now(),
+ user: float64(userTicks) / userHz,
+ sys: float64(sysTicks) / userHz,
+ cpus: r.getCPUCount(),
+ }
delta := ""
if r.lastCPUSample.hasData {
r.lastCPUSample = nextSample
}
-// Report stats periodically until r.done indicates someone called
-// Stop.
+// Report stats periodically until we learn (via r.done) that someone
+// called Stop.
func (r *Reporter) run() {
+ defer close(r.flushed)
+
r.reportedStatFile = make(map[string]string)
if !r.waitForCIDFile() || !r.waitForCgroup() {
}
r.lastNetSample = make(map[string]ioSample)
- r.lastDiskSample = make(map[string]ioSample)
+ r.lastDiskIOSample = make(map[string]ioSample)
+
+ if len(r.TempDir) == 0 {
+ // Temporary dir not provided, try to get it from the environment.
+ r.TempDir = os.Getenv("TMPDIR")
+ }
+ if len(r.TempDir) > 0 {
+ r.Logger.Printf("notice: monitoring temp dir %s\n", r.TempDir)
+ }
- ticker := time.NewTicker(r.Poll)
+ ticker := time.NewTicker(r.PollPeriod)
for {
r.doMemoryStats()
r.doCPUStats()
r.doBlkIOStats()
r.doNetworkStats()
+ r.doDiskSpaceStats()
select {
case <-r.done:
return
}
// If CID is empty, wait for it to appear in CIDFile. Return true if
-// we get it before r.done indicates someone called Stop.
+// we get it before we learn (via r.done) that someone called Stop.
func (r *Reporter) waitForCIDFile() bool {
if r.CID != "" || r.CIDFile == "" {
return true
select {
case <-ticker.C:
case <-r.done:
- r.Logger.Printf("CID never appeared in %+q: %v", r.CIDFile, err)
+ r.Logger.Printf("warning: CID never appeared in %+q: %v", r.CIDFile, err)
return false
}
}
func (r *Reporter) waitForCgroup() bool {
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
- warningTimer := time.After(r.Poll)
+ warningTimer := time.After(r.PollPeriod)
for {
c, err := r.openStatFile("cpuacct", "cgroup.procs", false)
if err == nil {
select {
case <-ticker.C:
case <-warningTimer:
- r.Logger.Printf("cgroup stats files have not appeared after %v (config error?) -- still waiting...", r.Poll)
+ r.Logger.Printf("warning: cgroup stats files have not appeared after %v (config error?) -- still waiting...", r.PollPeriod)
case <-r.done:
- r.Logger.Printf("cgroup stats files never appeared for %v", r.CID)
+ r.Logger.Printf("warning: cgroup stats files never appeared for %v", r.CID)
return false
}
}
projects / arvados.git / blobdiff