review: several fixes and addressing comments

pkg/apis/flowcontrol/validation/validation_test.go

@@ -554,7 +554,7 @@ func TestFlowSchemaValidation(t *testing.T) {
 					Name: "system-foo",
 				},
 				Spec: flowcontrol.FlowSchemaSpec{
-					MatchingPrecedence: 50000,
+					MatchingPrecedence: 10001,
 					PriorityLevelConfiguration: flowcontrol.PriorityLevelConfigurationReference{
 						Name: "system-bar",
 					},
@@ -579,7 +579,7 @@ func TestFlowSchemaValidation(t *testing.T) {
 				},
 			},
 			expectedErrors: field.ErrorList{
-				field.Invalid(field.NewPath("spec").Child("matchingPrecedence"), int32(50000), "must not be greater than 10000"),
+				field.Invalid(field.NewPath("spec").Child("matchingPrecedence"), int32(10001), "must not be greater than 10000"),
 			},
 		},
 	}

pkg/registry/flowcontrol/rest/BUILD

@@ -8,9 +8,11 @@ go_library(
     deps = [
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/apis/flowcontrol:go_default_library",
+        "//pkg/apis/flowcontrol/v1alpha1:go_default_library",
         "//pkg/registry/flowcontrol/flowschema/storage:go_default_library",
         "//pkg/registry/flowcontrol/prioritylevelconfiguration/storage:go_default_library",
         "//staging/src/k8s.io/api/flowcontrol/v1alpha1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
@@ -43,9 +45,11 @@ go_test(
     srcs = ["storage_flowcontrol_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//pkg/apis/flowcontrol/v1alpha1:go_default_library",
         "//staging/src/k8s.io/api/flowcontrol/v1alpha1:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/github.com/stretchr/testify/require:go_default_library",
     ],
 )

pkg/registry/flowcontrol/rest/storage_flowcontrol.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	flowcontrolv1alpha1 "k8s.io/api/flowcontrol/v1alpha1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -33,6 +34,7 @@ import (
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/apis/flowcontrol"
+	flowcontrolapisv1alpha1 "k8s.io/kubernetes/pkg/apis/flowcontrol/v1alpha1"
 	flowschemastore "k8s.io/kubernetes/pkg/registry/flowcontrol/flowschema/storage"
 	prioritylevelconfigurationstore "k8s.io/kubernetes/pkg/registry/flowcontrol/prioritylevelconfiguration/storage"
 )
@@ -94,20 +96,21 @@ func (p RESTStorageProvider) PostStartHook() (string, genericapiserver.PostStart
 			_ = wait.PollImmediateUntil(
 				retryCreatingSuggestedSettingsInterval,
 				func() (bool, error) {
-					shouldEnsureSuggested, err := shouldEnsureAllPredefined(flowcontrolClientSet)
+					shouldEnsureSuggested, err := lastMandatoryExists(flowcontrolClientSet)
 					if err != nil {
 						klog.Errorf("failed getting exempt flow-schema, will retry later: %v", err)
 						return false, nil
 					}
-					if shouldEnsureSuggested {
+					if !shouldEnsureSuggested {
-						err := ensure(
+						return true, nil
-							flowcontrolClientSet,
+					}
-							flowcontrolbootstrap.SuggestedFlowSchemas,
+					err = ensure(
-							flowcontrolbootstrap.SuggestedPriorityLevelConfigurations)
+						flowcontrolClientSet,
-						if err != nil {
+						flowcontrolbootstrap.SuggestedFlowSchemas,
-							klog.Errorf("failed ensuring suggested settings, will retry later: %v", err)
+						flowcontrolbootstrap.SuggestedPriorityLevelConfigurations)
-							return false, nil
+					if err != nil {
-						}
+						klog.Errorf("failed ensuring suggested settings, will retry later: %v", err)
+						return false, nil
 					}
 					return true, nil
 				},
@@ -124,7 +127,7 @@ func (p RESTStorageProvider) PostStartHook() (string, genericapiserver.PostStart
 						// the full initial set of objects from being created.
 						flowcontrolbootstrap.MandatoryPriorityLevelConfigurations,
 					); err != nil {
-						klog.Errorf("failed creating default flowcontrol settings: %v", err)
+						klog.Errorf("failed creating mandatory flowcontrol settings: %v", err)
 						return false, nil
 					}
 					return false, nil // always retry
@@ -138,7 +141,7 @@ func (p RESTStorageProvider) PostStartHook() (string, genericapiserver.PostStart
 
 // Returns false if there's a "exempt" priority-level existing in the cluster, otherwise returns a true
 // if the "exempt" priority-level is not found.
-func shouldEnsureAllPredefined(flowcontrolClientSet flowcontrolclient.FlowcontrolV1alpha1Interface) (bool, error) {
+func lastMandatoryExists(flowcontrolClientSet flowcontrolclient.FlowcontrolV1alpha1Interface) (bool, error) {
 	if _, err := flowcontrolClientSet.PriorityLevelConfigurations().Get(flowcontrol.PriorityLevelConfigurationNameExempt, metav1.GetOptions{}); err != nil {
 		if apierrors.IsNotFound(err) {
 			return true, nil
@@ -175,39 +178,75 @@ func ensure(flowcontrolClientSet flowcontrolclient.FlowcontrolV1alpha1Interface,
 }
 
 func upgrade(flowcontrolClientSet flowcontrolclient.FlowcontrolV1alpha1Interface, flowSchemas []*flowcontrolv1alpha1.FlowSchema, priorityLevels []*flowcontrolv1alpha1.PriorityLevelConfiguration) error {
-	for _, flowSchema := range flowSchemas {
+	for _, expectedFlowSchema := range flowSchemas {
-		_, err := flowcontrolClientSet.FlowSchemas().Get(flowSchema.Name, metav1.GetOptions{})
+		actualFlowSchema, err := flowcontrolClientSet.FlowSchemas().Get(expectedFlowSchema.Name, metav1.GetOptions{})
-		if err != nil {
+		if err == nil {
-			return fmt.Errorf("failed getting FlowSchema %s due to %v, will retry later", flowSchema.Name, err)
+			// TODO(yue9944882): extract existing version from label and compare
+			// TODO(yue9944882): create w/ version string attached
+			identical, err := flowSchemaHasWrongSpec(expectedFlowSchema, actualFlowSchema)
+			if err != nil {
+				return fmt.Errorf("failed checking if mandatory FlowSchema %s is up-to-date due to %v, will retry later", expectedFlowSchema.Name, err)
+			}
+			if !identical {
+				if _, err := flowcontrolClientSet.FlowSchemas().Update(expectedFlowSchema); err != nil {
+					return fmt.Errorf("failed upgrading mandatory FlowSchema %s due to %v, will retry later", expectedFlowSchema.Name, err)
+				}
+			}
+			continue
 		}
-		// TODO(yue9944882): extract existing version from label and compare
+		if !apierrors.IsNotFound(err) {
-		// TODO(yue9944882): create w/ version string attached
+			return fmt.Errorf("failed getting FlowSchema %s due to %v, will retry later", expectedFlowSchema.Name, err)
-		_, err = flowcontrolClientSet.FlowSchemas().Create(flowSchema)
+		}
+		_, err = flowcontrolClientSet.FlowSchemas().Create(expectedFlowSchema)
 		if apierrors.IsAlreadyExists(err) {
-			klog.V(3).Infof("system preset FlowSchema %s already exists, skipping creating", flowSchema.Name)
+			klog.V(3).Infof("system preset FlowSchema %s already exists, skipping creating", expectedFlowSchema.Name)
 			continue
 		}
 		if err != nil {
-			return fmt.Errorf("cannot create FlowSchema %s due to %v", flowSchema.Name, err)
+			return fmt.Errorf("cannot create FlowSchema %s due to %v", expectedFlowSchema.Name, err)
 		}
-		klog.V(3).Infof("created system preset FlowSchema %s", flowSchema.Name)
+		klog.V(3).Infof("created system preset FlowSchema %s", expectedFlowSchema.Name)
 	}
-	for _, priorityLevelConfiguration := range priorityLevels {
+	for _, expectedPriorityLevelConfiguration := range priorityLevels {
-		_, err := flowcontrolClientSet.FlowSchemas().Get(priorityLevelConfiguration.Name, metav1.GetOptions{})
+		actualPriorityLevelConfiguration, err := flowcontrolClientSet.PriorityLevelConfigurations().Get(expectedPriorityLevelConfiguration.Name, metav1.GetOptions{})
-		if err != nil {
+		if err == nil {
-			return fmt.Errorf("failed getting PriorityLevelConfiguration %s due to %v, will retry later", priorityLevelConfiguration.Name, err)
+			// TODO(yue9944882): extract existing version from label and compare
+			// TODO(yue9944882): create w/ version string attached
+			identical, err := priorityLevelHasWrongSpec(expectedPriorityLevelConfiguration, actualPriorityLevelConfiguration)
+			if err != nil {
+				return fmt.Errorf("failed checking if mandatory PriorityLevelConfiguration %s is up-to-date due to %v, will retry later", expectedPriorityLevelConfiguration.Name, err)
+			}
+			if !identical {
+				if _, err := flowcontrolClientSet.PriorityLevelConfigurations().Update(expectedPriorityLevelConfiguration); err != nil {
+					return fmt.Errorf("failed upgrading mandatory PriorityLevelConfiguration %s due to %v, will retry later", expectedPriorityLevelConfiguration.Name, err)
+				}
+			}
+			continue
 		}
-		// TODO(yue9944882): extract existing version from label and compare
+		if !apierrors.IsNotFound(err) {
-		// TODO(yue9944882): create w/ version string attached
+			return fmt.Errorf("failed getting PriorityLevelConfiguration %s due to %v, will retry later", expectedPriorityLevelConfiguration.Name, err)
-		_, err = flowcontrolClientSet.PriorityLevelConfigurations().Create(priorityLevelConfiguration)
+		}
+		_, err = flowcontrolClientSet.PriorityLevelConfigurations().Create(expectedPriorityLevelConfiguration)
 		if apierrors.IsAlreadyExists(err) {
-			klog.V(3).Infof("system preset PriorityLevelConfiguration %s already exists, skipping creating", priorityLevelConfiguration.Name)
+			klog.V(3).Infof("system preset PriorityLevelConfiguration %s already exists, skipping creating", expectedPriorityLevelConfiguration.Name)
 			continue
 		}
 		if err != nil {
-			return fmt.Errorf("cannot create PriorityLevelConfiguration %s due to %v", priorityLevelConfiguration.Name, err)
+			return fmt.Errorf("cannot create PriorityLevelConfiguration %s due to %v", expectedPriorityLevelConfiguration.Name, err)
 		}
-		klog.V(3).Infof("created system preset PriorityLevelConfiguration %s", priorityLevelConfiguration.Name)
+		klog.V(3).Infof("created system preset PriorityLevelConfiguration %s", expectedPriorityLevelConfiguration.Name)
 	}
 	return nil
 }
+
+func flowSchemaHasWrongSpec(expected, actual *flowcontrolv1alpha1.FlowSchema) (bool, error) {
+	copiedExpectedFlowSchema := expected.DeepCopy()
+	flowcontrolapisv1alpha1.SetObjectDefaults_FlowSchema(copiedExpectedFlowSchema)
+	return !equality.Semantic.DeepEqual(copiedExpectedFlowSchema.Spec, actual.Spec), nil
+}
+
+func priorityLevelHasWrongSpec(expected, actual *flowcontrolv1alpha1.PriorityLevelConfiguration) (bool, error) {
+	copiedExpectedPriorityLevel := expected.DeepCopy()
+	flowcontrolapisv1alpha1.SetObjectDefaults_PriorityLevelConfiguration(copiedExpectedPriorityLevel)
+	return !equality.Semantic.DeepEqual(copiedExpectedPriorityLevel.Spec, actual.Spec), nil
+}

pkg/registry/flowcontrol/rest/storage_flowcontrol_test.go

@@ -17,18 +17,20 @@ limitations under the License.
 package rest
 
 import (
+	"github.com/stretchr/testify/require"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	flowcontrol "k8s.io/api/flowcontrol/v1alpha1"
+	flowcontrolv1alpha1 "k8s.io/api/flowcontrol/v1alpha1"
 	"k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap"
 	"k8s.io/client-go/kubernetes/fake"
+	flowcontrolapisv1alpha1 "k8s.io/kubernetes/pkg/apis/flowcontrol/v1alpha1"
 )
 
 func TestShouldEnsurePredefinedSettings(t *testing.T) {
 	testCases := []struct {
 		name                  string
-		existingPriorityLevel *flowcontrol.PriorityLevelConfiguration
+		existingPriorityLevel *flowcontrolv1alpha1.PriorityLevelConfiguration
 		expected              bool
 	}{
 		{
@@ -49,9 +51,121 @@ func TestShouldEnsurePredefinedSettings(t *testing.T) {
 			if testCase.existingPriorityLevel != nil {
 				c.FlowcontrolV1alpha1().PriorityLevelConfigurations().Create(testCase.existingPriorityLevel)
 			}
-			should, err := shouldEnsureAllPredefined(c.FlowcontrolV1alpha1())
+			should, err := lastMandatoryExists(c.FlowcontrolV1alpha1())
 			assert.NoError(t, err)
 			assert.Equal(t, testCase.expected, should)
 		})
 	}
 }
+
+func TestFlowSchemaHasWrongSpec(t *testing.T) {
+	fs1 := &flowcontrolv1alpha1.FlowSchema{
+		Spec: flowcontrolv1alpha1.FlowSchemaSpec{},
+	}
+	fs2 := &flowcontrolv1alpha1.FlowSchema{
+		Spec: flowcontrolv1alpha1.FlowSchemaSpec{
+			MatchingPrecedence: 1,
+		},
+	}
+	fs1Defaulted := &flowcontrolv1alpha1.FlowSchema{
+		Spec: flowcontrolv1alpha1.FlowSchemaSpec{
+			MatchingPrecedence: flowcontrolapisv1alpha1.FlowSchemaDefaultMatchingPrecedence,
+		},
+	}
+	testCases := []struct {
+		name         string
+		expected     *flowcontrolv1alpha1.FlowSchema
+		actual       *flowcontrolv1alpha1.FlowSchema
+		hasWrongSpec bool
+	}{
+		{
+			name:         "identical flow-schemas should work",
+			expected:     bootstrap.MandatoryFlowSchemaCatchAll,
+			actual:       bootstrap.MandatoryFlowSchemaCatchAll,
+			hasWrongSpec: false,
+		},
+		{
+			name:         "defaulted flow-schemas should work",
+			expected:     fs1,
+			actual:       fs1Defaulted,
+			hasWrongSpec: false,
+		},
+		{
+			name:         "non-defaulted flow-schema has wrong spec",
+			expected:     fs1,
+			actual:       fs2,
+			hasWrongSpec: true,
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			w, err := flowSchemaHasWrongSpec(testCase.expected, testCase.actual)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.hasWrongSpec, w)
+		})
+	}
+}
+
+func TestPriorityLevelHasWrongSpec(t *testing.T) {
+	pl1 := &flowcontrolv1alpha1.PriorityLevelConfiguration{
+		Spec: flowcontrolv1alpha1.PriorityLevelConfigurationSpec{
+			Type: flowcontrolv1alpha1.PriorityLevelEnablementLimited,
+			Limited: &flowcontrolv1alpha1.LimitedPriorityLevelConfiguration{
+				LimitResponse: flowcontrolv1alpha1.LimitResponse{
+					Type: flowcontrolv1alpha1.LimitResponseTypeReject,
+				},
+			},
+		},
+	}
+	pl2 := &flowcontrolv1alpha1.PriorityLevelConfiguration{
+		Spec: flowcontrolv1alpha1.PriorityLevelConfigurationSpec{
+			Type: flowcontrolv1alpha1.PriorityLevelEnablementLimited,
+			Limited: &flowcontrolv1alpha1.LimitedPriorityLevelConfiguration{
+				AssuredConcurrencyShares: 1,
+			},
+		},
+	}
+	pl1Defaulted := &flowcontrolv1alpha1.PriorityLevelConfiguration{
+		Spec: flowcontrolv1alpha1.PriorityLevelConfigurationSpec{
+			Type: flowcontrolv1alpha1.PriorityLevelEnablementLimited,
+			Limited: &flowcontrolv1alpha1.LimitedPriorityLevelConfiguration{
+				AssuredConcurrencyShares: flowcontrolapisv1alpha1.PriorityLevelConfigurationDefaultAssuredConcurrencyShares,
+				LimitResponse: flowcontrolv1alpha1.LimitResponse{
+					Type: flowcontrolv1alpha1.LimitResponseTypeReject,
+				},
+			},
+		},
+	}
+	testCases := []struct {
+		name         string
+		expected     *flowcontrolv1alpha1.PriorityLevelConfiguration
+		actual       *flowcontrolv1alpha1.PriorityLevelConfiguration
+		hasWrongSpec bool
+	}{
+		{
+			name:         "identical priority-level should work",
+			expected:     bootstrap.MandatoryPriorityLevelConfigurationCatchAll,
+			actual:       bootstrap.MandatoryPriorityLevelConfigurationCatchAll,
+			hasWrongSpec: false,
+		},
+		{
+			name:         "defaulted priority-level should work",
+			expected:     pl1,
+			actual:       pl1Defaulted,
+			hasWrongSpec: false,
+		},
+		{
+			name:         "non-defaulted priority-level has wrong spec",
+			expected:     pl1,
+			actual:       pl2,
+			hasWrongSpec: true,
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			w, err := priorityLevelHasWrongSpec(testCase.expected, testCase.actual)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.hasWrongSpec, w)
+		})
+	}
+}

staging/src/k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap/default.go

@@ -63,6 +63,7 @@ var (
 		SuggestedFlowSchemaKubeControllerManager,     // references "workload-high" priority-level
 		SuggestedFlowSchemaKubeScheduler,             // references "workload-high" priority-level
 		SuggestedFlowSchemaKubeSystemServiceAccounts, // references "workload-high" priority-level
+		SuggestedFlowSchemaServiceAccounts,           // references "workload-low" priority-level
 	}
 )
 
@@ -98,7 +99,7 @@ var (
 	MandatoryFlowSchemaExempt = newFlowSchema(
 		"exempt",
 		flowcontrol.PriorityLevelConfigurationNameExempt,
-		0,  // matchingPrecedence
+		1,  // matchingPrecedence
 		"", // distinguisherMethodType
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: groups(user.SystemPrivilegedGroup),
@@ -221,7 +222,7 @@ var (
 // Suggested FlowSchema objects
 var (
 	SuggestedFlowSchemaSystemNodes = newFlowSchema(
-		"system-nodes", "system", 1500,
+		"system-nodes", "system", 500,
 		flowcontrol.FlowDistinguisherMethodByUserType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: groups(user.NodesGroup), // the nodes group
@@ -239,7 +240,7 @@ var (
 		},
 	)
 	SuggestedFlowSchemaSystemLeaderElection = newFlowSchema(
-		"system-leader-election", "leader-election", 2500,
+		"system-leader-election", "leader-election", 100,
 		flowcontrol.FlowDistinguisherMethodByUserType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: append(
@@ -262,19 +263,19 @@ var (
 		},
 	)
 	SuggestedFlowSchemaWorkloadLeaderElection = newFlowSchema(
-		"workload-leader-election", "leader-election", 2500,
+		"workload-leader-election", "leader-election", 200,
 		flowcontrol.FlowDistinguisherMethodByUserType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: kubeSystemServiceAccount(flowcontrol.NameAll),
 			ResourceRules: []flowcontrol.ResourcePolicyRule{
 				resourceRule(
-					[]string{flowcontrol.VerbAll},
+					[]string{"get", "create", "update"},
 					[]string{corev1.GroupName},
 					[]string{"endpoints", "configmaps"},
 					[]string{flowcontrol.NamespaceEvery},
 					false),
 				resourceRule(
-					[]string{flowcontrol.VerbAll},
+					[]string{"get", "create", "update"},
 					[]string{coordinationv1.GroupName},
 					[]string{"leases"},
 					[]string{flowcontrol.NamespaceEvery},
@@ -283,7 +284,7 @@ var (
 		},
 	)
 	SuggestedFlowSchemaKubeControllerManager = newFlowSchema(
-		"kube-controller-manager", "workload-high", 3500,
+		"kube-controller-manager", "workload-high", 800,
 		flowcontrol.FlowDistinguisherMethodByNamespaceType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: users(user.KubeControllerManager),
@@ -301,7 +302,7 @@ var (
 		},
 	)
 	SuggestedFlowSchemaKubeScheduler = newFlowSchema(
-		"kube-scheduler", "workload-high", 3500,
+		"kube-scheduler", "workload-high", 800,
 		flowcontrol.FlowDistinguisherMethodByNamespaceType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: users(user.KubeScheduler),
@@ -319,7 +320,7 @@ var (
 		},
 	)
 	SuggestedFlowSchemaKubeSystemServiceAccounts = newFlowSchema(
-		"kube-system-service-accounts", "workload-high", 3500,
+		"kube-system-service-accounts", "workload-high", 900,
 		flowcontrol.FlowDistinguisherMethodByNamespaceType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: kubeSystemServiceAccount(flowcontrol.NameAll),
@@ -337,7 +338,7 @@ var (
 		},
 	)
 	SuggestedFlowSchemaServiceAccounts = newFlowSchema(
-		"service-accounts", "workload-low", 7500,
+		"service-accounts", "workload-low", 9000,
 		flowcontrol.FlowDistinguisherMethodByUserType,
 		flowcontrol.PolicyRulesWithSubjects{
 			Subjects: groups(serviceaccount.AllServiceAccountsGroup),