X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/14f6c56b96fb8b7ccd104951f9e8374540f16fa5..refs/heads/5748-keepstore-leak:/services/crunchstat/crunchstat.go diff --git a/services/crunchstat/crunchstat.go b/services/crunchstat/crunchstat.go index d61871da64..1a22e264f1 100644 --- a/services/crunchstat/crunchstat.go +++ b/services/crunchstat/crunchstat.go @@ -2,6 +2,8 @@ package main import ( "bufio" + "bytes" + "errors" "flag" "fmt" "io" @@ -10,217 +12,349 @@ import ( "os" "os/exec" "os/signal" + "strconv" "strings" "syscall" "time" ) -func ReadLineByLine(inp io.ReadCloser, out chan string, finish chan bool) { - s := bufio.NewScanner(inp) - for s.Scan() { - out <- s.Text() - } - finish <- true +/* +#include +#include +#include +#include +*/ +import "C" + +// The above block of magic allows us to look up user_hz via _SC_CLK_TCK. + +type Cgroup struct { + root string + parent string + cid string } -func OutputChannel(stdout chan string, stderr chan string) { +var childLog = log.New(os.Stderr, "", 0) +var statLog = log.New(os.Stderr, "crunchstat: ", 0) + +const ( + MaxLogLine = 1 << 14 // Child stderr lines >16KiB will be split +) + +func CopyPipeToChildLog(in io.ReadCloser, done chan<- bool) { + reader := bufio.NewReaderSize(in, MaxLogLine) + var prefix string for { - select { - case s, ok := <-stdout: - if ok { - fmt.Fprintln(os.Stdout, s) - } else { - return - } - case s, ok := <-stderr: - if ok { - fmt.Fprintln(os.Stderr, s) - } else { - return - } + 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 = "" } } + done <- true + in.Close() } -func FindStat(cgroup_root string, cgroup_parent string, container_id string, statgroup string, stat string) string { - var path string - path = fmt.Sprintf("%s/%s/%s/%s/%s.%s", cgroup_root, statgroup, cgroup_parent, container_id, statgroup, stat) - if _, err := os.Stat(path); err == nil { - return path +func ReadAllOrWarn(in *os.File) ([]byte, error) { + content, err := ioutil.ReadAll(in) + if err != nil { + statLog.Printf("error reading %s: %s\n", in.Name(), err) } - path = fmt.Sprintf("%s/%s/%s/%s.%s", cgroup_root, cgroup_parent, container_id, statgroup, stat) - if _, err := os.Stat(path); err == nil { - return path + return content, err +} + +var reportedStatFile = map[string]string{} + +// Open the cgroup stats file in /sys/fs corresponding to the target +// cgroup, and return an *os.File. If no stats file is available, +// return nil. +// +// TODO: Instead of trying all options, choose a process in the +// 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(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), } - path = fmt.Sprintf("%s/%s/%s.%s", cgroup_root, statgroup, statgroup, stat) - if _, err := os.Stat(path); err == nil { - return path + 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 = "" + } } - path = fmt.Sprintf("%s/%s.%s", cgroup_root, statgroup, stat) - if _, err := os.Stat(path); err == nil { - return 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 + // happens 1 to 3 times per statistic, depending on + // 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 == "" { + 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) + } else { + statLog.Printf("error reading stats from %s\n", path) + } } - return "" + return file, err } -func PollCgroupStats(cgroup_root string, cgroup_parent string, container_id string, stderr chan string, poll int64) { - //var last_usage int64 = 0 - var last_user int64 = 0 - var last_sys int64 = 0 - var last_cpucount int64 = 0 - - type Disk struct { - last_read int64 - next_read int64 - last_write int64 - next_write int64 +func GetContainerNetStats(cgroup Cgroup) (io.Reader, error) { + procsFile, err := OpenStatFile(cgroup, "cpuacct", "cgroup.procs") + if err != nil { + return nil, err } + defer procsFile.Close() + reader := bufio.NewScanner(procsFile) + for reader.Scan() { + taskPid := reader.Text() + statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid) + stats, err := ioutil.ReadFile(statsFilename) + if err != nil { + statLog.Printf("read %s: %s\n", statsFilename, err) + continue + } + return strings.NewReader(string(stats)), nil + } + return nil, errors.New("Could not read stats for any proc in container") +} - disk := make(map[string]*Disk) - - //cpuacct_usage := FindStat(cgroup_path, "cpuacct", "usage") - cpuacct_stat := FindStat(cgroup_root, cgroup_parent, container_id, "cpuacct", "stat") - blkio_io_service_bytes := FindStat(cgroup_root, cgroup_parent, container_id, "blkio", "io_service_bytes") - cpuset_cpus := FindStat(cgroup_root, cgroup_parent, container_id, "cpuset", "cpus") - memory_stat := FindStat(cgroup_root, cgroup_parent, container_id, "memory", "stat") +type IoSample struct { + sampleTime time.Time + txBytes int64 + rxBytes int64 +} - if cpuacct_stat != "" { - stderr <- fmt.Sprintf("crunchstat: reading stats from %s", cpuacct_stat) - } - if blkio_io_service_bytes != "" { - stderr <- fmt.Sprintf("crunchstat: reading stats from %s", 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 } - if cpuset_cpus != "" { - stderr <- fmt.Sprintf("crunchstat: reading stats from %s", cpuset_cpus) + defer c.Close() + b := bufio.NewScanner(c) + var sampleTime = time.Now() + newSamples := make(map[string]IoSample) + for b.Scan() { + var device, op string + var val int64 + if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &val); err != nil { + continue + } + var thisSample IoSample + var ok bool + if thisSample, ok = newSamples[device]; !ok { + thisSample = IoSample{sampleTime, -1, -1} + } + switch op { + case "Read": + thisSample.rxBytes = val + case "Write": + thisSample.txBytes = val + } + newSamples[device] = thisSample } - if memory_stat != "" { - stderr <- fmt.Sprintf("crunchstat: reading stats from %s", memory_stat) + for dev, sample := range newSamples { + if sample.txBytes < 0 || sample.rxBytes < 0 { + continue + } + delta := "" + 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) + } + statLog.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta) + lastSample[dev] = sample } +} - var elapsed int64 = poll +type MemSample struct { + sampleTime time.Time + memStat map[string]int64 +} - for { - /*{ - c, _ := os.Open(cpuacct_usage) - b, _ := ioutil.ReadAll(c) - var next int64 - fmt.Sscanf(string(b), "%d", &next) - if last_usage != 0 { - stderr <- fmt.Sprintf("crunchstat: cpuacct.usage %v", (next-last_usage)/10000000) - } - //fmt.Printf("usage %d %d %d %d%%\n", last_usage, next, next-last_usage, (next-last_usage)/10000000) - last_usage = next - c.Close() - }*/ - var cpus int64 = 0 - if cpuset_cpus != "" { - c, _ := os.Open(cpuset_cpus) - b, _ := ioutil.ReadAll(c) - sp := strings.Split(string(b), ",") - for _, v := range sp { - var min, max int64 - n, _ := fmt.Sscanf(v, "%d-%d", &min, &max) - if n == 2 { - cpus += (max - min) + 1 - } else { - cpus += 1 - } - } +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", "swap", "pgmajfault", "rss"} + for b.Scan() { + var stat string + var val int64 + if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil { + continue + } + thisSample.memStat[stat] = val + } + var outstat bytes.Buffer + for _, key := range wantStats { + if val, ok := thisSample.memStat[key]; ok { + outstat.WriteString(fmt.Sprintf(" %d %s", val, key)) + } + } + statLog.Printf("mem%s\n", outstat.String()) +} - if cpus != last_cpucount { - stderr <- fmt.Sprintf("crunchstat: cpuset.cpus %v", cpus) - } - last_cpucount = cpus +func DoNetworkStats(cgroup Cgroup, lastSample map[string]IoSample) { + sampleTime := time.Now() + stats, err := GetContainerNetStats(cgroup) + if err != nil { + return + } - c.Close() + scanner := bufio.NewScanner(stats) + for scanner.Scan() { + var ifName string + var rx, tx int64 + words := strings.Fields(scanner.Text()) + if len(words) != 17 { + // Skip lines with wrong format + continue } - if cpus == 0 { - cpus = 1 + ifName = strings.TrimRight(words[0], ":") + if ifName == "lo" || ifName == "" { + // Skip loopback interface and lines with wrong format + continue } - if cpuacct_stat != "" { - c, _ := os.Open(cpuacct_stat) - b, _ := ioutil.ReadAll(c) - var next_user int64 - var next_sys int64 - fmt.Sscanf(string(b), "user %d\nsystem %d", &next_user, &next_sys) - c.Close() - - if last_user != 0 { - user_diff := next_user - last_user - sys_diff := next_sys - last_sys - // Assume we're reading stats based on 100 - // jiffies per second. Because the ellaspsed - // time is in milliseconds, we need to boost - // that to 1000 jiffies per second, then boost - // it by another 100x to get a percentage, then - // finally divide by the actual elapsed time - // and the number of cpus to get average load - // over the polling period. - user_pct := (user_diff * 10 * 100) / (elapsed * cpus) - sys_pct := (sys_diff * 10 * 100) / (elapsed * cpus) - - stderr <- fmt.Sprintf("crunchstat: cpuacct.stat user %v", user_pct) - stderr <- fmt.Sprintf("crunchstat: cpuacct.stat sys %v", sys_pct) - } - - /*fmt.Printf("user %d %d %d%%\n", last_user, next_user, next_user-last_user) - fmt.Printf("sys %d %d %d%%\n", last_sys, next_sys, next_sys-last_sys) - fmt.Printf("sum %d%%\n", (next_user-last_user)+(next_sys-last_sys))*/ - last_user = next_user - last_sys = next_sys + if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil { + continue } - if blkio_io_service_bytes != "" { - c, _ := os.Open(blkio_io_service_bytes) - b := bufio.NewScanner(c) - var device, op string - var next int64 - for b.Scan() { - if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &next); err == nil { - if disk[device] == nil { - disk[device] = new(Disk) - } - if op == "Read" { - disk[device].last_read = disk[device].next_read - disk[device].next_read = next - if disk[device].last_read > 0 && (disk[device].next_read != disk[device].last_read) { - stderr <- fmt.Sprintf("crunchstat: blkio.io_service_bytes %s read %v", device, disk[device].next_read-disk[device].last_read) - } - } - if op == "Write" { - disk[device].last_write = disk[device].next_write - disk[device].next_write = next - if disk[device].last_write > 0 && (disk[device].next_write != disk[device].last_write) { - stderr <- fmt.Sprintf("crunchstat: blkio.io_service_bytes %s write %v", device, disk[device].next_write-disk[device].last_write) - } - } - } - } - c.Close() + 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 prev, ok := lastSample[ifName]; ok { + interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds() + delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx", + interval, + tx-prev.txBytes, + rx-prev.rxBytes) + } + statLog.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta) + lastSample[ifName] = nextSample + } +} - if memory_stat != "" { - c, _ := os.Open(memory_stat) - b := bufio.NewScanner(c) - var stat string - var val int64 - for b.Scan() { - if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err == nil { - if stat == "rss" { - stderr <- fmt.Sprintf("crunchstat: memory.stat rss %v", val) - } - } - } - c.Close() +type CpuSample struct { + hasData bool // to distinguish the zero value from real data + sampleTime time.Time + user float64 + sys float64 + cpus int64 +} + +// Return the number of CPUs available in the container. Return 0 if +// we can't figure out the real number of 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) + sp := strings.Split(string(b), ",") + cpus := int64(0) + for _, v := range sp { + var min, max int64 + n, _ := fmt.Sscanf(v, "%d-%d", &min, &max) + if n == 2 { + cpus += (max - min) + 1 + } else { + cpus += 1 } + } + return cpus +} - bedtime := time.Now() - time.Sleep(time.Duration(poll) * time.Millisecond) - morning := time.Now() - elapsed = morning.Sub(bedtime).Nanoseconds() / int64(time.Millisecond) +func DoCpuStats(cgroup Cgroup, lastSample *CpuSample) { + statFile, err := OpenStatFile(cgroup, "cpuacct", "cpuacct.stat") + if err != nil { + return } + defer statFile.Close() + b, err := ReadAllOrWarn(statFile) + if err != nil { + return + } + + 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)) + nextSample.user = float64(userTicks) / user_hz + nextSample.sys = float64(sysTicks) / user_hz + + delta := "" + 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) + } + statLog.Printf("cpu %.4f user %.4f sys %d cpus%s\n", + nextSample.user, nextSample.sys, nextSample.cpus, delta) + *lastSample = nextSample } -func main() { +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() { + // Send periodic poll events. + poll_chan <- true + for { + time.Sleep(time.Duration(poll) * time.Millisecond) + poll_chan <- true + } + }() + for { + select { + case <-stop_poll_chan: + return + case <-poll_chan: + // Emit stats, then select again. + } + DoMemoryStats(cgroup) + DoCpuStats(cgroup, &lastCpuSample) + DoBlkIoStats(cgroup, lastDiskSample) + DoNetworkStats(cgroup, lastNetSample) + } +} + +func run(logger *log.Logger) error { var ( cgroup_root string @@ -238,102 +372,105 @@ func main() { flag.Parse() - logger := log.New(os.Stderr, "crunchstat: ", 0) - if cgroup_root == "" { - logger.Fatal("Must provide either -cgroup-root") + statLog.Fatal("error: must provide -cgroup-root") } - // Make output channel - stdout_chan := make(chan string) - stderr_chan := make(chan string) finish_chan := make(chan bool) - defer close(stdout_chan) - defer close(stderr_chan) defer close(finish_chan) - go OutputChannel(stdout_chan, stderr_chan) - 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) + cmd.Stdin = os.Stdin + 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) - - // Funnel stdout and stderr from subprocess to output channels - stdout_pipe, err := cmd.StdoutPipe() - if err != nil { - logger.Fatal(err) - } - go ReadLineByLine(stdout_pipe, stdout_chan, finish_chan) + 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 ReadLineByLine(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 + os.Stdin.Close() + os.Stdout.Close() } // Read the cid file var container_id string if cgroup_cidfile != "" { // wait up to 'wait' seconds for the cid file to appear + ok := false var i time.Duration for i = 0; i < time.Duration(wait)*time.Second; i += (100 * time.Millisecond) { - f, err := os.Open(cgroup_cidfile) - if err == nil { - cid, err2 := ioutil.ReadAll(f) - if err2 == nil && len(cid) > 0 { - container_id = string(cid) - f.Close() - break - } + cid, err := ioutil.ReadFile(cgroup_cidfile) + if err == nil && len(cid) > 0 { + ok = true + container_id = string(cid) + break } time.Sleep(100 * time.Millisecond) } - if cgroup_root == "" { - logger.Printf("Could not read cid file %s", cgroup_cidfile) + if !ok { + statLog.Println("error reading cid file:", cgroup_cidfile) } } - go PollCgroupStats(cgroup_root, cgroup_parent, container_id, stderr_chan, poll) + stop_poll_chan := make(chan bool, 1) + cgroup := Cgroup{cgroup_root, cgroup_parent, container_id} + go PollCgroupStats(cgroup, poll, stop_poll_chan) - // Wait for each of stdout and stderr to drain - <-finish_chan + // When the child exits, tell the polling goroutine to stop. + defer func() { stop_poll_chan <- true }() + + // Wait for CopyPipeToChan to consume child's stderr pipe <-finish_chan - if err := cmd.Wait(); err != nil { + return cmd.Wait() +} + +func main() { + logger := log.New(os.Stderr, "crunchstat: ", 0) + if err := run(logger); err != nil { if exiterr, ok := err.(*exec.ExitError); ok { // The program has exited with an exit code != 0 - // This works on both Unix and Windows. Although package - // syscall is generally platform dependent, WaitStatus is - // defined for both Unix and Windows and in both cases has - // an ExitStatus() method with the same signature. + // This works on both Unix and + // Windows. Although package syscall is + // generally platform dependent, WaitStatus is + // defined for both Unix and Windows and in + // both cases has an ExitStatus() method with + // the same signature. if status, ok := exiterr.Sys().(syscall.WaitStatus); ok { os.Exit(status.ExitStatus()) } } else { - logger.Fatalf("cmd.Wait: %v", err) + statLog.Fatalln("error in cmd.Wait:", err) } } }