Merge pull request #5951 from hashicorp/b-cpu-pid-scan

reduce CPU usage running large numbers of clients

drivers/shared/executor/executor.go

@@ -312,7 +312,7 @@ func (e *UniversalExecutor) Launch(command *ExecCommand) (*ProcessState, error)
 		return nil, fmt.Errorf("failed to start command path=%q --- args=%q: %v", path, e.childCmd.Args, err)
 	}
 
-	go e.pidCollector.collectPids(e.processExited, getAllPids)
+	go e.pidCollector.collectPids(e.processExited, e.getAllPids)
 	go e.wait()
 	return &ProcessState{Pid: e.childCmd.Process.Pid, ExitCode: -1, Time: time.Now()}, nil
 }

drivers/shared/executor/executor_basic.go

@@ -15,3 +15,7 @@ func NewExecutorWithIsolation(logger hclog.Logger) Executor {
 func (e *UniversalExecutor) configureResourceContainer(_ int) error { return nil }
 
 func (e *UniversalExecutor) runAs(_ string) error { return nil }
+
+func (e *UniversalExecutor) getAllPids() (map[int]*nomadPid, error) {
+	return getAllPidsByScanning()
+}

drivers/shared/executor/executor_universal_linux.go

@@ -75,11 +75,24 @@ func (e *UniversalExecutor) configureResourceContainer(pid int) error {
 		},
 	}
 
-	configureBasicCgroups(cfg)
+	err := configureBasicCgroups(cfg)
+	if err != nil {
+		e.logger.Warn("failed to create cgroup", "error", err)
+		return err
+	}
+
 	e.resConCtx.groups = cfg.Cgroups
 	return cgroups.EnterPid(cfg.Cgroups.Paths, pid)
 }
 
+func (e *UniversalExecutor) getAllPids() (map[int]*nomadPid, error) {
+	if e.resConCtx.isEmpty() {
+		return getAllPidsByScanning()
+	} else {
+		return e.resConCtx.getAllPidsByCgroup()
+	}
+}
+
 // DestroyCgroup kills all processes in the cgroup and removes the cgroup
 // configuration from the host. This function is idempotent.
 func DestroyCgroup(groups *lconfigs.Cgroup, executorPid int) error {

drivers/shared/executor/pid_collector.go

@@ -210,7 +210,7 @@ func aggregatedResourceUsage(systemCpuStats *stats.CpuStats, pidStats map[string
 	}
 }
 
-func getAllPids() (map[int]*nomadPid, error) {
+func getAllPidsByScanning() (map[int]*nomadPid, error) {
 	allProcesses, err := ps.Processes()
 	if err != nil {
 		return nil, err

drivers/shared/executor/resource_container_linux.go

@@ -4,6 +4,8 @@ import (
 	"os"
 	"sync"
 
+	"github.com/hashicorp/nomad/client/stats"
+	"github.com/opencontainers/runc/libcontainer/cgroups"
 	cgroupConfig "github.com/opencontainers/runc/libcontainer/configs"
 )
 
@@ -23,3 +25,38 @@ func (rc *resourceContainerContext) executorCleanup() error {
 	}
 	return nil
 }
+
+func (rc *resourceContainerContext) isEmpty() bool {
+	return rc.groups == nil
+}
+
+func (rc *resourceContainerContext) getAllPidsByCgroup() (map[int]*nomadPid, error) {
+	nPids := map[int]*nomadPid{}
+
+	if rc.groups == nil {
+		return nPids, nil
+	}
+
+	var path string
+	if p, ok := rc.groups.Paths["freezer"]; ok {
+		path = p
+	} else {
+		path = rc.groups.Path
+	}
+
+	pids, err := cgroups.GetAllPids(path)
+	if err != nil {
+		return nPids, err
+	}
+
+	for _, pid := range pids {
+		nPids[pid] = &nomadPid{
+			pid:           pid,
+			cpuStatsTotal: stats.NewCpuStats(),
+			cpuStatsUser:  stats.NewCpuStats(),
+			cpuStatsSys:   stats.NewCpuStats(),
+		}
+	}
+
+	return nPids, nil
+}