github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

This is a combination of 3 commits. 

refactor storage factory options

review: minor changes

1. make storage factory config complete with options
2. make BuildGenericConfig private

review: move codes
pull/8/head
yue9944882 2018-07-19 21:22:49 +08:00
parent 0e097af8d8
commit e8ae7887a5
5 changed files with 125 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cmd/kube-apiserver/app/BUILD
View File

@ -16,15 +16,6 @@ go_library(
deps = [
"//cmd/kube-apiserver/app/options:go_default_library",
"//pkg/api/legacyscheme:go_default_library",
"//pkg/apis/admissionregistration:go_default_library",
"//pkg/apis/apps:go_default_library",
"//pkg/apis/batch:go_default_library",
"//pkg/apis/core:go_default_library",
"//pkg/apis/events:go_default_library",
"//pkg/apis/extensions:go_default_library",
"//pkg/apis/networking:go_default_library",
"//pkg/apis/policy:go_default_library",
"//pkg/apis/storage:go_default_library",
"//pkg/capabilities:go_default_library",
"//pkg/client/clientset_generated/internalclientset:go_default_library",
"//pkg/client/informers/informers_generated/internalversion:go_default_library",
@ -72,7 +63,6 @@ go_library(
"//staging/src/k8s.io/apiserver/pkg/server/filters:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/healthz:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/options:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/storage:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/storage/etcd3/preflight:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",

87
cmd/kube-apiserver/app/server.go
View File

@ -37,7 +37,6 @@ import (
extensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
"k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
utilnet "k8s.io/apimachinery/pkg/util/net"
"k8s.io/apimachinery/pkg/util/sets"
@ -51,7 +50,6 @@ import (
genericapiserver "k8s.io/apiserver/pkg/server"
"k8s.io/apiserver/pkg/server/filters"
serveroptions "k8s.io/apiserver/pkg/server/options"
"k8s.io/apiserver/pkg/server/options/encryptionconfig"
serverstorage "k8s.io/apiserver/pkg/server/storage"
"k8s.io/apiserver/pkg/storage/etcd3/preflight"
utilfeature "k8s.io/apiserver/pkg/util/feature"
@ -66,15 +64,6 @@ import (
openapi "k8s.io/kube-openapi/pkg/common"
"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
"k8s.io/kubernetes/pkg/api/legacyscheme"
"k8s.io/kubernetes/pkg/apis/admissionregistration"
"k8s.io/kubernetes/pkg/apis/apps"
"k8s.io/kubernetes/pkg/apis/batch"
api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/apis/events"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/apis/networking"
"k8s.io/kubernetes/pkg/apis/policy"
"k8s.io/kubernetes/pkg/apis/storage"
"k8s.io/kubernetes/pkg/capabilities"
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
@ -284,7 +273,8 @@ func CreateKubeAPIServerConfig(
lastErr error,
) {
var genericConfig *genericapiserver.Config
genericConfig, sharedInformers, versionedInformers, insecureServingInfo, serviceResolver, pluginInitializers, admissionPostStartHook, lastErr = BuildGenericConfig(s.ServerRunOptions, proxyTransport)
var storageFactory *serverstorage.DefaultStorageFactory
genericConfig, sharedInformers, versionedInformers, insecureServingInfo, serviceResolver, pluginInitializers, admissionPostStartHook, storageFactory, lastErr = buildGenericConfig(s.ServerRunOptions, proxyTransport)
if lastErr != nil {
return
}
@ -312,11 +302,6 @@ func CreateKubeAPIServerConfig(
return
}
storageFactory, lastErr := BuildStorageFactory(s.ServerRunOptions, genericConfig.MergedResourceConfig)
if lastErr != nil {
return
}
clientCA, lastErr := readCAorNil(s.Authentication.ClientCert.ClientCA)
if lastErr != nil {
return
@ -412,7 +397,7 @@ func CreateKubeAPIServerConfig(
}
// BuildGenericConfig takes the master server options and produces the genericapiserver.Config associated with it
func BuildGenericConfig(
func buildGenericConfig(
s *options.ServerRunOptions,
proxyTransport *http.Transport,
) (
@ -423,9 +408,12 @@ func BuildGenericConfig(
serviceResolver aggregatorapiserver.ServiceResolver,
pluginInitializers []admission.PluginInitializer,
admissionPostStartHook genericapiserver.PostStartHookFunc,
storageFactory *serverstorage.DefaultStorageFactory,
lastErr error,
) {
genericConfig = genericapiserver.NewConfig(legacyscheme.Codecs)
genericConfig.MergedResourceConfig = master.DefaultAPIResourceConfigSource()
if lastErr = s.GenericServerRunOptions.ApplyTo(genericConfig); lastErr != nil {
return
}
@ -461,7 +449,14 @@ func BuildGenericConfig(
kubeVersion := version.Get()
genericConfig.Version = &kubeVersion
storageFactory, lastErr := BuildStorageFactory(s, genericConfig.MergedResourceConfig)
storageFactoryConfig := kubeapiserver.NewStorageFactoryConfig()
storageFactoryConfig.ApiResourceConfig = genericConfig.MergedResourceConfig
completedStorageFactoryConfig, err := storageFactoryConfig.Complete(s.Etcd, s.StorageSerialization)
if err != nil {
lastErr = err
return
}
storageFactory, lastErr = completedStorageFactoryConfig.New()
if lastErr != nil {
return
}
@ -627,60 +622,6 @@ func BuildAuthorizer(s *options.ServerRunOptions, sharedInformers informers.Shar
return authorizationConfig.New()
}
// BuildStorageFactory constructs the storage factory. If encryption at rest is used, it expects
// all supported KMS plugins to be registered in the KMS plugin registry before being called.
func BuildStorageFactory(s *options.ServerRunOptions, apiResourceConfig *serverstorage.ResourceConfig) (*serverstorage.DefaultStorageFactory, error) {
storageGroupsToEncodingVersion, err := s.StorageSerialization.StorageGroupsToEncodingVersion()
if err != nil {
return nil, fmt.Errorf("error generating storage version map: %s", err)
}
storageFactory, err := kubeapiserver.NewStorageFactory(
s.Etcd.StorageConfig, s.Etcd.DefaultStorageMediaType, legacyscheme.Codecs,
serverstorage.NewDefaultResourceEncodingConfig(legacyscheme.Scheme), storageGroupsToEncodingVersion,
// The list includes resources that need to be stored in a different
// group version than other resources in the groups.
// FIXME (soltysh): this GroupVersionResource override should be configurable
[]schema.GroupVersionResource{
batch.Resource("cronjobs").WithVersion("v1beta1"),
storage.Resource("volumeattachments").WithVersion("v1beta1"),
admissionregistration.Resource("initializerconfigurations").WithVersion("v1alpha1"),
},
apiResourceConfig)
if err != nil {
return nil, fmt.Errorf("error in initializing storage factory: %s", err)
}
storageFactory.AddCohabitatingResources(networking.Resource("networkpolicies"), extensions.Resource("networkpolicies"))
storageFactory.AddCohabitatingResources(apps.Resource("deployments"), extensions.Resource("deployments"))
storageFactory.AddCohabitatingResources(apps.Resource("daemonsets"), extensions.Resource("daemonsets"))
storageFactory.AddCohabitatingResources(apps.Resource("replicasets"), extensions.Resource("replicasets"))
storageFactory.AddCohabitatingResources(api.Resource("events"), events.Resource("events"))
storageFactory.AddCohabitatingResources(policy.Resource("podsecuritypolicies"), extensions.Resource("podsecuritypolicies"))
for _, override := range s.Etcd.EtcdServersOverrides {
tokens := strings.Split(override, "#")
apiresource := strings.Split(tokens[0], "/")
group := apiresource[0]
resource := apiresource[1]
groupResource := schema.GroupResource{Group: group, Resource: resource}
servers := strings.Split(tokens[1], ";")
storageFactory.SetEtcdLocation(groupResource, servers)
}
if len(s.Etcd.EncryptionProviderConfigFilepath) != 0 {
transformerOverrides, err := encryptionconfig.GetTransformerOverrides(s.Etcd.EncryptionProviderConfigFilepath)
if err != nil {
return nil, err
}
for groupResource, transformer := range transformerOverrides {
storageFactory.SetTransformer(groupResource, transformer)
}
}
return storageFactory, nil
}
// completedServerRunOptions is a private wrapper that enforces a call of Complete() before Run can be invoked.
type completedServerRunOptions struct {
*options.ServerRunOptions

13
pkg/kubeapiserver/BUILD
View File

@ -13,8 +13,21 @@ go_library(
],
importpath = "k8s.io/kubernetes/pkg/kubeapiserver",
deps = [
"//pkg/api/legacyscheme:go_default_library",
"//pkg/apis/admissionregistration:go_default_library",
"//pkg/apis/apps:go_default_library",
"//pkg/apis/batch:go_default_library",
"//pkg/apis/core:go_default_library",
"//pkg/apis/events:go_default_library",
"//pkg/apis/extensions:go_default_library",
"//pkg/apis/networking:go_default_library",
"//pkg/apis/policy:go_default_library",
"//pkg/apis/storage:go_default_library",
"//pkg/kubeapiserver/options:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/options:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/resourceconfig:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/server/storage:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",

110
pkg/kubeapiserver/default_storage_factory_builder.go
View File

@ -17,11 +17,27 @@ limitations under the License.
package kubeapiserver
import (
"strings"
"fmt"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
serveroptions "k8s.io/apiserver/pkg/server/options"
"k8s.io/apiserver/pkg/server/options/encryptionconfig"
"k8s.io/apiserver/pkg/server/resourceconfig"
serverstorage "k8s.io/apiserver/pkg/server/storage"
"k8s.io/apiserver/pkg/storage/storagebackend"
"k8s.io/kubernetes/pkg/api/legacyscheme"
"k8s.io/kubernetes/pkg/apis/admissionregistration"
"k8s.io/kubernetes/pkg/apis/apps"
"k8s.io/kubernetes/pkg/apis/batch"
api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/apis/events"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/apis/networking"
"k8s.io/kubernetes/pkg/apis/policy"
apisstorage "k8s.io/kubernetes/pkg/apis/storage"
kubeapiserveroptions "k8s.io/kubernetes/pkg/kubeapiserver/options"
)
// SpecialDefaultResourcePrefixes are prefixes compiled into Kubernetes.
@ -35,18 +51,84 @@ var SpecialDefaultResourcePrefixes = map[schema.GroupResource]string{
{Group: "policy", Resource: "podsecuritypolicies"}: "podsecuritypolicy",
}
// NewStorageFactory builds the DefaultStorageFactory.
// Merges defaultResourceEncoding with the user specified overrides.
func NewStorageFactory(
storageConfig storagebackend.Config,
defaultMediaType string,
serializer runtime.StorageSerializer,
defaultResourceEncoding *serverstorage.DefaultResourceEncodingConfig,
storageEncodingOverrides map[string]schema.GroupVersion,
resourceEncodingOverrides []schema.GroupVersionResource,
apiResourceConfig *serverstorage.ResourceConfig,
) (*serverstorage.DefaultStorageFactory, error) {
resourceEncodingConfig := resourceconfig.MergeGroupEncodingConfigs(defaultResourceEncoding, storageEncodingOverrides)
resourceEncodingConfig = resourceconfig.MergeResourceEncodingConfigs(resourceEncodingConfig, resourceEncodingOverrides)
return serverstorage.NewDefaultStorageFactory(storageConfig, defaultMediaType, serializer, resourceEncodingConfig, apiResourceConfig, SpecialDefaultResourcePrefixes), nil
func NewStorageFactoryConfig() *StorageFactoryConfig {
return &StorageFactoryConfig{
Serializer: legacyscheme.Codecs,
DefaultResourceEncoding: serverstorage.NewDefaultResourceEncodingConfig(legacyscheme.Scheme),
ResourceEncodingOverrides: []schema.GroupVersionResource{
batch.Resource("cronjobs").WithVersion("v1beta1"),
apisstorage.Resource("volumeattachments").WithVersion("v1beta1"),
admissionregistration.Resource("initializerconfigurations").WithVersion("v1alpha1"),
},
}
}
type StorageFactoryConfig struct {
StorageConfig storagebackend.Config
ApiResourceConfig *serverstorage.ResourceConfig
DefaultResourceEncoding *serverstorage.DefaultResourceEncodingConfig
DefaultStorageMediaType string
Serializer runtime.StorageSerializer
StorageEncodingOverrides map[string]schema.GroupVersion
ResourceEncodingOverrides []schema.GroupVersionResource
EtcdServersOverrides []string
EncryptionProviderConfigFilepath string
}
func (c *StorageFactoryConfig) Complete(etcdOptions *serveroptions.EtcdOptions, serializationOptions *kubeapiserveroptions.StorageSerializationOptions) (*completedStorageFactoryConfig, error) {
storageGroupsToEncodingVersion, err := serializationOptions.StorageGroupsToEncodingVersion()
if err != nil {
return nil, fmt.Errorf("error generating storage version map: %s", err)
}
c.StorageEncodingOverrides = storageGroupsToEncodingVersion
c.StorageConfig = etcdOptions.StorageConfig
c.DefaultStorageMediaType = etcdOptions.DefaultStorageMediaType
c.EtcdServersOverrides = etcdOptions.EtcdServersOverrides
c.EncryptionProviderConfigFilepath = etcdOptions.EncryptionProviderConfigFilepath
return &completedStorageFactoryConfig{c}, nil
}
type completedStorageFactoryConfig struct {
*StorageFactoryConfig
}
func (c *completedStorageFactoryConfig) New() (*serverstorage.DefaultStorageFactory, error) {
resourceEncodingConfig := resourceconfig.MergeGroupEncodingConfigs(c.DefaultResourceEncoding, c.StorageEncodingOverrides)
resourceEncodingConfig = resourceconfig.MergeResourceEncodingConfigs(resourceEncodingConfig, c.ResourceEncodingOverrides)
storageFactory := serverstorage.NewDefaultStorageFactory(
c.StorageConfig,
c.DefaultStorageMediaType,
c.Serializer,
resourceEncodingConfig,
c.ApiResourceConfig,
SpecialDefaultResourcePrefixes)
storageFactory.AddCohabitatingResources(networking.Resource("networkpolicies"), extensions.Resource("networkpolicies"))
storageFactory.AddCohabitatingResources(apps.Resource("deployments"), extensions.Resource("deployments"))
storageFactory.AddCohabitatingResources(apps.Resource("daemonsets"), extensions.Resource("daemonsets"))
storageFactory.AddCohabitatingResources(apps.Resource("replicasets"), extensions.Resource("replicasets"))
storageFactory.AddCohabitatingResources(api.Resource("events"), events.Resource("events"))
storageFactory.AddCohabitatingResources(policy.Resource("podsecuritypolicies"), extensions.Resource("podsecuritypolicies"))
for _, override := range c.EtcdServersOverrides {
tokens := strings.Split(override, "#")
apiresource := strings.Split(tokens[0], "/")
group := apiresource[0]
resource := apiresource[1]
groupResource := schema.GroupResource{Group: group, Resource: resource}
servers := strings.Split(tokens[1], ";")
storageFactory.SetEtcdLocation(groupResource, servers)
}
if len(c.EncryptionProviderConfigFilepath) != 0 {
transformerOverrides, err := encryptionconfig.GetTransformerOverrides(c.EncryptionProviderConfigFilepath)
if err != nil {
return nil, err
}
for groupResource, transformer := range transformerOverrides {
storageFactory.SetTransformer(groupResource, transformer)
}
}
return storageFactory, nil
}

6
pkg/kubeapiserver/options/storage_versions.go
View File

@ -17,14 +17,12 @@ limitations under the License.
package options
import (
"sort"
"strings"
"github.com/spf13/pflag"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/kubernetes/pkg/api/legacyscheme"
"sort"
"github.com/spf13/pflag"
)
const (