X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/200f7004f921a68ec40b407dfe31f1db95e98fb9..07baa0ed049746514495d1648c1aef0c40545141:/services/crunchstat/crunchstat.go

diff --git a/services/crunchstat/crunchstat.go b/services/crunchstat/crunchstat.go
index 1982223716..d28bee0f5e 100644
--- a/services/crunchstat/crunchstat.go
+++ b/services/crunchstat/crunchstat.go
@@ -1,478 +1,192 @@
+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: AGPL-3.0
+
 package main
 
 import (
 	"bufio"
-	"bytes"
-	"errors"
 	"flag"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"log"
 	"os"
 	"os/exec"
 	"os/signal"
-	"strconv"
-	"strings"
 	"syscall"
 	"time"
+
+	"git.arvados.org/arvados.git/lib/cmd"
+	"git.arvados.org/arvados.git/lib/crunchstat"
 )
 
-/*
-#include <unistd.h>
-#include <sys/types.h>
-#include <pwd.h>
-#include <stdlib.h>
-*/
-import "C"
+const MaxLogLine = 1 << 14 // Child stderr lines >16KiB will be split
 
-// The above block of magic allows us to look up user_hz via _SC_CLK_TCK.
+var (
+	signalOnDeadPPID  int = 15
+	ppidCheckInterval     = time.Second
+	version               = "dev"
+)
 
-type Cgroup struct {
-	root   string
-	parent string
-	cid    string
+type logger interface {
+	Printf(string, ...interface{})
 }
 
-var childLog = log.New(os.Stderr, "", 0)
-var statLog = log.New(os.Stderr, "crunchstat: ", 0)
+func main() {
+	reporter := crunchstat.Reporter{
+		Logger: log.New(os.Stderr, "crunchstat: ", 0),
+	}
+
+	flags := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
+	flags.StringVar(&reporter.CgroupRoot, "cgroup-root", "", "Root of cgroup tree")
+	flags.StringVar(&reporter.CgroupParent, "cgroup-parent", "", "Name of container parent under cgroup")
+	flags.StringVar(&reporter.CIDFile, "cgroup-cid", "", "Path to container id file")
+	flags.IntVar(&signalOnDeadPPID, "signal-on-dead-ppid", signalOnDeadPPID, "Signal to send child if crunchstat's parent process disappears (0 to disable)")
+	flags.DurationVar(&ppidCheckInterval, "ppid-check-interval", ppidCheckInterval, "Time between checks for parent process disappearance")
+	pollMsec := flags.Int64("poll", 1000, "Reporting interval, in milliseconds")
+	getVersion := flags.Bool("version", false, "Print version information and exit.")
+
+	if ok, code := cmd.ParseFlags(flags, os.Args[0], os.Args[1:], "program [args ...]", os.Stderr); !ok {
+		os.Exit(code)
+	} else if *getVersion {
+		fmt.Printf("crunchstat %s\n", version)
+		return
+	} else if flags.NArg() == 0 {
+		fmt.Fprintf(os.Stderr, "missing required argument: program (try -help)\n")
+		os.Exit(2)
+	}
 
-const (
-	MaxLogLine = 1 << 14 // Child stderr lines >16KiB will be split
-)
+	reporter.Logger.Printf("crunchstat %s started", version)
 
-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 = ""
-		}
+	if reporter.CgroupRoot == "" {
+		reporter.Logger.Printf("error: must provide -cgroup-root")
+		os.Exit(2)
+	} else if signalOnDeadPPID < 0 {
+		reporter.Logger.Printf("-signal-on-dead-ppid=%d is invalid (use a positive signal number, or 0 to disable)", signalOnDeadPPID)
+		os.Exit(2)
 	}
-	done <- true
-	in.Close()
-}
+	reporter.PollPeriod = time.Duration(*pollMsec) * time.Millisecond
 
-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
-}
+	reporter.Start()
+	err := runCommand(flags.Args(), reporter.Logger)
+	reporter.Stop()
 
-var reportedStatFile = map[string]string{}
+	if err, ok := err.(*exec.ExitError); ok {
+		// The program has exited with an exit code != 0
 
-// 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),
-	}
-	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 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)
+		// 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 := err.Sys().(syscall.WaitStatus); ok {
+			os.Exit(status.ExitStatus())
 		} else {
-			statLog.Printf("notice: reading stats from %s\n", path)
+			reporter.Logger.Printf("ExitError without WaitStatus: %v", err)
+			os.Exit(1)
 		}
-		reportedStatFile[stat] = path
+	} else if err != nil {
+		reporter.Logger.Printf("error running command: %v", err)
+		os.Exit(1)
 	}
-	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")
-}
+func runCommand(argv []string, logger logger) error {
+	cmd := exec.Command(argv[0], argv[1:]...)
 
-type IoSample struct {
-	sampleTime time.Time
-	txBytes    int64
-	rxBytes    int64
-}
+	logger.Printf("Running %v", argv)
 
-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
-}
+	// Child process will use our stdin and stdout pipes
+	// (we close our copies below)
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
 
-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))
+	// Forward SIGINT and SIGTERM to child process
+	sigChan := make(chan os.Signal, 1)
+	go func(sig <-chan os.Signal) {
+		catch := <-sig
+		if cmd.Process != nil {
+			cmd.Process.Signal(catch)
 		}
+		logger.Printf("notice: caught signal: %v", catch)
+	}(sigChan)
+	signal.Notify(sigChan, syscall.SIGTERM)
+	signal.Notify(sigChan, syscall.SIGINT)
+
+	// Kill our child proc if our parent process disappears
+	if signalOnDeadPPID != 0 {
+		go sendSignalOnDeadPPID(ppidCheckInterval, signalOnDeadPPID, os.Getppid(), cmd, logger)
 	}
-	statLog.Printf("mem%s\n", outstat.String())
-}
 
-func DoNetworkStats(cgroup Cgroup, lastSample map[string]IoSample) {
-	sampleTime := time.Now()
-	stats, err := GetContainerNetStats(cgroup)
+	// Funnel stderr through our channel
+	stderrPipe, err := cmd.StderrPipe()
 	if err != nil {
-		return
+		logger.Printf("error in StderrPipe: %v", err)
+		return err
 	}
 
-	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
+	// Run subprocess
+	if err := cmd.Start(); err != nil {
+		logger.Printf("error in cmd.Start: %v", err)
+		return err
 	}
-}
 
-type CpuSample struct {
-	hasData    bool // to distinguish the zero value from real data
-	sampleTime time.Time
-	user       float64
-	sys        float64
-	cpus       int64
-}
+	// Close stdin/stdout in this (parent) process
+	os.Stdin.Close()
+	os.Stdout.Close()
 
-// 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")
+	err = copyPipeToChildLog(stderrPipe, log.New(os.Stderr, "", 0))
 	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
-		}
+		cmd.Process.Kill()
+		return err
 	}
-	return cpus
-}
 
-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
+	return cmd.Wait()
 }
 
-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
+func sendSignalOnDeadPPID(intvl time.Duration, signum, ppidOrig int, cmd *exec.Cmd, logger logger) {
+	ticker := time.NewTicker(intvl)
+	for range ticker.C {
+		ppid := os.Getppid()
+		if ppid == ppidOrig {
+			continue
 		}
-	}()
-	for {
-		select {
-		case <-stop_poll_chan:
-			return
-		case <-poll_chan:
-			// Emit stats, then select again.
+		if cmd.Process == nil {
+			// Child process isn't running yet
+			continue
 		}
-		DoMemoryStats(cgroup)
-		DoCpuStats(cgroup, &lastCpuSample)
-		DoBlkIoStats(cgroup, lastDiskSample)
-		DoNetworkStats(cgroup, lastNetSample)
-	}
-}
-
-func run(logger *log.Logger) error {
-
-	var (
-		cgroup_root    string
-		cgroup_parent  string
-		cgroup_cidfile string
-		wait           int64
-		poll           int64
-	)
-
-	flag.StringVar(&cgroup_root, "cgroup-root", "", "Root of cgroup tree")
-	flag.StringVar(&cgroup_parent, "cgroup-parent", "", "Name of container parent under cgroup")
-	flag.StringVar(&cgroup_cidfile, "cgroup-cid", "", "Path to container id file")
-	flag.Int64Var(&wait, "wait", 5, "Maximum time (in seconds) to wait for cid file to show up")
-	flag.Int64Var(&poll, "poll", 1000, "Polling frequency, in milliseconds")
-
-	flag.Parse()
-
-	if cgroup_root == "" {
-		statLog.Fatal("error: must provide -cgroup-root")
-	}
-
-	finish_chan := make(chan bool)
-	defer close(finish_chan)
-
-	var cmd *exec.Cmd
-
-	if len(flag.Args()) > 0 {
-		// Set up subprocess
-		cmd = exec.Command(flag.Args()[0], flag.Args()[1:]...)
-
-		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
-		sigChan := make(chan os.Signal, 1)
-		go func(sig <-chan os.Signal) {
-			catch := <-sig
-			if cmd.Process != nil {
-				cmd.Process.Signal(catch)
-			}
-			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()
+		logger.Printf("notice: crunchstat ppid changed from %d to %d -- killing child pid %d with signal %d", ppidOrig, ppid, cmd.Process.Pid, signum)
+		err := cmd.Process.Signal(syscall.Signal(signum))
 		if err != nil {
-			statLog.Fatalln("error in StderrPipe:", err)
-		}
-		go CopyPipeToChildLog(stderr_pipe, finish_chan)
-
-		// Run subprocess
-		if err := cmd.Start(); err != nil {
-			statLog.Fatalln("error in cmd.Start:", err)
+			logger.Printf("error: sending signal: %s", err)
+			continue
 		}
-
-		// Close stdin/stdout in this (parent) process
-		os.Stdin.Close()
-		os.Stdout.Close()
+		ticker.Stop()
+		break
 	}
+}
 
-	// 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) {
-			cid, err := ioutil.ReadFile(cgroup_cidfile)
-			if err == nil && len(cid) > 0 {
-				ok = true
-				container_id = string(cid)
-				break
-			}
-			time.Sleep(100 * time.Millisecond)
+func copyPipeToChildLog(in io.ReadCloser, logger logger) error {
+	reader := bufio.NewReaderSize(in, MaxLogLine)
+	var prefix string
+	for {
+		line, isPrefix, err := reader.ReadLine()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return fmt.Errorf("error reading child stderr: %w", err)
 		}
-		if !ok {
-			statLog.Println("error reading cid file:", cgroup_cidfile)
+		var suffix string
+		if isPrefix {
+			suffix = "[...]"
 		}
-	}
-
-	stop_poll_chan := make(chan bool, 1)
-	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 }()
-
-	// Wait for CopyPipeToChan to consume child's stderr pipe
-	<-finish_chan
-
-	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.
-			if status, ok := exiterr.Sys().(syscall.WaitStatus); ok {
-				os.Exit(status.ExitStatus())
-			}
+		logger.Printf("%s%s%s", prefix, string(line), suffix)
+		// Set up prefix for following line
+		if isPrefix {
+			prefix = "[...]"
 		} else {
-			statLog.Fatalln("error in cmd.Wait:", err)
+			prefix = ""
 		}
 	}
+	return in.Close()
 }