Merge pull request #96798 from yue9944882/flaky/apnf-e2e-drown-test

E2E Flakiness: Eliminates client-side rate-limiting for AP&F drown-out test

test/e2e/apimachinery/BUILD

@@ -71,7 +71,6 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/version:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
-        "//staging/src/k8s.io/apiserver/pkg/authentication/user:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/discovery:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/features:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/names:go_default_library",
@@ -83,6 +82,7 @@ go_library(
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",
         "//staging/src/k8s.io/client-go/util/cert:go_default_library",
+        "//staging/src/k8s.io/client-go/util/flowcontrol:go_default_library",
         "//staging/src/k8s.io/client-go/util/keyutil:go_default_library",
         "//staging/src/k8s.io/client-go/util/retry:go_default_library",
         "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",

test/e2e/apimachinery/flowcontrol.go

@@ -32,8 +32,8 @@ import (
 
 	flowcontrol "k8s.io/api/flowcontrol/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/client-go/rest"
+	clientsideflowcontrol "k8s.io/client-go/util/flowcontrol"
 	"k8s.io/kubernetes/test/e2e/framework"
 )
 
@@ -43,7 +43,7 @@ const (
 )
 
 var _ = SIGDescribe("API priority and fairness", func() {
-	f := framework.NewDefaultFramework("flowschemas")
+	f := framework.NewDefaultFramework("apf")
 
 	ginkgo.It("should ensure that requests can be classified by testing flow-schemas/priority-levels", func() {
 		testingFlowSchemaName := "e2e-testing-flowschema"
@@ -56,7 +56,7 @@ var _ = SIGDescribe("API priority and fairness", func() {
 		defer cleanup()
 
 		ginkgo.By("creating a testing flowschema")
-		createdFlowSchema, cleanup := createFlowSchema(f, testingFlowSchemaName, 1000, testingPriorityLevelName, matchingUsername)
+		createdFlowSchema, cleanup := createFlowSchema(f, testingFlowSchemaName, 1000, testingPriorityLevelName, []string{matchingUsername})
 		defer cleanup()
 
 		ginkgo.By("checking response headers contain flow-schema/priority-level uid")
@@ -74,9 +74,12 @@ var _ = SIGDescribe("API priority and fairness", func() {
 	// higher QPS client cannot drown out the other one despite having higher
 	// priority.
 	ginkgo.It("should ensure that requests can't be drowned out (priority)", func() {
-		flowSchemaNamePrefix := "e2e-testing-flowschema"
+		flowSchemaNamePrefix := "e2e-testing-flowschema-" + f.UniqueName
-		priorityLevelNamePrefix := "e2e-testing-prioritylevel"
+		priorityLevelNamePrefix := "e2e-testing-prioritylevel-" + f.UniqueName
 		loadDuration := 10 * time.Second
+		highQPSClientName := "highqps-" + f.UniqueName
+		lowQPSClientName := "lowqps-" + f.UniqueName
+
 		type client struct {
 			username                    string
 			qps                         float64
@@ -86,6 +89,7 @@ var _ = SIGDescribe("API priority and fairness", func() {
 			flowSchemaName              string
 			matchingPrecedence          int32
 			completedRequests           int32
+			expectedCompletedPercentage float64
 		}
 		clients := []client{
 			// "highqps" refers to a client that creates requests at a much higher
@@ -93,8 +97,8 @@ var _ = SIGDescribe("API priority and fairness", func() {
 			// In contrast, "lowqps" stays under its concurrency shares.
 			// Additionally, the "highqps" client also has a higher matching
 			// precedence for its flow schema.
-			{username: "highqps", qps: 100.0, concurrencyMultiplier: 2.0, matchingPrecedence: 999},
+			{username: highQPSClientName, qps: 100.0, concurrencyMultiplier: 2.0, matchingPrecedence: 999, expectedCompletedPercentage: 0.75},
-			{username: "lowqps", qps: 5.0, concurrencyMultiplier: 0.5, matchingPrecedence: 1000},
+			{username: lowQPSClientName, qps: 5.0, concurrencyMultiplier: 0.5, matchingPrecedence: 1000, expectedCompletedPercentage: 0.75},
 		}
 
 		ginkgo.By("creating test priority levels and flow schemas")
@@ -106,7 +110,7 @@ var _ = SIGDescribe("API priority and fairness", func() {
 
 			clients[i].flowSchemaName = fmt.Sprintf("%s-%s", flowSchemaNamePrefix, clients[i].username)
 			framework.Logf("creating FlowSchema %q", clients[i].flowSchemaName)
-			_, cleanup = createFlowSchema(f, clients[i].flowSchemaName, clients[i].matchingPrecedence, clients[i].priorityLevelName, clients[i].username)
+			_, cleanup = createFlowSchema(f, clients[i].flowSchemaName, clients[i].matchingPrecedence, clients[i].priorityLevelName, []string{clients[i].username})
 			defer cleanup()
 		}
 
@@ -138,8 +142,8 @@ var _ = SIGDescribe("API priority and fairness", func() {
 			maxCompletedRequests := float64(client.concurrency) * client.qps * float64(loadDuration/time.Second)
 			fractionCompleted := float64(client.completedRequests) / maxCompletedRequests
 			framework.Logf("client %q completed %d/%d requests (%.1f%%)", client.username, client.completedRequests, int32(maxCompletedRequests), 100*fractionCompleted)
-			if fractionCompleted < 0.95 {
+			if fractionCompleted < client.expectedCompletedPercentage {
-				framework.Failf("client %q: got %.1f%% completed requests, want at least 95%%", client.username, 100*fractionCompleted)
+				framework.Failf("client %q: got %.1f%% completed requests, want at least %.1f%%", client.username, 100*fractionCompleted, 100*client.expectedCompletedPercentage)
 			}
 		}
 	})
@@ -150,16 +154,18 @@ var _ = SIGDescribe("API priority and fairness", func() {
 	// the two clients and not allow one client to drown out the other despite
 	// having a higher QPS.
 	ginkgo.It("should ensure that requests can't be drowned out (fairness)", func() {
-		priorityLevelName := "e2e-testing-prioritylevel"
+		priorityLevelName := "e2e-testing-prioritylevel-" + f.UniqueName
-		flowSchemaName := "e2e-testing-flowschema"
+		flowSchemaName := "e2e-testing-flowschema-" + f.UniqueName
 		loadDuration := 10 * time.Second
 
 		framework.Logf("creating PriorityLevel %q", priorityLevelName)
 		_, cleanup := createPriorityLevel(f, priorityLevelName, 1)
 		defer cleanup()
 
+		highQPSClientName := "highqps-" + f.UniqueName
+		lowQPSClientName := "lowqps-" + f.UniqueName
 		framework.Logf("creating FlowSchema %q", flowSchemaName)
-		_, cleanup = createFlowSchema(f, flowSchemaName, 1000, priorityLevelName, "*")
+		_, cleanup = createFlowSchema(f, flowSchemaName, 1000, priorityLevelName, []string{highQPSClientName, lowQPSClientName})
 		defer cleanup()
 
 		type client struct {
@@ -168,10 +174,11 @@ var _ = SIGDescribe("API priority and fairness", func() {
 			concurrencyMultiplier       float64
 			concurrency                 int32
 			completedRequests           int32
+			expectedCompletedPercentage float64
 		}
 		clients := []client{
-			{username: "highqps", qps: 100.0, concurrencyMultiplier: 2.0},
+			{username: highQPSClientName, qps: 100.0, concurrencyMultiplier: 2.0, expectedCompletedPercentage: 0.75},
-			{username: "lowqps", qps: 5.0, concurrencyMultiplier: 0.5},
+			{username: lowQPSClientName, qps: 5.0, concurrencyMultiplier: 0.5, expectedCompletedPercentage: 0.90},
 		}
 
 		framework.Logf("getting real concurrency")
@@ -202,8 +209,8 @@ var _ = SIGDescribe("API priority and fairness", func() {
 			maxCompletedRequests := float64(client.concurrency) * client.qps * float64(loadDuration/time.Second)
 			fractionCompleted := float64(client.completedRequests) / maxCompletedRequests
 			framework.Logf("client %q completed %d/%d requests (%.1f%%)", client.username, client.completedRequests, int32(maxCompletedRequests), 100*fractionCompleted)
-			if fractionCompleted < 0.95 {
+			if fractionCompleted < client.expectedCompletedPercentage {
-				framework.Failf("client %q: got %.1f%% completed requests, want at least 95%%", client.username, 100*fractionCompleted)
+				framework.Failf("client %q: got %.1f%% completed requests, want at least %.1f%%", client.username, 100*fractionCompleted, 100*client.expectedCompletedPercentage)
 			}
 		}
 	})
@@ -265,16 +272,9 @@ func getPriorityLevelConcurrency(f *framework.Framework, priorityLevelName strin
 
 // createFlowSchema creates a flow schema referring to a particular priority
 // level and matching the username provided.
-func createFlowSchema(f *framework.Framework, flowSchemaName string, matchingPrecedence int32, priorityLevelName string, matchingUsername string) (*flowcontrol.FlowSchema, func()) {
+func createFlowSchema(f *framework.Framework, flowSchemaName string, matchingPrecedence int32, priorityLevelName string, matchingUsernames []string) (*flowcontrol.FlowSchema, func()) {
 	var subjects []flowcontrol.Subject
-	if matchingUsername == "*" {
+	for _, matchingUsername := range matchingUsernames {
-		subjects = append(subjects, flowcontrol.Subject{
-			Kind: flowcontrol.SubjectKindGroup,
-			Group: &flowcontrol.GroupSubject{
-				Name: user.AllAuthenticated,
-			},
-		})
-	} else {
 		subjects = append(subjects, flowcontrol.Subject{
 			Kind: flowcontrol.SubjectKindUser,
 			User: &flowcontrol.UserSubject{
@@ -321,6 +321,7 @@ func createFlowSchema(f *framework.Framework, flowSchemaName string, matchingPre
 func makeRequest(f *framework.Framework, username string) *http.Response {
 	config := f.ClientConfig()
 	config.Impersonate.UserName = username
+	config.RateLimiter = clientsideflowcontrol.NewFakeAlwaysRateLimiter()
 	config.Impersonate.Groups = []string{"system:authenticated"}
 	roundTripper, err := rest.TransportFor(config)
 	framework.ExpectNoError(err)