Merge pull request cri-o#5514 from haircommander/taskset-prepend

prepend commands with taskset if InfraCtrCPUSet is configured

Makefile

@@ -311,6 +311,7 @@ testunit-bin:
 	done
 
 mockgen: \
+	mock-cmdrunner \
 	mock-containerstorage \
 	mock-criostorage \
 	mock-lib-config \

internal/config/conmonmgr/conmonmgr.go

@@ -19,14 +19,10 @@ type ConmonManager struct {
 
 // this function is heavily based on github.com/containers/common#probeConmon
 func New(conmonPath string) (*ConmonManager, error) {
-	return newWithCommandRunner(conmonPath, &cmdrunner.RealCommandRunner{})
-}
-
-func newWithCommandRunner(conmonPath string, runner cmdrunner.CommandRunner) (*ConmonManager, error) {
 	if !path.IsAbs(conmonPath) {
 		return nil, errors.Errorf("conmon path is not absolute: %s", conmonPath)
 	}
-	out, err := runner.CombinedOutput(conmonPath, "--version")
+	out, err := cmdrunner.CombinedOutput(conmonPath, "--version")
 	if err != nil {
 		return nil, errors.Wrapf(err, "get conmon version")
 	}

internal/config/conmonmgr/conmonmgr_test.go

@@ -2,6 +2,7 @@ package conmonmgr
 
 import (
 	runnerMock "github.com/cri-o/cri-o/test/mocks/cmdrunner"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/golang/mock/gomock"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -16,6 +17,7 @@ var _ = t.Describe("ConmonManager", func() {
 	t.Describe("New", func() {
 		BeforeEach(func() {
 			runner = runnerMock.NewMockCommandRunner(mockCtrl)
+			cmdrunner.SetMocked(runner)
 		})
 		It("should fail when path not absolute", func() {
 			// Given
@@ -24,7 +26,7 @@ var _ = t.Describe("ConmonManager", func() {
 			)
 
 			// When
-			mgr, err := newWithCommandRunner("", runner)
+			mgr, err := New("")
 
 			// Then
 			Expect(err).NotTo(BeNil())
@@ -37,7 +39,7 @@ var _ = t.Describe("ConmonManager", func() {
 			)
 
 			// When
-			mgr, err := newWithCommandRunner(validPath, runner)
+			mgr, err := New(validPath)
 
 			// Then
 			Expect(err).NotTo(BeNil())
@@ -50,7 +52,7 @@ var _ = t.Describe("ConmonManager", func() {
 			)
 
 			// When
-			mgr, err := newWithCommandRunner(validPath, runner)
+			mgr, err := New(validPath)
 
 			// Then
 			Expect(err).NotTo(BeNil())
@@ -63,7 +65,7 @@ var _ = t.Describe("ConmonManager", func() {
 			)
 
 			// When
-			mgr, err := newWithCommandRunner(validPath, runner)
+			mgr, err := New(validPath)
 
 			// Then
 			Expect(err).To(BeNil())
@@ -76,7 +78,7 @@ var _ = t.Describe("ConmonManager", func() {
 			)
 
 			// When
-			mgr, err := newWithCommandRunner(validPath, runner)
+			mgr, err := New(validPath)
 
 			// Then
 			Expect(err).To(BeNil())

internal/config/node/systemd.go

@@ -4,9 +4,9 @@
 package node
 
 import (
-	"os/exec"
 	"sync"
 
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/pkg/errors"
 )
 
@@ -37,7 +37,7 @@ func SystemdHasAllowedCPUs() bool {
 // systemdSupportsProperty checks whether systemd supports a property
 // It returns an error if it does not.
 func systemdSupportsProperty(property string) (bool, error) {
-	output, err := exec.Command("systemctl", "show", "-p", property, "systemd").Output()
+	output, err := cmdrunner.Command("systemctl", "show", "-p", property, "systemd").Output()
 	if err != nil {
 		return false, errors.Wrapf(err, "check systemd %s", property)
 	}

internal/config/nsmgr/nsmgr.go

@@ -4,14 +4,14 @@ import (
 	"bytes"
 	"fmt"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"strings"
 	"syscall"
 
 	nspkg "github.com/containernetworking/plugins/pkg/ns"
 	"github.com/containers/storage/pkg/idtools"
 	"github.com/cri-o/cri-o/utils"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -120,7 +120,7 @@ func (mgr *NamespaceManager) NewPodNamespaces(cfg *PodNamespacesConfig) ([]Names
 	}
 
 	logrus.Debugf("Calling pinns with %v", pinnsArgs)
-	output, err := exec.Command(mgr.pinnsPath, pinnsArgs...).CombinedOutput()
+	output, err := cmdrunner.Command(mgr.pinnsPath, pinnsArgs...).CombinedOutput()
 	if err != nil {
 		logrus.Warnf("Pinns %v failed: %s (%v)", pinnsArgs, string(output), err)
 		// cleanup the mounts

internal/dbusmgr/user.go

@@ -6,12 +6,12 @@ import (
 	"bufio"
 	"bytes"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"strconv"
 	"strings"
 
 	systemdDbus "github.com/coreos/go-systemd/v22/dbus"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	dbus "github.com/godbus/dbus/v5"
 	"github.com/opencontainers/runc/libcontainer/userns"
 	"github.com/pkg/errors"
@@ -55,7 +55,7 @@ func DetectUID() (int, error) {
 	if !userns.RunningInUserNS() {
 		return os.Getuid(), nil
 	}
-	b, err := exec.Command("busctl", "--user", "--no-pager", "status").CombinedOutput()
+	b, err := cmdrunner.Command("busctl", "--user", "--no-pager", "status").CombinedOutput()
 	if err != nil {
 		return -1, errors.Wrapf(err, "could not execute `busctl --user --no-pager status`: %q", string(b))
 	}
@@ -91,7 +91,7 @@ func DetectUserDbusSessionBusAddress() (string, error) {
 			return busAddress, nil
 		}
 	}
-	b, err := exec.Command("systemctl", "--user", "--no-pager", "show-environment").CombinedOutput()
+	b, err := cmdrunner.Command("systemctl", "--user", "--no-pager", "show-environment").CombinedOutput()
 	if err != nil {
 		return "", errors.Wrapf(err, "could not execute `systemctl --user --no-pager show-environment`, output=%q", string(b))
 	}

internal/oci/runtime_oci.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cri-o/cri-o/pkg/config"
 	"github.com/cri-o/cri-o/server/metrics"
 	"github.com/cri-o/cri-o/utils"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/fsnotify/fsnotify"
 	json "github.com/json-iterator/go"
 	rspec "github.com/opencontainers/runtime-spec/specs-go"
@@ -137,7 +138,7 @@ func (r *runtimeOCI) CreateContainer(ctx context.Context, c *Container, cgroupPa
 		"args": args,
 	}).Debugf("running conmon: %s", r.config.Conmon)
 
-	cmd := exec.Command(r.config.Conmon, args...) // nolint: gosec
+	cmd := cmdrunner.Command(r.config.Conmon, args...) // nolint: gosec
 	cmd.Dir = c.bundlePath
 	cmd.SysProcAttr = sysProcAttrPlatform()
 	cmd.Stdin = os.Stdin
@@ -353,7 +354,7 @@ func (r *runtimeOCI) ExecContainer(ctx context.Context, c *Container, cmd []stri
 
 	args := []string{rootFlag, r.root, "exec"}
 	args = append(args, "--process", processFile, c.ID())
-	execCmd := exec.Command(r.path, args...) // nolint: gosec
+	execCmd := cmdrunner.Command(r.path, args...) // nolint: gosec
 	if v, found := os.LookupEnv("XDG_RUNTIME_DIR"); found {
 		execCmd.Env = append(execCmd.Env, fmt.Sprintf("XDG_RUNTIME_DIR=%s", v))
 	}
@@ -487,7 +488,7 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
 		"--exec-process-spec", processFile,
 		"--runtime-arg", fmt.Sprintf("%s=%s", rootFlag, r.root))
 
-	cmd := exec.Command(r.config.Conmon, args...) // nolint: gosec
+	cmd := cmdrunner.Command(r.config.Conmon, args...) // nolint: gosec
 
 	var stdoutBuf, stderrBuf bytes.Buffer
 	cmd.Stdout = &stdoutBuf
@@ -598,7 +599,7 @@ func (r *runtimeOCI) UpdateContainer(ctx context.Context, c *Container, res *rsp
 		return nil
 	}
 
-	cmd := exec.Command(r.path, rootFlag, r.root, "update", "--resources", "-", c.ID()) // nolint: gosec
+	cmd := cmdrunner.Command(r.path, rootFlag, r.root, "update", "--resources", "-", c.ID()) // nolint: gosec
 	var stdout bytes.Buffer
 	var stderr bytes.Buffer
 	cmd.Stdout = &stdout
@@ -811,7 +812,7 @@ func (r *runtimeOCI) UpdateContainerStatus(ctx context.Context, c *Container) er
 	}
 
 	stateCmd := func() (*ContainerState, bool, error) {
-		cmd := exec.Command(r.path, rootFlag, r.root, "state", c.ID()) // nolint: gosec
+		cmd := cmdrunner.Command(r.path, rootFlag, r.root, "state", c.ID()) // nolint: gosec
 		if v, found := os.LookupEnv("XDG_RUNTIME_DIR"); found {
 			cmd.Env = append(cmd.Env, fmt.Sprintf("XDG_RUNTIME_DIR=%s", v))
 		}

internal/runtimehandlerhooks/high_performance_hooks.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cri-o/cri-o/internal/log"
 	"github.com/cri-o/cri-o/internal/oci"
 	crioannotations "github.com/cri-o/cri-o/pkg/annotations"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/opencontainers/runc/libcontainer/cgroups"
 	"github.com/opencontainers/runc/libcontainer/cgroups/systemd"
 	"github.com/pkg/errors"
@@ -263,7 +264,7 @@ func setIRQLoadBalancing(c *oci.Container, enable bool, irqSmpAffinityFile, irqB
 			return nil
 		}
 		// run irqbalance in daemon mode, so this won't cause delay
-		cmd := exec.Command(irqBalancedName, "--oneshot")
+		cmd := cmdrunner.Command(irqBalancedName, "--oneshot")
 		additionalEnv := irqBalanceBannedCpus + "=" + newIRQBalanceSetting
 		cmd.Env = append(os.Environ(), additionalEnv)
 		return cmd.Run()

internal/runtimehandlerhooks/utils.go

@@ -5,10 +5,10 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
-	"os/exec"
 	"strings"
 	"unicode"
 
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/sirupsen/logrus"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpuset"
 )
@@ -138,11 +138,11 @@ func UpdateIRQSmpAffinityMask(cpus, current string, set bool) (cpuMask, bannedCP
 }
 
 func restartIrqBalanceService() error {
-	return exec.Command("systemctl", "restart", "irqbalance").Run()
+	return cmdrunner.Command("systemctl", "restart", "irqbalance").Run()
 }
 
 func isServiceEnabled(serviceName string) bool {
-	cmd := exec.Command("systemctl", "is-enabled", serviceName)
+	cmd := cmdrunner.Command("systemctl", "is-enabled", serviceName)
 	status, err := cmd.CombinedOutput()
 	if err != nil {
 		logrus.Infof("Service %s is-enabled check returned with: %v", serviceName, err)

pkg/config/config.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cri-o/cri-o/server/metrics/collectors"
 	"github.com/cri-o/cri-o/server/useragent"
 	"github.com/cri-o/cri-o/utils"
+	"github.com/cri-o/cri-o/utils/cmdrunner"
 	"github.com/cri-o/ocicni/pkg/ocicni"
 	selinux "github.com/opencontainers/selinux/go-selinux"
 	"github.com/pkg/errors"
@@ -53,6 +54,7 @@ const (
 	defaultCtrStopTimeout      = 30 // seconds
 	defaultNamespacesDir       = "/var/run"
 	RuntimeTypeVMBinaryPattern = "containerd-shim-([a-zA-Z0-9\\-\\+])+-v2"
+	tasksetBinary              = "taskset"
 )
 
 // Config represents the entire set of configuration values that can be set for
@@ -969,9 +971,16 @@ func (c *RuntimeConfig) Validate(systemContext *types.SystemContext, onExecution
 	}
 
 	if c.InfraCtrCPUSet != "" {
-		if _, err := cpuset.Parse(c.InfraCtrCPUSet); err != nil {
+		set, err := cpuset.Parse(c.InfraCtrCPUSet)
+		if err != nil {
 			return errors.Wrap(err, "invalid infra_ctr_cpuset")
 		}
+
+		executable, err := exec.LookPath(tasksetBinary)
+		if err != nil {
+			return errors.Wrapf(err, "%q not found in $PATH", tasksetBinary)
+		}
+		cmdrunner.PrependCommandsWith(executable, "--cpu-list", set.String())
 	}
 
 	if err := c.Workloads.Validate(); err != nil {

utils/cmdrunner/cmdrunner.go

@@ -4,12 +4,68 @@ import (
 	"os/exec"
 )
 
+// Use a singleton instance because there are many modules that may want access
+// and having it all go through a shared object like the config or server would
+// add a lot of complexity.
+var commandRunner CommandRunner
+
+// CommandRunner is an interface for executing commands.
+// It gives the option to change the way commands are run server-wide.
 type CommandRunner interface {
+	Command(string, ...string) *exec.Cmd
 	CombinedOutput(string, ...string) ([]byte, error)
 }
 
-type RealCommandRunner struct{}
+// prependableCommandRunner is an implementation of CommandRunner.
+// It gives the option for all commands that are run to be prepended by another command
+// and arguments.
+type prependableCommandRunner struct {
+	prependCmd  string
+	prependArgs []string
+}
+
+// PrependCommandsWith updates the commandRunner singleton to have the configured prepended args and command.
+func PrependCommandsWith(prependCmd string, prependArgs ...string) {
+	commandRunner = &prependableCommandRunner{
+		prependCmd:  prependCmd,
+		prependArgs: prependArgs,
+	}
+}
+
+// CombinedOutput calls CombinedOutput on the defined commandRunner,
+// or the default implementation in the exec package if there's no commandRunner defined.
+func CombinedOutput(command string, args ...string) ([]byte, error) {
+	if commandRunner == nil {
+		return exec.Command(command, args...).CombinedOutput()
+	}
+	return commandRunner.CombinedOutput(command, args...)
+}
+
+// CombinedOutput returns the combined output of the command, given the prepended cmd/args that were defined.
+func (c *prependableCommandRunner) CombinedOutput(command string, args ...string) ([]byte, error) {
+	return c.Command(command, args...).CombinedOutput()
+}
+
+// Command calls Command on the defined commandRunner,
+// or the default implementation in the exec package if there's no commandRunner defined.
+func Command(cmd string, args ...string) *exec.Cmd {
+	if commandRunner == nil {
+		return exec.Command(cmd, args...)
+	}
+	return commandRunner.Command(cmd, args...)
+}
 
-func (c *RealCommandRunner) CombinedOutput(command string, args ...string) ([]byte, error) {
-	return exec.Command(command, args...).CombinedOutput()
+// Command creates an exec.Cmd object. If prependCmd is defined, the command will be prependCmd
+// and the args will be prependArgs + cmd + args.
+// Otherwise, cmd and args will be as inputted.
+func (c *prependableCommandRunner) Command(cmd string, args ...string) *exec.Cmd {
+	realCmd := cmd
+	realArgs := args
+	if c.prependCmd != "" {
+		realCmd = c.prependCmd
+		realArgs = c.prependArgs
+		realArgs = append(realArgs, cmd)
+		realArgs = append(realArgs, args...)
+	}
+	return exec.Command(realCmd, realArgs...)
 }