X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b7f67c80916c2efa0c234ab5f4e92c24d47223c5..682dd5b6cc23a455766a7651e3e841257660b31c:/services/crunchstat/crunchstat.go

diff --git a/services/crunchstat/crunchstat.go b/services/crunchstat/crunchstat.go
index 387f6478d9..e14912423d 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,79 +12,334 @@ import (
 	"os"
 	"os/exec"
 	"os/signal"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
 )
 
-func CopyPipeToChan(in io.Reader, out chan string, done chan<- bool) {
-	s := bufio.NewScanner(in)
-	for s.Scan() {
-		out <- s.Text()
+/*
+#include <unistd.h>
+#include <sys/types.h>
+#include <pwd.h>
+#include <stdlib.h>
+*/
+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
+}
+
+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 {
+		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 CopyChanToPipe(in <-chan string, out io.Writer) {
-	for s := range in {
-		fmt.Fprintln(out, s)
+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)
 	}
+	return content, err
 }
 
-func FindStat(cgroup_root string, cgroup_parent string, container_id string, statgroup string, stat string) string {
+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
+	if cgroup.cid != "" {
+		// Collect container's stats
+		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),
+		}
+	} else {
+		// Collect this host's stats
+		paths = []string{
+			fmt.Sprintf("%s/%s/%s", cgroup.root, statgroup, stat),
+			fmt.Sprintf("%s/%s", cgroup.root, stat),
+		}
+	}
 	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
+	var file *os.File
+	var err error
+	for _, path = range paths {
+		file, err = os.Open(path)
+		if err == nil {
+			break
+		} else {
+			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.
+		if path == "" {
+			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 {
+			statLog.Printf("notice: reading stats from %s\n", path)
+		}
+		reportedStatFile[stat] = path
 	}
-	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 file, err
+}
+
+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")
+}
+
+type IoSample struct {
+	sampleTime time.Time
+	txBytes    int64
+	rxBytes    int64
+}
+
+func DoBlkIoStats(cgroup Cgroup, lastSample map[string]IoSample) {
+	c, err := OpenStatFile(cgroup, "blkio", "blkio.io_service_bytes")
+	if err != nil {
+		return
+	}
+	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
+	}
+	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
+	}
+}
+
+type MemSample struct {
+	sampleTime time.Time
+	memStat    map[string]int64
+}
+
+func DoMemoryStats(cgroup Cgroup) {
+	c, err := OpenStatFile(cgroup, "memory", "memory.stat")
+	if err != nil {
+		return
 	}
-	path = fmt.Sprintf("%s/%s/%s.%s", cgroup_root, statgroup, statgroup, stat)
-	if _, err := os.Stat(path); err == nil {
-		return path
+	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
 	}
-	path = fmt.Sprintf("%s/%s.%s", cgroup_root, statgroup, stat)
-	if _, err := os.Stat(path); err == nil {
-		return path
+	var outstat bytes.Buffer
+	for _, key := range wantStats {
+		if val, ok := thisSample.memStat[key]; ok {
+			outstat.WriteString(fmt.Sprintf(" %d %s", val, key))
+		}
 	}
-	return ""
+	statLog.Printf("mem%s\n", outstat.String())
 }
 
-func PollCgroupStats(cgroup_root string, cgroup_parent string, container_id string, stderr chan string, poll int64, stop_poll_chan <-chan bool) {
-	//var last_usage int64 = 0
-	var last_user int64 = -1
-	var last_sys int64 = -1
-	var last_cpucount int64 = 0
-
-	type Disk struct {
-		last_read  int64
-		next_read  int64
-		last_write int64
-		next_write int64
+func DoNetworkStats(cgroup Cgroup, lastSample map[string]IoSample) {
+	sampleTime := time.Now()
+	stats, err := GetContainerNetStats(cgroup)
+	if err != nil {
+		return
 	}
 
-	disk := make(map[string]*Disk)
+	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
+		}
+		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 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
+	}
+}
 
-	//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 CpuSample struct {
+	hasData    bool // to distinguish the zero value from real data
+	sampleTime time.Time
+	user       float64
+	sys        float64
+	cpus       int64
+}
 
-	if cpuacct_stat != "" {
-		stderr <- fmt.Sprintf("crunchstat: reading stats from %s", cpuacct_stat)
+// 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
+		}
 	}
-	if blkio_io_service_bytes != "" {
-		stderr <- fmt.Sprintf("crunchstat: reading stats from %s", blkio_io_service_bytes)
+	return cpus
+}
+
+func DoCpuStats(cgroup Cgroup, lastSample *CpuSample) {
+	statFile, err := OpenStatFile(cgroup, "cpuacct", "cpuacct.stat")
+	if err != nil {
+		return
 	}
-	if cpuset_cpus != "" {
-		stderr <- fmt.Sprintf("crunchstat: reading stats from %s", cpuset_cpus)
+	defer statFile.Close()
+	b, err := ReadAllOrWarn(statFile)
+	if err != nil {
+		return
 	}
-	if memory_stat != "" {
-		stderr <- fmt.Sprintf("crunchstat: reading stats from %s", memory_stat)
+
+	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 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() {
@@ -94,152 +351,16 @@ func PollCgroupStats(cgroup_root string, cgroup_parent string, container_id stri
 		}
 	}()
 	for {
-		bedtime := time.Now()
 		select {
 		case <-stop_poll_chan:
 			return
 		case <-poll_chan:
 			// Emit stats, then select again.
 		}
-		morning := time.Now()
-		elapsed := morning.Sub(bedtime).Nanoseconds() / int64(time.Millisecond)
-		/*{
-			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, err := os.Open(cpuset_cpus)
-			if err != nil {
-				stderr <- fmt.Sprintf("open %s: %s", cpuset_cpus, err)
-				// cgroup probably gone -- skip other stats too.
-				continue
-			}
-			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
-				}
-			}
-
-			if cpus != last_cpucount {
-				stderr <- fmt.Sprintf("crunchstat: cpuset.cpus %v", cpus)
-			}
-			last_cpucount = cpus
-
-			c.Close()
-		}
-		if cpus == 0 {
-			cpus = 1
-		}
-		if cpuacct_stat != "" {
-			c, err := os.Open(cpuacct_stat)
-			if err != nil {
-				stderr <- fmt.Sprintf("open %s: %s", cpuacct_stat, err)
-				// Next time around, last_user would
-				// be >1 interval old, so stats will
-				// be incorrect. Start over instead.
-				last_user = -1
-
-				// cgroup probably gone -- skip other stats too.
-				continue
-			}
-			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 elapsed > 0 && last_user != -1 {
-				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 elapsed
-				// 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 blkio_io_service_bytes != "" {
-			c, err := os.Open(blkio_io_service_bytes)
-			if err != nil {
-				stderr <- fmt.Sprintf("open %s: %s", blkio_io_service_bytes, err)
-				// cgroup probably gone -- skip other stats too.
-				continue
-			}
-			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 memory_stat != "" {
-			c, err := os.Open(memory_stat)
-			if err != nil {
-				stderr <- fmt.Sprintf("open %s: %s", memory_stat, err)
-				// cgroup probably gone -- skip other stats too.
-				continue
-			}
-			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()
-		}
+		DoMemoryStats(cgroup)
+		DoCpuStats(cgroup, &lastCpuSample)
+		DoBlkIoStats(cgroup, lastDiskSample)
+		DoNetworkStats(cgroup, lastNetSample)
 	}
 }
 
@@ -262,23 +383,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)
@@ -286,27 +403,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
@@ -321,25 +438,22 @@ 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) {
-			f, err := os.Open(cgroup_cidfile)
-			if err == nil {
-				defer f.Close()
-				cid, err2 := ioutil.ReadAll(f)
-				if err2 == nil && len(cid) > 0 {
-					ok = true
-					container_id = string(cid)
-					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 !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)
-	go PollCgroupStats(cgroup_root, cgroup_parent, container_id, stderr_chan, poll, stop_poll_chan)
+	cgroup := Cgroup{cgroup_root, cgroup_parent, container_id}
+	go PollCgroupStats(cgroup, poll, stop_poll_chan)
 
 	// When the child exits, tell the polling goroutine to stop.
 	defer func() { stop_poll_chan <- true }()
@@ -366,7 +480,7 @@ func main() {
 				os.Exit(status.ExitStatus())
 			}
 		} else {
-			logger.Fatalf("cmd.Wait: %v", err)
+			statLog.Fatalln("error in cmd.Wait:", err)
 		}
 	}
 }