diff --git a/test/e2e/apimachinery/flowcontrol.go b/test/e2e/apimachinery/flowcontrol.go index 1fb0340973b..65d1057c847 100644 --- a/test/e2e/apimachinery/flowcontrol.go +++ b/test/e2e/apimachinery/flowcontrol.go @@ -51,73 +51,14 @@ var _ = SIGDescribe("API priority and fairness", func() { nonMatchingUsername := "foo" ginkgo.By("creating a testing prioritylevel") - createdPriorityLevel, err := f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Create( - context.TODO(), - &flowcontrol.PriorityLevelConfiguration{ - ObjectMeta: metav1.ObjectMeta{ - Name: testingPriorityLevelName, - }, - Spec: flowcontrol.PriorityLevelConfigurationSpec{ - Type: flowcontrol.PriorityLevelEnablementLimited, - Limited: &flowcontrol.LimitedPriorityLevelConfiguration{ - AssuredConcurrencyShares: 1, // will have at minimum 1 concurrency share - LimitResponse: flowcontrol.LimitResponse{ - Type: flowcontrol.LimitResponseTypeReject, - }, - }, - }, - }, - metav1.CreateOptions{}) - framework.ExpectNoError(err) - - defer func() { - // clean-ups - err := f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Delete(context.TODO(), testingPriorityLevelName, metav1.DeleteOptions{}) - framework.ExpectNoError(err) - err = f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Delete(context.TODO(), testingFlowSchemaName, metav1.DeleteOptions{}) - framework.ExpectNoError(err) - }() + createdPriorityLevel, cleanup := createPriorityLevel(f, testingPriorityLevelName, 1) + defer cleanup() ginkgo.By("creating a testing flowschema") - createdFlowSchema, err := f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Create( - context.TODO(), - &flowcontrol.FlowSchema{ - ObjectMeta: metav1.ObjectMeta{ - Name: testingFlowSchemaName, - }, - Spec: flowcontrol.FlowSchemaSpec{ - MatchingPrecedence: 1000, // a rather higher precedence to ensure it make effect - PriorityLevelConfiguration: flowcontrol.PriorityLevelConfigurationReference{ - Name: testingPriorityLevelName, - }, - DistinguisherMethod: &flowcontrol.FlowDistinguisherMethod{ - Type: flowcontrol.FlowDistinguisherMethodByUserType, - }, - Rules: []flowcontrol.PolicyRulesWithSubjects{ - { - Subjects: []flowcontrol.Subject{ - { - Kind: flowcontrol.SubjectKindUser, - User: &flowcontrol.UserSubject{ - Name: matchingUsername, - }, - }, - }, - NonResourceRules: []flowcontrol.NonResourcePolicyRule{ - { - Verbs: []string{flowcontrol.VerbAll}, - NonResourceURLs: []string{flowcontrol.NonResourceAll}, - }, - }, - }, - }, - }, - }, - metav1.CreateOptions{}) - framework.ExpectNoError(err) - - ginkgo.By("response headers should contain flow-schema/priority-level uid") + createdFlowSchema, cleanup := createFlowSchema(f, testingFlowSchemaName, 1000, testingPriorityLevelName, matchingUsername) + defer cleanup() + ginkgo.By("checking response headers contain flow-schema/priority-level uid") if !testResponseHeaderMatches(f, matchingUsername, string(createdPriorityLevel.UID), string(createdFlowSchema.UID)) { framework.Failf("matching user doesnt received UID for the testing priority-level and flow-schema") } @@ -154,99 +95,23 @@ var _ = SIGDescribe("API priority and fairness", func() { for i := range clients { clients[i].priorityLevelName = fmt.Sprintf("%s-%s", priorityLevelNamePrefix, clients[i].username) framework.Logf("creating PriorityLevel %q", clients[i].priorityLevelName) - _, err := f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Create( - context.TODO(), - &flowcontrol.PriorityLevelConfiguration{ - ObjectMeta: metav1.ObjectMeta{ - Name: clients[i].priorityLevelName, - }, - Spec: flowcontrol.PriorityLevelConfigurationSpec{ - Type: flowcontrol.PriorityLevelEnablementLimited, - Limited: &flowcontrol.LimitedPriorityLevelConfiguration{ - AssuredConcurrencyShares: 1, - LimitResponse: flowcontrol.LimitResponse{ - Type: flowcontrol.LimitResponseTypeReject, - }, - }, - }, - }, - metav1.CreateOptions{}) - framework.ExpectNoError(err) - defer func(name string) { - framework.ExpectNoError(f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Delete(context.TODO(), name, metav1.DeleteOptions{})) - }(clients[i].priorityLevelName) + _, cleanup := createPriorityLevel(f, clients[i].priorityLevelName, 1) + defer cleanup() + clients[i].flowSchemaName = fmt.Sprintf("%s-%s", flowSchemaNamePrefix, clients[i].username) framework.Logf("creating FlowSchema %q", clients[i].flowSchemaName) - _, err = f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Create( - context.TODO(), - &flowcontrol.FlowSchema{ - ObjectMeta: metav1.ObjectMeta{ - Name: clients[i].flowSchemaName, - }, - Spec: flowcontrol.FlowSchemaSpec{ - MatchingPrecedence: clients[i].matchingPrecedence, - PriorityLevelConfiguration: flowcontrol.PriorityLevelConfigurationReference{ - Name: clients[i].priorityLevelName, - }, - DistinguisherMethod: &flowcontrol.FlowDistinguisherMethod{ - Type: flowcontrol.FlowDistinguisherMethodByUserType, - }, - Rules: []flowcontrol.PolicyRulesWithSubjects{ - { - Subjects: []flowcontrol.Subject{ - { - Kind: flowcontrol.SubjectKindUser, - User: &flowcontrol.UserSubject{ - Name: clients[i].username, - }, - }, - }, - NonResourceRules: []flowcontrol.NonResourcePolicyRule{ - { - Verbs: []string{flowcontrol.VerbAll}, - NonResourceURLs: []string{flowcontrol.NonResourceAll}, - }, - }, - }, - }, - }, - }, - metav1.CreateOptions{}) - framework.ExpectNoError(err) - defer func(name string) { - framework.ExpectNoError(f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Delete(context.TODO(), name, metav1.DeleteOptions{})) - }(clients[i].flowSchemaName) + _, cleanup = createFlowSchema(f, clients[i].flowSchemaName, clients[i].matchingPrecedence, clients[i].priorityLevelName, clients[i].username) + defer cleanup() } ginkgo.By("getting request concurrency from metrics") for i := range clients { - resp, err := f.ClientSet.CoreV1().RESTClient().Get().RequestURI("/metrics").DoRaw(context.TODO()) - framework.ExpectNoError(err) - sampleDecoder := expfmt.SampleDecoder{ - Dec: expfmt.NewDecoder(bytes.NewBuffer(resp), expfmt.FmtText), - Opts: &expfmt.DecodeOptions{}, - } - for { - var v model.Vector - err := sampleDecoder.Decode(&v) - if err == io.EOF { - break - } - framework.ExpectNoError(err) - for _, metric := range v { - if string(metric.Metric[model.MetricNameLabel]) != requestConcurrencyLimitMetricName { - continue - } - if string(metric.Metric[requestConcurrencyLimitMetricLabelName]) != clients[i].priorityLevelName { - continue - } - clients[i].concurrency = int32(float64(metric.Value) * clients[i].concurrencyMultiplier) - if clients[i].concurrency < 1 { - clients[i].concurrency = 1 - } - framework.Logf("request concurrency for %q will be %d (concurrency share = %d)", clients[i].username, clients[i].concurrency, int32(metric.Value)) - } + realConcurrency := getPriorityLevelConcurrency(f, clients[i].priorityLevelName) + clients[i].concurrency = int32(float64(realConcurrency) * clients[i].concurrencyMultiplier) + if clients[i].concurrency < 1 { + clients[i].concurrency = 1 } + framework.Logf("request concurrency for %q will be %d (concurrency share = %d)", clients[i].username, clients[i].concurrency, realConcurrency) } ginkgo.By(fmt.Sprintf("starting uniform QPS load for %s", loadDuration.String())) @@ -274,6 +139,104 @@ var _ = SIGDescribe("API priority and fairness", func() { }) }) +// createPriorityLevel creates a priority level with the provided assured +// concurrency share. +func createPriorityLevel(f *framework.Framework, priorityLevelName string, assuredConcurrencyShares int32) (*flowcontrol.PriorityLevelConfiguration, func()) { + createdPriorityLevel, err := f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Create( + context.TODO(), + &flowcontrol.PriorityLevelConfiguration{ + ObjectMeta: metav1.ObjectMeta{ + Name: priorityLevelName, + }, + Spec: flowcontrol.PriorityLevelConfigurationSpec{ + Type: flowcontrol.PriorityLevelEnablementLimited, + Limited: &flowcontrol.LimitedPriorityLevelConfiguration{ + AssuredConcurrencyShares: assuredConcurrencyShares, + LimitResponse: flowcontrol.LimitResponse{ + Type: flowcontrol.LimitResponseTypeReject, + }, + }, + }, + }, + metav1.CreateOptions{}) + framework.ExpectNoError(err) + return createdPriorityLevel, func() { + framework.ExpectNoError(f.ClientSet.FlowcontrolV1beta1().PriorityLevelConfigurations().Delete(context.TODO(), priorityLevelName, metav1.DeleteOptions{})) + } +} + +func getPriorityLevelConcurrency(f *framework.Framework, priorityLevelName string) int32 { + resp, err := f.ClientSet.CoreV1().RESTClient().Get().RequestURI("/metrics").DoRaw(context.TODO()) + framework.ExpectNoError(err) + sampleDecoder := expfmt.SampleDecoder{ + Dec: expfmt.NewDecoder(bytes.NewBuffer(resp), expfmt.FmtText), + Opts: &expfmt.DecodeOptions{}, + } + for { + var v model.Vector + err := sampleDecoder.Decode(&v) + if err == io.EOF { + break + } + framework.ExpectNoError(err) + for _, metric := range v { + if string(metric.Metric[model.MetricNameLabel]) != requestConcurrencyLimitMetricName { + continue + } + if string(metric.Metric[requestConcurrencyLimitMetricLabelName]) != priorityLevelName { + continue + } + return int32(metric.Value) + } + } + framework.ExpectNoError(fmt.Errorf("cannot find metric %q with matching priority level name label %q", requestConcurrencyLimitMetricName, priorityLevelName)) + return 0 +} + +// 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()) { + createdFlowSchema, err := f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Create( + context.TODO(), + &flowcontrol.FlowSchema{ + ObjectMeta: metav1.ObjectMeta{ + Name: flowSchemaName, + }, + Spec: flowcontrol.FlowSchemaSpec{ + MatchingPrecedence: matchingPrecedence, + PriorityLevelConfiguration: flowcontrol.PriorityLevelConfigurationReference{ + Name: priorityLevelName, + }, + DistinguisherMethod: &flowcontrol.FlowDistinguisherMethod{ + Type: flowcontrol.FlowDistinguisherMethodByUserType, + }, + Rules: []flowcontrol.PolicyRulesWithSubjects{ + { + Subjects: []flowcontrol.Subject{ + { + Kind: flowcontrol.SubjectKindUser, + User: &flowcontrol.UserSubject{ + Name: matchingUsername, + }, + }, + }, + NonResourceRules: []flowcontrol.NonResourcePolicyRule{ + { + Verbs: []string{flowcontrol.VerbAll}, + NonResourceURLs: []string{flowcontrol.NonResourceAll}, + }, + }, + }, + }, + }, + }, + metav1.CreateOptions{}) + framework.ExpectNoError(err) + return createdFlowSchema, func() { + framework.ExpectNoError(f.ClientSet.FlowcontrolV1beta1().FlowSchemas().Delete(context.TODO(), flowSchemaName, metav1.DeleteOptions{})) + } +} + // makeRequests creates a request to the API server and returns the response. func makeRequest(f *framework.Framework, username string) *http.Response { config := rest.CopyConfig(f.ClientConfig())