123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461 |
- /*
- Copyright 2016 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 e2e_node
- import (
- "encoding/json"
- "flag"
- "fmt"
- "io/ioutil"
- "net/http"
- "os/exec"
- "regexp"
- "strings"
- "time"
- "golang.org/x/net/context"
- "k8s.io/klog"
- apiv1 "k8s.io/api/core/v1"
- apiequality "k8s.io/apimachinery/pkg/api/equality"
- metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/apimachinery/pkg/util/sets"
- utilfeature "k8s.io/apiserver/pkg/util/feature"
- "k8s.io/client-go/kubernetes/scheme"
- "k8s.io/component-base/featuregate"
- internalapi "k8s.io/cri-api/pkg/apis"
- kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
- "k8s.io/kubernetes/pkg/features"
- kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
- "k8s.io/kubernetes/pkg/kubelet/apis/podresources"
- podresourcesapi "k8s.io/kubernetes/pkg/kubelet/apis/podresources/v1alpha1"
- stats "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
- "k8s.io/kubernetes/pkg/kubelet/cm"
- kubeletconfigcodec "k8s.io/kubernetes/pkg/kubelet/kubeletconfig/util/codec"
- kubeletmetrics "k8s.io/kubernetes/pkg/kubelet/metrics"
- "k8s.io/kubernetes/pkg/kubelet/remote"
- "k8s.io/kubernetes/pkg/kubelet/util"
- "k8s.io/kubernetes/test/e2e/framework"
- e2elog "k8s.io/kubernetes/test/e2e/framework/log"
- frameworkmetrics "k8s.io/kubernetes/test/e2e/framework/metrics"
- imageutils "k8s.io/kubernetes/test/utils/image"
- . "github.com/onsi/ginkgo"
- . "github.com/onsi/gomega"
- )
- // TODO(random-liu): Get this automatically from kubelet flag.
- var kubeletAddress = flag.String("kubelet-address", "http://127.0.0.1:10255", "Host and port of the kubelet")
- var startServices = flag.Bool("start-services", true, "If true, start local node services")
- var stopServices = flag.Bool("stop-services", true, "If true, stop local node services after running tests")
- var busyboxImage = imageutils.GetE2EImage(imageutils.BusyBox)
- var perlImage = imageutils.GetE2EImage(imageutils.Perl)
- const (
- // Kubelet internal cgroup name for node allocatable cgroup.
- defaultNodeAllocatableCgroup = "kubepods"
- // defaultPodResourcesPath is the path to the local endpoint serving the podresources GRPC service.
- defaultPodResourcesPath = "/var/lib/kubelet/pod-resources"
- defaultPodResourcesTimeout = 10 * time.Second
- defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb
- )
- func getNodeSummary() (*stats.Summary, error) {
- req, err := http.NewRequest("GET", *kubeletAddress+"/stats/summary", nil)
- if err != nil {
- return nil, fmt.Errorf("failed to build http request: %v", err)
- }
- req.Header.Add("Accept", "application/json")
- client := &http.Client{}
- resp, err := client.Do(req)
- if err != nil {
- return nil, fmt.Errorf("failed to get /stats/summary: %v", err)
- }
- defer resp.Body.Close()
- contentsBytes, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return nil, fmt.Errorf("failed to read /stats/summary: %+v", resp)
- }
- decoder := json.NewDecoder(strings.NewReader(string(contentsBytes)))
- summary := stats.Summary{}
- err = decoder.Decode(&summary)
- if err != nil {
- return nil, fmt.Errorf("failed to parse /stats/summary to go struct: %+v", resp)
- }
- return &summary, nil
- }
- func getNodeDevices() (*podresourcesapi.ListPodResourcesResponse, error) {
- endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
- if err != nil {
- return nil, fmt.Errorf("Error getting local endpoint: %v", err)
- }
- client, conn, err := podresources.GetClient(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
- if err != nil {
- return nil, fmt.Errorf("Error getting grpc client: %v", err)
- }
- defer conn.Close()
- ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
- defer cancel()
- resp, err := client.List(ctx, &podresourcesapi.ListPodResourcesRequest{})
- if err != nil {
- return nil, fmt.Errorf("%v.Get(_) = _, %v", client, err)
- }
- return resp, nil
- }
- // Returns the current KubeletConfiguration
- func getCurrentKubeletConfig() (*kubeletconfig.KubeletConfiguration, error) {
- resp := pollConfigz(5*time.Minute, 5*time.Second)
- kubeCfg, err := decodeConfigz(resp)
- if err != nil {
- return nil, err
- }
- return kubeCfg, nil
- }
- // Must be called within a Context. Allows the function to modify the KubeletConfiguration during the BeforeEach of the context.
- // The change is reverted in the AfterEach of the context.
- // Returns true on success.
- func tempSetCurrentKubeletConfig(f *framework.Framework, updateFunction func(initialConfig *kubeletconfig.KubeletConfiguration)) {
- var oldCfg *kubeletconfig.KubeletConfiguration
- BeforeEach(func() {
- configEnabled, err := isKubeletConfigEnabled(f)
- framework.ExpectNoError(err)
- Expect(configEnabled).To(BeTrue(), "The Dynamic Kubelet Configuration feature is not enabled.\n"+
- "Pass --feature-gates=DynamicKubeletConfig=true to the Kubelet to enable this feature.\n"+
- "For `make test-e2e-node`, you can set `TEST_ARGS='--feature-gates=DynamicKubeletConfig=true'`.")
- oldCfg, err = getCurrentKubeletConfig()
- framework.ExpectNoError(err)
- newCfg := oldCfg.DeepCopy()
- updateFunction(newCfg)
- if apiequality.Semantic.DeepEqual(*newCfg, *oldCfg) {
- return
- }
- framework.ExpectNoError(setKubeletConfiguration(f, newCfg))
- })
- AfterEach(func() {
- if oldCfg != nil {
- err := setKubeletConfiguration(f, oldCfg)
- framework.ExpectNoError(err)
- }
- })
- }
- // Returns true if kubeletConfig is enabled, false otherwise or if we cannot determine if it is.
- func isKubeletConfigEnabled(f *framework.Framework) (bool, error) {
- cfgz, err := getCurrentKubeletConfig()
- if err != nil {
- return false, fmt.Errorf("could not determine whether 'DynamicKubeletConfig' feature is enabled, err: %v", err)
- }
- v, ok := cfgz.FeatureGates[string(features.DynamicKubeletConfig)]
- if !ok {
- return true, nil
- }
- return v, nil
- }
- // Creates or updates the configmap for KubeletConfiguration, waits for the Kubelet to restart
- // with the new configuration. Returns an error if the configuration after waiting for restartGap
- // doesn't match what you attempted to set, or if the dynamic configuration feature is disabled.
- // You should only call this from serial tests.
- func setKubeletConfiguration(f *framework.Framework, kubeCfg *kubeletconfig.KubeletConfiguration) error {
- const (
- restartGap = 40 * time.Second
- pollInterval = 5 * time.Second
- )
- // make sure Dynamic Kubelet Configuration feature is enabled on the Kubelet we are about to reconfigure
- if configEnabled, err := isKubeletConfigEnabled(f); err != nil {
- return err
- } else if !configEnabled {
- return fmt.Errorf("The Dynamic Kubelet Configuration feature is not enabled.\n" +
- "Pass --feature-gates=DynamicKubeletConfig=true to the Kubelet to enable this feature.\n" +
- "For `make test-e2e-node`, you can set `TEST_ARGS='--feature-gates=DynamicKubeletConfig=true'`.")
- }
- // create the ConfigMap with the new configuration
- cm, err := createConfigMap(f, kubeCfg)
- if err != nil {
- return err
- }
- // create the reference and set Node.Spec.ConfigSource
- src := &apiv1.NodeConfigSource{
- ConfigMap: &apiv1.ConfigMapNodeConfigSource{
- Namespace: "kube-system",
- Name: cm.Name,
- KubeletConfigKey: "kubelet",
- },
- }
- // set the source, retry a few times in case we are competing with other writers
- Eventually(func() error {
- if err := setNodeConfigSource(f, src); err != nil {
- return err
- }
- return nil
- }, time.Minute, time.Second).Should(BeNil())
- // poll for new config, for a maximum wait of restartGap
- Eventually(func() error {
- newKubeCfg, err := getCurrentKubeletConfig()
- if err != nil {
- return fmt.Errorf("failed trying to get current Kubelet config, will retry, error: %v", err)
- }
- if !apiequality.Semantic.DeepEqual(*kubeCfg, *newKubeCfg) {
- return fmt.Errorf("still waiting for new configuration to take effect, will continue to watch /configz")
- }
- klog.Infof("new configuration has taken effect")
- return nil
- }, restartGap, pollInterval).Should(BeNil())
- return nil
- }
- // sets the current node's configSource, this should only be called from Serial tests
- func setNodeConfigSource(f *framework.Framework, source *apiv1.NodeConfigSource) error {
- // since this is a serial test, we just get the node, change the source, and then update it
- // this prevents any issues with the patch API from affecting the test results
- nodeclient := f.ClientSet.CoreV1().Nodes()
- // get the node
- node, err := nodeclient.Get(framework.TestContext.NodeName, metav1.GetOptions{})
- if err != nil {
- return err
- }
- // set new source
- node.Spec.ConfigSource = source
- // update to the new source
- _, err = nodeclient.Update(node)
- if err != nil {
- return err
- }
- return nil
- }
- // Causes the test to fail, or returns a status 200 response from the /configz endpoint
- func pollConfigz(timeout time.Duration, pollInterval time.Duration) *http.Response {
- endpoint := fmt.Sprintf("http://127.0.0.1:8080/api/v1/nodes/%s/proxy/configz", framework.TestContext.NodeName)
- client := &http.Client{}
- req, err := http.NewRequest("GET", endpoint, nil)
- framework.ExpectNoError(err)
- req.Header.Add("Accept", "application/json")
- var resp *http.Response
- Eventually(func() bool {
- resp, err = client.Do(req)
- if err != nil {
- klog.Errorf("Failed to get /configz, retrying. Error: %v", err)
- return false
- }
- if resp.StatusCode != 200 {
- klog.Errorf("/configz response status not 200, retrying. Response was: %+v", resp)
- return false
- }
- return true
- }, timeout, pollInterval).Should(Equal(true))
- return resp
- }
- // Decodes the http response from /configz and returns a kubeletconfig.KubeletConfiguration (internal type).
- func decodeConfigz(resp *http.Response) (*kubeletconfig.KubeletConfiguration, error) {
- // This hack because /configz reports the following structure:
- // {"kubeletconfig": {the JSON representation of kubeletconfigv1beta1.KubeletConfiguration}}
- type configzWrapper struct {
- ComponentConfig kubeletconfigv1beta1.KubeletConfiguration `json:"kubeletconfig"`
- }
- configz := configzWrapper{}
- kubeCfg := kubeletconfig.KubeletConfiguration{}
- contentsBytes, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return nil, err
- }
- err = json.Unmarshal(contentsBytes, &configz)
- if err != nil {
- return nil, err
- }
- err = scheme.Scheme.Convert(&configz.ComponentConfig, &kubeCfg, nil)
- if err != nil {
- return nil, err
- }
- return &kubeCfg, nil
- }
- // creates a configmap containing kubeCfg in kube-system namespace
- func createConfigMap(f *framework.Framework, internalKC *kubeletconfig.KubeletConfiguration) (*apiv1.ConfigMap, error) {
- cmap := newKubeletConfigMap("testcfg", internalKC)
- cmap, err := f.ClientSet.CoreV1().ConfigMaps("kube-system").Create(cmap)
- if err != nil {
- return nil, err
- }
- return cmap, nil
- }
- // constructs a ConfigMap, populating one of its keys with the KubeletConfiguration. Always uses GenerateName to generate a suffix.
- func newKubeletConfigMap(name string, internalKC *kubeletconfig.KubeletConfiguration) *apiv1.ConfigMap {
- data, err := kubeletconfigcodec.EncodeKubeletConfig(internalKC, kubeletconfigv1beta1.SchemeGroupVersion)
- framework.ExpectNoError(err)
- cmap := &apiv1.ConfigMap{
- ObjectMeta: metav1.ObjectMeta{GenerateName: name + "-"},
- Data: map[string]string{
- "kubelet": string(data),
- },
- }
- return cmap
- }
- func logPodEvents(f *framework.Framework) {
- e2elog.Logf("Summary of pod events during the test:")
- err := framework.ListNamespaceEvents(f.ClientSet, f.Namespace.Name)
- framework.ExpectNoError(err)
- }
- func logNodeEvents(f *framework.Framework) {
- e2elog.Logf("Summary of node events during the test:")
- err := framework.ListNamespaceEvents(f.ClientSet, "")
- framework.ExpectNoError(err)
- }
- func getLocalNode(f *framework.Framework) *apiv1.Node {
- nodeList := framework.GetReadySchedulableNodesOrDie(f.ClientSet)
- Expect(len(nodeList.Items)).To(Equal(1), "Unexpected number of node objects for node e2e. Expects only one node.")
- return &nodeList.Items[0]
- }
- // logKubeletLatencyMetrics logs KubeletLatencyMetrics computed from the Prometheus
- // metrics exposed on the current node and identified by the metricNames.
- // The Kubelet subsystem prefix is automatically prepended to these metric names.
- func logKubeletLatencyMetrics(metricNames ...string) {
- metricSet := sets.NewString()
- for _, key := range metricNames {
- metricSet.Insert(kubeletmetrics.KubeletSubsystem + "_" + key)
- }
- metric, err := frameworkmetrics.GrabKubeletMetricsWithoutProxy(framework.TestContext.NodeName+":10255", "/metrics")
- if err != nil {
- e2elog.Logf("Error getting kubelet metrics: %v", err)
- } else {
- e2elog.Logf("Kubelet Metrics: %+v", framework.GetKubeletLatencyMetrics(metric, metricSet))
- }
- }
- // returns config related metrics from the local kubelet, filtered to the filterMetricNames passed in
- func getKubeletMetrics(filterMetricNames sets.String) (frameworkmetrics.KubeletMetrics, error) {
- // grab Kubelet metrics
- ms, err := frameworkmetrics.GrabKubeletMetricsWithoutProxy(framework.TestContext.NodeName+":10255", "/metrics")
- if err != nil {
- return nil, err
- }
- filtered := frameworkmetrics.NewKubeletMetrics()
- for name := range ms {
- if !filterMetricNames.Has(name) {
- continue
- }
- filtered[name] = ms[name]
- }
- return filtered, nil
- }
- // runCommand runs the cmd and returns the combined stdout and stderr, or an
- // error if the command failed.
- func runCommand(cmd ...string) (string, error) {
- output, err := exec.Command(cmd[0], cmd[1:]...).CombinedOutput()
- if err != nil {
- return "", fmt.Errorf("failed to run %q: %s (%s)", strings.Join(cmd, " "), err, output)
- }
- return string(output), nil
- }
- // getCRIClient connects CRI and returns CRI runtime service clients and image service client.
- func getCRIClient() (internalapi.RuntimeService, internalapi.ImageManagerService, error) {
- // connection timeout for CRI service connection
- const connectionTimeout = 2 * time.Minute
- runtimeEndpoint := framework.TestContext.ContainerRuntimeEndpoint
- r, err := remote.NewRemoteRuntimeService(runtimeEndpoint, connectionTimeout)
- if err != nil {
- return nil, nil, err
- }
- imageManagerEndpoint := runtimeEndpoint
- if framework.TestContext.ImageServiceEndpoint != "" {
- //ImageServiceEndpoint is the same as ContainerRuntimeEndpoint if not
- //explicitly specified
- imageManagerEndpoint = framework.TestContext.ImageServiceEndpoint
- }
- i, err := remote.NewRemoteImageService(imageManagerEndpoint, connectionTimeout)
- if err != nil {
- return nil, nil, err
- }
- return r, i, nil
- }
- // TODO: Find a uniform way to deal with systemctl/initctl/service operations. #34494
- func restartKubelet() {
- stdout, err := exec.Command("sudo", "systemctl", "list-units", "kubelet*", "--state=running").CombinedOutput()
- framework.ExpectNoError(err)
- regex := regexp.MustCompile("(kubelet-\\w+)")
- matches := regex.FindStringSubmatch(string(stdout))
- Expect(len(matches)).NotTo(BeZero())
- kube := matches[0]
- e2elog.Logf("Get running kubelet with systemctl: %v, %v", string(stdout), kube)
- stdout, err = exec.Command("sudo", "systemctl", "restart", kube).CombinedOutput()
- framework.ExpectNoError(err, "Failed to restart kubelet with systemctl: %v, %v", err, stdout)
- }
- func toCgroupFsName(cgroupName cm.CgroupName) string {
- if framework.TestContext.KubeletConfig.CgroupDriver == "systemd" {
- return cgroupName.ToSystemd()
- } else {
- return cgroupName.ToCgroupfs()
- }
- }
- // reduceAllocatableMemoryUsage uses memory.force_empty (https://lwn.net/Articles/432224/)
- // to make the kernel reclaim memory in the allocatable cgroup
- // the time to reduce pressure may be unbounded, but usually finishes within a second
- func reduceAllocatableMemoryUsage() {
- cmd := fmt.Sprintf("echo 0 > /sys/fs/cgroup/memory/%s/memory.force_empty", toCgroupFsName(cm.NewCgroupName(cm.RootCgroupName, defaultNodeAllocatableCgroup)))
- _, err := exec.Command("sudo", "sh", "-c", cmd).CombinedOutput()
- framework.ExpectNoError(err)
- }
- // Equivalent of featuregatetesting.SetFeatureGateDuringTest
- // which can't be used here because we're not in a Testing context.
- // This must be in a non-"_test" file to pass
- // make verify WHAT=test-featuregates
- func withFeatureGate(feature featuregate.Feature, desired bool) func() {
- current := utilfeature.DefaultFeatureGate.Enabled(feature)
- utilfeature.DefaultMutableFeatureGate.Set(fmt.Sprintf("%s=%v", string(feature), desired))
- return func() {
- utilfeature.DefaultMutableFeatureGate.Set(fmt.Sprintf("%s=%v", string(feature), current))
- }
- }
|