cpu manager: specify the container CPU set during the creation

We can set the container cpuset.cpus diring the creation and it
will not need to call to update resources after the container creation.

Additional side effect of the change, that the runc process that responsible
to create the container will run with the same CPU affinity because the
runc runs on the cpuset provided in the config.json arg.

It will allow to prevent undesirable interupts on isolated CPUs.

Signed-off-by: Artyom Lukianov <alukiano@redhat.com>

pkg/kubelet/cm/BUILD

@@ -15,6 +15,9 @@ go_library(
         "helpers_linux.go",
         "helpers_unsupported.go",
         "internal_container_lifecycle.go",
+        "internal_container_lifecycle_linux.go",
+        "internal_container_lifecycle_unsupported.go",
+        "internal_container_lifecycle_windows.go",
         "node_container_manager_linux.go",
         "pod_container_manager_linux.go",
         "pod_container_manager_stub.go",
@@ -42,6 +45,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/cri-api/pkg/apis:go_default_library",
+        "//staging/src/k8s.io/cri-api/pkg/apis/runtime/v1alpha2:go_default_library",
         "//staging/src/k8s.io/kubelet/pkg/apis/podresources/v1:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
     ] + select({

pkg/kubelet/cm/cpumanager/cpu_manager.go

@@ -60,10 +60,9 @@ type Manager interface {
 	// e.g. at pod admission time.
 	Allocate(pod *v1.Pod, container *v1.Container) error
 
-	// AddContainer is called between container create and container start
+	// AddContainer adds the mapping between container ID to pod UID and the container name
-	// so that initial CPU affinity settings can be written through to the
+	// The mapping used to remove the CPU allocation during the container removal
-	// container runtime before the first process begins to execute.
+	AddContainer(p *v1.Pod, c *v1.Container, containerID string)
-	AddContainer(p *v1.Pod, c *v1.Container, containerID string) error
 
 	// RemoveContainer is called after Kubelet decides to kill or delete a
 	// container. After this call, the CPU manager stops trying to reconcile
@@ -237,29 +236,10 @@ func (m *manager) Allocate(p *v1.Pod, c *v1.Container) error {
 	return nil
 }
 
-func (m *manager) AddContainer(p *v1.Pod, c *v1.Container, containerID string) error {
+func (m *manager) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) {
 	m.Lock()
-	// Get the CPUs assigned to the container during Allocate()
+	defer m.Unlock()
-	// (or fall back to the default CPUSet if none were assigned).
+	m.containerMap.Add(string(pod.UID), container.Name, containerID)
-	cpus := m.state.GetCPUSetOrDefault(string(p.UID), c.Name)
-	m.Unlock()
-
-	if !cpus.IsEmpty() {
-		err := m.updateContainerCPUSet(containerID, cpus)
-		if err != nil {
-			klog.Errorf("[cpumanager] AddContainer error: error updating CPUSet for container (pod: %s, container: %s, container id: %s, err: %v)", p.Name, c.Name, containerID, err)
-			m.Lock()
-			err := m.policyRemoveContainerByRef(string(p.UID), c.Name)
-			if err != nil {
-				klog.Errorf("[cpumanager] AddContainer rollback state error: %v", err)
-			}
-			m.Unlock()
-		}
-		return err
-	}
-
-	klog.V(5).Infof("[cpumanager] update container resources is skipped due to cpu set is empty")
-	return nil
 }
 
 func (m *manager) RemoveContainer(containerID string) error {

pkg/kubelet/cm/cpumanager/cpu_manager_test.go

@@ -252,14 +252,6 @@ func TestCPUManagerAdd(t *testing.T) {
 			expAllocateErr:     fmt.Errorf("fake reg error"),
 			expAddContainerErr: nil,
 		},
-		{
-			description:        "cpu manager add - container update error",
-			updateErr:          fmt.Errorf("fake update error"),
-			policy:             testPolicy,
-			expCPUSet:          cpuset.NewCPUSet(1, 2, 3, 4),
-			expAllocateErr:     nil,
-			expAddContainerErr: fmt.Errorf("fake update error"),
-		},
 	}
 
 	for _, testCase := range testCases {
@@ -287,7 +279,8 @@ func TestCPUManagerAdd(t *testing.T) {
 				testCase.description, testCase.expAllocateErr, err)
 		}
 
-		err = mgr.AddContainer(pod, container, "fakeID")
+		mgr.AddContainer(pod, container, "fakeID")
+		_, _, err = mgr.containerMap.GetContainerRef("fakeID")
 		if !reflect.DeepEqual(err, testCase.expAddContainerErr) {
 			t.Errorf("CPU Manager AddContainer() error (%v). expected error: %v but got: %v",
 				testCase.description, testCase.expAddContainerErr, err)
@@ -520,7 +513,9 @@ func TestCPUManagerAddWithInitContainers(t *testing.T) {
 				t.Errorf("StaticPolicy Allocate() error (%v). unexpected error for container id: %v: %v",
 					testCase.description, containerIDs[i], err)
 			}
-			err = mgr.AddContainer(testCase.pod, &containers[i], containerIDs[i])
+
+			mgr.AddContainer(testCase.pod, &containers[i], containerIDs[i])
+			_, _, err = mgr.containerMap.GetContainerRef(containerIDs[i])
 			if err != nil {
 				t.Errorf("StaticPolicy AddContainer() error (%v). unexpected error for container id: %v: %v",
 					testCase.description, containerIDs[i], err)
@@ -1018,14 +1013,6 @@ func TestCPUManagerAddWithResvList(t *testing.T) {
 			expAllocateErr:     nil,
 			expAddContainerErr: nil,
 		},
-		{
-			description:        "cpu manager add - container update error",
-			updateErr:          fmt.Errorf("fake update error"),
-			policy:             testPolicy,
-			expCPUSet:          cpuset.NewCPUSet(0, 1, 2, 3),
-			expAllocateErr:     nil,
-			expAddContainerErr: fmt.Errorf("fake update error"),
-		},
 	}
 
 	for _, testCase := range testCases {
@@ -1053,7 +1040,8 @@ func TestCPUManagerAddWithResvList(t *testing.T) {
 				testCase.description, testCase.expAllocateErr, err)
 		}
 
-		err = mgr.AddContainer(pod, container, "fakeID")
+		mgr.AddContainer(pod, container, "fakeID")
+		_, _, err = mgr.containerMap.GetContainerRef("fakeID")
 		if !reflect.DeepEqual(err, testCase.expAddContainerErr) {
 			t.Errorf("CPU Manager AddContainer() error (%v). expected error: %v but got: %v",
 				testCase.description, testCase.expAddContainerErr, err)

pkg/kubelet/cm/cpumanager/fake_cpu_manager.go

@@ -46,9 +46,8 @@ func (m *fakeManager) Allocate(pod *v1.Pod, container *v1.Container) error {
 	return nil
 }
 
-func (m *fakeManager) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) error {
+func (m *fakeManager) AddContainer(pod *v1.Pod, container *v1.Container, containerID string) {
 	klog.Infof("[fake cpumanager] AddContainer (pod: %s, container: %s, container id: %s)", pod.Name, container.Name, containerID)
-	return nil
 }
 
 func (m *fakeManager) RemoveContainer(containerID string) error {

pkg/kubelet/cm/fake_internal_container_lifecycle.go

@@ -18,6 +18,7 @@ package cm
 
 import (
 	"k8s.io/api/core/v1"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 )
 
 func NewFakeInternalContainerLifecycle() *fakeInternalContainerLifecycle {
@@ -26,6 +27,10 @@ func NewFakeInternalContainerLifecycle() *fakeInternalContainerLifecycle {
 
 type fakeInternalContainerLifecycle struct{}
 
+func (f *fakeInternalContainerLifecycle) PreCreateContainer(pod *v1.Pod, container *v1.Container, containerConfig *runtimeapi.ContainerConfig) error {
+	return nil
+}
+
 func (f *fakeInternalContainerLifecycle) PreStartContainer(pod *v1.Pod, container *v1.Container, containerID string) error {
 	return nil
 }

pkg/kubelet/cm/internal_container_lifecycle.go

@@ -18,14 +18,15 @@ package cm
 
 import (
 	"k8s.io/api/core/v1"
-
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 	kubefeatures "k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm/topologymanager"
 )
 
 type InternalContainerLifecycle interface {
+	PreCreateContainer(pod *v1.Pod, container *v1.Container, containerConfig *runtimeapi.ContainerConfig) error
 	PreStartContainer(pod *v1.Pod, container *v1.Container, containerID string) error
 	PreStopContainer(containerID string) error
 	PostStopContainer(containerID string) error
@@ -39,11 +40,9 @@ type internalContainerLifecycleImpl struct {
 
 func (i *internalContainerLifecycleImpl) PreStartContainer(pod *v1.Pod, container *v1.Container, containerID string) error {
 	if i.cpuManager != nil {
-		err := i.cpuManager.AddContainer(pod, container, containerID)
+		i.cpuManager.AddContainer(pod, container, containerID)
-		if err != nil {
-			return err
-		}
 	}
+
 	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.TopologyManager) {
 		err := i.topologyManager.AddContainer(pod, containerID)
 		if err != nil {

pkg/kubelet/cm/internal_container_lifecycle_linux.go

@@ -0,0 +1,35 @@
+// +build linux
+
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cm
+
+import (
+	"k8s.io/api/core/v1"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+)
+
+func (i *internalContainerLifecycleImpl) PreCreateContainer(pod *v1.Pod, container *v1.Container, containerConfig *runtimeapi.ContainerConfig) error {
+	if i.cpuManager != nil {
+		allocatedCPUs := i.cpuManager.GetCPUs(string(pod.UID), container.Name)
+		if !allocatedCPUs.IsEmpty() {
+			containerConfig.Linux.Resources.CpusetCpus = allocatedCPUs.String()
+		}
+	}
+
+	return nil
+}

pkg/kubelet/cm/internal_container_lifecycle_unsupported.go

@@ -0,0 +1,28 @@
+// +build !linux,!windows
+
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cm
+
+import (
+	"k8s.io/api/core/v1"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+)
+
+func (i *internalContainerLifecycleImpl) PreCreateContainer(pod *v1.Pod, container *v1.Container, containerConfig *runtimeapi.ContainerConfig) error {
+	return nil
+}

pkg/kubelet/cm/internal_container_lifecycle_windows.go

@@ -0,0 +1,28 @@
+// +build windows
+
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cm
+
+import (
+	"k8s.io/api/core/v1"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+)
+
+func (i *internalContainerLifecycleImpl) PreCreateContainer(pod *v1.Pod, container *v1.Container, containerConfig *runtimeapi.ContainerConfig) error {
+	return nil
+}

pkg/kubelet/dockershim/helpers_linux.go

@@ -118,6 +118,7 @@ func (ds *dockerService) updateCreateConfig(
 				CPUShares:  rOpts.CpuShares,
 				CPUQuota:   rOpts.CpuQuota,
 				CPUPeriod:  rOpts.CpuPeriod,
+				CpusetCpus: rOpts.CpusetCpus,
 			}
 			createConfig.HostConfig.OomScoreAdj = int(rOpts.OomScoreAdj)
 		}

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -56,6 +56,8 @@ import (
 var (
 	// ErrCreateContainerConfig - failed to create container config
 	ErrCreateContainerConfig = errors.New("CreateContainerConfigError")
+	// ErrPreCreateHook - failed to execute PreCreateHook
+	ErrPreCreateHook = errors.New("PreCreateHookError")
 	// ErrCreateContainer - failed to create container
 	ErrCreateContainer = errors.New("CreateContainerError")
 	// ErrPreStartHook - failed to execute PreStartHook
@@ -167,6 +169,13 @@ func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandb
 		return s.Message(), ErrCreateContainerConfig
 	}
 
+	err = m.internalLifecycle.PreCreateContainer(pod, container, containerConfig)
+	if err != nil {
+		s, _ := grpcstatus.FromError(err)
+		m.recordContainerEvent(pod, container, "", v1.EventTypeWarning, events.FailedToCreateContainer, "Internal PreCreateContainer hook failed: %v", s.Message())
+		return s.Message(), ErrPreCreateHook
+	}
+
 	containerID, err := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
 	if err != nil {
 		s, _ := grpcstatus.FromError(err)