e2e test fix

Author: carreter

cmd/server/app/server.go

@@ -168,6 +168,7 @@ func (p *Proxy) Run(o *options.ProxyRunOptions, stopCh <-chan struct{}) error {
 		)
 		klog.V(1).Infoln("Starting lease acquisition and garbage collection controller.")
 		leaseController.Run(ctx)
+		defer leaseController.Stop()
 	}
 
 	klog.V(1).Infoln("Starting admin server for debug connections.")

e2e/lease_count_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"path"
@@ -13,9 +14,14 @@ import (
 	"time"
 
 	appsv1api "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/e2e-framework/klient/k8s"
 	"sigs.k8s.io/e2e-framework/klient/wait"
 	"sigs.k8s.io/e2e-framework/klient/wait/conditions"
 	"sigs.k8s.io/e2e-framework/pkg/env"
@@ -48,7 +54,7 @@ func TestMain(m *testing.M) {
 	kindCluster := kind.NewCluster(kindClusterName).WithOpts(kind.WithImage(*kindImage))
 
 	testenv.Setup(
-		envfuncs.CreateCluster(kindCluster, kindClusterName),
+		envfuncs.CreateClusterWithConfig(kindCluster, kindClusterName, "templates/kind.config"),
 		envfuncs.LoadImageToCluster(kindClusterName, *agentImage),
 		envfuncs.LoadImageToCluster(kindClusterName, *serverImage),
 		renderAndApplyManifests,
@@ -84,15 +90,16 @@ func renderTemplate(file string, params any) (client.Object, *schema.GroupVersio
 	return obj.(client.Object), gvk, nil
 }
 
-type KeyValue struct {
-	Key   string
-	Value string
+type CLIFlag struct {
+	Flag       string
+	Value      string
+	EmptyValue bool
 }
 
 type DeploymentConfig struct {
 	Replicas int
 	Image    string
-	Args     []KeyValue
+	Args     []CLIFlag
 }
 
 func renderAndApplyManifests(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
@@ -111,10 +118,6 @@ func renderAndApplyManifests(ctx context.Context, cfg *envconf.Config) (context.
 	if err != nil {
 		return ctx, err
 	}
-	agentService, _, err := renderTemplate("agent/service.yaml", struct{}{})
-	if err != nil {
-		return ctx, err
-	}
 
 	// Submit agent RBAC templates to k8s.
 	err = client.Resources().Create(ctx, agentServiceAccount)
@@ -129,10 +132,6 @@ func renderAndApplyManifests(ctx context.Context, cfg *envconf.Config) (context.
 	if err != nil {
 		return ctx, err
 	}
-	err = client.Resources().Create(ctx, agentService)
-	if err != nil {
-		return ctx, err
-	}
 
 	// Render server RBAC and Service templates.
 	serverClusterRoleBinding, _, err := renderTemplate("server/clusterrolebinding.yaml", struct{}{})
@@ -157,50 +156,172 @@ func renderAndApplyManifests(ctx context.Context, cfg *envconf.Config) (context.
 	return ctx, nil
 }
 
-func deployAndWaitForDeployment(obj client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
+func createDeployment(obj client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
 	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
-		client := cfg.Client()
-		err := client.Resources().Create(ctx, obj)
+		deployment, ok := obj.(*appsv1api.Deployment)
+		if !ok {
+			t.Fatalf("object %q is not a deployment", obj.GetName())
+		}
+
+		err := cfg.Client().Resources(deployment.Namespace).Create(ctx, deployment)
+		if err != nil {
+			t.Fatalf("could not create deployment %q: %v", deployment.Name, err)
+		}
+
+		newDeployment := &appsv1api.Deployment{}
+		err = cfg.Client().Resources(deployment.Namespace).Get(ctx, deployment.Name, deployment.Namespace, newDeployment)
+		if err != nil {
+			t.Fatalf("could not get deployment %q after creation: %v", deployment.Name, err)
+		}
+
+		return ctx
+	}
+}
+
+func deleteDeployment(obj client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
+	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+		deployment, ok := obj.(*appsv1api.Deployment)
+		if !ok {
+			t.Fatalf("object %q is not a deployment", obj.GetName())
+		}
+
+		k8sClient := kubernetes.NewForConfigOrDie(cfg.Client().RESTConfig())
+		pods, err := k8sClient.CoreV1().Pods(deployment.Namespace).List(ctx, metav1.ListOptions{LabelSelector: labels.FormatLabels(deployment.Spec.Selector.MatchLabels)})
 		if err != nil {
-			t.Fatalf("could not create Deployment: %v", err)
+			t.Fatalf("could not get pods for deployment %q: %v", deployment.Name, err)
 		}
 
+		cfg.Client().Resources(deployment.Namespace).Delete(ctx, deployment)
+
 		err = wait.For(
-			conditions.New(client.Resources()).DeploymentAvailable(obj.GetName(), obj.GetNamespace()),
-			wait.WithTimeout(1*time.Minute),
-			wait.WithInterval(10*time.Second),
+			conditions.New(cfg.Client().Resources(deployment.Namespace)).ResourcesDeleted(pods),
+			wait.WithTimeout(60*time.Second),
+			wait.WithInterval(5*time.Second),
+		)
+		if err != nil {
+			pods, err := k8sClient.CoreV1().Pods(deployment.Namespace).List(ctx, metav1.ListOptions{LabelSelector: labels.FormatLabels(deployment.Spec.Selector.MatchLabels)})
+			if err != nil {
+				t.Fatalf("could not get pods for deployment %q: %v", deployment.Name, err)
+			}
+
+			for _, pod := range pods.Items {
+				logs, err := dumpPodLogs(ctx, k8sClient, pod.Namespace, pod.Name)
+				if err != nil {
+					t.Fatalf("could not dump logs for pod %q: %v", pod.Name, err)
+				}
+				t.Errorf("logs for pod %q: %v", pod.Name, logs)
+			}
+			t.Fatalf("waiting for deletion of pods for deployment %q failed, dumped pod logs", deployment.Name)
+		}
+
+		return ctx
+	}
+}
+
+func sleepFor(duration time.Duration) func(context.Context, *testing.T, *envconf.Config) context.Context {
+	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+		time.Sleep(duration)
+		return ctx
+	}
+}
+
+func waitForDeployment(obj client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
+	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+		deployment, ok := obj.(*appsv1api.Deployment)
+		if !ok {
+			t.Fatalf("object %q is not a deployment", obj.GetName())
+		}
+
+		k8sClient := kubernetes.NewForConfigOrDie(cfg.Client().RESTConfig())
+		err := wait.For(
+			conditions.New(cfg.Client().Resources(deployment.Namespace)).DeploymentAvailable(deployment.Name, deployment.Namespace),
+			wait.WithTimeout(60*time.Second),
+			wait.WithInterval(5*time.Second),
 		)
 		if err != nil {
-			t.Fatalf("waiting for Deployment failed: %v", err)
+			pods, err := k8sClient.CoreV1().Pods(deployment.Namespace).List(ctx, metav1.ListOptions{LabelSelector: labels.FormatLabels(deployment.Spec.Selector.MatchLabels)})
+			if err != nil {
+				t.Fatalf("could not get pods for deployment %q: %v", deployment.Name, err)
+			}
+
+			for _, pod := range pods.Items {
+				if isPodReady(&pod) {
+					continue
+				}
+
+				logs, err := dumpPodLogs(ctx, k8sClient, pod.Namespace, pod.Name)
+				if err != nil {
+					t.Fatalf("could not dump logs for pod %q: %v", pod.Name, err)
+				}
+				t.Errorf("logs for pod %q: %v", pod.Name, logs)
+			}
+			t.Fatalf("waiting for deployment %q failed, dumped pod logs", deployment.Name)
 		}
 
 		return ctx
 	}
 }
 
+func isPodReady(pod *corev1.Pod) bool {
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == corev1.PodReady {
+			return condition.Status == corev1.ConditionTrue
+		}
+	}
+	return false
+}
+
+func dumpPodLogs(ctx context.Context, k8sClient kubernetes.Interface, namespace, name string) (string, error) {
+	req := k8sClient.CoreV1().Pods(namespace).GetLogs(name, &corev1.PodLogOptions{})
+	podLogs, err := req.Stream(ctx)
+	if err != nil {
+		return "", fmt.Errorf("could not stream logs for pod %q in namespace %q: %w", name, namespace, err)
+	}
+	defer podLogs.Close()
+
+	b, err := io.ReadAll(podLogs)
+	if err != nil {
+		return "", fmt.Errorf("could not read logs for pod %q in namespace %q from stream: %w", name, namespace, err)
+	}
+
+	return string(b), nil
+}
+
 func scaleDeployment(obj client.Object, replicas int) func(context.Context, *testing.T, *envconf.Config) context.Context {
 	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
 		deployment, ok := obj.(*appsv1api.Deployment)
 		if !ok {
 			t.Fatalf("provided object is not a deployment")
 		}
 
+		err := cfg.Client().Resources().Get(ctx, deployment.Name, deployment.Namespace, deployment)
+		if err != nil {
+			t.Fatalf("could not get Deployment to update (name: %q, namespace: %q): %v", deployment.Name, deployment.Namespace, err)
+		}
+
 		newReplicas := int32(replicas)
 		deployment.Spec.Replicas = &newReplicas
 
 		client := cfg.Client()
-		err := client.Resources().Update(ctx, deployment)
+		err = client.Resources().Update(ctx, deployment)
 		if err != nil {
 			t.Fatalf("could not update Deployment replicas: %v", err)
 		}
 
 		err = wait.For(
-			conditions.New(client.Resources()).DeploymentAvailable(deployment.GetName(), deployment.GetNamespace()),
-			wait.WithTimeout(1*time.Minute),
-			wait.WithInterval(10*time.Second),
+			conditions.New(cfg.Client().Resources(deployment.Namespace)).ResourceScaled(deployment, func(obj k8s.Object) int32 {
+				deployment, ok := obj.(*appsv1api.Deployment)
+				if !ok {
+					t.Fatalf("provided object is not a deployment")
+				}
+
+				return deployment.Status.AvailableReplicas
+			}, int32(replicas)),
+			wait.WithTimeout(60*time.Second),
+			wait.WithInterval(5*time.Second),
 		)
 		if err != nil {
-			t.Fatalf("waiting for Deployment failed: %v", err)
+			t.Fatalf("waiting for deployment %q to scale failed", deployment.Name)
 		}
 
 		return ctx