Merge pull request kubernetes#24787 from nikhiljindal/storageVersion

Automatic merge from submit-queue

Moving StorageFactory building logic to genericapiserver

Adding a DefaultStorageFactoryBuilder which builds the required StorageFactory.
This allows us to remove the duplicated code between `cmd/kube-apiserver` and `federation/cmd/federated-apiserver`

cc @deads2k @lavalamp @jianhuiz

Author: k8s-merge-robot

cmd/kube-apiserver/app/options/options.go

@@ -23,10 +23,7 @@ import (
 
 	"k8s.io/kubernetes/pkg/admission"
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/api/unversioned"
-	apiutil "k8s.io/kubernetes/pkg/api/util"
 	"k8s.io/kubernetes/pkg/api/validation"
-	"k8s.io/kubernetes/pkg/apimachinery/registered"
 	"k8s.io/kubernetes/pkg/apiserver"
 	"k8s.io/kubernetes/pkg/genericapiserver"
 	kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
@@ -46,7 +43,6 @@ type APIServer struct {
 	BasicAuthFile              string
 	DefaultStorageMediaType    string
 	DeleteCollectionWorkers    int
-	DeprecatedStorageVersion   string
 	EtcdServersOverrides       []string
 	EventTTL                   time.Duration
 	KeystoneURL                string
@@ -62,13 +58,8 @@ type APIServer struct {
 	SSHUser                    string
 	ServiceAccountKeyFile      string
 	ServiceAccountLookup       bool
-	StorageVersions            string
-	// The default values for StorageVersions. StorageVersions overrides
-	// these; you can change this if you want to change the defaults (e.g.,
-	// for testing). This is not actually exposed as a flag.
-	DefaultStorageVersions string
-	TokenAuthFile          string
-	WatchCacheSizes        []string
+	TokenAuthFile              string
+	WatchCacheSizes            []string
 }
 
 // NewAPIServer creates a new APIServer object with default parameters
@@ -81,8 +72,6 @@ func NewAPIServer() *APIServer {
 		DeleteCollectionWorkers: 1,
 		EventTTL:                1 * time.Hour,
 		MasterServiceNamespace:  api.NamespaceDefault,
-		StorageVersions:         registered.AllPreferredGroupVersions(),
-		DefaultStorageVersions:  registered.AllPreferredGroupVersions(),
 		KubeletConfig: kubeletclient.KubeletClientConfig{
 			Port:        ports.KubeletPort,
 			EnableHttps: true,
@@ -92,69 +81,12 @@ func NewAPIServer() *APIServer {
 	return &s
 }
 
-// dest must be a map of group to groupVersion.
-func mergeGroupVersionIntoMap(gvList string, dest map[string]unversioned.GroupVersion) error {
-	for _, gvString := range strings.Split(gvList, ",") {
-		if gvString == "" {
-			continue
-		}
-		// We accept two formats. "group/version" OR
-		// "group=group/version". The latter is used when types
-		// move between groups.
-		if !strings.Contains(gvString, "=") {
-			gv, err := unversioned.ParseGroupVersion(gvString)
-			if err != nil {
-				return err
-			}
-			dest[gv.Group] = gv
-
-		} else {
-			parts := strings.SplitN(gvString, "=", 2)
-			gv, err := unversioned.ParseGroupVersion(parts[1])
-			if err != nil {
-				return err
-			}
-			dest[parts[0]] = gv
-		}
-	}
-
-	return nil
-}
-
-// StorageGroupsToEncodingVersion returns a map from group name to group version,
-// computed from the s.DeprecatedStorageVersion and s.StorageVersions flags.
-// TODO: can we move the whole storage version concept to the generic apiserver?
-func (s *APIServer) StorageGroupsToEncodingVersion() (map[string]unversioned.GroupVersion, error) {
-	storageVersionMap := map[string]unversioned.GroupVersion{}
-	if s.DeprecatedStorageVersion != "" {
-		storageVersionMap[""] = unversioned.GroupVersion{Group: apiutil.GetGroup(s.DeprecatedStorageVersion), Version: apiutil.GetVersion(s.DeprecatedStorageVersion)}
-	}
-
-	// First, get the defaults.
-	if err := mergeGroupVersionIntoMap(s.DefaultStorageVersions, storageVersionMap); err != nil {
-		return nil, err
-	}
-	// Override any defaults with the user settings.
-	if err := mergeGroupVersionIntoMap(s.StorageVersions, storageVersionMap); err != nil {
-		return nil, err
-	}
-
-	return storageVersionMap, nil
-}
-
 // AddFlags adds flags for a specific APIServer to the specified FlagSet
 func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
 	// Add the generic flags.
 	s.ServerRunOptions.AddFlags(fs)
 	// Note: the weird ""+ in below lines seems to be the only way to get gofmt to
 	// arrange these text blocks sensibly. Grrr.
-	fs.StringVar(&s.DeprecatedStorageVersion, "storage-version", s.DeprecatedStorageVersion, "The version to store the legacy v1 resources with. Defaults to server preferred")
-	fs.MarkDeprecated("storage-version", "--storage-version is deprecated and will be removed when the v1 API is retired. See --storage-versions instead.")
-	fs.StringVar(&s.StorageVersions, "storage-versions", s.StorageVersions, "The per-group version to store resources in. "+
-		"Specified in the format \"group1/version1,group2/version2,...\". "+
-		"In the case where objects are moved from one group to the other, you may specify the format \"group1=group2/v1beta1,group3/v1beta1,...\". "+
-		"You only need to pass the groups you wish to change from the defaults. "+
-		"It defaults to a list of preferred versions of all registered groups, which is derived from the KUBE_API_VERSIONS environment variable.")
 	fs.StringVar(&s.DefaultStorageMediaType, "storage-media-type", s.DefaultStorageMediaType, "The media type to use to store objects in storage. Defaults to application/json. Some resources may only support a specific media type and will ignore this setting.")
 	fs.DurationVar(&s.EventTTL, "event-ttl", s.EventTTL, "Amount of time to retain events. Default 1 hour.")
 	fs.StringVar(&s.BasicAuthFile, "basic-auth-file", s.BasicAuthFile, "If set, the file that will be used to admit requests to the secure port of the API server via http basic authentication.")

cmd/kube-apiserver/app/options/options_test.go

@@ -17,72 +17,11 @@ limitations under the License.
 package options
 
 import (
-	"reflect"
 	"testing"
 
 	"github.com/spf13/pflag"
-
-	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/api/unversioned"
-	"k8s.io/kubernetes/pkg/apis/autoscaling"
-	"k8s.io/kubernetes/pkg/apis/extensions"
 )
 
-func TestGenerateStorageVersionMap(t *testing.T) {
-	testCases := []struct {
-		legacyVersion   string
-		storageVersions string
-		defaultVersions string
-		expectedMap     map[string]unversioned.GroupVersion
-	}{
-		{
-			legacyVersion:   "v1",
-			storageVersions: "v1,extensions/v1beta1",
-			expectedMap: map[string]unversioned.GroupVersion{
-				api.GroupName:        {Version: "v1"},
-				extensions.GroupName: {Group: "extensions", Version: "v1beta1"},
-			},
-		},
-		{
-			legacyVersion:   "",
-			storageVersions: "extensions/v1beta1,v1",
-			expectedMap: map[string]unversioned.GroupVersion{
-				api.GroupName:        {Version: "v1"},
-				extensions.GroupName: {Group: "extensions", Version: "v1beta1"},
-			},
-		},
-		{
-			legacyVersion:   "",
-			storageVersions: "autoscaling=extensions/v1beta1,v1",
-			defaultVersions: "extensions/v1beta1,v1,autoscaling/v1",
-			expectedMap: map[string]unversioned.GroupVersion{
-				api.GroupName:         {Version: "v1"},
-				autoscaling.GroupName: {Group: "extensions", Version: "v1beta1"},
-				extensions.GroupName:  {Group: "extensions", Version: "v1beta1"},
-			},
-		},
-		{
-			legacyVersion:   "",
-			storageVersions: "",
-			expectedMap:     map[string]unversioned.GroupVersion{},
-		},
-	}
-	for i, test := range testCases {
-		s := APIServer{
-			DeprecatedStorageVersion: test.legacyVersion,
-			StorageVersions:          test.storageVersions,
-			DefaultStorageVersions:   test.defaultVersions,
-		}
-		output, err := s.StorageGroupsToEncodingVersion()
-		if err != nil {
-			t.Errorf("%v: unexpected error: %v", i, err)
-		}
-		if !reflect.DeepEqual(test.expectedMap, output) {
-			t.Errorf("%v: unexpected error. expect: %v, got: %v", i, test.expectedMap, output)
-		}
-	}
-}
-
 func TestAddFlagsFlag(t *testing.T) {
 	// TODO: This only tests the enable-swagger-ui flag for now.
 	// Expand the test to include other flags as well.

cmd/kube-apiserver/app/server.go

@@ -21,7 +21,6 @@ package app
 
 import (
 	"crypto/tls"
-	"fmt"
 	"net"
 	"net/url"
 	"strconv"
@@ -35,26 +34,18 @@ import (
 	"k8s.io/kubernetes/pkg/admission"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
-	apiv1 "k8s.io/kubernetes/pkg/api/v1"
-	appsapi "k8s.io/kubernetes/pkg/apis/apps/v1alpha1"
 	"k8s.io/kubernetes/pkg/apis/autoscaling"
-	autoscalingapiv1 "k8s.io/kubernetes/pkg/apis/autoscaling/v1"
 	"k8s.io/kubernetes/pkg/apis/batch"
-	batchapiv1 "k8s.io/kubernetes/pkg/apis/batch/v1"
 	"k8s.io/kubernetes/pkg/apis/extensions"
-	extensionsapiv1beta1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 	"k8s.io/kubernetes/pkg/apiserver"
 	"k8s.io/kubernetes/pkg/apiserver/authenticator"
 	"k8s.io/kubernetes/pkg/capabilities"
-	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
-	"k8s.io/kubernetes/pkg/client/restclient"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	"k8s.io/kubernetes/pkg/genericapiserver"
 	kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
 	"k8s.io/kubernetes/pkg/master"
 	"k8s.io/kubernetes/pkg/registry/cachesize"
-	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/serviceaccount"
 )
 
@@ -132,44 +123,17 @@ func Run(s *options.APIServer) error {
 		glog.Fatalf("Failure to start kubelet client: %v", err)
 	}
 
-	apiResourceConfigSource, err := parseRuntimeConfig(s)
+	storageGroupsToEncodingVersion, err := s.StorageGroupsToEncodingVersion()
 	if err != nil {
-		glog.Fatalf("error in parsing runtime-config: %s", err)
+		glog.Fatalf("error generating storage version map: %s", err)
 	}
-
-	clientConfig := &restclient.Config{
-		Host: net.JoinHostPort(s.InsecureBindAddress.String(), strconv.Itoa(s.InsecurePort)),
-		// Increase QPS limits. The client is currently passed to all admission plugins,
-		// and those can be throttled in case of higher load on apiserver - see #22340 and #22422
-		// for more details. Once #22422 is fixed, we may want to remove it.
-		QPS:   50,
-		Burst: 100,
-	}
-	if len(s.DeprecatedStorageVersion) != 0 {
-		gv, err := unversioned.ParseGroupVersion(s.DeprecatedStorageVersion)
-		if err != nil {
-			glog.Fatalf("error in parsing group version: %s", err)
-		}
-		clientConfig.GroupVersion = &gv
-	}
-
-	client, err := clientset.NewForConfig(clientConfig)
+	storageFactory, err := genericapiserver.BuildDefaultStorageFactory(
+		s.StorageConfig, s.DefaultStorageMediaType, api.Codecs,
+		genericapiserver.NewDefaultResourceEncodingConfig(), storageGroupsToEncodingVersion,
+		master.DefaultAPIResourceConfigSource(), s.RuntimeConfig)
 	if err != nil {
-		glog.Errorf("Failed to create clientset: %v", err)
-	}
-
-	resourceEncoding := genericapiserver.NewDefaultResourceEncodingConfig()
-	groupToEncoding, err := s.StorageGroupsToEncodingVersion()
-	if err != nil {
-		glog.Fatalf("error getting group encoding: %s", err)
-	}
-	for group, storageEncodingVersion := range groupToEncoding {
-		resourceEncoding.SetVersionEncoding(group, storageEncodingVersion, unversioned.GroupVersion{Group: group, Version: runtime.APIVersionInternal})
+		glog.Fatalf("error in initializing storage factory: %s", err)
 	}
-
-	storageFactory := genericapiserver.NewDefaultStorageFactory(s.StorageConfig, s.DefaultStorageMediaType, api.Codecs, resourceEncoding, apiResourceConfigSource)
-	// third party resources are always serialized to storage using JSON
-	storageFactory.SetSerializer(extensions.Resource("thirdpartyresources"), "application/json", nil)
 	storageFactory.AddCohabitatingResources(batch.Resource("jobs"), extensions.Resource("jobs"))
 	storageFactory.AddCohabitatingResources(autoscaling.Resource("horizontalpodautoscalers"), extensions.Resource("horizontalpodautoscalers"))
 	for _, override := range s.EtcdServersOverrides {
@@ -238,6 +202,10 @@ func Run(s *options.APIServer) error {
 	}
 
 	admissionControlPluginNames := strings.Split(s.AdmissionControl, ",")
+	client, err := s.NewSelfClient()
+	if err != nil {
+		glog.Errorf("Failed to create clientset: %v", err)
+	}
 	admissionController := admission.NewFromPlugins(client, admissionControlPluginNames, s.AdmissionControlConfigFile)
 
 	genericConfig := genericapiserver.NewConfig(s.ServerRunOptions)
@@ -247,7 +215,7 @@ func Run(s *options.APIServer) error {
 	genericConfig.SupportsBasicAuth = len(s.BasicAuthFile) > 0
 	genericConfig.Authorizer = authorizer
 	genericConfig.AdmissionControl = admissionController
-	genericConfig.APIResourceConfigSource = apiResourceConfigSource
+	genericConfig.APIResourceConfigSource = storageFactory.APIResourceConfigSource
 	genericConfig.MasterServiceNamespace = s.MasterServiceNamespace
 	genericConfig.ProxyDialer = proxyDialerFn
 	genericConfig.ProxyTLSClientConfig = proxyTLSClientConfig
@@ -275,85 +243,3 @@ func Run(s *options.APIServer) error {
 	m.Run(s.ServerRunOptions)
 	return nil
 }
-
-func getRuntimeConfigValue(s *options.APIServer, apiKey string, defaultValue bool) bool {
-	flagValue, ok := s.RuntimeConfig[apiKey]
-	if ok {
-		if flagValue == "" {
-			return true
-		}
-		boolValue, err := strconv.ParseBool(flagValue)
-		if err != nil {
-			glog.Fatalf("Invalid value of %s: %s, err: %v", apiKey, flagValue, err)
-		}
-		return boolValue
-	}
-	return defaultValue
-}
-
-// Parses the given runtime-config and formats it into genericapiserver.APIResourceConfigSource
-func parseRuntimeConfig(s *options.APIServer) (genericapiserver.APIResourceConfigSource, error) {
-	v1GroupVersionString := "api/v1"
-	extensionsGroupVersionString := extensionsapiv1beta1.SchemeGroupVersion.String()
-	versionToResourceSpecifier := map[unversioned.GroupVersion]string{
-		apiv1.SchemeGroupVersion:                v1GroupVersionString,
-		extensionsapiv1beta1.SchemeGroupVersion: extensionsGroupVersionString,
-		batchapiv1.SchemeGroupVersion:           batchapiv1.SchemeGroupVersion.String(),
-		autoscalingapiv1.SchemeGroupVersion:     autoscalingapiv1.SchemeGroupVersion.String(),
-		appsapi.SchemeGroupVersion:              appsapi.SchemeGroupVersion.String(),
-	}
-
-	resourceConfig := master.DefaultAPIResourceConfigSource()
-
-	// "api/all=false" allows users to selectively enable specific api versions.
-	enableAPIByDefault := true
-	allAPIFlagValue, ok := s.RuntimeConfig["api/all"]
-	if ok && allAPIFlagValue == "false" {
-		enableAPIByDefault = false
-	}
-
-	// "api/legacy=false" allows users to disable legacy api versions.
-	disableLegacyAPIs := false
-	legacyAPIFlagValue, ok := s.RuntimeConfig["api/legacy"]
-	if ok && legacyAPIFlagValue == "false" {
-		disableLegacyAPIs = true
-	}
-	_ = disableLegacyAPIs // hush the compiler while we don't have legacy APIs to disable.
-
-	// "<resourceSpecifier>={true|false} allows users to enable/disable API.
-	// This takes preference over api/all and api/legacy, if specified.
-	for version, resourceSpecifier := range versionToResourceSpecifier {
-		enableVersion := getRuntimeConfigValue(s, resourceSpecifier, enableAPIByDefault)
-		if enableVersion {
-			resourceConfig.EnableVersions(version)
-		} else {
-			resourceConfig.DisableVersions(version)
-		}
-	}
-
-	for key := range s.RuntimeConfig {
-		tokens := strings.Split(key, "/")
-		if len(tokens) != 3 {
-			continue
-		}
-
-		switch {
-		case strings.HasPrefix(key, extensionsGroupVersionString+"/"):
-			if !resourceConfig.AnyResourcesForVersionEnabled(extensionsapiv1beta1.SchemeGroupVersion) {
-				return nil, fmt.Errorf("%v is disabled, you cannot configure its resources individually", extensionsapiv1beta1.SchemeGroupVersion)
-			}
-
-			resource := strings.TrimPrefix(key, extensionsGroupVersionString+"/")
-			if getRuntimeConfigValue(s, key, false) {
-				resourceConfig.EnableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource(resource))
-			} else {
-				resourceConfig.DisableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource(resource))
-			}
-
-		default:
-			// TODO enable individual resource capability for all GroupVersionResources
-			return nil, fmt.Errorf("%v resources cannot be enabled/disabled individually", key)
-		}
-	}
-	return resourceConfig, nil
-}