Merge pull request #94723 from knight42/refactor/e2e-node

refactor: migrate node e2e tests off insecure port

test/e2e_node/e2e_node_suite_test.go

@@ -56,18 +56,21 @@ import (
 	"k8s.io/klog/v2"
 )
 
-var e2es *services.E2EServices
+var (
+	e2es *services.E2EServices
 
-// TODO(random-liu): Change the following modes to sub-command.
-var runServicesMode = flag.Bool("run-services-mode", false, "If true, only run services (etcd, apiserver) in current process, and not run test.")
-var runKubeletMode = flag.Bool("run-kubelet-mode", false, "If true, only start kubelet, and not run test.")
-var systemValidateMode = flag.Bool("system-validate-mode", false, "If true, only run system validation in current process, and not run test.")
-var systemSpecFile = flag.String("system-spec-file", "", "The name of the system spec file that will be used for node conformance test. If it's unspecified or empty, the default system spec (system.DefaultSysSpec) will be used.")
+	// TODO(random-liu): Change the following modes to sub-command.
+	runServicesMode    = flag.Bool("run-services-mode", false, "If true, only run services (etcd, apiserver) in current process, and not run test.")
+	runKubeletMode     = flag.Bool("run-kubelet-mode", false, "If true, only start kubelet, and not run test.")
+	systemValidateMode = flag.Bool("system-validate-mode", false, "If true, only run system validation in current process, and not run test.")
+	systemSpecFile     = flag.String("system-spec-file", "", "The name of the system spec file that will be used for node conformance test. If it's unspecified or empty, the default system spec (system.DefaultSysSpec) will be used.")
+)
 
 // registerNodeFlags registers flags specific to the node e2e test suite.
 func registerNodeFlags(flags *flag.FlagSet) {
 	// Mark the test as node e2e when node flags are api.Registry.
 	framework.TestContext.NodeE2E = true
+	flags.StringVar(&framework.TestContext.BearerToken, "bearer-token", "", "The bearer token to authenticate with. If not specified, it would be a random token. Currently this token is only used in node e2e tests.")
 	flags.StringVar(&framework.TestContext.NodeName, "node-name", "", "Name of the node to run tests on.")
 	// TODO(random-liu): Move kubelet start logic out of the test.
 	// TODO(random-liu): Move log fetch logic out of the test.
@@ -205,8 +208,12 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
 	// Reference common test to make the import valid.
 	commontest.CurrentSuite = commontest.NodeE2E
 
-	return nil
-}, func([]byte) {
+	// ginkgo would spawn multiple processes to run tests.
+	// Since the bearer token is generated randomly at run time,
+	// we need to distribute the bearer token to other processes to make them use the same token.
+	return []byte(framework.TestContext.BearerToken)
+}, func(token []byte) {
+	framework.TestContext.BearerToken = string(token)
 	// update test context with node configuration.
 	gomega.Expect(updateTestContext()).To(gomega.Succeed(), "update test context with node config.")
 })

test/e2e_node/services/apiserver.go

@@ -18,18 +18,17 @@ package services
 
 import (
 	"fmt"
+	"io/ioutil"
 	"net"
 
 	"k8s.io/apiserver/pkg/storage/storagebackend"
+
 	apiserver "k8s.io/kubernetes/cmd/kube-apiserver/app"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
+	"k8s.io/kubernetes/test/e2e/framework"
 )
 
-const (
-	clusterIPRange          = "10.0.0.1/24"
-	apiserverClientURL      = "http://localhost:8080"
-	apiserverHealthCheckURL = apiserverClientURL + "/healthz"
-)
+const clusterIPRange = "10.0.0.1/24"
 
 // APIServer is a server which manages apiserver.
 type APIServer struct {
@@ -47,14 +46,23 @@ func NewAPIServer(storageConfig storagebackend.Config) *APIServer {
 
 // Start starts the apiserver, returns when apiserver is ready.
 func (a *APIServer) Start() error {
+	const tokenFilePath = "known_tokens.csv"
+
 	o := options.NewServerRunOptions()
 	o.Etcd.StorageConfig = a.storageConfig
 	_, ipnet, err := net.ParseCIDR(clusterIPRange)
 	if err != nil {
 		return err
 	}
+	o.SecureServing.BindAddress = net.ParseIP("127.0.0.1")
+	// Disable insecure serving
+	o.InsecureServing.BindPort = 0
 	o.ServiceClusterIPRanges = ipnet.String()
 	o.AllowPrivileged = true
+	if err := generateTokenFile(tokenFilePath); err != nil {
+		return fmt.Errorf("failed to generate token file %s: %v", tokenFilePath, err)
+	}
+	o.Authentication.TokenFile.TokenFile = tokenFilePath
 	o.Admission.GenericAdmission.DisablePlugins = []string{"ServiceAccount", "TaintNodesByCondition"}
 	errCh := make(chan error)
 	go func() {
@@ -71,7 +79,7 @@ func (a *APIServer) Start() error {
 		}
 	}()
 
-	err = readinessCheck("apiserver", []string{apiserverHealthCheckURL}, errCh)
+	err = readinessCheck("apiserver", []string{getAPIServerHealthCheckURL()}, errCh)
 	if err != nil {
 		return err
 	}
@@ -96,9 +104,14 @@ func (a *APIServer) Name() string {
 }
 
 func getAPIServerClientURL() string {
-	return apiserverClientURL
+	return framework.TestContext.Host
 }
 
 func getAPIServerHealthCheckURL() string {
-	return apiserverHealthCheckURL
+	return framework.TestContext.Host + "/healthz"
+}
+
+func generateTokenFile(tokenFilePath string) error {
+	tokenFile := fmt.Sprintf("%s,kubelet,uid,system:masters\n", framework.TestContext.BearerToken)
+	return ioutil.WriteFile(tokenFilePath, []byte(tokenFile), 0644)
 }

test/e2e_node/services/kubelet.go

@@ -27,12 +27,12 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
-	"k8s.io/klog/v2"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	cliflag "k8s.io/component-base/cli/flag"
+	"k8s.io/klog/v2"
 	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
+
 	"k8s.io/kubernetes/cmd/kubelet/app/options"
 	"k8s.io/kubernetes/pkg/features"
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
@@ -356,13 +356,15 @@ func createPodDirectory() (string, error) {
 
 // createKubeconfig creates a kubeconfig file at the fully qualified `path`. The parent dirs must exist.
 func createKubeconfig(path string) error {
-	kubeconfig := []byte(`apiVersion: v1
+	kubeconfig := []byte(fmt.Sprintf(`apiVersion: v1
 kind: Config
 users:
 - name: kubelet
+  user:
+    token: %s
 clusters:
 - cluster:
-    server: ` + getAPIServerClientURL() + `
+    server: %s
     insecure-skip-tls-verify: true
   name: local
 contexts:
@@ -370,7 +372,7 @@ contexts:
     cluster: local
     user: kubelet
   name: local-context
-current-context: local-context`)
+current-context: local-context`, framework.TestContext.BearerToken, getAPIServerClientURL()))
 
 	if err := ioutil.WriteFile(path, kubeconfig, 0666); err != nil {
 		return err

test/e2e_node/services/namespace_controller.go

@@ -19,12 +19,13 @@ package services
 import (
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/metadata"
 	restclient "k8s.io/client-go/rest"
 	namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
+	"k8s.io/kubernetes/test/e2e/framework"
 )
 
 const (
@@ -49,7 +50,13 @@ func NewNamespaceController(host string) *NamespaceController {
 
 // Start starts the namespace controller.
 func (n *NamespaceController) Start() error {
-	config := restclient.AddUserAgent(&restclient.Config{Host: n.host}, ncName)
+	config := restclient.AddUserAgent(&restclient.Config{
+		Host:        n.host,
+		BearerToken: framework.TestContext.BearerToken,
+		TLSClientConfig: restclient.TLSClientConfig{
+			Insecure: true,
+		},
+	}, ncName)
 
 	// the namespace cleanup controller is very chatty.  It makes lots of discovery calls and then it makes lots of delete calls.
 	config.QPS = 50

test/e2e_node/services/services.go

@@ -24,9 +24,9 @@ import (
 	"path"
 	"testing"
 
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog/v2"
 
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/test/e2e/framework"
 )
 
@@ -65,14 +65,16 @@ func NewE2EServices(monitorParent bool) *E2EServices {
 func (e *E2EServices) Start() error {
 	var err error
 	if !framework.TestContext.NodeConformance {
+		if e.services, err = e.startInternalServices(); err != nil {
+			return fmt.Errorf("failed to start internal services: %v", err)
+		}
 		// Start kubelet
 		e.kubelet, err = e.startKubelet()
 		if err != nil {
 			return fmt.Errorf("failed to start kubelet: %v", err)
 		}
 	}
-	e.services, err = e.startInternalServices()
-	return err
+	return nil
 }
 
 // Stop stops the e2e services.
@@ -129,7 +131,11 @@ func (e *E2EServices) startInternalServices() (*server, error) {
 		return nil, fmt.Errorf("can't get current binary: %v", err)
 	}
 	// Pass all flags into the child process, so that it will see the same flag set.
-	startCmd := exec.Command(testBin, append([]string{"--run-services-mode"}, os.Args[1:]...)...)
+	startCmd := exec.Command(testBin,
+		append(
+			[]string{"--run-services-mode", fmt.Sprintf("--bearer-token=%s", framework.TestContext.BearerToken)},
+			os.Args[1:]...,
+		)...)
 	server := newServer("services", startCmd, nil, nil, getServicesHealthCheckURLs(), servicesLogFile, e.monitorParent, false)
 	return server, server.start()
 }

test/e2e_node/services/util.go

@@ -17,13 +17,17 @@ limitations under the License.
 package services
 
 import (
+	"crypto/tls"
 	"fmt"
-	"k8s.io/klog/v2"
 	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
+
+	"k8s.io/klog/v2"
+
+	"k8s.io/kubernetes/test/e2e/framework"
 )
 
 // terminationSignals are signals that cause the program to exit in the
@@ -42,6 +46,13 @@ func waitForTerminationSignal() {
 // and return the error.
 func readinessCheck(name string, urls []string, errCh <-chan error) error {
 	klog.Infof("Running readiness check for service %q", name)
+
+	insecureTransport := http.DefaultTransport.(*http.Transport).Clone()
+	insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+	insecureHTTPClient := &http.Client{
+		Transport: insecureTransport,
+	}
+
 	endTime := time.Now().Add(*serverStartTimeout)
 	blockCh := make(chan error)
 	defer close(blockCh)
@@ -67,8 +78,7 @@ func readinessCheck(name string, urls []string, errCh <-chan error) error {
 		case <-time.After(time.Second):
 			ready := true
 			for _, url := range urls {
-				resp, err := http.Head(url)
-				if err != nil || resp.StatusCode != http.StatusOK {
+				if !healthCheck(insecureHTTPClient, url) {
 					ready = false
 					break
 				}
@@ -80,3 +90,20 @@ func readinessCheck(name string, urls []string, errCh <-chan error) error {
 	}
 	return fmt.Errorf("e2e service %q readiness check timeout %v", name, *serverStartTimeout)
 }
+
+// Perform a health check. Anything other than a 200-response is treated as a failure.
+// Only returns non-recoverable errors.
+func healthCheck(client *http.Client, url string) bool {
+	req, err := http.NewRequest("HEAD", url, nil)
+	if err != nil {
+		return false
+	}
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", framework.TestContext.BearerToken))
+	resp, err := client.Do(req)
+	if err != nil {
+		klog.Warningf("Health check on %q failed, error=%v", url, err)
+	} else if resp.StatusCode != http.StatusOK {
+		klog.Warningf("Health check on %q failed, status=%d", url, resp.StatusCode)
+	}
+	return err == nil && resp.StatusCode == http.StatusOK
+}