Improve integration test performance in scheduler

pkg/controller/daemon/daemon_controller_test.go

@@ -480,11 +480,12 @@ func TestDeleteFinalStateUnknown(t *testing.T) {
 }
 
 func TestExpectationsOnRecreate(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	client := fake.NewSimpleClientset()
-	stopCh := make(chan struct{})
-	defer close(stopCh)
 
-	_, ctx := ktesting.NewTestContext(t)
 	f := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
 	dsc, err := NewDaemonSetsController(
 		ctx,
@@ -550,8 +551,8 @@ func TestExpectationsOnRecreate(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	f.Start(stopCh)
-	for ty, ok := range f.WaitForCacheSync(stopCh) {
+	f.Start(ctx.Done())
+	for ty, ok := range f.WaitForCacheSync(ctx.Done()) {
 		if !ok {
 			t.Fatalf("caches failed to sync: %v", ty)
 		}

test/integration/apiserver/certreload/certreload_test.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/component-base/cli/flag"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/ktesting"
 )
 
 type caWithClient struct {
@@ -135,6 +136,10 @@ func TestClientCARecreate(t *testing.T) {
 }
 
 func testClientCA(t *testing.T, recreate bool) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	frontProxyCA, err := newTestCAWithClient(
 		pkix.Name{
 			CommonName: "test-front-proxy-ca",
@@ -170,7 +175,7 @@ func testClientCA(t *testing.T, recreate bool) {
 	clientCAFilename := ""
 	frontProxyCAFilename := ""
 
-	kubeClient, kubeconfig, tearDownFn := framework.StartTestServer(t, framework.TestServerSetup{
+	kubeClient, kubeconfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			opts.GenericServerRunOptions.MaxRequestBodyBytes = 1024 * 1024
 			clientCAFilename = opts.Authentication.ClientCert.ClientCA
@@ -300,7 +305,7 @@ func testClientCA(t *testing.T, recreate bool) {
 	}
 
 	// Call an endpoint to make sure we are authenticated
-	_, err = testClient.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
+	_, err = testClient.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		t.Error(err)
 	}
@@ -468,9 +473,13 @@ func TestServingCertRecreate(t *testing.T) {
 }
 
 func testServingCert(t *testing.T, recreate bool) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	var servingCertPath string
 
-	_, kubeconfig, tearDownFn := framework.StartTestServer(t, framework.TestServerSetup{
+	_, kubeconfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			opts.GenericServerRunOptions.MaxRequestBodyBytes = 1024 * 1024
 			servingCertPath = opts.SecureServing.ServerCert.CertDirectory
@@ -509,7 +518,11 @@ func testServingCert(t *testing.T, recreate bool) {
 func TestSNICert(t *testing.T) {
 	var servingCertPath string
 
-	_, kubeconfig, tearDownFn := framework.StartTestServer(t, framework.TestServerSetup{
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	_, kubeconfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			opts.GenericServerRunOptions.MaxRequestBodyBytes = 1024 * 1024
 			servingCertPath = opts.SecureServing.ServerCert.CertDirectory

test/integration/apiserver/export_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package apiserver
 
 import (
-	"context"
 	"net/http"
 	"testing"
 
@@ -27,31 +26,32 @@ import (
 
 // Tests that the apiserver rejects the export param
 func TestExportRejection(t *testing.T) {
-	clientSet, _, tearDownFn := setup(t)
+	ctx, clientSet, _, tearDownFn := setup(t)
 	defer tearDownFn()
 
-	_, err := clientSet.CoreV1().Namespaces().Create(context.Background(), &corev1.Namespace{
+	_, err := clientSet.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{Name: "export-fail"},
 	}, metav1.CreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer func() {
-		clientSet.CoreV1().Namespaces().Delete(context.Background(), "export-fail", metav1.DeleteOptions{})
+		if err := clientSet.CoreV1().Namespaces().Delete(ctx, "export-fail", metav1.DeleteOptions{}); err != nil {
+			t.Errorf("error whiling deleting the namespace, err: %v", err)
+		}
 	}()
 
-	result := clientSet.Discovery().RESTClient().Get().AbsPath("/api/v1/namespaces/export-fail").Param("export", "true").Do(context.Background())
+	result := clientSet.Discovery().RESTClient().Get().AbsPath("/api/v1/namespaces/export-fail").Param("export", "true").Do(ctx)
 	statusCode := 0
 	result.StatusCode(&statusCode)
 	if statusCode != http.StatusBadRequest {
 		t.Errorf("expected %v, got %v", http.StatusBadRequest, statusCode)
 	}
 
-	result = clientSet.Discovery().RESTClient().Get().AbsPath("/api/v1/namespaces/export-fail").Param("export", "false").Do(context.Background())
+	result = clientSet.Discovery().RESTClient().Get().AbsPath("/api/v1/namespaces/export-fail").Param("export", "false").Do(ctx)
 	statusCode = 0
 	result.StatusCode(&statusCode)
 	if statusCode != http.StatusOK {
 		t.Errorf("expected %v, got %v", http.StatusOK, statusCode)
 	}
-
 }

test/integration/apiserver/flowcontrol/concurrency_test.go

@@ -38,6 +38,7 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/ktesting"
 )
 
 const (
@@ -48,22 +49,30 @@ const (
 	timeout                           = time.Second * 10
 )
 
-func setup(t testing.TB, maxReadonlyRequestsInFlight, MaxMutatingRequestsInFlight int) (*rest.Config, framework.TearDownFunc) {
-	_, kubeConfig, tearDownFn := framework.StartTestServer(t, framework.TestServerSetup{
+func setup(t testing.TB, maxReadonlyRequestsInFlight, maxMutatingRequestsInFlight int) (context.Context, *rest.Config, framework.TearDownFunc) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+
+	_, kubeConfig, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			// Ensure all clients are allowed to send requests.
 			opts.Authorization.Modes = []string{"AlwaysAllow"}
 			opts.GenericServerRunOptions.MaxRequestsInFlight = maxReadonlyRequestsInFlight
-			opts.GenericServerRunOptions.MaxMutatingRequestsInFlight = MaxMutatingRequestsInFlight
+			opts.GenericServerRunOptions.MaxMutatingRequestsInFlight = maxMutatingRequestsInFlight
 		},
 	})
-	return kubeConfig, tearDownFn
+
+	newTeardown := func() {
+		cancel()
+		tearDownFn()
+	}
+	return ctx, kubeConfig, newTeardown
 }
 
 func TestPriorityLevelIsolation(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.APIPriorityAndFairness, true)()
 	// NOTE: disabling the feature should fail the test
-	kubeConfig, closeFn := setup(t, 1, 1)
+	ctx, kubeConfig, closeFn := setup(t, 1, 1)
 	defer closeFn()
 
 	loopbackClient := clientset.NewForConfigOrDie(kubeConfig)
@@ -106,15 +115,15 @@ func TestPriorityLevelIsolation(t *testing.T) {
 	// "elephant"
 	wg.Add(concurrencyShares + queueLength)
 	streamRequests(concurrencyShares+queueLength, func() {
-		_, err := noxu1Client.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
+		_, err := noxu1Client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
 		if err != nil {
 			t.Error(err)
 		}
 	}, &wg, stopCh)
 	// "mouse"
 	wg.Add(3)
 	streamRequests(3, func() {
-		_, err := noxu2Client.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
+		_, err := noxu2Client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
 		if err != nil {
 			t.Error(err)
 		}

test/integration/apiserver/flowcontrol/concurrency_util_test.go

@@ -38,6 +38,7 @@ import (
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/pkg/controlplane"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/ktesting"
 )
 
 const (
@@ -147,10 +148,14 @@ func (d *noxuDelayingAuthorizer) Authorize(ctx context.Context, a authorizer.Att
 // Secondarily, this test also checks the observed seat utilizations against values derived from expecting that
 // the throughput observed by the client equals the execution throughput observed by the server.
 func TestConcurrencyIsolation(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.APIPriorityAndFairness, true)()
 	// NOTE: disabling the feature should fail the test
 
-	_, kubeConfig, closeFn := framework.StartTestServer(t, framework.TestServerSetup{
+	_, kubeConfig, closeFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			// Ensure all clients are allowed to send requests.
 			opts.Authorization.Modes = []string{"AlwaysAllow"}
@@ -191,7 +196,7 @@ func TestConcurrencyIsolation(t *testing.T) {
 	wg.Add(noxu1NumGoroutines)
 	streamRequests(noxu1NumGoroutines, func() {
 		start := time.Now()
-		_, err := noxu1Client.CoreV1().Namespaces().Get(context.Background(), "default", metav1.GetOptions{})
+		_, err := noxu1Client.CoreV1().Namespaces().Get(ctx, "default", metav1.GetOptions{})
 		duration := time.Since(start).Seconds()
 		noxu1LatMeasure.update(duration)
 		if err != nil {
@@ -204,7 +209,7 @@ func TestConcurrencyIsolation(t *testing.T) {
 	wg.Add(noxu2NumGoroutines)
 	streamRequests(noxu2NumGoroutines, func() {
 		start := time.Now()
-		_, err := noxu2Client.CoreV1().Namespaces().Get(context.Background(), "default", metav1.GetOptions{})
+		_, err := noxu2Client.CoreV1().Namespaces().Get(ctx, "default", metav1.GetOptions{})
 		duration := time.Since(start).Seconds()
 		noxu2LatMeasure.update(duration)
 		if err != nil {

test/integration/apiserver/flowcontrol/fight_test.go

@@ -172,7 +172,7 @@ func (ft *fightTest) evaluate(tBeforeCreate, tAfterCreate time.Time) {
 }
 func TestConfigConsumerFight(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.APIPriorityAndFairness, true)()
-	kubeConfig, closeFn := setup(t, 100, 100)
+	_, kubeConfig, closeFn := setup(t, 100, 100)
 	defer closeFn()
 	const teamSize = 3
 	ft := newFightTest(t, kubeConfig, teamSize)

test/integration/apiserver/flowcontrol/fs_condition_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package flowcontrol
 
 import (
-	"context"
 	"encoding/json"
 	"testing"
 	"time"
@@ -38,15 +37,11 @@ import (
 func TestConditionIsolation(t *testing.T) {
 	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, genericfeatures.APIPriorityAndFairness, true)()
 	// NOTE: disabling the feature should fail the test
-	kubeConfig, closeFn := setup(t, 10, 10)
+	ctx, kubeConfig, closeFn := setup(t, 10, 10)
 	defer closeFn()
 
 	loopbackClient := clientset.NewForConfigOrDie(kubeConfig)
 
-	stopCh := make(chan struct{})
-	defer close(stopCh)
-	ctx := context.Background()
-
 	fsOrig := fcboot.SuggestedFlowSchemas[0]
 	t.Logf("Testing Status Condition isolation in FlowSchema %q", fsOrig.Name)
 	fsClient := loopbackClient.FlowcontrolV1beta3().FlowSchemas()
@@ -60,7 +55,7 @@ func TestConditionIsolation(t *testing.T) {
 		}
 		dangleOrig = getCondition(fsGot.Status.Conditions, flowcontrol.FlowSchemaConditionDangling)
 		return dangleOrig != nil, nil
-	}, stopCh)
+	}, ctx.Done())
 
 	ssaType := flowcontrol.FlowSchemaConditionType("test-ssa")
 	patchSSA := flowcontrolapply.FlowSchema(fsOrig.Name).

test/integration/apiserver/max_json_patch_operations_test.go

@@ -28,11 +28,16 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/ktesting"
 )
 
 // Tests that the apiserver limits the number of operations in a json patch.
 func TestMaxJSONPatchOperations(t *testing.T) {
-	clientSet, _, tearDownFn := framework.StartTestServer(t, framework.TestServerSetup{
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	clientSet, _, tearDownFn := framework.StartTestServer(ctx, t, framework.TestServerSetup{
 		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
 			opts.GenericServerRunOptions.MaxRequestBodyBytes = 1024 * 1024
 		},
@@ -50,13 +55,13 @@ func TestMaxJSONPatchOperations(t *testing.T) {
 			Name: "test",
 		},
 	}
-	_, err := clientSet.CoreV1().Secrets("default").Create(context.TODO(), secret, metav1.CreateOptions{})
+	_, err := clientSet.CoreV1().Secrets("default").Create(ctx, secret, metav1.CreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	err = c.Patch(types.JSONPatchType).AbsPath(fmt.Sprintf("/api/v1/namespaces/default/secrets/test")).
-		Body(hugePatch).Do(context.TODO()).Error()
+		Body(hugePatch).Do(ctx).Error()
 	if err == nil {
 		t.Fatalf("unexpected no error")
 	}