Repository: kata-containers/agent (go)

Issue:
k8s: vCPUs are not connected
# Description of problem

* Run a container with a CPU constraints of 3 CPUs.
* Inside the container at least 3 CPUs should be visible and connected.

*~/cpu-demo.yaml*
```yaml
apiVersion: v1
kind: Pod
metadata:
  name: cpu-demo
  namespace: sandbox
spec:
  containers:
  - name: cpu0
    image: debian
    resources:
      limits:
        cpu: "3"
    args:
    - "bash"
    tty: true
    stdin: true
```

```
$ sudo -E kubectl create -f ~/cpu-demo.yaml
```


# Expected result

```
$ nproc
4
```

# Actual result

```
$ nrpoc
1
```


Pull request description:
grpc: honour CPU constraints in Kubernetes
Once all vCPUs have been connected, cpuset cgroup MUST BE updated,
to achieve that, each cpuset cgroup parent of the container
MUST BE updated with the actual range of vCPUs.

fixes #239

Signed-off-by: Julio Montes <julio.montes@intel.com>

Patch:
diff --git a/agent.go b/agent.go
index 09105ea..6a67371 100644
--- a/agent.go
+++ b/agent.go
@@ -7,7 +7,6 @@
 package main
 
 import (
-	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -18,6 +17,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"runtime"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"syscall"
@@ -97,11 +97,6 @@ var agentLog = logrus.WithFields(agentFields)
 // version is the agent version. This variable is populated at build time.
 var version = "unknown"
 
-var debug = false
-
-// if true, coredump when an internal error occurs or a fatal signal is received
-var crashOnError = false
-
 // This is the list of file descriptors we can properly close after the process
 // has been started. When the new process is exec(), those file descriptors are
 // duplicated and it is our responsibility to close them since we have opened
@@ -364,60 +359,35 @@ func (s *sandbox) listenToUdevEvents() {
 }
 
 // This loop is meant to be run inside a separate Go routine.
-func (s *sandbox) signalHandlerLoop(sigCh chan os.Signal) {
+func (s *sandbox) reaperLoop(sigCh chan os.Signal) {
 	for sig := range sigCh {
-		logger := agentLog.WithField("signal", sig)
-
-		if sig == unix.SIGCHLD {
+		switch sig {
+		case unix.SIGCHLD:
 			if err := s.subreaper.reap(); err != nil {
-				logger.WithError(err).Error("failed to reap")
-				continue
+				agentLog.Error(err)
+				return
 			}
+		default:
+			agentLog.Infof("Unexpected signal %s, nothing to do...", sig.String())
 		}
-
-		nativeSignal, ok := sig.(syscall.Signal)
-		if !ok {
-			err := errors.New("unknown signal")
-			logger.WithError(err).Error("failed to handle signal")
-			continue
-		}
-
-		if fatalSignal(nativeSignal) {
-			logger.Error("received fatal signal")
-			die()
-		}
-
-		if debug && nonFatalSignal(nativeSignal) {
-			logger.Debug("handling signal")
-			backtrace()
-			continue
-		}
-
-		logger.Info("ignoring unexpected signal")
 	}
 }
 
-func (s *sandbox) setupSignalHandler() error {
-	// Set agent as subreaper
-	err := unix.Prctl(unix.PR_SET_CHILD_SUBREAPER, uintptr(1), 0, 0, 0)
-	if err != nil {
+func (s *sandbox) setSubreaper() error {
+	if err := unix.Prctl(unix.PR_SET_CHILD_SUBREAPER, uintptr(1), 0, 0, 0); err != nil {
 		return err
 	}
 
 	sigCh := make(chan os.Signal, 512)
 	signal.Notify(sigCh, unix.SIGCHLD)
 
-	for _, sig := range handledSignals() {
-		signal.Notify(sigCh, sig)
-	}
-
-	go s.signalHandlerLoop(sigCh)
+	go s.reaperLoop(sigCh)
 
 	return nil
 }
 
 // getMemory returns a string containing the total amount of memory reported
-// by the kernel. The string includes a suffix denoting the units the memory
+// by the kernell. The string includes a suffix denoting the units the memory
 // is measured in.
 func getMemory() (string, error) {
 	bytes, err := ioutil.ReadFile(meminfo)
@@ -674,6 +644,9 @@ func initAgentAsInit() error {
 }
 
 func init() {
+	// Force full stacktrace on internal error
+	debug.SetTraceback("system")
+
 	if len(os.Args) > 1 && os.Args[1] == "init" {
 		runtime.GOMAXPROCS(1)
 		runtime.LockOSThread()
@@ -685,7 +658,7 @@ func init() {
 	}
 }
 
-func realMain() {
+func main() {
 	var err error
 	var showVersion bool
 
@@ -730,19 +703,17 @@ func realMain() {
 	}
 
 	if err = s.initLogger(); err != nil {
-		agentLog.WithError(err).Error("failed to setup logger")
 		return
 	}
 
-	if err = s.setupSignalHandler(); err != nil {
-		agentLog.WithError(err).Error("failed to setup signal handler")
+	// Set agent as subreaper.
+	if err = s.setSubreaper(); err != nil {
 		return
 	}
 
 	// Check for vsock vs serial. This will fill the sandbox structure with
 	// information about the channel.
 	if err = s.initChannel(); err != nil {
-		agentLog.WithError(err).Error("failed to setup channels")
 		return
 	}
 
@@ -753,8 +724,3 @@ func realMain() {
 
 	s.wg.Wait()
 }
-
-func main() {
-	defer handlePanic()
-	realMain()
-}
diff --git a/config.go b/config.go
index c5e842d..226efc2 100644
--- a/config.go
+++ b/config.go
@@ -18,7 +18,6 @@ import (
 const (
 	optionPrefix      = "agent."
 	logLevelFlag      = optionPrefix + "log"
-	devModeFlag       = optionPrefix + "devmode"
 	kernelCmdlineFile = "/proc/cmdline"
 )
 
@@ -71,13 +70,6 @@ func (c *agentConfig) parseCmdlineOption(option string) error {
 		optionSeparator = "="
 	)
 
-	if option == devModeFlag {
-		crashOnError = true
-		debug = true
-
-		return nil
-	}
-
 	split := strings.Split(option, optionSeparator)
 
 	if len(split) < valuePosition+1 {
@@ -91,9 +83,6 @@ func (c *agentConfig) parseCmdlineOption(option string) error {
 			return err
 		}
 		c.logLevel = level
-		if level == logrus.DebugLevel {
-			debug = true
-		}
 	default:
 		if strings.HasPrefix(split[optionPosition], optionPrefix) {
 			return grpcStatus.Errorf(codes.NotFound, "Unknown option %s", split[optionPosition])
diff --git a/grpc.go b/grpc.go
index 21ccf27..fc251d6 100644
--- a/grpc.go
+++ b/grpc.go
@@ -17,7 +17,6 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
 	"syscall"
 	"time"
 
@@ -57,7 +56,7 @@ var (
 	sysfsCPUOnlinePath     = "/sys/devices/system/cpu"
 	sysfsMemOnlinePath     = "/sys/devices/system/memory"
 	sysfsConnectedCPUsPath = filepath.Join(sysfsCPUOnlinePath, "online")
-	sysfsDockerCpusetPath  = "/sys/fs/cgroup/cpuset/docker/cpuset.cpus"
+	sysfsCpusetPath        = "/sys/fs/cgroup/cpuset"
 )
 
 type onlineResource struct {
@@ -67,17 +66,15 @@ type onlineResource struct {
 
 var emptyResp = &gpb.Empty{}
 
-var onlineCPUMemLock sync.Mutex
-
 const onlineCPUMemWaitTime = 100 * time.Millisecond
 
 const onlineCPUMaxTries = 10
 
-const dockerCpusetMode = 0644
+const cpusetMode = 0644
 
-// handleError will log the specified error if wait is false
+// handleError will log the specified error if wait is true
 func handleError(wait bool, err error) error {
-	if !wait {
+	if wait {
 		agentLog.WithError(err).Error()
 	}
 
@@ -163,9 +160,11 @@ func (a *agentGRPC) onlineCPUMem(req *pb.OnlineCPUMemRequest) error {
 		return handleError(req.Wait, fmt.Errorf("requested number of CPUs '%d' must be greater than 0", req.NbCpus))
 	}
 
-	onlineCPUMemLock.Lock()
-	defer onlineCPUMemLock.Unlock()
+	// we are going to update the containers of the sandbox, we have to lock it
+	a.sandbox.Lock()
+	defer a.sandbox.Unlock()
 
+	agentLog.WithField("vcpus-to-connect", req.NbCpus).Debug("connecting vCPUs")
 	if err := onlineCPUResources(req.NbCpus); err != nil {
 		return handleError(req.Wait, err)
 	}
@@ -181,29 +180,51 @@ func (a *agentGRPC) onlineCPUMem(req *pb.OnlineCPUMemRequest) error {
 		return handleError(req.Wait, fmt.Errorf("Could not get the actual range of connected CPUs: %v", err))
 	}
 	connectedCpus := strings.Trim(string(cpus), "\t\n ")
+	agentLog.WithField("range-of-vcpus", connectedCpus).Debug("connecting vCPUs")
 
-	// In order to update container's cpuset cgroups, docker's cpuset cgroup MUST BE updated with
-	// the actual number of connected CPUs
-	if err := ioutil.WriteFile(sysfsDockerCpusetPath, []byte(connectedCpus), dockerCpusetMode); err != nil {
-		return handleError(req.Wait, fmt.Errorf("Could not update docker cpuset cgroup '%s': %v", connectedCpus, err))
-	}
+	mapCgroupPaths := make(map[string]bool)
 
 	// Now that we know the actual range of connected CPUs, we need to iterate over
 	// all containers an update each cpuset cgroup. This is not required in docker
 	// containers since they don't hot add/remove CPUs.
 	for _, c := range a.sandbox.containers {
+		agentLog.WithField("container", c.container.ID()).Debug("updating cpuset cgroup")
 		contConfig := c.container.Config()
+
 		// Don't update cpuset cgroup if one was already defined.
 		if contConfig.Cgroups.Resources.CpusetCpus != "" {
+			agentLog.WithField("cpuset", contConfig.Cgroups.Resources.CpusetCpus).Debug("cpuset value is not empty")
 			continue
 		}
-		contConfig.Cgroups.Resources = &configs.Resources{
-			CpusetCpus: connectedCpus,
-		}
-		if err := c.container.Set(contConfig); err != nil {
-			return handleError(req.Wait, err)
-		}
 
+		// Each cpuset cgroup MUST BE updated with the actual number of vCPUs.
+		cpusetPath := sysfsCpusetPath
+		cgroupsPaths := strings.Split(contConfig.Cgroups.Path, "/")
+		for _, path := range cgroupsPaths {
+			// Skip if empty.
+			if path == "" {
+				continue
+			}
+
+			cpusetPath = filepath.Join(cpusetPath, path)
+
+			// check if the cgroup was already updated.
+			if mapCgroupPaths[cpusetPath] == true {
+				agentLog.WithField("path", cpusetPath).Debug("cpuset cgroup already updated")
+				continue
+			}
+
+			// Don't use c.container.Set because of it will modify container's config.
+			// c.container.Set MUST BE used only on update.
+			cpusetCpusPath := filepath.Join(cpusetPath, "cpuset.cpus")
+			agentLog.WithField("path", cpusetPath).Debug("updating cpuset cgroup")
+			if err := ioutil.WriteFile(cpusetCpusPath, []byte(connectedCpus), cpusetMode); err != nil {
+				return handleError(req.Wait, fmt.Errorf("Could not update cpuset cgroup '%s': %v", connectedCpus, err))
+			}
+
+			// add cgroup path to the map.
+			mapCgroupPaths[cpusetPath] = true
+		}
 	}
 
 	return nil
diff --git a/mockcontainer.go b/mockcontainer.go
index bcc5fb4..cbda967 100644
--- a/mockcontainer.go
+++ b/mockcontainer.go
@@ -7,6 +7,7 @@
 package main
 
 import (
+	"fmt"
 	"os"
 
 	"github.com/opencontainers/runc/libcontainer"
@@ -37,6 +38,7 @@ func (m *mockContainer) Config() configs.Config {
 		Capabilities: &configs.Capabilities{},
 		Cgroups: &configs.Cgroup{
 			Resources: &configs.Resources{},
+			Path:      fmt.Sprintf("/cgroup/%s", m.id),
 		},
 		Seccomp: &configs.Seccomp{},
 	}
diff --git a/signals.go b/signals.go
deleted file mode 100644
index 1be77d4..0000000
--- a/signals.go
+++ /dev/null
@@ -1,97 +0,0 @@
-// Copyright 2018 Intel Corporation.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-
-package main
-
-import (
-	"bytes"
-	"fmt"
-	"os"
-	"os/signal"
-	"runtime/pprof"
-	"strings"
-	"syscall"
-)
-
-// List of handled signals.
-//
-// The value is true if receiving the signal should be fatal.
-var handledSignalsMap = map[syscall.Signal]bool{
-	syscall.SIGABRT:   true,
-	syscall.SIGBUS:    true,
-	syscall.SIGILL:    true,
-	syscall.SIGQUIT:   true,
-	syscall.SIGSEGV:   true,
-	syscall.SIGSTKFLT: true,
-	syscall.SIGSYS:    true,
-	syscall.SIGTRAP:   true,
-	syscall.SIGUSR1:   false,
-}
-
-func handlePanic() {
-	r := recover()
-
-	if r != nil {
-		msg := fmt.Sprintf("%s", r)
-		agentLog.WithField("panic", msg).Error("fatal error")
-
-		die()
-	}
-}
-
-func backtrace() {
-	profiles := pprof.Profiles()
-
-	buf := &bytes.Buffer{}
-
-	for _, p := range profiles {
-		// The magic number requests a full stacktrace. See
-		// https://golang.org/pkg/runtime/pprof/#Profile.WriteTo.
-		pprof.Lookup(p.Name()).WriteTo(buf, 2)
-	}
-
-	for _, line := range strings.Split(buf.String(), "\n") {
-		agentLog.Error(line)
-	}
-}
-
-func fatalSignal(sig syscall.Signal) bool {
-	s, exists := handledSignalsMap[sig]
-	if !exists {
-		return false
-	}
-
-	return s
-}
-
-func nonFatalSignal(sig syscall.Signal) bool {
-	s, exists := handledSignalsMap[sig]
-	if !exists {
-		return false
-	}
-
-	return !s
-}
-
-func handledSignals() []syscall.Signal {
-	var signals []syscall.Signal
-
-	for sig := range handledSignalsMap {
-		signals = append(signals, sig)
-	}
-
-	return signals
-}
-
-func die() {
-	backtrace()
-
-	if crashOnError {
-		signal.Reset(syscall.SIGABRT)
-		syscall.Kill(0, syscall.SIGABRT)
-	}
-
-	os.Exit(1)
-}


Test patch:
diff --git a/config_test.go b/config_test.go
index ad68a09..ca115ed 100644
--- a/config_test.go
+++ b/config_test.go
@@ -71,15 +71,10 @@ func TestParseCmdlineOptionCorrectOptions(t *testing.T) {
 	logFlagList := []string{"debug", "info", "warn", "error", "fatal", "panic"}
 
 	for _, logFlag := range logFlagList {
-		debug = false
 		option := logLevelFlag + "=" + logFlag
 
 		err := a.parseCmdlineOption(option)
 		assert.NoError(err, "%v", err)
-
-		if logFlag == "debug" {
-			assert.True(debug)
-		}
 	}
 }
 
@@ -98,45 +93,6 @@ func TestParseCmdlineOptionIncorrectOptions(t *testing.T) {
 	}
 }
 
-func TestParseCmdlineOptionDevMode(t *testing.T) {
-	assert := assert.New(t)
-
-	a := &agentConfig{}
-
-	type testData struct {
-		option               string
-		expectDevModeEnabled bool
-	}
-
-	data := []testData{
-		{"agent.Devmode", false},
-		{"agent.DevMode", false},
-		{"devmode", false},
-		{"DevMode", false},
-		{"agent.devmodel", false},
-		{"agent.devmode.", false},
-		{"agent.devmode-", false},
-		{"agent.devmode:", false},
-
-		{"agent.devmode", true},
-	}
-
-	for i, d := range data {
-		debug = false
-		crashOnError = false
-
-		err := a.parseCmdlineOption(d.option)
-		assert.NoError(err)
-
-		if !d.expectDevModeEnabled {
-			continue
-		}
-
-		assert.True(debug, "test %d (%+v)", i, d)
-		assert.True(crashOnError, "test %d (%+v)", i, d)
-	}
-}
-
 func TestGetConfigEmptyFileName(t *testing.T) {
 	assert := assert.New(t)
 
diff --git a/grpc_test.go b/grpc_test.go
index 99b6594..41a39bb 100644
--- a/grpc_test.go
+++ b/grpc_test.go
@@ -8,7 +8,6 @@ package main
 
 import (
 	"context"
-	"errors"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -117,6 +116,17 @@ func TestOnlineCPUMem(t *testing.T) {
 		},
 	}
 
+	containerID := "1"
+	containerID2 := "2"
+	container := &container{
+		container: &mockContainer{
+			id:        containerID,
+			processes: []int{1},
+		},
+	}
+	a.sandbox.containers[containerID] = container
+	a.sandbox.containers[containerID2] = container
+
 	req := &pb.OnlineCPUMemRequest{
 		NbCpus: 1,
 		Wait:   true,
@@ -166,10 +176,13 @@ func TestOnlineCPUMem(t *testing.T) {
 	_, err = a.OnlineCPUMem(context.TODO(), req)
 	assert.Error(err, "docker cgroup path does not exist")
 
-	dockerCpusetPath, err := ioutil.TempDir("", "docker")
+	sysfsCpusetPath, err = ioutil.TempDir("", "cgroup")
+	assert.NoError(err)
+	cfg := container.container.Config()
+	cgroupPath := filepath.Join(sysfsCpusetPath, cfg.Cgroups.Path)
+	err = os.MkdirAll(cgroupPath, 0777)
 	assert.NoError(err)
-	defer os.RemoveAll(dockerCpusetPath)
-	sysfsDockerCpusetPath = filepath.Join(dockerCpusetPath, "cpuset.cpus")
+	defer os.RemoveAll(sysfsCpusetPath)
 
 	err = ioutil.WriteFile(memory0Online, []byte("0"), 0755)
 	assert.NoError(err)
@@ -322,20 +335,3 @@ func TestStatsContainer(t *testing.T) {
 	assert.NotNil(r)
 
 }
-
-func TestHandleError(t *testing.T) {
-	assert := assert.New(t)
-
-	err := errors.New("")
-	e := handleError(true, err)
-	assert.Error(e)
-
-	e = handleError(true, nil)
-	assert.NoError(e)
-
-	e = handleError(false, err)
-	assert.Error(e)
-
-	e = handleError(false, nil)
-	assert.NoError(e)
-}
diff --git a/signals_test.go b/signals_test.go
deleted file mode 100644
index d33ae77..0000000
--- a/signals_test.go
+++ /dev/null
@@ -1,134 +0,0 @@
-// Copyright (c) 2018 Intel Corporation
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-
-package main
-
-import (
-	"bytes"
-	"reflect"
-	goruntime "runtime"
-	"sort"
-	"strings"
-	"syscall"
-	"testing"
-
-	"github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestSignalFatalSignal(t *testing.T) {
-	assert := assert.New(t)
-
-	for sig, fatal := range handledSignalsMap {
-		result := nonFatalSignal(sig)
-		if fatal {
-			assert.False(result)
-		} else {
-			assert.True(result)
-		}
-	}
-}
-
-func TestSignalHandledSignalsMap(t *testing.T) {
-	assert := assert.New(t)
-
-	for sig, fatal := range handledSignalsMap {
-		result := fatalSignal(sig)
-		if fatal {
-			assert.True(result)
-		} else {
-			assert.False(result)
-		}
-	}
-}
-
-func TestSignalHandledSignals(t *testing.T) {
-	assert := assert.New(t)
-
-	var expected []syscall.Signal
-
-	for sig := range handledSignalsMap {
-		expected = append(expected, sig)
-	}
-
-	got := handledSignals()
-
-	sort.Slice(expected, func(i, j int) bool {
-		return int(expected[i]) < int(expected[j])
-	})
-
-	sort.Slice(got, func(i, j int) bool {
-		return int(got[i]) < int(got[j])
-	})
-
-	assert.True(reflect.DeepEqual(expected, got))
-}
-
-func TestSignalNonFatalSignal(t *testing.T) {
-	assert := assert.New(t)
-
-	for sig, fatal := range handledSignalsMap {
-		result := nonFatalSignal(sig)
-		if fatal {
-			assert.False(result)
-		} else {
-			assert.True(result)
-		}
-	}
-}
-
-func TestSignalFatalSignalInvalidSignal(t *testing.T) {
-	assert := assert.New(t)
-
-	sig := syscall.SIGXCPU
-
-	result := fatalSignal(sig)
-	assert.False(result)
-}
-
-func TestSignalNonFatalSignalInvalidSignal(t *testing.T) {
-	assert := assert.New(t)
-
-	sig := syscall.SIGXCPU
-
-	result := nonFatalSignal(sig)
-	assert.False(result)
-}
-
-func TestSignalBacktrace(t *testing.T) {
-	assert := assert.New(t)
-
-	// create buffer to save logger output
-	buf := &bytes.Buffer{}
-
-	savedLog := agentLog
-	defer func() {
-		agentLog = savedLog
-	}()
-
-	agentLog = logrus.WithField("test-agent-logger", true)
-
-	agentLog.Logger.Formatter = &logrus.TextFormatter{
-		DisableColors: true,
-	}
-
-	// capture output to buffer
-	agentLog.Logger.Out = buf
-
-	// determine name of *this* function
-	pc := make([]uintptr, 1)
-	goruntime.Callers(1, pc)
-	fn := goruntime.FuncForPC(pc[0])
-	name := fn.Name()
-
-	backtrace()
-
-	b := buf.String()
-
-	// very basic tests to check if a backtrace was produced
-	assert.True(strings.Contains(b, "contention:"))
-	assert.True(strings.Contains(b, `level=error`))
-	assert.True(strings.Contains(b, name))
-}