github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-24 20:24:09 +00:00
Code Issues Projects Releases Wiki Activity

e2e_node: refactor code to use a single method to update the kubelet config

Browse Source 
Signed-off-by: Artyom Lukianov <alukiano@redhat.com>
This commit is contained in:
Artyom Lukianov 2021-11-04 15:24:44 +02:00
parent ca35bdb403
commit 50fdcdfc59
7 changed files with 380 additions and 631 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
test/e2e_node/cpu_manager_test.go
View File

@ -38,10 +38,8 @@ import (
"github.com/onsi/ginkgo" "github.com/onsi/ginkgo"
"github.com/onsi/gomega" "github.com/onsi/gomega"
"k8s.io/kubernetes/test/e2e/framework" "k8s.io/kubernetes/test/e2e/framework"
e2enode "k8s.io/kubernetes/test/e2e/framework/node"
e2epod "k8s.io/kubernetes/test/e2e/framework/pod" e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper" e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
e2enodekubelet "k8s.io/kubernetes/test/e2e_node/kubeletconfig"
) )
// Helper for makeCPUManagerPod(). // Helper for makeCPUManagerPod().
@ -186,60 +184,15 @@ func getCoreSiblingList(cpuRes int64) string {
return string(out) return string(out)
} }
func deleteStateFile() {
err := exec.Command("/bin/sh", "-c", "rm -f /var/lib/kubelet/cpu_manager_state").Run()
framework.ExpectNoError(err, "error deleting state file")
}
func setOldKubeletConfig(oldCfg *kubeletconfig.KubeletConfiguration) {
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
// Delete the CPU Manager state file so that the old Kubelet configuration
// can take effect
deleteStateFile()
if oldCfg != nil {
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(oldCfg))
}
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
}
type cpuManagerKubeletArguments struct { type cpuManagerKubeletArguments struct {
policyName string policyName string
cleanStateFile bool
enableCPUManager bool enableCPUManager bool
enableCPUManagerOptions bool enableCPUManagerOptions bool
reservedSystemCPUs cpuset.CPUSet reservedSystemCPUs cpuset.CPUSet
options map[string]string options map[string]string
} }
func disableCPUManagerInKubelet(f *framework.Framework) (oldCfg *kubeletconfig.KubeletConfiguration) { func configureCPUManagerInKubelet(oldCfg *kubeletconfig.KubeletConfiguration, kubeletArguments *cpuManagerKubeletArguments) *kubeletconfig.KubeletConfiguration {
// Disable CPU Manager in Kubelet.
oldCfg = configureCPUManagerInKubelet(f, &cpuManagerKubeletArguments{
enableCPUManager: false,
})
return oldCfg
}
func configureCPUManagerInKubelet(f *framework.Framework, kubeletArguments *cpuManagerKubeletArguments) (oldCfg *kubeletconfig.KubeletConfiguration) {
// Enable CPU Manager in Kubelet with static policy.
oldCfg, err := getCurrentKubeletConfig()
framework.ExpectNoError(err)
newCfg := oldCfg.DeepCopy() newCfg := oldCfg.DeepCopy()
if newCfg.FeatureGates == nil { if newCfg.FeatureGates == nil {
newCfg.FeatureGates = make(map[string]bool) newCfg.FeatureGates = make(map[string]bool)
@ -275,44 +228,6 @@ func configureCPUManagerInKubelet(f *framework.Framework, kubeletArguments *cpuM
} }
} }
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
// After graduation of the CPU Manager feature to Beta, the CPU Manager
// "none" policy is ON by default. But when we set the CPU Manager policy to
// "static" in this test and the Kubelet is restarted so that "static"
// policy can take effect, there will always be a conflict with the state
// checkpointed in the disk (i.e., the policy checkpointed in the disk will
// be "none" whereas we are trying to restart Kubelet with "static"
// policy). Therefore, we delete the state file so that we can proceed
// with the tests.
// Only delete the state file at the begin of the tests.
if kubeletArguments.cleanStateFile {
deleteStateFile()
}
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(newCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
return oldCfg return oldCfg
} }
@ -615,6 +530,12 @@ func runCPUManagerTests(f *framework.Framework) {
var ctnAttrs []ctnAttribute var ctnAttrs []ctnAttribute
var pod *v1.Pod var pod *v1.Pod
ginkgo.BeforeEach(func() {
var err error
oldCfg, err = getCurrentKubeletConfig()
framework.ExpectNoError(err)
})
ginkgo.It("should assign CPUs as expected based on the Pod spec", func() { ginkgo.It("should assign CPUs as expected based on the Pod spec", func() {
cpuCap, cpuAlloc, _ = getLocalNodeCPUDetails(f) cpuCap, cpuAlloc, _ = getLocalNodeCPUDetails(f)
@ -624,12 +545,12 @@ func runCPUManagerTests(f *framework.Framework) {
} }
// Enable CPU Manager in the kubelet. // Enable CPU Manager in the kubelet.
oldCfg = configureCPUManagerInKubelet(f, &cpuManagerKubeletArguments{ newCfg := configureCPUManagerInKubelet(oldCfg, &cpuManagerKubeletArguments{
policyName: string(cpumanager.PolicyStatic), policyName: string(cpumanager.PolicyStatic),
enableCPUManager: true, enableCPUManager: true,
reservedSystemCPUs: cpuset.CPUSet{}, reservedSystemCPUs: cpuset.CPUSet{},
cleanStateFile: true,
}) })
updateKubeletConfig(f, newCfg, true)
ginkgo.By("running a non-Gu pod") ginkgo.By("running a non-Gu pod")
runNonGuPodTest(f, cpuCap) runNonGuPodTest(f, cpuCap)
@ -689,18 +610,24 @@ func runCPUManagerTests(f *framework.Framework) {
pod.Spec.Containers[0].Name, pod.Name) pod.Spec.Containers[0].Name, pod.Name)
ginkgo.By("disable cpu manager in kubelet") ginkgo.By("disable cpu manager in kubelet")
disableCPUManagerInKubelet(f) newCfg = configureCPUManagerInKubelet(oldCfg, &cpuManagerKubeletArguments{
policyName: string(cpumanager.PolicyStatic),
enableCPUManager: false,
reservedSystemCPUs: cpuset.CPUSet{},
})
updateKubeletConfig(f, newCfg, false)
ginkgo.By("by deleting the pod and waiting for container removal") ginkgo.By("by deleting the pod and waiting for container removal")
deletePods(f, []string{pod.Name}) deletePods(f, []string{pod.Name})
waitForContainerRemoval(pod.Spec.Containers[0].Name, pod.Name, pod.Namespace) waitForContainerRemoval(pod.Spec.Containers[0].Name, pod.Name, pod.Namespace)
ginkgo.By("enable cpu manager in kubelet without delete state file") ginkgo.By("enable cpu manager in kubelet without delete state file")
configureCPUManagerInKubelet(f, &cpuManagerKubeletArguments{ newCfg = configureCPUManagerInKubelet(oldCfg, &cpuManagerKubeletArguments{
policyName: string(cpumanager.PolicyStatic), policyName: string(cpumanager.PolicyStatic),
enableCPUManager: true, enableCPUManager: true,
reservedSystemCPUs: cpuset.CPUSet{}, reservedSystemCPUs: cpuset.CPUSet{},
}) })
updateKubeletConfig(f, newCfg, false)
ginkgo.By("wait for the deleted pod to be cleaned up from the state file") ginkgo.By("wait for the deleted pod to be cleaned up from the state file")
waitForStateFileCleanedUp() waitForStateFileCleanedUp()
@ -729,15 +656,16 @@ func runCPUManagerTests(f *framework.Framework) {
cpuPolicyOptions := map[string]string{ cpuPolicyOptions := map[string]string{
cpumanager.FullPCPUsOnlyOption: "true", cpumanager.FullPCPUsOnlyOption: "true",
} }
oldCfg = configureCPUManagerInKubelet(f, newCfg := configureCPUManagerInKubelet(oldCfg,
&cpuManagerKubeletArguments{ &cpuManagerKubeletArguments{
policyName: string(cpumanager.PolicyStatic), policyName: string(cpumanager.PolicyStatic),
enableCPUManager: true, enableCPUManager: true,
cleanStateFile: true,
reservedSystemCPUs: cpuset.NewCPUSet(0), reservedSystemCPUs: cpuset.NewCPUSet(0),
enableCPUManagerOptions: true, enableCPUManagerOptions: true,
options: cpuPolicyOptions, options: cpuPolicyOptions,
}) },
)
updateKubeletConfig(f, newCfg, true)
// the order between negative and positive doesn't really matter // the order between negative and positive doesn't really matter
runSMTAlignmentNegativeTests(f) runSMTAlignmentNegativeTests(f)
@ -745,7 +673,7 @@ func runCPUManagerTests(f *framework.Framework) {
}) })
ginkgo.AfterEach(func() { ginkgo.AfterEach(func() {
setOldKubeletConfig(oldCfg) updateKubeletConfig(f, oldCfg, true)
}) })
} }

18
test/e2e_node/device_manager_test.go
View File

@ -61,15 +61,6 @@ var _ = SIGDescribe("Device Manager [Serial] [Feature:DeviceManager][NodeFeatur
e2eskipper.Skipf("this test is meant to run on a system with at least one configured VF from SRIOV device") e2eskipper.Skipf("this test is meant to run on a system with at least one configured VF from SRIOV device")
} }
oldCfg := enablePodResourcesFeatureGateInKubelet(f)
defer func() {
// restore kubelet config
setOldKubeletConfig(f, oldCfg)
// Delete state file to allow repeated runs
deleteStateFile()
}()
configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile) configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile)
sd := setupSRIOVConfigOrFail(f, configMap) sd := setupSRIOVConfigOrFail(f, configMap)
@ -167,15 +158,6 @@ var _ = SIGDescribe("Device Manager [Serial] [Feature:DeviceManager][NodeFeatur
e2eskipper.Skipf("this test is meant to run on a system with at least one configured VF from SRIOV device") e2eskipper.Skipf("this test is meant to run on a system with at least one configured VF from SRIOV device")
} }
oldCfg := enablePodResourcesFeatureGateInKubelet(f)
defer func() {
// restore kubelet config
setOldKubeletConfig(f, oldCfg)
// Delete state file to allow repeated runs
deleteStateFile()
}()
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err) framework.ExpectNoError(err)

1
test/e2e_node/device_plugin_test.go
View File

@ -110,7 +110,6 @@ func readDaemonSetV1OrDie(objBytes []byte) *appsv1.DaemonSet {
func testDevicePlugin(f *framework.Framework, pluginSockDir string) { func testDevicePlugin(f *framework.Framework, pluginSockDir string) {
pluginSockDir = filepath.Join(pluginSockDir) + "/" pluginSockDir = filepath.Join(pluginSockDir) + "/"
ginkgo.Context("DevicePlugin", func() { ginkgo.Context("DevicePlugin", func() {
ginkgo.By("Enabling support for Kubelet Plugins Watcher")
ginkgo.It("[Flaky] Verifies the Kubelet device plugin functionality.", func() { ginkgo.It("[Flaky] Verifies the Kubelet device plugin functionality.", func() {
ginkgo.By("Wait for node is ready to start with") ginkgo.By("Wait for node is ready to start with")
e2enode.WaitForNodeToBeReady(f.ClientSet, framework.TestContext.NodeName, 5*time.Minute) e2enode.WaitForNodeToBeReady(f.ClientSet, framework.TestContext.NodeName, 5*time.Minute)

223
test/e2e_node/memory_manager_test.go
View File

@ -41,9 +41,7 @@ import (
"k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/state" "k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/state"
"k8s.io/kubernetes/pkg/kubelet/util" "k8s.io/kubernetes/pkg/kubelet/util"
"k8s.io/kubernetes/test/e2e/framework" "k8s.io/kubernetes/test/e2e/framework"
e2enode "k8s.io/kubernetes/test/e2e/framework/node"
e2epod "k8s.io/kubernetes/test/e2e/framework/pod" e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
e2enodekubelet "k8s.io/kubernetes/test/e2e_node/kubeletconfig"
"k8s.io/utils/pointer" "k8s.io/utils/pointer"
"github.com/onsi/ginkgo" "github.com/onsi/ginkgo"
@ -51,11 +49,10 @@ import (
) )
const ( const (
evictionHardMemory = "memory.available" evictionHardMemory = "memory.available"
memoryManagerStateFile = "/var/lib/kubelet/memory_manager_state" resourceMemory = "memory"
resourceMemory = "memory" staticPolicy = "Static"
staticPolicy = "Static" nonePolicy = "None"
nonePolicy = "None"
) )
// Helper for makeMemoryManagerPod(). // Helper for makeMemoryManagerPod().
@ -138,11 +135,6 @@ func makeMemoryManagerPod(podName string, initCtnAttributes, ctnAttributes []mem
return pod return pod
} }
func deleteMemoryManagerStateFile() {
err := exec.Command("/bin/sh", "-c", fmt.Sprintf("rm -f %s", memoryManagerStateFile)).Run()
framework.ExpectNoError(err, "failed to delete the state file")
}
func getMemoryManagerState() (*state.MemoryManagerCheckpoint, error) { func getMemoryManagerState() (*state.MemoryManagerCheckpoint, error) {
if _, err := os.Stat(memoryManagerStateFile); os.IsNotExist(err) { if _, err := os.Stat(memoryManagerStateFile); os.IsNotExist(err) {
return nil, fmt.Errorf("the memory manager state file %s does not exist", memoryManagerStateFile) return nil, fmt.Errorf("the memory manager state file %s does not exist", memoryManagerStateFile)
@ -187,76 +179,44 @@ type kubeletParams struct {
evictionHard map[string]string evictionHard map[string]string
} }
func getUpdatedKubeletConfig(oldCfg *kubeletconfig.KubeletConfiguration, params *kubeletParams) *kubeletconfig.KubeletConfiguration { func updateKubeletConfigWithMemoryManagerParams(initialCfg *kubeletconfig.KubeletConfiguration, params *kubeletParams) {
newCfg := oldCfg.DeepCopy() if initialCfg.FeatureGates == nil {
initialCfg.FeatureGates = map[string]bool{}
if newCfg.FeatureGates == nil {
newCfg.FeatureGates = map[string]bool{}
} }
newCfg.MemoryManagerPolicy = params.memoryManagerPolicy initialCfg.MemoryManagerPolicy = params.memoryManagerPolicy
// update system-reserved // update system-reserved
if newCfg.SystemReserved == nil { if initialCfg.SystemReserved == nil {
newCfg.SystemReserved = map[string]string{} initialCfg.SystemReserved = map[string]string{}
} }
for resourceName, value := range params.systemReserved { for resourceName, value := range params.systemReserved {
newCfg.SystemReserved[resourceName] = value initialCfg.SystemReserved[resourceName] = value
} }
// update kube-reserved // update kube-reserved
if newCfg.KubeReserved == nil { if initialCfg.KubeReserved == nil {
newCfg.KubeReserved = map[string]string{} initialCfg.KubeReserved = map[string]string{}
} }
for resourceName, value := range params.kubeReserved { for resourceName, value := range params.kubeReserved {
newCfg.KubeReserved[resourceName] = value initialCfg.KubeReserved[resourceName] = value
} }
// update hard eviction threshold // update hard eviction threshold
if newCfg.EvictionHard == nil { if initialCfg.EvictionHard == nil {
newCfg.EvictionHard = map[string]string{} initialCfg.EvictionHard = map[string]string{}
} }
for resourceName, value := range params.evictionHard { for resourceName, value := range params.evictionHard {
newCfg.EvictionHard[resourceName] = value initialCfg.EvictionHard[resourceName] = value
} }
// update reserved memory // update reserved memory
if newCfg.ReservedMemory == nil { if initialCfg.ReservedMemory == nil {
newCfg.ReservedMemory = []kubeletconfig.MemoryReservation{} initialCfg.ReservedMemory = []kubeletconfig.MemoryReservation{}
} }
for _, memoryReservation := range params.systemReservedMemory { for _, memoryReservation := range params.systemReservedMemory {
newCfg.ReservedMemory = append(newCfg.ReservedMemory, memoryReservation) initialCfg.ReservedMemory = append(initialCfg.ReservedMemory, memoryReservation)
} }
return newCfg
}
func updateKubeletConfig(f *framework.Framework, cfg *kubeletconfig.KubeletConfiguration) {
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(cfg))
deleteMemoryManagerStateFile()
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
} }
func getAllNUMANodes() []int { func getAllNUMANodes() []int {
@ -288,8 +248,6 @@ var _ = SIGDescribe("Memory Manager [Disruptive] [Serial] [Feature:MemoryManager
ctnParams, initCtnParams []memoryManagerCtnAttributes ctnParams, initCtnParams []memoryManagerCtnAttributes
is2MiHugepagesSupported *bool is2MiHugepagesSupported *bool
isMultiNUMASupported *bool isMultiNUMASupported *bool
kubeParams *kubeletParams
oldCfg *kubeletconfig.KubeletConfiguration
testPod *v1.Pod testPod *v1.Pod
) )
@ -364,7 +322,6 @@ var _ = SIGDescribe("Memory Manager [Disruptive] [Serial] [Feature:MemoryManager
// dynamically update the kubelet configuration // dynamically update the kubelet configuration
ginkgo.JustBeforeEach(func() { ginkgo.JustBeforeEach(func() {
var err error
hugepagesCount := 8 hugepagesCount := 8
// allocate hugepages // allocate hugepages
@ -373,22 +330,9 @@ var _ = SIGDescribe("Memory Manager [Disruptive] [Serial] [Feature:MemoryManager
gomega.Eventually(func() error { gomega.Eventually(func() error {
return configureHugePages(hugepagesSize2M, hugepagesCount) return configureHugePages(hugepagesSize2M, hugepagesCount)
}, 30*time.Second, framework.Poll).Should(gomega.BeNil()) }, 30*time.Second, framework.Poll).Should(gomega.BeNil())
}
// get the old kubelet config restartKubelet(true)
if oldCfg == nil {
gomega.Eventually(func() error {
oldCfg, err = e2enodekubelet.GetCurrentKubeletConfigFromFile()
return err
}, 5*time.Minute, 15*time.Second).Should(gomega.BeNil())
}
// update the kubelet config with new parameters
newCfg := getUpdatedKubeletConfig(oldCfg, kubeParams)
updateKubeletConfig(f, newCfg)
// request hugepages resources under the container
if *is2MiHugepagesSupported {
ginkgo.By("Waiting for hugepages resource to become available on the local node") ginkgo.By("Waiting for hugepages resource to become available on the local node")
waitingForHugepages(hugepagesCount) waitingForHugepages(hugepagesCount)
@ -414,24 +358,18 @@ var _ = SIGDescribe("Memory Manager [Disruptive] [Serial] [Feature:MemoryManager
gomega.Eventually(func() error { gomega.Eventually(func() error {
return configureHugePages(hugepagesSize2M, 0) return configureHugePages(hugepagesSize2M, 0)
}, 90*time.Second, 15*time.Second).ShouldNot(gomega.HaveOccurred(), "failed to release hugepages") }, 90*time.Second, 15*time.Second).ShouldNot(gomega.HaveOccurred(), "failed to release hugepages")
}
// update the kubelet config with old parameters restartKubelet(true)
if oldCfg != nil {
updateKubeletConfig(f, oldCfg)
}
if *is2MiHugepagesSupported {
waitingForHugepages(0) waitingForHugepages(0)
} }
}) })
ginkgo.Context("with static policy", func() { ginkgo.Context("with static policy", func() {
ginkgo.BeforeEach(func() { tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
// override kubelet configuration parameters kubeParams := *defaultKubeParams
tmpParams := *defaultKubeParams kubeParams.memoryManagerPolicy = staticPolicy
tmpParams.memoryManagerPolicy = staticPolicy updateKubeletConfigWithMemoryManagerParams(initialConfig, &kubeParams)
kubeParams = &tmpParams
}) })
ginkgo.JustAfterEach(func() { ginkgo.JustAfterEach(func() {
@ -706,70 +644,75 @@ var _ = SIGDescribe("Memory Manager [Disruptive] [Serial] [Feature:MemoryManager
}) })
ginkgo.Context("with none policy", func() { ginkgo.Context("with none policy", func() {
ginkgo.BeforeEach(func() { tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
tmpParams := *defaultKubeParams kubeParams := *defaultKubeParams
tmpParams.memoryManagerPolicy = nonePolicy kubeParams.memoryManagerPolicy = nonePolicy
kubeParams = &tmpParams updateKubeletConfigWithMemoryManagerParams(initialConfig, &kubeParams)
// override pod parameters
ctnParams = []memoryManagerCtnAttributes{
{
ctnName: "memory-manager-none",
cpus: "100m",
memory: "128Mi",
},
}
}) })
// TODO: move the test to pod resource API test suite, see - https://github.com/kubernetes/kubernetes/issues/101945 // empty context to configure same container parameters for all tests
ginkgo.It("should not report any memory data during request to pod resources GetAllocatableResources", func() { ginkgo.Context("", func() {
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) ginkgo.BeforeEach(func() {
framework.ExpectNoError(err) // override pod parameters
ctnParams = []memoryManagerCtnAttributes{
{
ctnName: "memory-manager-none",
cpus: "100m",
memory: "128Mi",
},
}
})
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) // TODO: move the test to pod resource API test suite, see - https://github.com/kubernetes/kubernetes/issues/101945
framework.ExpectNoError(err) ginkgo.It("should not report any memory data during request to pod resources GetAllocatableResources", func() {
defer conn.Close() endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
resp, err := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{}) cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err) framework.ExpectNoError(err)
defer conn.Close()
gomega.Expect(resp.Memory).To(gomega.BeEmpty()) resp, err := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{})
}) framework.ExpectNoError(err)
// TODO: move the test to pod resource API test suite, see - https://github.com/kubernetes/kubernetes/issues/101945 gomega.Expect(resp.Memory).To(gomega.BeEmpty())
ginkgo.It("should not report any memory data during request to pod resources List", func() { })
testPod = f.PodClient().CreateSync(testPod)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) // TODO: move the test to pod resource API test suite, see - https://github.com/kubernetes/kubernetes/issues/101945
framework.ExpectNoError(err) ginkgo.It("should not report any memory data during request to pod resources List", func() {
testPod = f.PodClient().CreateSync(testPod)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err) framework.ExpectNoError(err)
defer conn.Close()
resp, err := cli.List(context.TODO(), &kubeletpodresourcesv1.ListPodResourcesRequest{}) cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err) framework.ExpectNoError(err)
defer conn.Close()
for _, podResource := range resp.PodResources { resp, err := cli.List(context.TODO(), &kubeletpodresourcesv1.ListPodResourcesRequest{})
if podResource.Name != testPod.Name { framework.ExpectNoError(err)
continue
for _, podResource := range resp.PodResources {
if podResource.Name != testPod.Name {
continue
}
for _, containerResource := range podResource.Containers {
gomega.Expect(containerResource.Memory).To(gomega.BeEmpty())
}
}
})
ginkgo.It("should succeed to start the pod", func() {
testPod = f.PodClient().CreateSync(testPod)
// it no taste to verify NUMA pinning when the node has only one NUMA node
if !*isMultiNUMASupported {
return
} }
for _, containerResource := range podResource.Containers { verifyMemoryPinning(testPod, allNUMANodes)
gomega.Expect(containerResource.Memory).To(gomega.BeEmpty()) })
}
}
})
ginkgo.It("should succeed to start the pod", func() {
testPod = f.PodClient().CreateSync(testPod)
// it no taste to verify NUMA pinning when the node has only one NUMA node
if !*isMultiNUMASupported {
return
}
verifyMemoryPinning(testPod, allNUMANodes)
}) })
}) })
}) })

533
test/e2e_node/podresources_test.go
View File

@ -28,7 +28,6 @@ import (
v1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilfeature "k8s.io/apiserver/pkg/util/feature"
kubeletpodresourcesv1 "k8s.io/kubelet/pkg/apis/podresources/v1" kubeletpodresourcesv1 "k8s.io/kubelet/pkg/apis/podresources/v1"
kubefeatures "k8s.io/kubernetes/pkg/features" kubefeatures "k8s.io/kubernetes/pkg/features"
kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config" kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
@ -38,15 +37,13 @@ import (
"k8s.io/kubernetes/pkg/kubelet/util" "k8s.io/kubernetes/pkg/kubelet/util"
testutils "k8s.io/kubernetes/test/utils" testutils "k8s.io/kubernetes/test/utils"
"github.com/onsi/ginkgo"
"github.com/onsi/gomega"
"k8s.io/kubernetes/test/e2e/framework" "k8s.io/kubernetes/test/e2e/framework"
e2enode "k8s.io/kubernetes/test/e2e/framework/node" e2enode "k8s.io/kubernetes/test/e2e/framework/node"
e2epod "k8s.io/kubernetes/test/e2e/framework/pod" e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper" e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
e2etestfiles "k8s.io/kubernetes/test/e2e/framework/testfiles" e2etestfiles "k8s.io/kubernetes/test/e2e/framework/testfiles"
e2enodekubelet "k8s.io/kubernetes/test/e2e_node/kubeletconfig"
"github.com/onsi/ginkgo"
"github.com/onsi/gomega"
) )
type podDesc struct { type podDesc struct {
@ -536,240 +533,255 @@ var _ = SIGDescribe("POD Resources [Serial] [Feature:PodResources][NodeFeature:P
reservedSystemCPUs := cpuset.MustParse("1") reservedSystemCPUs := cpuset.MustParse("1")
ginkgo.Context("With SRIOV devices in the system", func() { ginkgo.Context("with SRIOV devices in the system", func() {
ginkgo.It("should return the expected responses with cpumanager static policy enabled", func() { ginkgo.BeforeEach(func() {
// this is a very rough check. We just want to rule out system that does NOT have enough resources
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
if cpuAlloc < minCoreCount {
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount)
}
requireSRIOVDevices() requireSRIOVDevices()
onlineCPUs, err := getOnlineCPUs()
framework.ExpectNoError(err)
// Make sure all the feature gates and the right settings are in place.
oldCfg := configurePodResourcesInKubelet(f, true, reservedSystemCPUs)
defer func() {
// restore kubelet config
setOldKubeletConfig(oldCfg)
// Delete state file to allow repeated runs
deleteStateFile()
}()
configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile)
sd := setupSRIOVConfigOrFail(f, configMap)
defer teardownSRIOVConfigOrFail(f, sd)
waitForSRIOVResources(f, sd)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
waitForSRIOVResources(f, sd)
ginkgo.By("checking List()")
podresourcesListTests(f, cli, sd)
ginkgo.By("checking GetAllocatableResources()")
podresourcesGetAllocatableResourcesTests(cli, sd, onlineCPUs, reservedSystemCPUs)
}) })
ginkgo.It("should return the expected responses with cpumanager none policy", func() { ginkgo.Context("with CPU manager Static policy", func() {
// current default is "none" policy - no need to restart the kubelet ginkgo.BeforeEach(func() {
// this is a very rough check. We just want to rule out system that does NOT have enough resources
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
requireSRIOVDevices() if cpuAlloc < minCoreCount {
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount)
}
})
oldCfg := enablePodResourcesFeatureGateInKubelet(f) // empty context to apply kubelet config changes
defer func() { ginkgo.Context("", func() {
// restore kubelet config tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
setOldKubeletConfig(oldCfg) // Set the CPU Manager policy to static.
initialConfig.CPUManagerPolicy = string(cpumanager.PolicyStatic)
// Delete state file to allow repeated runs // Set the CPU Manager reconcile period to 1 second.
deleteStateFile() initialConfig.CPUManagerReconcilePeriod = metav1.Duration{Duration: 1 * time.Second}
}()
configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile) cpus := reservedSystemCPUs.String()
sd := setupSRIOVConfigOrFail(f, configMap) framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus)
defer teardownSRIOVConfigOrFail(f, sd) initialConfig.ReservedSystemCPUs = cpus
})
waitForSRIOVResources(f, sd) ginkgo.It("should return the expected responses", func() {
onlineCPUs, err := getOnlineCPUs()
framework.ExpectNoError(err)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile)
framework.ExpectNoError(err) sd := setupSRIOVConfigOrFail(f, configMap)
defer teardownSRIOVConfigOrFail(f, sd)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) waitForSRIOVResources(f, sd)
framework.ExpectNoError(err)
defer conn.Close()
waitForSRIOVResources(f, sd) endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
// intentionally passing empty cpuset instead of onlineCPUs because with none policy cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
// we should get no allocatable cpus - no exclusively allocatable CPUs, depends on policy static framework.ExpectNoError(err)
podresourcesGetAllocatableResourcesTests(cli, sd, cpuset.CPUSet{}, cpuset.CPUSet{}) defer conn.Close()
waitForSRIOVResources(f, sd)
ginkgo.By("checking List()")
podresourcesListTests(f, cli, sd)
ginkgo.By("checking GetAllocatableResources()")
podresourcesGetAllocatableResourcesTests(cli, sd, onlineCPUs, reservedSystemCPUs)
})
})
}) })
}) ginkgo.Context("with CPU manager None policy", func() {
ginkgo.It("should return the expected responses", func() {
// current default is "none" policy - no need to restart the kubelet
ginkgo.Context("Without SRIOV devices in the system", func() { requireSRIOVDevices()
ginkgo.It("should return the expected responses with cpumanager static policy enabled", func() {
// this is a very rough check. We just want to rule out system that does NOT have enough resources
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
if cpuAlloc < minCoreCount { configMap := getSRIOVDevicePluginConfigMap(framework.TestContext.SriovdpConfigMapFile)
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount) sd := setupSRIOVConfigOrFail(f, configMap)
} defer teardownSRIOVConfigOrFail(f, sd)
requireLackOfSRIOVDevices() waitForSRIOVResources(f, sd)
onlineCPUs, err := getOnlineCPUs() endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err) framework.ExpectNoError(err)
// Make sure all the feature gates and the right settings are in place. cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
oldCfg := configurePodResourcesInKubelet(f, true, reservedSystemCPUs) framework.ExpectNoError(err)
defer func() { defer conn.Close()
// restore kubelet config
setOldKubeletConfig(oldCfg)
// Delete state file to allow repeated runs waitForSRIOVResources(f, sd)
deleteStateFile()
}()
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) // intentionally passing empty cpuset instead of onlineCPUs because with none policy
framework.ExpectNoError(err) // we should get no allocatable cpus - no exclusively allocatable CPUs, depends on policy static
podresourcesGetAllocatableResourcesTests(cli, sd, cpuset.CPUSet{}, cpuset.CPUSet{})
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) })
framework.ExpectNoError(err)
defer conn.Close()
podresourcesListTests(f, cli, nil)
podresourcesGetAllocatableResourcesTests(cli, nil, onlineCPUs, reservedSystemCPUs)
})
ginkgo.It("should return the expected responses with cpumanager none policy", func() {
// current default is "none" policy - no need to restart the kubelet
requireLackOfSRIOVDevices()
oldCfg := enablePodResourcesFeatureGateInKubelet(f)
defer func() {
// restore kubelet config
setOldKubeletConfig(oldCfg)
// Delete state file to allow repeated runs
deleteStateFile()
}()
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
// intentionally passing empty cpuset instead of onlineCPUs because with none policy
// we should get no allocatable cpus - no exclusively allocatable CPUs, depends on policy static
podresourcesGetAllocatableResourcesTests(cli, nil, cpuset.CPUSet{}, cpuset.CPUSet{})
})
ginkgo.It("should return the expected error with the feature gate disabled", func() {
e2eskipper.SkipIfFeatureGateEnabled(kubefeatures.KubeletPodResourcesGetAllocatable)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
ginkgo.By("checking GetAllocatableResources fail if the feature gate is not enabled")
allocatableRes, err := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{})
framework.Logf("GetAllocatableResources result: %v, err: %v", allocatableRes, err)
framework.ExpectError(err, "With feature gate disabled, the call must fail")
}) })
}) })
ginkgo.Context("Use KubeVirt device plugin, which reports resources w/o hardware topology", func() { ginkgo.Context("without SRIOV devices in the system", func() {
ginkgo.It("should return proper podresources the same as before the restart of kubelet", func() { ginkgo.BeforeEach(func() {
requireLackOfSRIOVDevices()
})
if !utilfeature.DefaultFeatureGate.Enabled(kubefeatures.KubeletPodResources) { ginkgo.Context("with CPU manager Static policy", func() {
e2eskipper.Skipf("this test is meant to run with the POD Resources Extensions feature gate enabled") ginkgo.BeforeEach(func() {
} // this is a very rough check. We just want to rule out system that does NOT have enough resources
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
if cpuAlloc < minCoreCount {
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount)
}
})
// empty context to apply kubelet config changes
ginkgo.Context("", func() {
tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
// Set the CPU Manager policy to static.
initialConfig.CPUManagerPolicy = string(cpumanager.PolicyStatic)
// Set the CPU Manager reconcile period to 1 second.
initialConfig.CPUManagerReconcilePeriod = metav1.Duration{Duration: 1 * time.Second}
cpus := reservedSystemCPUs.String()
framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus)
initialConfig.ReservedSystemCPUs = cpus
})
ginkgo.It("should return the expected responses", func() {
onlineCPUs, err := getOnlineCPUs()
framework.ExpectNoError(err)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
podresourcesListTests(f, cli, nil)
podresourcesGetAllocatableResourcesTests(cli, nil, onlineCPUs, reservedSystemCPUs)
})
})
})
ginkgo.Context("with CPU manager None policy", func() {
ginkgo.It("should return the expected responses", func() {
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
// intentionally passing empty cpuset instead of onlineCPUs because with none policy
// we should get no allocatable cpus - no exclusively allocatable CPUs, depends on policy static
podresourcesGetAllocatableResourcesTests(cli, nil, cpuset.CPUSet{}, cpuset.CPUSet{})
})
})
ginkgo.Context("with disabled KubeletPodResourcesGetAllocatable feature gate", func() {
tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
if initialConfig.FeatureGates == nil {
initialConfig.FeatureGates = make(map[string]bool)
}
initialConfig.FeatureGates[string(kubefeatures.KubeletPodResourcesGetAllocatable)] = false
})
ginkgo.It("should return the expected error with the feature gate disabled", func() {
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
ginkgo.By("checking GetAllocatableResources fail if the feature gate is not enabled")
allocatableRes, err := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{})
framework.Logf("GetAllocatableResources result: %v, err: %v", allocatableRes, err)
framework.ExpectError(err, "With feature gate disabled, the call must fail")
})
})
})
ginkgo.Context("with KubeVirt device plugin, which reports resources w/o hardware topology", func() {
ginkgo.BeforeEach(func() {
_, err := os.Stat("/dev/kvm") _, err := os.Stat("/dev/kvm")
if errors.Is(err, os.ErrNotExist) { if errors.Is(err, os.ErrNotExist) {
e2eskipper.Skipf("KubeVirt device plugin could work only in kvm based environment") e2eskipper.Skipf("KubeVirt device plugin could work only in kvm based environment")
} }
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
if cpuAlloc < minCoreCount {
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount)
}
// Make sure all the feature gates and the right settings are in place.
oldCfg := configurePodResourcesInKubelet(f, true, reservedSystemCPUs)
defer func() {
// restore kubelet config
setOldKubeletConfig(oldCfg)
// Delete state file to allow repeated runs
deleteStateFile()
}()
dpPod := setupKubeVirtDevicePluginOrFail(f)
defer teardownKubeVirtDevicePluginOrFail(f, dpPod)
waitForKubeVirtResources(f, dpPod)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
ginkgo.By("checking List and resources kubevirt resource should be without topology")
allocatableResponse, _ := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{})
for _, dev := range allocatableResponse.GetDevices() {
if dev.ResourceName != KubeVirtResourceName {
continue
}
framework.ExpectEqual(dev.Topology == nil, true, "Topology is expected to be empty for kubevirt resources")
}
// Run pod which requires KubeVirtResourceName
desc := podDesc{
podName: "pod-01",
cntName: "cnt-01",
resourceName: KubeVirtResourceName,
resourceAmount: 1,
cpuRequest: 1000,
}
tpd := newTestPodData()
tpd.createPodsForTest(f, []podDesc{
desc,
})
expectPodResources(1, cli, []podDesc{desc})
ginkgo.By("Restarting Kubelet")
restartKubelet(true)
framework.WaitForAllNodesSchedulable(f.ClientSet, framework.TestContext.NodeSchedulableTimeout)
expectPodResources(1, cli, []podDesc{desc})
tpd.deletePodsForTest(f)
}) })
ginkgo.Context("with CPU manager Static policy", func() {
ginkgo.BeforeEach(func() {
// this is a very rough check. We just want to rule out system that does NOT have enough resources
_, cpuAlloc, _ := getLocalNodeCPUDetails(f)
if cpuAlloc < minCoreCount {
e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount)
}
})
// empty context to apply kubelet config changes
ginkgo.Context("", func() {
tempSetCurrentKubeletConfig(f, func(initialConfig *kubeletconfig.KubeletConfiguration) {
// Set the CPU Manager policy to static.
initialConfig.CPUManagerPolicy = string(cpumanager.PolicyStatic)
// Set the CPU Manager reconcile period to 1 second.
initialConfig.CPUManagerReconcilePeriod = metav1.Duration{Duration: 1 * time.Second}
cpus := reservedSystemCPUs.String()
framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus)
initialConfig.ReservedSystemCPUs = cpus
})
ginkgo.It("should return proper podresources the same as before the restart of kubelet", func() {
dpPod := setupKubeVirtDevicePluginOrFail(f)
defer teardownKubeVirtDevicePluginOrFail(f, dpPod)
waitForKubeVirtResources(f, dpPod)
endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
framework.ExpectNoError(err)
cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
framework.ExpectNoError(err)
defer conn.Close()
ginkgo.By("checking List and resources kubevirt resource should be without topology")
allocatableResponse, _ := cli.GetAllocatableResources(context.TODO(), &kubeletpodresourcesv1.AllocatableResourcesRequest{})
for _, dev := range allocatableResponse.GetDevices() {
if dev.ResourceName != KubeVirtResourceName {
continue
}
framework.ExpectEqual(dev.Topology == nil, true, "Topology is expected to be empty for kubevirt resources")
}
// Run pod which requires KubeVirtResourceName
desc := podDesc{
podName: "pod-01",
cntName: "cnt-01",
resourceName: KubeVirtResourceName,
resourceAmount: 1,
cpuRequest: 1000,
}
tpd := newTestPodData()
tpd.createPodsForTest(f, []podDesc{
desc,
})
expectPodResources(1, cli, []podDesc{desc})
ginkgo.By("Restarting Kubelet")
restartKubelet(true)
framework.WaitForAllNodesSchedulable(f.ClientSet, framework.TestContext.NodeSchedulableTimeout)
expectPodResources(1, cli, []podDesc{desc})
tpd.deletePodsForTest(f)
})
})
})
}) })
}) })
@ -787,113 +799,6 @@ func getOnlineCPUs() (cpuset.CPUSet, error) {
return cpuset.Parse(strings.TrimSpace(string(onlineCPUList))) return cpuset.Parse(strings.TrimSpace(string(onlineCPUList)))
} }
func configurePodResourcesInKubelet(f *framework.Framework, cleanStateFile bool, reservedSystemCPUs cpuset.CPUSet) (oldCfg *kubeletconfig.KubeletConfiguration) {
// we also need CPUManager with static policy to be able to do meaningful testing
oldCfg, err := getCurrentKubeletConfig()
framework.ExpectNoError(err)
newCfg := oldCfg.DeepCopy()
newCfg.FeatureGates[string(kubefeatures.KubeletPodResourcesGetAllocatable)] = true
// After graduation of the CPU Manager feature to Beta, the CPU Manager
// "none" policy is ON by default. But when we set the CPU Manager policy to
// "static" in this test and the Kubelet is restarted so that "static"
// policy can take effect, there will always be a conflict with the state
// checkpointed in the disk (i.e., the policy checkpointed in the disk will
// be "none" whereas we are trying to restart Kubelet with "static"
// policy). Therefore, we delete the state file so that we can proceed
// with the tests.
// Only delete the state file at the begin of the tests.
if cleanStateFile {
deleteStateFile()
}
// Set the CPU Manager policy to static.
newCfg.CPUManagerPolicy = string(cpumanager.PolicyStatic)
// Set the CPU Manager reconcile period to 1 second.
newCfg.CPUManagerReconcilePeriod = metav1.Duration{Duration: 1 * time.Second}
if reservedSystemCPUs.Size() > 0 {
cpus := reservedSystemCPUs.String()
framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus)
newCfg.ReservedSystemCPUs = cpus
} else {
// The Kubelet panics if either kube-reserved or system-reserved is not set
// when CPU Manager is enabled. Set cpu in kube-reserved > 0 so that
// kubelet doesn't panic.
if newCfg.KubeReserved == nil {
newCfg.KubeReserved = map[string]string{}
}
if _, ok := newCfg.KubeReserved["cpu"]; !ok {
newCfg.KubeReserved["cpu"] = "200m"
}
}
// Update the Kubelet configuration.
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(newCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
return oldCfg
}
func enablePodResourcesFeatureGateInKubelet(f *framework.Framework) (oldCfg *kubeletconfig.KubeletConfiguration) {
oldCfg, err := getCurrentKubeletConfig()
framework.ExpectNoError(err)
newCfg := oldCfg.DeepCopy()
if newCfg.FeatureGates == nil {
newCfg.FeatureGates = make(map[string]bool)
}
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(newCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
return oldCfg
}
func setupKubeVirtDevicePluginOrFail(f *framework.Framework) *v1.Pod { func setupKubeVirtDevicePluginOrFail(f *framework.Framework) *v1.Pod {
e2enode.WaitForNodeToBeReady(f.ClientSet, framework.TestContext.NodeName, 5*time.Minute) e2enode.WaitForNodeToBeReady(f.ClientSet, framework.TestContext.NodeName, 5*time.Minute)

38
test/e2e_node/topology_manager_test.go
View File

@ -42,7 +42,6 @@ import (
e2epod "k8s.io/kubernetes/test/e2e/framework/pod" e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper" e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
e2etestfiles "k8s.io/kubernetes/test/e2e/framework/testfiles" e2etestfiles "k8s.io/kubernetes/test/e2e/framework/testfiles"
e2enodekubelet "k8s.io/kubernetes/test/e2e_node/kubeletconfig"
testutils "k8s.io/kubernetes/test/utils" testutils "k8s.io/kubernetes/test/utils"
"github.com/onsi/ginkgo" "github.com/onsi/ginkgo"
@ -194,7 +193,7 @@ func findNUMANodeWithoutSRIOVDevices(configMap *v1.ConfigMap, numaNodes int) (in
return findNUMANodeWithoutSRIOVDevicesFromSysfs(numaNodes) return findNUMANodeWithoutSRIOVDevicesFromSysfs(numaNodes)
} }
func configureTopologyManagerInKubelet(f *framework.Framework, oldCfg *kubeletconfig.KubeletConfiguration, policy, scope string, configMap *v1.ConfigMap, numaNodes int) string { func configureTopologyManagerInKubelet(oldCfg *kubeletconfig.KubeletConfiguration, policy, scope string, configMap *v1.ConfigMap, numaNodes int) (*kubeletconfig.KubeletConfiguration, string) {
// Configure Topology Manager in Kubelet with policy. // Configure Topology Manager in Kubelet with policy.
newCfg := oldCfg.DeepCopy() newCfg := oldCfg.DeepCopy()
if newCfg.FeatureGates == nil { if newCfg.FeatureGates == nil {
@ -204,8 +203,6 @@ func configureTopologyManagerInKubelet(f *framework.Framework, oldCfg *kubeletco
newCfg.FeatureGates["CPUManager"] = true newCfg.FeatureGates["CPUManager"] = true
newCfg.FeatureGates["TopologyManager"] = true newCfg.FeatureGates["TopologyManager"] = true
deleteStateFile()
// Set the Topology Manager policy // Set the Topology Manager policy
newCfg.TopologyManagerPolicy = policy newCfg.TopologyManagerPolicy = policy
@ -237,27 +234,7 @@ func configureTopologyManagerInKubelet(f *framework.Framework, oldCfg *kubeletco
// Dump the config -- debug // Dump the config -- debug
framework.Logf("New kubelet config is %s", *newCfg) framework.Logf("New kubelet config is %s", *newCfg)
ginkgo.By("Stopping the kubelet") return newCfg, newCfg.ReservedSystemCPUs
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(newCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
return newCfg.ReservedSystemCPUs
} }
// getSRIOVDevicePluginPod returns the Device Plugin pod for sriov resources in e2e tests. // getSRIOVDevicePluginPod returns the Device Plugin pod for sriov resources in e2e tests.
@ -910,7 +887,8 @@ func runTopologyManagerTests(f *framework.Framework) {
ginkgo.By(fmt.Sprintf("by configuring Topology Manager policy to %s", policy)) ginkgo.By(fmt.Sprintf("by configuring Topology Manager policy to %s", policy))
framework.Logf("Configuring topology Manager policy to %s", policy) framework.Logf("Configuring topology Manager policy to %s", policy)
configureTopologyManagerInKubelet(f, oldCfg, policy, scope, nil, 0) newCfg, _ := configureTopologyManagerInKubelet(oldCfg, policy, scope, nil, 0)
updateKubeletConfig(f, newCfg, true)
// Run the tests // Run the tests
runTopologyManagerPolicySuiteTests(f) runTopologyManagerPolicySuiteTests(f)
} }
@ -933,7 +911,8 @@ func runTopologyManagerTests(f *framework.Framework) {
ginkgo.By(fmt.Sprintf("by configuring Topology Manager policy to %s", policy)) ginkgo.By(fmt.Sprintf("by configuring Topology Manager policy to %s", policy))
framework.Logf("Configuring topology Manager policy to %s", policy) framework.Logf("Configuring topology Manager policy to %s", policy)
reservedSystemCPUs := configureTopologyManagerInKubelet(f, oldCfg, policy, scope, configMap, numaNodes) newCfg, reservedSystemCPUs := configureTopologyManagerInKubelet(oldCfg, policy, scope, configMap, numaNodes)
updateKubeletConfig(f, newCfg, true)
runTopologyManagerNodeAlignmentSuiteTests(f, sd, reservedSystemCPUs, policy, numaNodes, coreCount) runTopologyManagerNodeAlignmentSuiteTests(f, sd, reservedSystemCPUs, policy, numaNodes, coreCount)
} }
@ -950,14 +929,15 @@ func runTopologyManagerTests(f *framework.Framework) {
policy := topologymanager.PolicySingleNumaNode policy := topologymanager.PolicySingleNumaNode
scope := podScopeTopology scope := podScopeTopology
reservedSystemCPUs := configureTopologyManagerInKubelet(f, oldCfg, policy, scope, configMap, numaNodes) newCfg, reservedSystemCPUs := configureTopologyManagerInKubelet(oldCfg, policy, scope, configMap, numaNodes)
updateKubeletConfig(f, newCfg, true)
runTMScopeResourceAlignmentTestSuite(f, configMap, reservedSystemCPUs, policy, numaNodes, coreCount) runTMScopeResourceAlignmentTestSuite(f, configMap, reservedSystemCPUs, policy, numaNodes, coreCount)
}) })
ginkgo.AfterEach(func() { ginkgo.AfterEach(func() {
// restore kubelet config // restore kubelet config
setOldKubeletConfig(oldCfg) updateKubeletConfig(f, oldCfg, true)
}) })
} }

82
test/e2e_node/util.go
View File

@ -75,6 +75,9 @@ const (
defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb
kubeletReadOnlyPort = "10255" kubeletReadOnlyPort = "10255"
kubeletHealthCheckURL = "http://127.0.0.1:" + kubeletReadOnlyPort + "/healthz" kubeletHealthCheckURL = "http://127.0.0.1:" + kubeletReadOnlyPort + "/healthz"
// state files
cpuManagerStateFile = "/var/lib/kubelet/cpu_manager_state"
memoryManagerStateFile = "/var/lib/kubelet/memory_manager_state"
) )
func getNodeSummary() (*stats.Summary, error) { func getNodeSummary() (*stats.Summary, error) {
@ -158,58 +161,67 @@ func getCurrentKubeletConfig() (*kubeletconfig.KubeletConfiguration, error) {
// Returns true on success. // Returns true on success.
func tempSetCurrentKubeletConfig(f *framework.Framework, updateFunction func(initialConfig *kubeletconfig.KubeletConfiguration)) { func tempSetCurrentKubeletConfig(f *framework.Framework, updateFunction func(initialConfig *kubeletconfig.KubeletConfiguration)) {
var oldCfg *kubeletconfig.KubeletConfiguration var oldCfg *kubeletconfig.KubeletConfiguration
ginkgo.BeforeEach(func() { ginkgo.BeforeEach(func() {
oldCfg, err := getCurrentKubeletConfig() oldCfg, err := getCurrentKubeletConfig()
framework.ExpectNoError(err) framework.ExpectNoError(err)
newCfg := oldCfg.DeepCopy() newCfg := oldCfg.DeepCopy()
updateFunction(newCfg) updateFunction(newCfg)
if apiequality.Semantic.DeepEqual(*newCfg, *oldCfg) { if apiequality.Semantic.DeepEqual(*newCfg, *oldCfg) {
return return
} }
// Update the Kubelet configuration. updateKubeletConfig(f, newCfg, true)
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(newCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
}) })
ginkgo.AfterEach(func() { ginkgo.AfterEach(func() {
if oldCfg != nil { if oldCfg != nil {
// Update the Kubelet configuration. // Update the Kubelet configuration.
ginkgo.By("Stopping the kubelet") updateKubeletConfig(f, oldCfg, true)
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(oldCfg))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
} }
}) })
} }
func updateKubeletConfig(f *framework.Framework, kubeletConfig *kubeletconfig.KubeletConfiguration, deleteStateFiles bool) {
// Update the Kubelet configuration.
ginkgo.By("Stopping the kubelet")
startKubelet := stopKubelet()
// wait until the kubelet health check will fail
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, time.Minute, time.Second).Should(gomega.BeFalse())
// Delete CPU and memory manager state files to be sure it will not prevent the kubelet restart
if deleteStateFiles {
deleteStateFile(cpuManagerStateFile)
deleteStateFile(memoryManagerStateFile)
}
framework.ExpectNoError(e2enodekubelet.WriteKubeletConfigFile(kubeletConfig))
ginkgo.By("Starting the kubelet")
startKubelet()
// wait until the kubelet health check will succeed
gomega.Eventually(func() bool {
return kubeletHealthCheck(kubeletHealthCheckURL)
}, 2*time.Minute, 5*time.Second).Should(gomega.BeTrue())
// Wait for the Kubelet to be ready.
gomega.Eventually(func() bool {
nodes, err := e2enode.TotalReady(f.ClientSet)
framework.ExpectNoError(err)
return nodes == 1
}, time.Minute, time.Second).Should(gomega.BeTrue())
}
func deleteStateFile(stateFileName string) {
err := exec.Command("/bin/sh", "-c", fmt.Sprintf("rm -f %s", stateFileName)).Run()
framework.ExpectNoError(err, "failed to delete the state file")
}
// Returns true if kubeletConfig is enabled, false otherwise or if we cannot determine if it is. // Returns true if kubeletConfig is enabled, false otherwise or if we cannot determine if it is.
func isKubeletConfigEnabled(f *framework.Framework) (bool, error) { func isKubeletConfigEnabled(f *framework.Framework) (bool, error) {
cfgz, err := getCurrentKubeletConfig() cfgz, err := getCurrentKubeletConfig()