Merge branch '4956-limit-request-size' refs #4956
[arvados.git] / services / crunchstat / crunchstat.go
1 package main
2
3 import (
4         "bufio"
5         "bytes"
6         "errors"
7         "flag"
8         "fmt"
9         "io"
10         "io/ioutil"
11         "log"
12         "os"
13         "os/exec"
14         "os/signal"
15         "strconv"
16         "strings"
17         "syscall"
18         "time"
19 )
20
21 /*
22 #include <unistd.h>
23 #include <sys/types.h>
24 #include <pwd.h>
25 #include <stdlib.h>
26 */
27 import "C"
28
29 // The above block of magic allows us to look up user_hz via _SC_CLK_TCK.
30
31 type Cgroup struct {
32         root   string
33         parent string
34         cid    string
35 }
36
37 var childLog = log.New(os.Stderr, "", 0)
38 var statLog = log.New(os.Stderr, "crunchstat: ", 0)
39
40 const (
41         MaxLogLine = 1 << 14 // Child stderr lines >16KiB will be split
42 )
43
44 func CopyPipeToChildLog(in io.ReadCloser, done chan<- bool) {
45         reader := bufio.NewReaderSize(in, MaxLogLine)
46         var prefix string
47         for {
48                 line, isPrefix, err := reader.ReadLine()
49                 if err == io.EOF {
50                         break
51                 } else if err != nil {
52                         statLog.Fatal("error reading child stderr:", err)
53                 }
54                 var suffix string
55                 if isPrefix {
56                         suffix = "[...]"
57                 }
58                 childLog.Print(prefix, string(line), suffix)
59                 // Set up prefix for following line
60                 if isPrefix {
61                         prefix = "[...]"
62                 } else {
63                         prefix = ""
64                 }
65         }
66         done <- true
67         in.Close()
68 }
69
70 func ReadAllOrWarn(in *os.File) ([]byte, error) {
71         content, err := ioutil.ReadAll(in)
72         if err != nil {
73                 statLog.Printf("error reading %s: %s\n", in.Name(), err)
74         }
75         return content, err
76 }
77
78 var reportedStatFile = map[string]string{}
79
80 // Open the cgroup stats file in /sys/fs corresponding to the target
81 // cgroup, and return an *os.File. If no stats file is available,
82 // return nil.
83 //
84 // TODO: Instead of trying all options, choose a process in the
85 // container, and read /proc/PID/cgroup to determine the appropriate
86 // cgroup root for the given statgroup. (This will avoid falling back
87 // to host-level stats during container setup and teardown.)
88 func OpenStatFile(cgroup Cgroup, statgroup string, stat string) (*os.File, error) {
89         var paths = []string{
90                 fmt.Sprintf("%s/%s/%s/%s/%s", cgroup.root, statgroup, cgroup.parent, cgroup.cid, stat),
91                 fmt.Sprintf("%s/%s/%s/%s", cgroup.root, cgroup.parent, cgroup.cid, stat),
92                 fmt.Sprintf("%s/%s/%s", cgroup.root, statgroup, stat),
93                 fmt.Sprintf("%s/%s", cgroup.root, stat),
94         }
95         var path string
96         var file *os.File
97         var err error
98         for _, path = range paths {
99                 file, err = os.Open(path)
100                 if err == nil {
101                         break
102                 } else {
103                         path = ""
104                 }
105         }
106         if pathWas, ok := reportedStatFile[stat]; !ok || pathWas != path {
107                 // Log whenever we start using a new/different cgroup
108                 // stat file for a given statistic. This typically
109                 // happens 1 to 3 times per statistic, depending on
110                 // whether we happen to collect stats [a] before any
111                 // processes have been created in the container and
112                 // [b] after all contained processes have exited.
113                 reportedStatFile[stat] = path
114                 if path == "" {
115                         statLog.Printf("error finding stats file: stat %s, statgroup %s, cid %s, parent %s, root %s\n", stat, statgroup, cgroup.cid, cgroup.parent, cgroup.root)
116                 } else {
117                         statLog.Printf("error reading stats from %s\n", path)
118                 }
119         }
120         return file, err
121 }
122
123 func GetContainerNetStats(cgroup Cgroup) (io.Reader, error) {
124         procsFile, err := OpenStatFile(cgroup, "cpuacct", "cgroup.procs")
125         if err != nil {
126                 return nil, err
127         }
128         defer procsFile.Close()
129         reader := bufio.NewScanner(procsFile)
130         for reader.Scan() {
131                 taskPid := reader.Text()
132                 statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid)
133                 stats, err := ioutil.ReadFile(statsFilename)
134                 if err != nil {
135                         statLog.Printf("read %s: %s\n", statsFilename, err)
136                         continue
137                 }
138                 return strings.NewReader(string(stats)), nil
139         }
140         return nil, errors.New("Could not read stats for any proc in container")
141 }
142
143 type IoSample struct {
144         sampleTime time.Time
145         txBytes    int64
146         rxBytes    int64
147 }
148
149 func DoBlkIoStats(cgroup Cgroup, lastSample map[string]IoSample) {
150         c, err := OpenStatFile(cgroup, "blkio", "blkio.io_service_bytes")
151         if err != nil {
152                 return
153         }
154         defer c.Close()
155         b := bufio.NewScanner(c)
156         var sampleTime = time.Now()
157         newSamples := make(map[string]IoSample)
158         for b.Scan() {
159                 var device, op string
160                 var val int64
161                 if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &val); err != nil {
162                         continue
163                 }
164                 var thisSample IoSample
165                 var ok bool
166                 if thisSample, ok = newSamples[device]; !ok {
167                         thisSample = IoSample{sampleTime, -1, -1}
168                 }
169                 switch op {
170                 case "Read":
171                         thisSample.rxBytes = val
172                 case "Write":
173                         thisSample.txBytes = val
174                 }
175                 newSamples[device] = thisSample
176         }
177         for dev, sample := range newSamples {
178                 if sample.txBytes < 0 || sample.rxBytes < 0 {
179                         continue
180                 }
181                 delta := ""
182                 if prev, ok := lastSample[dev]; ok {
183                         delta = fmt.Sprintf(" -- interval %.4f seconds %d write %d read",
184                                 sample.sampleTime.Sub(prev.sampleTime).Seconds(),
185                                 sample.txBytes-prev.txBytes,
186                                 sample.rxBytes-prev.rxBytes)
187                 }
188                 statLog.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta)
189                 lastSample[dev] = sample
190         }
191 }
192
193 type MemSample struct {
194         sampleTime time.Time
195         memStat    map[string]int64
196 }
197
198 func DoMemoryStats(cgroup Cgroup) {
199         c, err := OpenStatFile(cgroup, "memory", "memory.stat")
200         if err != nil {
201                 return
202         }
203         defer c.Close()
204         b := bufio.NewScanner(c)
205         thisSample := MemSample{time.Now(), make(map[string]int64)}
206         wantStats := [...]string{"cache", "swap", "pgmajfault", "rss"}
207         for b.Scan() {
208                 var stat string
209                 var val int64
210                 if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil {
211                         continue
212                 }
213                 thisSample.memStat[stat] = val
214         }
215         var outstat bytes.Buffer
216         for _, key := range wantStats {
217                 if val, ok := thisSample.memStat[key]; ok {
218                         outstat.WriteString(fmt.Sprintf(" %d %s", val, key))
219                 }
220         }
221         statLog.Printf("mem%s\n", outstat.String())
222 }
223
224 func DoNetworkStats(cgroup Cgroup, lastSample map[string]IoSample) {
225         sampleTime := time.Now()
226         stats, err := GetContainerNetStats(cgroup)
227         if err != nil {
228                 return
229         }
230
231         scanner := bufio.NewScanner(stats)
232         for scanner.Scan() {
233                 var ifName string
234                 var rx, tx int64
235                 words := strings.Fields(scanner.Text())
236                 if len(words) != 17 {
237                         // Skip lines with wrong format
238                         continue
239                 }
240                 ifName = strings.TrimRight(words[0], ":")
241                 if ifName == "lo" || ifName == "" {
242                         // Skip loopback interface and lines with wrong format
243                         continue
244                 }
245                 if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil {
246                         continue
247                 }
248                 if rx, err = strconv.ParseInt(words[1], 10, 64); err != nil {
249                         continue
250                 }
251                 nextSample := IoSample{}
252                 nextSample.sampleTime = sampleTime
253                 nextSample.txBytes = tx
254                 nextSample.rxBytes = rx
255                 var delta string
256                 if prev, ok := lastSample[ifName]; ok {
257                         interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
258                         delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx",
259                                 interval,
260                                 tx-prev.txBytes,
261                                 rx-prev.rxBytes)
262                 }
263                 statLog.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta)
264                 lastSample[ifName] = nextSample
265         }
266 }
267
268 type CpuSample struct {
269         hasData    bool // to distinguish the zero value from real data
270         sampleTime time.Time
271         user       float64
272         sys        float64
273         cpus       int64
274 }
275
276 // Return the number of CPUs available in the container. Return 0 if
277 // we can't figure out the real number of CPUs.
278 func GetCpuCount(cgroup Cgroup) int64 {
279         cpusetFile, err := OpenStatFile(cgroup, "cpuset", "cpuset.cpus")
280         if err != nil {
281                 return 0
282         }
283         defer cpusetFile.Close()
284         b, err := ReadAllOrWarn(cpusetFile)
285         sp := strings.Split(string(b), ",")
286         cpus := int64(0)
287         for _, v := range sp {
288                 var min, max int64
289                 n, _ := fmt.Sscanf(v, "%d-%d", &min, &max)
290                 if n == 2 {
291                         cpus += (max - min) + 1
292                 } else {
293                         cpus += 1
294                 }
295         }
296         return cpus
297 }
298
299 func DoCpuStats(cgroup Cgroup, lastSample *CpuSample) {
300         statFile, err := OpenStatFile(cgroup, "cpuacct", "cpuacct.stat")
301         if err != nil {
302                 return
303         }
304         defer statFile.Close()
305         b, err := ReadAllOrWarn(statFile)
306         if err != nil {
307                 return
308         }
309
310         nextSample := CpuSample{true, time.Now(), 0, 0, GetCpuCount(cgroup)}
311         var userTicks, sysTicks int64
312         fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks)
313         user_hz := float64(C.sysconf(C._SC_CLK_TCK))
314         nextSample.user = float64(userTicks) / user_hz
315         nextSample.sys = float64(sysTicks) / user_hz
316
317         delta := ""
318         if lastSample.hasData {
319                 delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys",
320                         nextSample.sampleTime.Sub(lastSample.sampleTime).Seconds(),
321                         nextSample.user-lastSample.user,
322                         nextSample.sys-lastSample.sys)
323         }
324         statLog.Printf("cpu %.4f user %.4f sys %d cpus%s\n",
325                 nextSample.user, nextSample.sys, nextSample.cpus, delta)
326         *lastSample = nextSample
327 }
328
329 func PollCgroupStats(cgroup Cgroup, poll int64, stop_poll_chan <-chan bool) {
330         var lastNetSample = map[string]IoSample{}
331         var lastDiskSample = map[string]IoSample{}
332         var lastCpuSample = CpuSample{}
333
334         poll_chan := make(chan bool, 1)
335         go func() {
336                 // Send periodic poll events.
337                 poll_chan <- true
338                 for {
339                         time.Sleep(time.Duration(poll) * time.Millisecond)
340                         poll_chan <- true
341                 }
342         }()
343         for {
344                 select {
345                 case <-stop_poll_chan:
346                         return
347                 case <-poll_chan:
348                         // Emit stats, then select again.
349                 }
350                 DoMemoryStats(cgroup)
351                 DoCpuStats(cgroup, &lastCpuSample)
352                 DoBlkIoStats(cgroup, lastDiskSample)
353                 DoNetworkStats(cgroup, lastNetSample)
354         }
355 }
356
357 func run(logger *log.Logger) error {
358
359         var (
360                 cgroup_root    string
361                 cgroup_parent  string
362                 cgroup_cidfile string
363                 wait           int64
364                 poll           int64
365         )
366
367         flag.StringVar(&cgroup_root, "cgroup-root", "", "Root of cgroup tree")
368         flag.StringVar(&cgroup_parent, "cgroup-parent", "", "Name of container parent under cgroup")
369         flag.StringVar(&cgroup_cidfile, "cgroup-cid", "", "Path to container id file")
370         flag.Int64Var(&wait, "wait", 5, "Maximum time (in seconds) to wait for cid file to show up")
371         flag.Int64Var(&poll, "poll", 1000, "Polling frequency, in milliseconds")
372
373         flag.Parse()
374
375         if cgroup_root == "" {
376                 statLog.Fatal("error: must provide -cgroup-root")
377         }
378
379         finish_chan := make(chan bool)
380         defer close(finish_chan)
381
382         var cmd *exec.Cmd
383
384         if len(flag.Args()) > 0 {
385                 // Set up subprocess
386                 cmd = exec.Command(flag.Args()[0], flag.Args()[1:]...)
387
388                 childLog.Println("Running", flag.Args())
389
390                 // Child process will use our stdin and stdout pipes
391                 // (we close our copies below)
392                 cmd.Stdin = os.Stdin
393                 cmd.Stdout = os.Stdout
394
395                 // Forward SIGINT and SIGTERM to inner process
396                 sigChan := make(chan os.Signal, 1)
397                 go func(sig <-chan os.Signal) {
398                         catch := <-sig
399                         if cmd.Process != nil {
400                                 cmd.Process.Signal(catch)
401                         }
402                         statLog.Println("caught signal:", catch)
403                 }(sigChan)
404                 signal.Notify(sigChan, syscall.SIGTERM)
405                 signal.Notify(sigChan, syscall.SIGINT)
406
407                 // Funnel stderr through our channel
408                 stderr_pipe, err := cmd.StderrPipe()
409                 if err != nil {
410                         statLog.Fatalln("error in StderrPipe:", err)
411                 }
412                 go CopyPipeToChildLog(stderr_pipe, finish_chan)
413
414                 // Run subprocess
415                 if err := cmd.Start(); err != nil {
416                         statLog.Fatalln("error in cmd.Start:", err)
417                 }
418
419                 // Close stdin/stdout in this (parent) process
420                 os.Stdin.Close()
421                 os.Stdout.Close()
422         }
423
424         // Read the cid file
425         var container_id string
426         if cgroup_cidfile != "" {
427                 // wait up to 'wait' seconds for the cid file to appear
428                 ok := false
429                 var i time.Duration
430                 for i = 0; i < time.Duration(wait)*time.Second; i += (100 * time.Millisecond) {
431                         cid, err := ioutil.ReadFile(cgroup_cidfile)
432                         if err == nil && len(cid) > 0 {
433                                 ok = true
434                                 container_id = string(cid)
435                                 break
436                         }
437                         time.Sleep(100 * time.Millisecond)
438                 }
439                 if !ok {
440                         statLog.Println("error reading cid file:", cgroup_cidfile)
441                 }
442         }
443
444         stop_poll_chan := make(chan bool, 1)
445         cgroup := Cgroup{cgroup_root, cgroup_parent, container_id}
446         go PollCgroupStats(cgroup, poll, stop_poll_chan)
447
448         // When the child exits, tell the polling goroutine to stop.
449         defer func() { stop_poll_chan <- true }()
450
451         // Wait for CopyPipeToChan to consume child's stderr pipe
452         <-finish_chan
453
454         return cmd.Wait()
455 }
456
457 func main() {
458         logger := log.New(os.Stderr, "crunchstat: ", 0)
459         if err := run(logger); err != nil {
460                 if exiterr, ok := err.(*exec.ExitError); ok {
461                         // The program has exited with an exit code != 0
462
463                         // This works on both Unix and
464                         // Windows. Although package syscall is
465                         // generally platform dependent, WaitStatus is
466                         // defined for both Unix and Windows and in
467                         // both cases has an ExitStatus() method with
468                         // the same signature.
469                         if status, ok := exiterr.Sys().(syscall.WaitStatus); ok {
470                                 os.Exit(status.ExitStatus())
471                         }
472                 } else {
473                         statLog.Fatalln("error in cmd.Wait:", err)
474                 }
475         }
476 }