1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
5 // Package crunchstat reports resource usage (CPU, memory, disk,
6 // network) for a cgroup.
23 // This magically allows us to look up userHz via _SC_CLK_TCK:
27 #include <sys/types.h>
33 // A Reporter gathers statistics for a cgroup and writes them to a
35 type Reporter struct {
36 // CID of the container to monitor. If empty, read the CID
37 // from CIDFile (first waiting until a non-empty file appears
38 // at CIDFile). If CIDFile is also empty, report host
42 // Path to a file we can read CID from.
45 // Where cgroup accounting files live on this system, e.g.,
49 // Parent cgroup, e.g., "docker".
52 // Interval between samples. Must be positive.
53 PollPeriod time.Duration
55 // Where to write statistics. Must not be nil.
58 reportedStatFile map[string]string
59 lastNetSample map[string]ioSample
60 lastDiskSample map[string]ioSample
61 lastCPUSample cpuSample
63 done chan struct{} // closed when we should stop reporting
64 flushed chan struct{} // closed when we have made our last report
67 // Start starts monitoring in a new goroutine, and returns
70 // The monitoring goroutine waits for a non-empty CIDFile to appear
71 // (unless CID is non-empty). Then it waits for the accounting files
72 // to appear for the monitored container. Then it collects and reports
73 // statistics until Stop is called.
75 // Callers should not call Start more than once.
77 // Callers should not modify public data fields after calling Start.
78 func (r *Reporter) Start() {
79 r.done = make(chan struct{})
80 r.flushed = make(chan struct{})
84 // Stop reporting. Do not call more than once, or before calling
87 // Nothing will be logged after Stop returns.
88 func (r *Reporter) Stop() {
93 func (r *Reporter) readAllOrWarn(in io.Reader) ([]byte, error) {
94 content, err := ioutil.ReadAll(in)
96 r.Logger.Printf("warning: %v", err)
101 // Open the cgroup stats file in /sys/fs corresponding to the target
102 // cgroup, and return an io.ReadCloser. If no stats file is available,
105 // Log the file that was opened, if it isn't the same file opened on
106 // the last openStatFile for this stat.
108 // Log "not available" if no file is found and either this stat has
109 // been available in the past, or verbose==true.
111 // TODO: Instead of trying all options, choose a process in the
112 // container, and read /proc/PID/cgroup to determine the appropriate
113 // cgroup root for the given statgroup. (This will avoid falling back
114 // to host-level stats during container setup and teardown.)
115 func (r *Reporter) openStatFile(statgroup, stat string, verbose bool) (io.ReadCloser, error) {
118 // Collect container's stats
120 fmt.Sprintf("%s/%s/%s/%s/%s", r.CgroupRoot, statgroup, r.CgroupParent, r.CID, stat),
121 fmt.Sprintf("%s/%s/%s/%s", r.CgroupRoot, r.CgroupParent, r.CID, stat),
124 // Collect this host's stats
126 fmt.Sprintf("%s/%s/%s", r.CgroupRoot, statgroup, stat),
127 fmt.Sprintf("%s/%s", r.CgroupRoot, stat),
133 for _, path = range paths {
134 file, err = os.Open(path)
141 if pathWas := r.reportedStatFile[stat]; pathWas != path {
142 // Log whenever we start using a new/different cgroup
143 // stat file for a given statistic. This typically
144 // happens 1 to 3 times per statistic, depending on
145 // whether we happen to collect stats [a] before any
146 // processes have been created in the container and
147 // [b] after all contained processes have exited.
148 if path == "" && verbose {
149 r.Logger.Printf("notice: stats not available: stat %s, statgroup %s, cid %s, parent %s, root %s\n", stat, statgroup, r.CID, r.CgroupParent, r.CgroupRoot)
150 } else if pathWas != "" {
151 r.Logger.Printf("notice: stats moved from %s to %s\n", r.reportedStatFile[stat], path)
153 r.Logger.Printf("notice: reading stats from %s\n", path)
155 r.reportedStatFile[stat] = path
160 func (r *Reporter) getContainerNetStats() (io.Reader, error) {
161 procsFile, err := r.openStatFile("cpuacct", "cgroup.procs", true)
165 defer procsFile.Close()
166 reader := bufio.NewScanner(procsFile)
168 taskPid := reader.Text()
169 statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid)
170 stats, err := ioutil.ReadFile(statsFilename)
172 r.Logger.Printf("notice: %v", err)
175 return strings.NewReader(string(stats)), nil
177 return nil, errors.New("Could not read stats for any proc in container")
180 type ioSample struct {
186 func (r *Reporter) doBlkIOStats() {
187 c, err := r.openStatFile("blkio", "blkio.io_service_bytes", true)
192 b := bufio.NewScanner(c)
193 var sampleTime = time.Now()
194 newSamples := make(map[string]ioSample)
196 var device, op string
198 if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &val); err != nil {
201 var thisSample ioSample
203 if thisSample, ok = newSamples[device]; !ok {
204 thisSample = ioSample{sampleTime, -1, -1}
208 thisSample.rxBytes = val
210 thisSample.txBytes = val
212 newSamples[device] = thisSample
214 for dev, sample := range newSamples {
215 if sample.txBytes < 0 || sample.rxBytes < 0 {
219 if prev, ok := r.lastDiskSample[dev]; ok {
220 delta = fmt.Sprintf(" -- interval %.4f seconds %d write %d read",
221 sample.sampleTime.Sub(prev.sampleTime).Seconds(),
222 sample.txBytes-prev.txBytes,
223 sample.rxBytes-prev.rxBytes)
225 r.Logger.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta)
226 r.lastDiskSample[dev] = sample
230 type memSample struct {
232 memStat map[string]int64
235 func (r *Reporter) doMemoryStats() {
236 c, err := r.openStatFile("memory", "memory.stat", true)
241 b := bufio.NewScanner(c)
242 thisSample := memSample{time.Now(), make(map[string]int64)}
243 wantStats := [...]string{"cache", "swap", "pgmajfault", "rss"}
247 if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil {
250 thisSample.memStat[stat] = val
252 var outstat bytes.Buffer
253 for _, key := range wantStats {
254 if val, ok := thisSample.memStat[key]; ok {
255 outstat.WriteString(fmt.Sprintf(" %d %s", val, key))
258 r.Logger.Printf("mem%s\n", outstat.String())
261 func (r *Reporter) doNetworkStats() {
262 sampleTime := time.Now()
263 stats, err := r.getContainerNetStats()
268 scanner := bufio.NewScanner(stats)
272 words := strings.Fields(scanner.Text())
273 if len(words) != 17 {
274 // Skip lines with wrong format
277 ifName = strings.TrimRight(words[0], ":")
278 if ifName == "lo" || ifName == "" {
279 // Skip loopback interface and lines with wrong format
282 if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil {
285 if rx, err = strconv.ParseInt(words[1], 10, 64); err != nil {
288 nextSample := ioSample{}
289 nextSample.sampleTime = sampleTime
290 nextSample.txBytes = tx
291 nextSample.rxBytes = rx
293 if prev, ok := r.lastNetSample[ifName]; ok {
294 interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
295 delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx",
300 r.Logger.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta)
301 r.lastNetSample[ifName] = nextSample
305 type cpuSample struct {
306 hasData bool // to distinguish the zero value from real data
313 // Return the number of CPUs available in the container. Return 0 if
314 // we can't figure out the real number of CPUs.
315 func (r *Reporter) getCPUCount() int64 {
316 cpusetFile, err := r.openStatFile("cpuset", "cpuset.cpus", true)
320 defer cpusetFile.Close()
321 b, err := r.readAllOrWarn(cpusetFile)
325 sp := strings.Split(string(b), ",")
327 for _, v := range sp {
329 n, _ := fmt.Sscanf(v, "%d-%d", &min, &max)
331 cpus += (max - min) + 1
339 func (r *Reporter) doCPUStats() {
340 statFile, err := r.openStatFile("cpuacct", "cpuacct.stat", true)
344 defer statFile.Close()
345 b, err := r.readAllOrWarn(statFile)
350 var userTicks, sysTicks int64
351 fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks)
352 userHz := float64(C.sysconf(C._SC_CLK_TCK))
353 nextSample := cpuSample{
355 sampleTime: time.Now(),
356 user: float64(userTicks) / userHz,
357 sys: float64(sysTicks) / userHz,
358 cpus: r.getCPUCount(),
362 if r.lastCPUSample.hasData {
363 delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys",
364 nextSample.sampleTime.Sub(r.lastCPUSample.sampleTime).Seconds(),
365 nextSample.user-r.lastCPUSample.user,
366 nextSample.sys-r.lastCPUSample.sys)
368 r.Logger.Printf("cpu %.4f user %.4f sys %d cpus%s\n",
369 nextSample.user, nextSample.sys, nextSample.cpus, delta)
370 r.lastCPUSample = nextSample
373 // Report stats periodically until we learn (via r.done) that someone
375 func (r *Reporter) run() {
376 defer close(r.flushed)
378 r.reportedStatFile = make(map[string]string)
380 if !r.waitForCIDFile() || !r.waitForCgroup() {
384 r.lastNetSample = make(map[string]ioSample)
385 r.lastDiskSample = make(map[string]ioSample)
387 ticker := time.NewTicker(r.PollPeriod)
401 // If CID is empty, wait for it to appear in CIDFile. Return true if
402 // we get it before we learn (via r.done) that someone called Stop.
403 func (r *Reporter) waitForCIDFile() bool {
404 if r.CID != "" || r.CIDFile == "" {
408 ticker := time.NewTicker(100 * time.Millisecond)
411 cid, err := ioutil.ReadFile(r.CIDFile)
412 if err == nil && len(cid) > 0 {
419 r.Logger.Printf("warning: CID never appeared in %+q: %v", r.CIDFile, err)
425 // Wait for the cgroup stats files to appear in cgroup_root. Return
426 // true if they appear before r.done indicates someone called Stop. If
427 // they don't appear within one poll interval, log a warning and keep
429 func (r *Reporter) waitForCgroup() bool {
430 ticker := time.NewTicker(100 * time.Millisecond)
432 warningTimer := time.After(r.PollPeriod)
434 c, err := r.openStatFile("cpuacct", "cgroup.procs", false)
442 r.Logger.Printf("warning: cgroup stats files have not appeared after %v (config error?) -- still waiting...", r.PollPeriod)
444 r.Logger.Printf("warning: cgroup stats files never appeared for %v", r.CID)