【Kubernetes】e2e 测试框架

End to End (e2e) 测试模拟用户行为操作 Kubernetes，用于保证 Kubernetes 服务或集群的行为完全符合设计。

E2E Framework

如何工作

查看 Kubernetes 源码即可了解如何使用 E2E 框架，入口点：

func TestE2E(t *testing.T) {
    RunE2ETests(t)
}

可以看到这是一个标准的Go Test，它调用 e2e.go 中定义的：

import (
    "k8s.io/klog"
 
    "github.com/onsi/ginkgo"
    "github.com/onsi/ginkgo/config"
    "github.com/onsi/ginkgo/reporters"
    "github.com/onsi/gomega"
 
    v1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    runtimeutils "k8s.io/apimachinery/pkg/util/runtime"
    "k8s.io/component-base/logs"
    "k8s.io/component-base/version"
    commontest "k8s.io/kubernetes/test/e2e/common"
    "k8s.io/kubernetes/test/e2e/framework"
    e2elog "k8s.io/kubernetes/test/e2e/framework/log"
    e2enode "k8s.io/kubernetes/test/e2e/framework/node"
    e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
    "k8s.io/kubernetes/test/e2e/manifest"
    e2ereporters "k8s.io/kubernetes/test/e2e/reporters"
    testutils "k8s.io/kubernetes/test/utils"
    utilnet "k8s.io/utils/net"
 
    clientset "k8s.io/client-go/kubernetes"
    // 确保Auth插件加载
    _ "k8s.io/client-go/plugin/pkg/client/auth"
 
    // ensure that cloud providers are loaded
    _ "k8s.io/kubernetes/test/e2e/framework/providers/aws"
    _ "k8s.io/kubernetes/test/e2e/framework/providers/azure"
    _ "k8s.io/kubernetes/test/e2e/framework/providers/gce"
    _ "k8s.io/kubernetes/test/e2e/framework/providers/kubemark"
    _ "k8s.io/kubernetes/test/e2e/framework/providers/openstack"
    _ "k8s.io/kubernetes/test/e2e/framework/providers/vsphere"
)
 
func RunE2ETests(t *testing.T) {
    // 控制HandleCrash函数的行为，设置为True导致panic
    runtimeutils.ReallyCrash = true
    // 初始化日志
    logs.InitLogs()
    // 刷空日志
    defer logs.FlushLogs()
    // 断言失败处理
    gomega.RegisterFailHandler(e2elog.Fail)
    // 除非明确通过命令行参数要求，否则跳过测试
    if config.GinkgoConfig.FocusString == "" && config.GinkgoConfig.SkipString == "" {
        config.GinkgoConfig.SkipString = `\[Flaky\]|\[Feature:.+\]`
    }
 
    // 初始化Reporter
    var r []ginkgo.Reporter
    if framework.TestContext.ReportDir != "" {
        if err := os.MkdirAll(framework.TestContext.ReportDir, 0755); err != nil {
            klog.Errorf("Failed creating report directory: %v", err)
        } else {
            r = append(r, reporters.NewJUnitReporter(path.Join(framework.TestContext.ReportDir, fmt.Sprintf("junit_%v%02d.xml", framework.TestContext.ReportPrefix, config.GinkgoConfig.ParallelNode))))
        }
    }
 
    // 测试进度信息输出到控制台，以及可选的外部URL
    r = append(r, e2ereporters.NewProgressReporter(framework.TestContext.ProgressReportURL))
    
    // 启动测试套件，使用Ginkgo默认Reporter + 自定义的Reporter
    ginkgo.RunSpecsWithDefaultAndCustomReporters(t, "Kubernetes e2e suite", r)
}

上述代码主要就是启动测试套件。实际执行Spec之前会调用下面的函数启动进行准备工作：

func setupSuite() {
    // Run only on Ginkgo node 1
 
    // GCE/GKE特殊处理
    switch framework.TestContext.Provider {
    case "gce", "gke":
        framework.LogClusterImageSources()
    }
    // 创建Clientset
    c, err := framework.LoadClientset()
    if err != nil {
        klog.Fatal("Error loading client: ", err)
    }
 
    // 删除所有K8S自带的命名空间，清除上次测试的残余
    if framework.TestContext.CleanStart {
                        // E2E框架提供大量便捷的API
        deleted, err := framework.DeleteNamespaces(c, nil, /* 支持过滤器 */
            []string{
                metav1.NamespaceSystem,
                metav1.NamespaceDefault,
                metav1.NamespacePublic,
                v1.NamespaceNodeLease,
            })
        if err != nil {
            // 直接失败
            framework.Failf("Error deleting orphaned namespaces: %v", err)
        }
        klog.Infof("Waiting for deletion of the following namespaces: %v", deleted)
                            // 有很多类似的，等待操作完成的函数
                                                                // 很多可用的常量
        if err := framework.WaitForNamespacesDeleted(c, deleted, framework.NamespaceCleanupTimeout); err != nil {
            framework.Failf("Failed to delete orphaned namespaces %v: %v", deleted, err)
        }
    }
 
    // 对于大型集群，执行到这里时，可能很多节点的路由还没有同步，因此不支持调度
    // 下面的方法等待直到所有节点可调度
    framework.ExpectNoError(framework.WaitForAllNodesSchedulable(c, framework.TestContext.NodeSchedulableTimeout))
 
    // 如果没有指定节点数量，自动计算
    if framework.TestContext.CloudConfig.NumNodes == framework.DefaultNumNodes {
        //            获取可调度节点数
        nodes, err := e2enode.GetReadySchedulableNodes(c)
        // 断言
        framework.ExpectNoError(err)
        framework.TestContext.CloudConfig.NumNodes = len(nodes.Items)
    }
 
    // 在测试之前，确保所有Pod已经Running/Ready，否则没有就绪的集群基础设施Pod可能阻止测试Pod
    // 运行
    podStartupTimeout := framework.TestContext.SystemPodsStartupTimeout
    if err := e2epod.WaitForPodsRunningReady(c, metav1.NamespaceSystem, int32(framework.TestContext.MinStartupPods), int32(framework.TestContext.AllowedNotReadyNodes), podStartupTimeout, map[string]string{}); err != nil {
        // Dump出指定命名空间的事件、Pod、节点信息
        framework.DumpAllNamespaceInfo(c, metav1.NamespaceSystem)
        // 对失败的容器执行kubectl logs                             Logf为Info级别日志器
        framework.LogFailedContainers(c, metav1.NamespaceSystem, framework.Logf)
        // 运行一个测试容器，尝试连接到API Server，等待此容器Ready，打印其标准输出，退出
        runKubernetesServiceTestContainer(c, metav1.NamespaceDefault)
        framework.Failf("Error waiting for all pods to be running and ready: %v", err)
    }
    //                  等待DaemonSets全部就绪
    if err := framework.WaitForDaemonSets(c, metav1.NamespaceSystem, int32(framework.TestContext.AllowedNotReadyNodes), framework.TestContext.SystemDaemonsetStartupTimeout); err != nil {
        framework.Logf("WARNING: Waiting for all daemonsets to be ready failed: %v", err)
    }
 
    // 打印服务器和客户端版本信息
    framework.Logf("e2e test version: %s", version.Get().GitVersion)
 
    dc := c.DiscoveryClient
 
    serverVersion, serverErr := dc.ServerVersion()
    if serverErr != nil {
        framework.Logf("Unexpected server error retrieving version: %v", serverErr)
    }
    if serverVersion != nil {
        framework.Logf("kube-apiserver version: %s", serverVersion.GitVersion)
    }
 
    if framework.TestContext.NodeKiller.Enabled {
        nodeKiller := framework.NewNodeKiller(framework.TestContext.NodeKiller, c, framework.TestContext.Provider)
        // NodeKiller负责周期性的模拟节点失败
        go nodeKiller.Run(framework.TestContext.NodeKiller.NodeKillerStopCh)
    }
}

setupSuite执行完毕之后，Ginkgo会运行子目录中的数千个Specs。

编写 spec

Kubernetes项目中有大量例子可以参考，例如：

package e2e
 
import (
    "fmt"
    "path/filepath"
    "sync"
    "time"
 
    rbacv1 "k8s.io/api/rbac/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime/schema"
    "k8s.io/apiserver/pkg/authentication/serviceaccount"
    clientset "k8s.io/client-go/kubernetes"
    podutil "k8s.io/kubernetes/pkg/api/v1/pod"
    commonutils "k8s.io/kubernetes/test/e2e/common"
    "k8s.io/kubernetes/test/e2e/framework"
    "k8s.io/kubernetes/test/e2e/framework/auth"
    e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
    "k8s.io/kubernetes/test/e2e/framework/testfiles"
 
    "github.com/onsi/ginkgo"
)
 
const (
    serverStartTimeout = framework.PodStartTimeout + 3*time.Minute
)
 
// 声明一个 ginkgo.Describe 块，自动添加[k8s.io] 标签
var _ = framework.KubeDescribe("[Feature:Example]", func() {
    // 创建一个新的Framework对象，自动提供：
    //   BeforeEach：创建K8S客户端、创建命名空间、启动资源用量收集器、指标收集器
    //   AfterEach：调用cleanupHandle、删除命名空间
    f := framework.NewDefaultFramework("examples")
 
    var c clientset.Interface
    var ns string
    // 自己可以添加额外的Setup/Teardown块
    ginkgo.BeforeEach(func() {
        // 获取客户端、使用的命名空间
        c = f.ClientSet
        ns = f.Namespace.Name
 
        // 在命名空间级别绑定RBAC权限，给default服务账号授权
        err := auth.BindClusterRoleInNamespace(c.RbacV1(), "edit", f.Namespace.Name,
            rbacv1.Subject{Kind: rbacv1.ServiceAccountKind, Namespace: f.Namespace.Name, Name: "default"})
        framework.ExpectNoError(err)
        // 等待操作完成
        err = auth.WaitForAuthorizationUpdate(c.AuthorizationV1(),
            serviceaccount.MakeUsername(f.Namespace.Name, "default"),
            f.Namespace.Name, "create", schema.GroupResource{Resource: "pods"}, true)
        // 断言
        framework.ExpectNoError(err)
    })
 
    // 嵌套的Describe
    framework.KubeDescribe("Liveness", func() {
        // 第一个Spec：测试健康检查失败的Pod能否自动重启
        ginkgo.It("liveness pods should be automatically restarted", func() {
            test := "test/fixtures/doc-yaml/user-guide/liveness"
            // 读取文件，Go Template形式，并解析为YAML资源清单
            execYaml := readFile(test, "exec-liveness.yaml.in")
            httpYaml := readFile(test, "http-liveness.yaml.in")
            nsFlag := fmt.Sprintf("--namespace=%v", ns)
 
            // 调用Kubectl来创建资源
            framework.RunKubectlOrDieInput(execYaml, "create", "-f", "-", nsFlag)
            framework.RunKubectlOrDieInput(httpYaml, "create", "-f", "-", nsFlag)
 
            // 并行测试
            var wg sync.WaitGroup
            passed := true
            // 此函数检查发生了重启
            checkRestart := func(podName string, timeout time.Duration) {
                // 等待Pod就绪
                err := e2epod.WaitForPodNameRunningInNamespace(c, podName, ns)
                framework.ExpectNoError(err)
                // 轮询知道重启次数大于0
                for t := time.Now(); time.Since(t) < timeout; time.Sleep(framework.Poll) {
                    pod, err := c.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
                    framework.ExpectNoError(err, fmt.Sprintf("getting pod %s", podName))
                    stat := podutil.GetExistingContainerStatus(pod.Status.ContainerStatuses, podName)
                    framework.Logf("Pod: %s, restart count:%d", stat.Name, stat.RestartCount)
                    if stat.RestartCount > 0 {
                        framework.Logf("Saw %v restart, succeeded...", podName)
                        wg.Done()
                        return
                    }
                }
                framework.Logf("Failed waiting for %v restart! ", podName)
                passed = false
                wg.Done()
            }
            // By用于添加一段文档说明
            ginkgo.By("Check restarts")
 
            // 检查两个Pod
            wg.Add(2)
            for _, c := range []string{"liveness-http", "liveness-exec"} {
                go checkRestart(c, 2*time.Minute)
            }
            wg.Wait()
            // 断言
            if !passed {
                framework.Failf("At least one liveness example failed.  See the logs above.")
            }
        })
    })
 
    framework.KubeDescribe("Secret", func() {
        // 第二个Spec，测试Pod能读取一个保密字典
        ginkgo.It("should create a pod that reads a secret", func() {
            test := "test/fixtures/doc-yaml/user-guide/secrets"
            secretYaml := readFile(test, "secret.yaml")
            podYaml := readFile(test, "secret-pod.yaml.in")
 
            nsFlag := fmt.Sprintf("--namespace=%v", ns)
            podName := "secret-test-pod"
 
            ginkgo.By("creating secret and pod")
            // 创建一个Secret，以及会读取此Secret并打印的Pod
            framework.RunKubectlOrDieInput(secretYaml, "create", "-f", "-", nsFlag)
            framework.RunKubectlOrDieInput(podYaml, "create", "-f", "-", nsFlag)
            // 等待Pod退出
            err := e2epod.WaitForPodNoLongerRunningInNamespace(c, podName, ns)
            framework.ExpectNoError(err)
 
            ginkgo.By("checking if secret was read correctly")
            // 检查Pod日志
            _, err = framework.LookForStringInLog(ns, "secret-test-pod", "test-container", "value-1", serverStartTimeout)
            framework.ExpectNoError(err)
        })
    })
})
 
func readFile(test, file string) string {
    from := filepath.Join(test, file)
    return commonutils.SubstituteImageName(string(testfiles.ReadOrDie(from)))
}

测试分类

可以为E2E测试添加标签，以区分类别：

标签	说明
无	测试可以快速（5m以内）完成，支持并行测试，具有一致性
[Slow]	执行比较慢的用例
[Serial]	不支持并发的测试用例，比如占用太多资源，还比如需要重启Node的
[Disruptive]	可能影响（例如重启组件、Taint节点）不是该测试自己创建的工作负载。任何Disruptive测试自动是Serial的
[Flaky]	标记测试中的问题难以短期修复。这种测试默认情况下不会运行，除非使用focus/skip参数
[Feature:.+]	如果一个测试运行/处理非核心功能，因此需要排除出标准测试套件，使用此标签。
[LinuxOnly]	需要使用Linux特有的特性

此外，任何测试都必须归属于某个SIG，并具有对应的 [sig-<name>] 标签。每个e2e的子包都在framework.go中SIGDescribe函数，来添加此标签：

// SIGDescribe annotates the test with the SIG label.
func SIGDescribe(text string, body func()) bool {
    return framework.KubeDescribe("[sig-node] "+text, body)
}

当然除了以上标签，还有个比较重要的标签就是 [Conformance], 此标签用于验收Kubernetes集群最小功能集。所以如果你有个私有部署的k8s集群，就可以通过这套用例来搞验收。方法也很简单，通过下面几步就可以执行：

# under kubernetes folder, compile test cases and ginkgo tool
make WHAT=test/e2e/e2e.test && make ginkgo

# setup for conformance tests
export KUBECONFIG=/path/to/kubeconfig
export KUBERNETES_CONFORMANCE_TEST=y
export KUBERNETES_PROVIDER=skeleton

# run all conformance tests
go run hack/e2e.go -v --test --test_args="--ginkgo.focus=\[Conformance\]"

注意，kubernetes的测试使用的镜像都放在GCR上了，如果你的集群在国内，且还不带FQ功能，那可能会发现pod会因为下载不了镜像而启动失败。

E2E Utils

此库是定义在 kubernetes/test/e2e/framework/util.go 中的一系列常量、变量、函数：

1. 常量：各种操作的超时值、集群节点数量、CPU剖析采样间隔
2. 变量：各种常用镜像的URL，例如BusyBox
3. 函数：
   1. 命名空间操控：
      1. CreateTestingNS：创建一个新的，供当前测试使用的命名空间
      2. DeleteNamespaces：删除命名空间
      3. CheckTestingNSDeletedExcept：检查所有e2e测试创建的命名空间出于Terminating状态，并且阻塞直到删除
      4. Cleanup：读取文件中的清单，从指定命名空间中删除它们，并检查命名空间中匹配指定selector的资源正确停止
   2. 节点操控：
      1. 设置Label、设置Taint
      2. RemoveLabelOffNode、RemoveTaintOffNode 删除Label/Taint
      3. AllNodesReady：检查所有节点是否就绪
      4. GetMasterHost：获取哦Master的主机名
      5. NodeHasTaint：判断节点是否具有Taint
   3. Pod操控：
      1. CreateEmptyFileOnPod：在Pod中创建文件
      2. LookForStringInPodExec：在Pod中执行命令并搜索输出
      3. LookForStringInLog：搜索Pod日志
      4. WaitForAllNodesSchedulabl：e等待节点可调度
   4. 日志和调试：
      1. CoreDump：登陆所有节点并保存日志到指定目录
      2. DumpDebugInfo：输出测试的调试信息
      3. DumpNodeDebugInfo：输出节点的调试信息
   5. 网络操控：
      1. BlockNetwork：通过操控Iptables阻塞两个节点之间的网络
      2. UnblockNetwork：解除阻塞
   6. 控制平面操控：
      1. RestartApiserver：重启API Server
      2. RestartKubelet：重启Kubelet
   7. 断言，若干Expect、Fail函数
   8. 收集CPU、内存等剖析信息Gather
   9. 执行Kubectl命令：KubectlCmd
   10. 创建K8S客户端：
       1. LoadConfig：加载K8S配置
       2. LoadClientset：创建客户端对象
   11. Ginkgo API封装：
       1. KubeDescribe
   12. 等待各种资源达到某种状态：Wait
   13. 其它杂项：
       1. OpenWebSocketForURL：打开WebSocket连接
       2. PrettyPrintJSON：格式化JSON
       3. Run：运行命令

其文档位于 https://godoc.org/k8s.io/kubernetes/test/e2e/framework 。

实战

下载 k8s 代码，并checkout到指定分支，这里以 release-1.18 为例

git clone https://github.com/kubernetes/kubernetes.git

git co -b release-1.18 origin/release-1.18

build e2e测试 manifest

hack/generate-bindata.sh

生成 e2e.test 二进制

cd test/e2e && GOOS=linux GOARCH=amd64 go test -c -o e2e.test -v

登陆到待测试k8s集群节点，上传e2e.test到指定节点，配置e2e.test需要的镜像仓库列表

export KUBE_TEST_REPO_LIST=/home/ubuntu/e2e/repos.yaml //导出环境变量
cat > repos.yaml <<EOF
promoterE2eRegistry: ccr.ccs.tencentyun.com/e2e-test-images
EOF

执行e2e测试，这里以networkpolicy为例

./e2e.test -ginkgo.focus=NetworkPolicy --disable-log-dump --provider="skeleton" --kubeconfig="/root/.kube/config" > >(tee e2e.log)

也可以使用 ginkgo

./ginkgo --focus="NetworkPolicy" ./e2e.test -- --disable-log-dump --provider="skeleton" --kubeconfig="/root/.kube/config" > >(tee e2e.log)

总结

研究Kubernetes的 e2e 测试框架，然后类比我们以往的经验，下面几点特性还是值得借鉴的:

All e2e compiled into one binary

在对服务端程序进行API测试时，我们经常会针对每个服务都创建一个ginkgo suite来框定测试用例的范围，这样做的好处是用例目标非常清晰，但是随着服务数量的增多，这样的suite会越来越来多。从组织上，看起来就稍显杂乱，而且不利于测试服务的输出。

比如，我们考虑这么一个场景，QA需要对新机房部署，或者私有机房进行服务验证。这时候，就通常需要copy所有代码到指定集群在运行了，非常的不方便，而且也容易造成代码泄露。

kubernetes显然也会有这个需求，所以他们改变写法，将所有的测试用例都编译进一个 e2e.test 的二进制，这样针对上面场景时，就可以直接使用这个可执行文件来操作，非常的方便。

当然可执行文件的方便少不了外部参数的自由注入，以及整体测试用例的精心标记。否则，测试代码写的不规范，需要频繁的针对特定环境修改，也是特别不方便的。

Each case has a uniqe namespace

为每条测试用例创建一个独立的空间，是kubernetes e2e framework的一大精华。每条测试用例独享一个空间，彼此不冲突，从而根本上避免并发困扰，借助ginkgo的CLI来运行，会极大的提高执行效率。

而且这处代码的方式也非常优美，很有借鉴价值:

func NewFramework(baseName string, options FrameworkOptions, client clientset.Interface) *Framework {
   f := &Framework{
      BaseName:                 baseName,
      AddonResourceConstraints: make(map[string]ResourceConstraint),
      Options:                  options,
      ClientSet:                client,
   }
 
   BeforeEach(f.BeforeEach)
   AfterEach(f.AfterEach)
 
   return f
}

利用 ginkgo 的 BeforeEach 的嵌套特定，虽然在Describe下就定义framework的初始化(如下)，但是在每个 It 执行前，上面的BeforeEach才会真正执行，所以并不会有冲突:

var _ = framework.KubeDescribe("GKE local SSD [Feature:GKELocalSSD]", func() {
   f := framework.NewDefaultFramework("localssd")
   It("should write and read from node local SSD [Feature:GKELocalSSD]", func() {
   ...
   })
})

当然 e2e 框架还负责case执行完的环境清理，并且是按需灵活配置。比如你希望，case失败保留现场,不删除namespace，那么就可以设置flag 参数 delete-namespace-on-failure 为false来实现。

Asynchronous wait

几乎所有的Kubernetes操作都是异步的，所以不管是产品代码还是测试用例，都广泛的使用了这个异步等待库：kubernetes/vendor/k8s.io/apimachinery/pkg/util/wait。这个库，实现简单，精悍，非常值得学习。

另外，针对测试的异步验证，其实ginkgo(gomega)本身提供的Eventualy，也是非常好用的。

Suitable logs

Kubernetes e2e 主要使用两种方式输出log，一个是使用glog库，另一个则是 framework.Logf 方法。glog本身是golang官方提供的log库，使用比较灵活。但是这里主要推荐的还是Framework.Logf。因为使用此方法的log会输出到GinkgoWriter里面，这样当我们使用ginkgo.RunSpecsWithDefaultAndCustomReporters方法时，log不光输出到控制台，也会保存在junit格式的xml文件里，非常方便在jenkins里展示测试结果。

Clean code

我们从Kubernetes e2e能看到很多好的借鉴，比如：

• 抽取主干方法，以突出测试用例主体
• 采用数据驱动方式书写共性测试用例
• 注释工整，多少适宜
• 不输出低级别log
• 代码行长短适宜
• 方法名定义清晰，可读性强

参考资料

• Kubernetes End-to-end Testing for Everyone
• KEP: NetworkPolicy verification rearchitecture
• https://github.com/kubernetes/kubernetes/blob/master/test/e2e/network/network_policy.go
• https://blog.csdn.net/shida_csdn/article/details/107198325
• https://events19.linuxfoundation.org/wp-content/uploads/2017/11/Verify-Your-Kubernetes-Clusters-with-Upstream-e2e-Tests-Kenichi-Omichi-NEC.pdf
• https://jimmysong.io/kubernetes-handbook/develop/testing.html
• https://blog.gmem.cc/kubernetes-e2e-test