Display user's processes info in task log when high memory consumption is detected, for debugging reasons (#594)

* Display user's processes information in task log when high memory consumption is detected, for debugging reasons

Author: harshita-chaudhary

runner/execer/os/execer.go

@@ -1,7 +1,6 @@
 package os
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"math"
@@ -118,15 +117,15 @@ func (e *execer) Exec(command scootexecer.Command) (scootexecer.Process, error)
 
 	proc := &process{cmd: cmd, wg: &wg, ats: AbortTimeoutSec, LogTags: command.LogTags}
 	if e.memCap > 0 {
-		go e.monitorMem(proc, command.MemCh)
+		go e.monitorMem(proc, command.MemCh, command.Stderr)
 	}
 
 	return proc, nil
 }
 
 // Periodically check to make sure memory constraints are respected,
 // and clean up after ourselves when the process has completed
-func (e *execer) monitorMem(p *process, memCh chan scootexecer.ProcessStatus) {
+func (e *execer) monitorMem(p *process, memCh chan scootexecer.ProcessStatus, stderr io.Writer) {
 	pid := p.cmd.Process.Pid
 	pgid, err := syscall.Getpgid(pid)
 	if err != nil {
@@ -168,6 +167,8 @@ func (e *execer) monitorMem(p *process, memCh chan scootexecer.ProcessStatus) {
 			Error:    msg,
 			ExitCode: errors.HighInitialMemoryUtilizationExitCode,
 		}
+		// log the process snapshot in worker log, as well as task stderr log
+		e.pw.LogProcs(p, log.ErrorLevel, stderr)
 		p.mutex.Unlock()
 		e.memCapKill(p, mem, memCh)
 		return
@@ -212,6 +213,7 @@ func (e *execer) monitorMem(p *process, memCh chan scootexecer.ProcessStatus) {
 					Error:    msg,
 					ExitCode: 1,
 				}
+				e.pw.LogProcs(p, log.ErrorLevel, stderr)
 				p.mutex.Unlock()
 				e.memCapKill(p, mem, memCh)
 				return
@@ -230,7 +232,7 @@ func (e *execer) monitorMem(p *process, memCh chan scootexecer.ProcessStatus) {
 						"jobID":       p.JobID,
 						"taskID":      p.TaskID,
 					}).Infof("Memory utilization increased to %d%%, pid: %d", int(memUsagePct*100), pid)
-				debugProcesses(p)
+				e.pw.LogProcs(p, log.DebugLevel, nil)
 				for memUsagePct > reportThresholds[thresholdsIdx] {
 					thresholdsIdx++
 				}
@@ -265,26 +267,6 @@ func (e *execer) memCapKill(p *process, mem scootexecer.Memory, memCh chan scoot
 	e.stat.Gauge(stats.WorkerMemory).Update(int64(postKillMem))
 }
 
-// debugProcesses logs a snapshot of the current processes at debug level.
-func debugProcesses(p *process) {
-	// Debug log output with timeout since it seems CombinedOutput() sometimes fails to return.
-	if log.IsLevelEnabled(log.DebugLevel) {
-		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-		ps, err := exec.CommandContext(ctx, "ps", "-u", os.Getenv("USER"), "-opid,sess,ppid,pgid,rss,args").CombinedOutput()
-		log.WithFields(
-			log.Fields{
-				"pid":    p.cmd.Process.Pid,
-				"ps":     string(ps),
-				"err":    err,
-				"errCtx": ctx.Err(),
-				"tag":    p.Tag,
-				"jobID":  p.JobID,
-				"taskID": p.TaskID,
-			}).Debugf("ps after increased memory utilization for pid %d", p.cmd.Process.Pid)
-		cancel()
-	}
-}
-
 // Kill process along with all child processes, assuming no child processes called setpgid
 func cleanupProcs(pgid int) (err error) {
 	log.WithFields(

runner/execer/os/process_watcher.go

@@ -1,18 +1,24 @@
 package os
 
 import (
+	"context"
 	"fmt"
+	"io"
+	"os"
+	"time"
 
 	"os/exec"
 	"strings"
 
+	log "github.com/sirupsen/logrus"
 	scootexecer "github.com/twitter/scoot/runner/execer"
 )
 
 // Used for mocking memCap monitoring
 type ProcessWatcher interface {
 	GetProcs() (map[int]ProcInfo, error)
 	MemUsage(int) (scootexecer.Memory, error)
+	LogProcs(*process, log.Level, io.Writer)
 }
 
 type ProcInfo struct {
@@ -95,6 +101,36 @@ func (opw *procWatcher) MemUsage(pid int) (scootexecer.Memory, error) {
 	return scootexecer.Memory(total * bytesToKB), nil
 }
 
+// LogProcs logs the process snapshot of the current process along with other running processes for the user in the worker log,
+// at the specified level. Also writes to the writer, if provided
+func (opw *procWatcher) LogProcs(p *process, level log.Level, w io.Writer) {
+	if !log.IsLevelEnabled(level) {
+		return
+	}
+
+	// log output with timeout since it seems CombinedOutput() sometimes fails to return.
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	ps, err := exec.CommandContext(ctx, "ps", "-u", os.Getenv("USER"), "-opid,sess,ppid,pgid,rss,args", "--sort=-rss").CombinedOutput()
+
+	log.WithFields(
+		log.Fields{
+			"pid":    p.cmd.Process.Pid,
+			"ps":     string(ps),
+			"err":    err,
+			"errCtx": ctx.Err(),
+			"tag":    p.Tag,
+			"jobID":  p.JobID,
+			"taskID": p.TaskID,
+		}).Log(level, fmt.Sprintf("ps after increased memory utilization for pid %d", p.cmd.Process.Pid))
+
+	if w != nil {
+		w.Write([]byte(fmt.Sprintf("\nps after increased memory utilization for pid %d:\n\n", p.cmd.Process.Pid)))
+		w.Write(ps)
+	}
+
+	cancel()
+}
+
 // Format processes into pgid and ppid groups for summation of memory usage
 func parseProcs(procs []string) (allProcesses map[int]ProcInfo, processGroups map[int][]ProcInfo,
 	parentProcesses map[int][]ProcInfo, err error) {

runner/runners/invoke.go

@@ -329,7 +329,7 @@ func (inv *Invoker) run(cmd *runner.Command, id runner.RunID, abortCh chan struc
 				"tag":    cmd.Tag,
 				"jobID":  cmd.JobID,
 				"taskID": cmd.TaskID,
-			}).Info("Run timedout")
+			}).Error("Run timedout")
 		runStatus = runner.TimeoutStatus(id,
 			tags.LogTags{JobID: cmd.JobID, TaskID: cmd.TaskID, Tag: cmd.Tag})
 	case st := <-memCh:
@@ -345,7 +345,7 @@ func (inv *Invoker) run(cmd *runner.Command, id runner.RunID, abortCh chan struc
 				"taskID":   cmd.TaskID,
 				"status":   st,
 				"checkout": co.Path(),
-			}).Infof(st.Error)
+			}).Errorf(st.Error)
 		inv.stat.Counter(stats.WorkerMemoryCapExceeded).Inc(1)
 		runStatus = getPostExecRunStatus(st, id, cmd)
 		runStatus.Error = st.Error