Merge pull request kubernetes#84705 from whypro/cpumanager-panic-master

Return error instead of panic when cpu manager fails on startup.

Author: k8s-ci-robot

pkg/kubelet/cm/container_manager_linux.go

@@ -579,7 +579,10 @@ func (cm *containerManagerImpl) Start(node *v1.Node,
 		if err != nil {
 			return fmt.Errorf("failed to build map of initial containers from runtime: %v", err)
 		}
-		cm.cpuManager.Start(cpumanager.ActivePodsFunc(activePods), sourcesReady, podStatusProvider, runtimeService, containerMap)
+		err = cm.cpuManager.Start(cpumanager.ActivePodsFunc(activePods), sourcesReady, podStatusProvider, runtimeService, containerMap)
+		if err != nil {
+			return fmt.Errorf("start cpu manager error: %v", err)
+		}
 	}
 
 	// cache the node Info including resource capacity and

pkg/kubelet/cm/cpumanager/cpu_manager.go

@@ -53,7 +53,7 @@ const cpuManagerStateFileName = "cpu_manager_state"
 // Manager interface provides methods for Kubelet to manage pod cpus.
 type Manager interface {
 	// Start is called during Kubelet initialization.
-	Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap)
+	Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap) error
 
 	// AddContainer is called between container create and container start
 	// so that initial CPU affinity settings can be written through to the
@@ -155,7 +155,10 @@ func NewManager(cpuPolicyName string, reconcilePeriod time.Duration, machineInfo
 		// exclusively allocated.
 		reservedCPUsFloat := float64(reservedCPUs.MilliValue()) / 1000
 		numReservedCPUs := int(math.Ceil(reservedCPUsFloat))
-		policy = NewStaticPolicy(topo, numReservedCPUs, specificCPUs, affinity)
+		policy, err = NewStaticPolicy(topo, numReservedCPUs, specificCPUs, affinity)
+		if err != nil {
+			return nil, fmt.Errorf("new static policy error: %v", err)
+		}
 
 	default:
 		return nil, fmt.Errorf("unknown policy: \"%s\"", cpuPolicyName)
@@ -173,7 +176,7 @@ func NewManager(cpuPolicyName string, reconcilePeriod time.Duration, machineInfo
 	return manager, nil
 }
 
-func (m *manager) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap) {
+func (m *manager) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap) error {
 	klog.Infof("[cpumanager] starting with %s policy", m.policy.Name())
 	klog.Infof("[cpumanager] reconciling every %v", m.reconcilePeriod)
 	m.sourcesReady = sourcesReady
@@ -183,16 +186,22 @@ func (m *manager) Start(activePods ActivePodsFunc, sourcesReady config.SourcesRe
 
 	stateImpl, err := state.NewCheckpointState(m.stateFileDirectory, cpuManagerStateFileName, m.policy.Name(), initialContainers)
 	if err != nil {
-		klog.Errorf("[cpumanager] could not initialize checkpoint manager: %v\n", err)
-		panic("[cpumanager] - please drain node and remove policy state file")
+		klog.Errorf("[cpumanager] could not initialize checkpoint manager: %v, please drain node and remove policy state file", err)
+		return err
 	}
 	m.state = stateImpl
 
-	m.policy.Start(m.state)
+	err = m.policy.Start(m.state)
+	if err != nil {
+		klog.Errorf("[cpumanager] policy start error: %v", err)
+		return err
+	}
+
 	if m.policy.Name() == string(PolicyNone) {
-		return
+		return nil
 	}
 	go wait.Until(func() { m.reconcileState() }, m.reconcilePeriod, wait.NeverStop)
+	return nil
 }
 
 func (m *manager) AddContainer(p *v1.Pod, c *v1.Container, containerID string) error {

pkg/kubelet/cm/cpumanager/cpu_manager_test.go

@@ -100,7 +100,8 @@ func (p *mockPolicy) Name() string {
 	return "mock"
 }
 
-func (p *mockPolicy) Start(s state.State) {
+func (p *mockPolicy) Start(s state.State) error {
+	return p.err
 }
 
 func (p *mockPolicy) AddContainer(s state.State, pod *v1.Pod, container *v1.Container) error {
@@ -206,7 +207,7 @@ func makeMultiContainerPod(initCPUs, appCPUs []struct{ request, limit string })
 }
 
 func TestCPUManagerAdd(t *testing.T) {
-	testPolicy := NewStaticPolicy(
+	testPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{
 			NumCPUs:    4,
 			NumSockets: 1,
@@ -464,7 +465,7 @@ func TestCPUManagerAddWithInitContainers(t *testing.T) {
 	}
 
 	for _, testCase := range testCases {
-		policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+		policy, _ := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
 
 		state := &mockState{
 			assignments:   testCase.stAssignments,
@@ -941,7 +942,7 @@ func TestReconcileState(t *testing.T) {
 // above test cases are without kubelet --reserved-cpus cmd option
 // the following tests are with --reserved-cpus configured
 func TestCPUManagerAddWithResvList(t *testing.T) {
-	testPolicy := NewStaticPolicy(
+	testPolicy, _ := NewStaticPolicy(
 		&topology.CPUTopology{
 			NumCPUs:    4,
 			NumSockets: 1,

pkg/kubelet/cm/cpumanager/fake_cpu_manager.go

@@ -30,8 +30,9 @@ type fakeManager struct {
 	state state.State
 }
 
-func (m *fakeManager) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap) {
+func (m *fakeManager) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, podStatusProvider status.PodStatusProvider, containerRuntime runtimeService, initialContainers containermap.ContainerMap) error {
 	klog.Info("[fake cpumanager] Start()")
+	return nil
 }
 
 func (m *fakeManager) Policy() Policy {

pkg/kubelet/cm/cpumanager/policy.go

@@ -25,7 +25,7 @@ import (
 // Policy implements logic for pod container to CPU assignment.
 type Policy interface {
 	Name() string
-	Start(s state.State)
+	Start(s state.State) error
 	// AddContainer call is idempotent
 	AddContainer(s state.State, pod *v1.Pod, container *v1.Container) error
 	// RemoveContainer call is idempotent

pkg/kubelet/cm/cpumanager/policy_none.go

@@ -39,8 +39,9 @@ func (p *nonePolicy) Name() string {
 	return string(PolicyNone)
 }
 
-func (p *nonePolicy) Start(s state.State) {
+func (p *nonePolicy) Start(s state.State) error {
 	klog.Info("[cpumanager] none policy: Start")
+	return nil
 }
 
 func (p *nonePolicy) AddContainer(s state.State, pod *v1.Pod, container *v1.Container) error {

pkg/kubelet/cm/cpumanager/policy_static.go

@@ -85,7 +85,7 @@ var _ Policy = &staticPolicy{}
 // NewStaticPolicy returns a CPU manager policy that does not change CPU
 // assignments for exclusively pinned guaranteed containers after the main
 // container process starts.
-func NewStaticPolicy(topology *topology.CPUTopology, numReservedCPUs int, reservedCPUs cpuset.CPUSet, affinity topologymanager.Store) Policy {
+func NewStaticPolicy(topology *topology.CPUTopology, numReservedCPUs int, reservedCPUs cpuset.CPUSet, affinity topologymanager.Store) (Policy, error) {
 	allCPUs := topology.CPUDetails.CPUs()
 	var reserved cpuset.CPUSet
 	if reservedCPUs.Size() > 0 {
@@ -100,7 +100,8 @@ func NewStaticPolicy(topology *topology.CPUTopology, numReservedCPUs int, reserv
 	}
 
 	if reserved.Size() != numReservedCPUs {
-		panic(fmt.Sprintf("[cpumanager] unable to reserve the required amount of CPUs (size of %s did not equal %d)", reserved, numReservedCPUs))
+		err := fmt.Errorf("[cpumanager] unable to reserve the required amount of CPUs (size of %s did not equal %d)", reserved, numReservedCPUs)
+		return nil, err
 	}
 
 	klog.Infof("[cpumanager] reserved %d CPUs (\"%s\") not available for exclusive assignment", reserved.Size(), reserved)
@@ -109,18 +110,19 @@ func NewStaticPolicy(topology *topology.CPUTopology, numReservedCPUs int, reserv
 		topology: topology,
 		reserved: reserved,
 		affinity: affinity,
-	}
+	}, nil
 }
 
 func (p *staticPolicy) Name() string {
 	return string(PolicyStatic)
 }
 
-func (p *staticPolicy) Start(s state.State) {
+func (p *staticPolicy) Start(s state.State) error {
 	if err := p.validateState(s); err != nil {
-		klog.Errorf("[cpumanager] static policy invalid state: %s\n", err.Error())
-		panic("[cpumanager] - please drain node and remove policy state file")
+		klog.Errorf("[cpumanager] static policy invalid state: %v, please drain node and remove policy state file", err)
+		return err
 	}
+	return nil
 }
 
 func (p *staticPolicy) validateState(s state.State) error {

pkg/kubelet/cm/cpumanager/policy_static_test.go

@@ -41,11 +41,10 @@ type staticPolicyTest struct {
 	expErr          error
 	expCPUAlloc     bool
 	expCSet         cpuset.CPUSet
-	expPanic        bool
 }
 
 func TestStaticPolicyName(t *testing.T) {
-	policy := NewStaticPolicy(topoSingleSocketHT, 1, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+	policy, _ := NewStaticPolicy(topoSingleSocketHT, 1, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
 
 	policyName := policy.Name()
 	if policyName != "static" {
@@ -81,7 +80,7 @@ func TestStaticPolicyStart(t *testing.T) {
 			numReservedCPUs: 2,
 			stAssignments:   state.ContainerCPUAssignments{},
 			stDefaultCPUSet: cpuset.NewCPUSet(0, 1),
-			expPanic:        true,
+			expErr:          fmt.Errorf("not all reserved cpus: \"0,6\" are present in defaultCpuSet: \"0-1\""),
 		},
 		{
 			description: "assigned core 2 is still present in available cpuset",
@@ -92,7 +91,7 @@ func TestStaticPolicyStart(t *testing.T) {
 				},
 			},
 			stDefaultCPUSet: cpuset.NewCPUSet(2, 3, 4, 5, 6, 7, 8, 9, 10, 11),
-			expPanic:        true,
+			expErr:          fmt.Errorf("pod: fakePod, container: 0 cpuset: \"0-2\" overlaps with default cpuset \"2-11\""),
 		},
 		{
 			description: "core 12 is not present in topology but is in state cpuset",
@@ -104,7 +103,7 @@ func TestStaticPolicyStart(t *testing.T) {
 				},
 			},
 			stDefaultCPUSet: cpuset.NewCPUSet(5, 6, 7, 8, 9, 10, 11, 12),
-			expPanic:        true,
+			expErr:          fmt.Errorf("current set of available CPUs \"0-11\" doesn't match with CPUs in state \"0-12\""),
 		},
 		{
 			description: "core 11 is present in topology but is not in state cpuset",
@@ -116,26 +115,25 @@ func TestStaticPolicyStart(t *testing.T) {
 				},
 			},
 			stDefaultCPUSet: cpuset.NewCPUSet(5, 6, 7, 8, 9, 10),
-			expPanic:        true,
+			expErr:          fmt.Errorf("current set of available CPUs \"0-11\" doesn't match with CPUs in state \"0-10\""),
 		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.description, func(t *testing.T) {
-			defer func() {
-				if err := recover(); err != nil {
-					if !testCase.expPanic {
-						t.Errorf("unexpected panic occurred: %q", err)
-					}
-				} else if testCase.expPanic {
-					t.Error("expected panic doesn't occurred")
-				}
-			}()
-			policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager()).(*staticPolicy)
+			p, _ := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+			policy := p.(*staticPolicy)
 			st := &mockState{
 				assignments:   testCase.stAssignments,
 				defaultCPUSet: testCase.stDefaultCPUSet,
 			}
-			policy.Start(st)
+			err := policy.Start(st)
+			if !reflect.DeepEqual(err, testCase.expErr) {
+				t.Errorf("StaticPolicy Start() error (%v). expected error: %v but got: %v",
+					testCase.description, testCase.expErr, err)
+			}
+			if err != nil {
+				return
+			}
 
 			if !testCase.stDefaultCPUSet.IsEmpty() {
 				for cpuid := 1; cpuid < policy.topology.NumCPUs; cpuid++ {
@@ -438,7 +436,7 @@ func TestStaticPolicyAdd(t *testing.T) {
 	}
 
 	for _, testCase := range testCases {
-		policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+		policy, _ := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
 
 		st := &mockState{
 			assignments:   testCase.stAssignments,
@@ -539,7 +537,7 @@ func TestStaticPolicyRemove(t *testing.T) {
 	}
 
 	for _, testCase := range testCases {
-		policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+		policy, _ := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
 
 		st := &mockState{
 			assignments:   testCase.stAssignments,
@@ -629,12 +627,17 @@ func TestTopologyAwareAllocateCPUs(t *testing.T) {
 		},
 	}
 	for _, tc := range testCases {
-		policy := NewStaticPolicy(tc.topo, 0, cpuset.NewCPUSet(), topologymanager.NewFakeManager()).(*staticPolicy)
+		p, _ := NewStaticPolicy(tc.topo, 0, cpuset.NewCPUSet(), topologymanager.NewFakeManager())
+		policy := p.(*staticPolicy)
 		st := &mockState{
 			assignments:   tc.stAssignments,
 			defaultCPUSet: tc.stDefaultCPUSet,
 		}
-		policy.Start(st)
+		err := policy.Start(st)
+		if err != nil {
+			t.Errorf("StaticPolicy Start() error (%v)", err)
+			continue
+		}
 
 		cset, err := policy.allocateCPUs(st, tc.numRequested, tc.socketMask)
 		if err != nil {
@@ -661,9 +664,9 @@ type staticPolicyTestWithResvList struct {
 	stDefaultCPUSet cpuset.CPUSet
 	pod             *v1.Pod
 	expErr          error
+	expNewErr       error
 	expCPUAlloc     bool
 	expCSet         cpuset.CPUSet
-	expPanic        bool
 }
 
 func TestStaticPolicyStartWithResvList(t *testing.T) {
@@ -684,7 +687,7 @@ func TestStaticPolicyStartWithResvList(t *testing.T) {
 			reserved:        cpuset.NewCPUSet(0, 1),
 			stAssignments:   state.ContainerCPUAssignments{},
 			stDefaultCPUSet: cpuset.NewCPUSet(2, 3, 4, 5),
-			expPanic:        true,
+			expErr:          fmt.Errorf("not all reserved cpus: \"0-1\" are present in defaultCpuSet: \"2-5\""),
 		},
 		{
 			description:     "inconsistency between numReservedCPUs and reserved",
@@ -693,26 +696,32 @@ func TestStaticPolicyStartWithResvList(t *testing.T) {
 			reserved:        cpuset.NewCPUSet(0, 1),
 			stAssignments:   state.ContainerCPUAssignments{},
 			stDefaultCPUSet: cpuset.NewCPUSet(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11),
-			expPanic:        true,
+			expNewErr:       fmt.Errorf("[cpumanager] unable to reserve the required amount of CPUs (size of 0-1 did not equal 1)"),
 		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.description, func(t *testing.T) {
-			defer func() {
-				if err := recover(); err != nil {
-					if !testCase.expPanic {
-						t.Errorf("unexpected panic occurred: %q", err)
-					}
-				} else if testCase.expPanic {
-					t.Error("expected panic doesn't occurred")
-				}
-			}()
-			policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, testCase.reserved, topologymanager.NewFakeManager()).(*staticPolicy)
+			p, err := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, testCase.reserved, topologymanager.NewFakeManager())
+			if !reflect.DeepEqual(err, testCase.expNewErr) {
+				t.Errorf("StaticPolicy Start() error (%v). expected error: %v but got: %v",
+					testCase.description, testCase.expNewErr, err)
+			}
+			if err != nil {
+				return
+			}
+			policy := p.(*staticPolicy)
 			st := &mockState{
 				assignments:   testCase.stAssignments,
 				defaultCPUSet: testCase.stDefaultCPUSet,
 			}
-			policy.Start(st)
+			err = policy.Start(st)
+			if !reflect.DeepEqual(err, testCase.expErr) {
+				t.Errorf("StaticPolicy Start() error (%v). expected error: %v but got: %v",
+					testCase.description, testCase.expErr, err)
+			}
+			if err != nil {
+				return
+			}
 
 			if !st.GetDefaultCPUSet().Equals(testCase.expCSet) {
 				t.Errorf("State CPUSet is different than expected. Have %q wants: %q", st.GetDefaultCPUSet(),
@@ -769,7 +778,7 @@ func TestStaticPolicyAddWithResvList(t *testing.T) {
 	}
 
 	for _, testCase := range testCases {
-		policy := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, testCase.reserved, topologymanager.NewFakeManager())
+		policy, _ := NewStaticPolicy(testCase.topo, testCase.numReservedCPUs, testCase.reserved, topologymanager.NewFakeManager())
 
 		st := &mockState{
 			assignments:   testCase.stAssignments,