X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/becd16125599abeefa3d2d1203279bf7eee69669..ae6b514a98d89ce0bdfd47edd4508c42761cb049:/services/crunchstat/crunchstat.go diff --git a/services/crunchstat/crunchstat.go b/services/crunchstat/crunchstat.go index 8f698a2139..1982223716 100644 --- a/services/crunchstat/crunchstat.go +++ b/services/crunchstat/crunchstat.go @@ -3,8 +3,8 @@ package main import ( "bufio" "bytes" - "flag" "errors" + "flag" "fmt" "io" "io/ioutil" @@ -12,6 +12,7 @@ import ( "os" "os/exec" "os/signal" + "strconv" "strings" "syscall" "time" @@ -24,6 +25,7 @@ import ( #include */ import "C" + // The above block of magic allows us to look up user_hz via _SC_CLK_TCK. type Cgroup struct { @@ -32,36 +34,43 @@ type Cgroup struct { cid string } -func CopyPipeToChan(in io.Reader, out chan string, done chan<- bool) { - s := bufio.NewScanner(in) - for s.Scan() { - out <- s.Text() - } - done <- true -} +var childLog = log.New(os.Stderr, "", 0) +var statLog = log.New(os.Stderr, "crunchstat: ", 0) -func CopyChanToPipe(in <-chan string, out io.Writer) { - for s := range in { - fmt.Fprintln(out, s) - } -} +const ( + MaxLogLine = 1 << 14 // Child stderr lines >16KiB will be split +) -func OpenAndReadAll(filename string, log_chan chan<- string) ([]byte, error) { - in, err := os.Open(filename) - if err != nil { - if log_chan != nil { - log_chan <- fmt.Sprintf("crunchstat: open %s: %s", filename, err) +func CopyPipeToChildLog(in io.ReadCloser, done chan<- bool) { + reader := bufio.NewReaderSize(in, MaxLogLine) + var prefix string + for { + line, isPrefix, err := reader.ReadLine() + if err == io.EOF { + break + } else if err != nil { + statLog.Fatal("error reading child stderr:", err) + } + var suffix string + if isPrefix { + suffix = "[...]" + } + childLog.Print(prefix, string(line), suffix) + // Set up prefix for following line + if isPrefix { + prefix = "[...]" + } else { + prefix = "" } - return nil, err } - defer in.Close() - return ReadAllOrWarn(in, log_chan) + done <- true + in.Close() } -func ReadAllOrWarn(in *os.File, log_chan chan<- string) ([]byte, error) { +func ReadAllOrWarn(in *os.File) ([]byte, error) { content, err := ioutil.ReadAll(in) - if err != nil && log_chan != nil { - log_chan <- fmt.Sprintf("crunchstat: read %s: %s", in.Name(), err) + if err != nil { + statLog.Printf("error reading %s: %s\n", in.Name(), err) } return content, err } @@ -76,26 +85,24 @@ var reportedStatFile = map[string]string{} // container, and read /proc/PID/cgroup to determine the appropriate // cgroup root for the given statgroup. (This will avoid falling back // to host-level stats during container setup and teardown.) -func OpenStatFile(stderr chan<- string, cgroup Cgroup, statgroup string, stat string) (*os.File, error) { - var path string - path = fmt.Sprintf("%s/%s/%s/%s/%s", cgroup.root, statgroup, cgroup.parent, cgroup.cid, stat) - file, err := os.Open(path) - if err != nil { - path = fmt.Sprintf("%s/%s/%s/%s", cgroup.root, cgroup.parent, cgroup.cid, stat) - file, err = os.Open(path) - } - if err != nil { - path = fmt.Sprintf("%s/%s/%s", cgroup.root, statgroup, stat) - file, err = os.Open(path) +func OpenStatFile(cgroup Cgroup, statgroup string, stat string) (*os.File, error) { + var paths = []string{ + fmt.Sprintf("%s/%s/%s/%s/%s", cgroup.root, statgroup, cgroup.parent, cgroup.cid, stat), + fmt.Sprintf("%s/%s/%s/%s", cgroup.root, cgroup.parent, cgroup.cid, stat), + fmt.Sprintf("%s/%s/%s", cgroup.root, statgroup, stat), + fmt.Sprintf("%s/%s", cgroup.root, stat), } - if err != nil { - path = fmt.Sprintf("%s/%s", cgroup.root, stat) + var path string + var file *os.File + var err error + for _, path = range paths { file, err = os.Open(path) + if err == nil { + break + } else { + path = "" + } } - if err != nil { - file = nil - path = "" - } if pathWas, ok := reportedStatFile[stat]; !ok || pathWas != path { // Log whenever we start using a new/different cgroup // stat file for a given statistic. This typically @@ -103,18 +110,20 @@ func OpenStatFile(stderr chan<- string, cgroup Cgroup, statgroup string, stat st // whether we happen to collect stats [a] before any // processes have been created in the container and // [b] after all contained processes have exited. - reportedStatFile[stat] = path if path == "" { - stderr <- fmt.Sprintf("crunchstat: did not find stats file (root %s, parent %s, cid %s, statgroup %s, stat %s)", cgroup.root, cgroup.parent, cgroup.cid, statgroup, stat) + statLog.Printf("notice: stats not available: stat %s, statgroup %s, cid %s, parent %s, root %s\n", stat, statgroup, cgroup.cid, cgroup.parent, cgroup.root) + } else if ok { + statLog.Printf("notice: stats moved from %s to %s\n", reportedStatFile[stat], path) } else { - stderr <- fmt.Sprintf("crunchstat: reading stats from %s", path) + statLog.Printf("notice: reading stats from %s\n", path) } + reportedStatFile[stat] = path } return file, err } -func GetContainerNetStats(stderr chan<- string, cgroup Cgroup) (io.Reader, error) { - procsFile, err := OpenStatFile(stderr, cgroup, "cpuacct", "cgroup.procs") +func GetContainerNetStats(cgroup Cgroup) (io.Reader, error) { + procsFile, err := OpenStatFile(cgroup, "cpuacct", "cgroup.procs") if err != nil { return nil, err } @@ -123,8 +132,9 @@ func GetContainerNetStats(stderr chan<- string, cgroup Cgroup) (io.Reader, error for reader.Scan() { taskPid := reader.Text() statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid) - stats, err := OpenAndReadAll(statsFilename, stderr) + stats, err := ioutil.ReadFile(statsFilename) if err != nil { + statLog.Printf("read %s: %s\n", statsFilename, err) continue } return strings.NewReader(string(stats)), nil @@ -138,10 +148,10 @@ type IoSample struct { rxBytes int64 } -func DoBlkIoStats(stderr chan<- string, cgroup Cgroup, lastSample map[string]IoSample) (map[string]IoSample) { - c, err := OpenStatFile(stderr, cgroup, "blkio", "blkio.io_service_bytes") +func DoBlkIoStats(cgroup Cgroup, lastSample map[string]IoSample) { + c, err := OpenStatFile(cgroup, "blkio", "blkio.io_service_bytes") if err != nil { - return lastSample + return } defer c.Close() b := bufio.NewScanner(c) @@ -166,9 +176,6 @@ func DoBlkIoStats(stderr chan<- string, cgroup Cgroup, lastSample map[string]IoS } newSamples[device] = thisSample } - if lastSample == nil { - lastSample = make(map[string]IoSample) - } for dev, sample := range newSamples { if sample.txBytes < 0 || sample.rxBytes < 0 { continue @@ -177,13 +184,12 @@ func DoBlkIoStats(stderr chan<- string, cgroup Cgroup, lastSample map[string]IoS if prev, ok := lastSample[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) + sample.txBytes-prev.txBytes, + sample.rxBytes-prev.rxBytes) } - stderr <- fmt.Sprintf("crunchstat: blkio:%s %d write %d read%s", dev, sample.txBytes, sample.rxBytes, delta) + statLog.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta) lastSample[dev] = sample } - return lastSample } type MemSample struct { @@ -191,15 +197,15 @@ type MemSample struct { memStat map[string]int64 } -func DoMemoryStats(stderr chan<- string, cgroup Cgroup) { - c, err := OpenStatFile(stderr, cgroup, "memory", "memory.stat") +func DoMemoryStats(cgroup Cgroup) { + c, err := OpenStatFile(cgroup, "memory", "memory.stat") if err != nil { return } defer c.Close() b := bufio.NewScanner(c) thisSample := MemSample{time.Now(), make(map[string]int64)} - wantStats := [...]string{"cache", "pgmajfault", "rss"} + wantStats := [...]string{"cache", "swap", "pgmajfault", "rss"} for b.Scan() { var stat string var val int64 @@ -214,64 +220,55 @@ func DoMemoryStats(stderr chan<- string, cgroup Cgroup) { outstat.WriteString(fmt.Sprintf(" %d %s", val, key)) } } - stderr <- fmt.Sprintf("crunchstat: mem%s", outstat.String()) + statLog.Printf("mem%s\n", outstat.String()) } -func DoNetworkStats(stderr chan<- string, cgroup Cgroup, lastSample map[string]IoSample) (map[string]IoSample) { +func DoNetworkStats(cgroup Cgroup, lastSample map[string]IoSample) { sampleTime := time.Now() - stats, err := GetContainerNetStats(stderr, cgroup) - if err != nil { return lastSample } - - if lastSample == nil { - lastSample = make(map[string]IoSample) + stats, err := GetContainerNetStats(cgroup) + if err != nil { + return } + scanner := bufio.NewScanner(stats) - Iface: for scanner.Scan() { + for scanner.Scan() { var ifName string var rx, tx int64 - words := bufio.NewScanner(strings.NewReader(scanner.Text())) - words.Split(bufio.ScanWords) - wordIndex := 0 - for words.Scan() { - word := words.Text() - switch wordIndex { - case 0: - ifName = strings.TrimRight(word, ":") - case 1: - if _, err := fmt.Sscanf(word, "%d", &rx); err != nil { - continue Iface - } - case 9: - if _, err := fmt.Sscanf(word, "%d", &tx); err != nil { - continue Iface - } - } - wordIndex++ + words := strings.Fields(scanner.Text()) + if len(words) != 17 { + // Skip lines with wrong format + continue } - if ifName == "lo" || ifName == "" || wordIndex != 17 { + ifName = strings.TrimRight(words[0], ":") + if ifName == "lo" || ifName == "" { // Skip loopback interface and lines with wrong format continue } + if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil { + continue + } + if rx, err = strconv.ParseInt(words[1], 10, 64); err != nil { + continue + } nextSample := IoSample{} nextSample.sampleTime = sampleTime nextSample.txBytes = tx nextSample.rxBytes = rx var delta string - if lastSample, ok := lastSample[ifName]; ok { - interval := nextSample.sampleTime.Sub(lastSample.sampleTime).Seconds() + if prev, ok := lastSample[ifName]; ok { + interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds() delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx", interval, - tx - lastSample.txBytes, - rx - lastSample.rxBytes) + tx-prev.txBytes, + rx-prev.rxBytes) } - stderr <- fmt.Sprintf("crunchstat: net:%s %d tx %d rx%s", - ifName, tx, rx, delta) + statLog.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta) lastSample[ifName] = nextSample } - return lastSample } type CpuSample struct { + hasData bool // to distinguish the zero value from real data sampleTime time.Time user float64 sys float64 @@ -280,13 +277,13 @@ type CpuSample struct { // Return the number of CPUs available in the container. Return 0 if // we can't figure out the real number of CPUs. -func GetCpuCount(stderr chan<- string, cgroup Cgroup) (int64) { - cpusetFile, err := OpenStatFile(stderr, cgroup, "cpuset", "cpuset.cpus") +func GetCpuCount(cgroup Cgroup) int64 { + cpusetFile, err := OpenStatFile(cgroup, "cpuset", "cpuset.cpus") if err != nil { return 0 } defer cpusetFile.Close() - b, err := ReadAllOrWarn(cpusetFile, stderr) + b, err := ReadAllOrWarn(cpusetFile) sp := strings.Split(string(b), ",") cpus := int64(0) for _, v := range sp { @@ -301,18 +298,18 @@ func GetCpuCount(stderr chan<- string, cgroup Cgroup) (int64) { return cpus } -func DoCpuStats(stderr chan<- string, cgroup Cgroup, lastSample *CpuSample) (*CpuSample) { - statFile, err := OpenStatFile(stderr, cgroup, "cpuacct", "cpuacct.stat") +func DoCpuStats(cgroup Cgroup, lastSample *CpuSample) { + statFile, err := OpenStatFile(cgroup, "cpuacct", "cpuacct.stat") if err != nil { - return lastSample + return } defer statFile.Close() - b, err := ReadAllOrWarn(statFile, stderr) + b, err := ReadAllOrWarn(statFile) if err != nil { - return lastSample + return } - nextSample := &CpuSample{time.Now(), 0, 0, GetCpuCount(stderr, cgroup)} + nextSample := CpuSample{true, time.Now(), 0, 0, GetCpuCount(cgroup)} var userTicks, sysTicks int64 fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks) user_hz := float64(C.sysconf(C._SC_CLK_TCK)) @@ -320,21 +317,21 @@ func DoCpuStats(stderr chan<- string, cgroup Cgroup, lastSample *CpuSample) (*Cp nextSample.sys = float64(sysTicks) / user_hz delta := "" - if lastSample != nil { + if lastSample.hasData { delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys", nextSample.sampleTime.Sub(lastSample.sampleTime).Seconds(), - nextSample.user - lastSample.user, - nextSample.sys - lastSample.sys) + nextSample.user-lastSample.user, + nextSample.sys-lastSample.sys) } - stderr <- fmt.Sprintf("crunchstat: cpu %.4f user %.4f sys %d cpus%s", + statLog.Printf("cpu %.4f user %.4f sys %d cpus%s\n", nextSample.user, nextSample.sys, nextSample.cpus, delta) - return nextSample + *lastSample = nextSample } -func PollCgroupStats(cgroup Cgroup, stderr chan string, poll int64, stop_poll_chan <-chan bool) { - var lastNetSample map[string]IoSample = nil - var lastDiskSample map[string]IoSample = nil - var lastCpuSample *CpuSample = nil +func PollCgroupStats(cgroup Cgroup, poll int64, stop_poll_chan <-chan bool) { + var lastNetSample = map[string]IoSample{} + var lastDiskSample = map[string]IoSample{} + var lastCpuSample = CpuSample{} poll_chan := make(chan bool, 1) go func() { @@ -352,10 +349,10 @@ func PollCgroupStats(cgroup Cgroup, stderr chan string, poll int64, stop_poll_ch case <-poll_chan: // Emit stats, then select again. } - DoMemoryStats(stderr, cgroup) - lastCpuSample = DoCpuStats(stderr, cgroup, lastCpuSample) - lastDiskSample = DoBlkIoStats(stderr, cgroup, lastDiskSample) - lastNetSample = DoNetworkStats(stderr, cgroup, lastNetSample) + DoMemoryStats(cgroup) + DoCpuStats(cgroup, &lastCpuSample) + DoBlkIoStats(cgroup, lastDiskSample) + DoNetworkStats(cgroup, lastNetSample) } } @@ -378,23 +375,19 @@ func run(logger *log.Logger) error { flag.Parse() if cgroup_root == "" { - logger.Fatal("Must provide -cgroup-root") + statLog.Fatal("error: must provide -cgroup-root") } - stderr_chan := make(chan string, 1) - defer close(stderr_chan) finish_chan := make(chan bool) defer close(finish_chan) - go CopyChanToPipe(stderr_chan, os.Stderr) - var cmd *exec.Cmd if len(flag.Args()) > 0 { // Set up subprocess cmd = exec.Command(flag.Args()[0], flag.Args()[1:]...) - logger.Print("Running ", flag.Args()) + childLog.Println("Running", flag.Args()) // Child process will use our stdin and stdout pipes // (we close our copies below) @@ -402,27 +395,27 @@ func run(logger *log.Logger) error { cmd.Stdout = os.Stdout // Forward SIGINT and SIGTERM to inner process - term := make(chan os.Signal, 1) + sigChan := make(chan os.Signal, 1) go func(sig <-chan os.Signal) { catch := <-sig if cmd.Process != nil { cmd.Process.Signal(catch) } - logger.Print("caught signal: ", catch) - }(term) - signal.Notify(term, syscall.SIGTERM) - signal.Notify(term, syscall.SIGINT) + statLog.Println("caught signal:", catch) + }(sigChan) + signal.Notify(sigChan, syscall.SIGTERM) + signal.Notify(sigChan, syscall.SIGINT) // Funnel stderr through our channel stderr_pipe, err := cmd.StderrPipe() if err != nil { - logger.Fatal(err) + statLog.Fatalln("error in StderrPipe:", err) } - go CopyPipeToChan(stderr_pipe, stderr_chan, finish_chan) + go CopyPipeToChildLog(stderr_pipe, finish_chan) // Run subprocess if err := cmd.Start(); err != nil { - logger.Fatal(err) + statLog.Fatalln("error in cmd.Start:", err) } // Close stdin/stdout in this (parent) process @@ -437,7 +430,7 @@ func run(logger *log.Logger) error { ok := false var i time.Duration for i = 0; i < time.Duration(wait)*time.Second; i += (100 * time.Millisecond) { - cid, err := OpenAndReadAll(cgroup_cidfile, nil) + cid, err := ioutil.ReadFile(cgroup_cidfile) if err == nil && len(cid) > 0 { ok = true container_id = string(cid) @@ -446,13 +439,13 @@ func run(logger *log.Logger) error { time.Sleep(100 * time.Millisecond) } if !ok { - logger.Printf("Could not read cid file %s", cgroup_cidfile) + statLog.Println("error reading cid file:", cgroup_cidfile) } } stop_poll_chan := make(chan bool, 1) cgroup := Cgroup{cgroup_root, cgroup_parent, container_id} - go PollCgroupStats(cgroup, stderr_chan, poll, stop_poll_chan) + go PollCgroupStats(cgroup, poll, stop_poll_chan) // When the child exits, tell the polling goroutine to stop. defer func() { stop_poll_chan <- true }() @@ -479,7 +472,7 @@ func main() { os.Exit(status.ExitStatus()) } } else { - logger.Fatalf("cmd.Wait: %v", err) + statLog.Fatalln("error in cmd.Wait:", err) } } }