github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-09 12:07:47 +00:00
Code Issues Projects Releases Wiki Activity

pkg/controlplane: wire core storage as RESTStorageProvider

Browse Source
This commit is contained in:
Dr. Stefan Schimanski 2023-07-03 12:36:42 +02:00
parent 65b34221eb
commit c733c57962
No known key found for this signature in database
GPG Key ID: 4C68E0F19F95EC33
3 changed files with 72 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
pkg/controlplane/instance.go
View File

@ -395,12 +395,6 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
ClusterAuthenticationInfo: c.ExtraConfig.ClusterAuthenticationInfo,
}
// install legacy rest storage
if err := m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter); err != nil {
return nil, err
}
clientset, err := kubernetes.NewForConfig(c.GenericConfig.LoopbackClientConfig)
if err != nil {
return nil, err
@ -417,6 +411,23 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
// TODO: describe the priority all the way down in the RESTStorageProviders and plumb it back through the various discovery
// handlers that we have.
restStorageProviders := []RESTStorageProvider{
corerest.LegacyRESTStorageProvider{
GenericLegacyRESTStorageProvider: corerest.GenericLegacyRESTStorageProvider{
StorageFactory: c.ExtraConfig.StorageFactory,
EventTTL: c.ExtraConfig.EventTTL,
LoopbackClientConfig: c.GenericConfig.LoopbackClientConfig,
ServiceAccountIssuer: c.ExtraConfig.ServiceAccountIssuer,
ExtendExpiration: c.ExtraConfig.ExtendExpiration,
ServiceAccountMaxExpiration: c.ExtraConfig.ServiceAccountMaxExpiration,
APIAudiences: c.GenericConfig.Authentication.APIAudiences,
Informers: c.ExtraConfig.VersionedInformers,
},
ProxyTransport: c.ExtraConfig.ProxyTransport,
KubeletClientConfig: c.ExtraConfig.KubeletClientConfig,
ServiceIPRange: c.ExtraConfig.ServiceIPRange,
SecondaryServiceIPRange: c.ExtraConfig.SecondaryServiceIPRange,
ServiceNodePortRange: c.ExtraConfig.ServiceNodePortRange,
},
apiserverinternalrest.StorageProvider{},
authenticationrest.RESTStorageProvider{Authenticator: c.GenericConfig.Authentication.Authenticator, APIAudiences: c.GenericConfig.Authentication.APIAudiences},
authorizationrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorization.Authorizer, RuleResolver: c.GenericConfig.RuleResolver},
@ -443,12 +454,13 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
return nil, err
}
m.GenericAPIServer.AddPostStartHookOrDie("start-system-namespaces-controller", func(hookContext genericapiserver.PostStartHookContext) error {
go systemnamespaces.NewController(clientset, c.ExtraConfig.VersionedInformers.Core().V1().Namespaces()).Run(hookContext.StopCh)
return nil
})
m.GenericAPIServer.AddPostStartHookOrDie("start-cluster-authentication-info-controller", func(hookContext genericapiserver.PostStartHookContext) error {
kubeClient, err := kubernetes.NewForConfig(hookContext.LoopbackClientConfig)
if err != nil {
return err
}
controller := clusterauthenticationtrust.NewClusterAuthenticationTrustController(m.ClusterAuthenticationInfo, kubeClient)
controller := clusterauthenticationtrust.NewClusterAuthenticationTrustController(m.ClusterAuthenticationInfo, clientset)
// generate a context from stopCh. This is to avoid modifying files which are relying on apiserver
// TODO: See if we can pass ctx to the current method
@ -572,59 +584,6 @@ func labelAPIServerHeartbeatFunc(identity string) lease.ProcessLeaseFunc {
}
}
// InstallLegacyAPI will install the legacy APIs for the restStorageProviders if they are enabled.
func (m *Instance) InstallLegacyAPI(c *completedConfig, restOptionsGetter generic.RESTOptionsGetter) error {
legacyRESTStorageProvider := corerest.LegacyRESTStorageProvider{
GenericLegacyRESTStorageProvider: corerest.GenericLegacyRESTStorageProvider{
StorageFactory: c.ExtraConfig.StorageFactory,
EventTTL: c.ExtraConfig.EventTTL,
LoopbackClientConfig: c.GenericConfig.LoopbackClientConfig,
ServiceAccountIssuer: c.ExtraConfig.ServiceAccountIssuer,
ExtendExpiration: c.ExtraConfig.ExtendExpiration,
ServiceAccountMaxExpiration: c.ExtraConfig.ServiceAccountMaxExpiration,
APIAudiences: c.GenericConfig.Authentication.APIAudiences,
Informers: c.ExtraConfig.VersionedInformers,
},
ProxyTransport: c.ExtraConfig.ProxyTransport,
KubeletClientConfig: c.ExtraConfig.KubeletClientConfig,
ServiceIPRange: c.ExtraConfig.ServiceIPRange,
SecondaryServiceIPRange: c.ExtraConfig.SecondaryServiceIPRange,
ServiceNodePortRange: c.ExtraConfig.ServiceNodePortRange,
}
rangeRegistries, apiGroupInfo, err := legacyRESTStorageProvider.NewLegacyRESTStorage(c.ExtraConfig.APIResourceConfigSource, restOptionsGetter)
if err != nil {
return fmt.Errorf("error building core storage: %v", err)
}
if len(apiGroupInfo.VersionedResourcesStorageMap) == 0 { // if all core storage is disabled, return.
return nil
}
controllerName := "bootstrap-controller"
client := kubernetes.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig)
// Kubernetes clusters contains the following system namespaces:
// kube-system, kube-node-lease, kube-public, default
m.GenericAPIServer.AddPostStartHookOrDie("start-system-namespaces-controller", func(hookContext genericapiserver.PostStartHookContext) error {
go systemnamespaces.NewController(client, c.ExtraConfig.VersionedInformers.Core().V1().Namespaces()).Run(hookContext.StopCh)
return nil
})
kubenetesserviceConfig, err := c.newKubernetesServiceControllerConfig(client)
if err != nil {
return err
}
bootstrapController, err := kubernetesservice.New(*kubenetesserviceConfig, rangeRegistries)
if err != nil {
return fmt.Errorf("error creating bootstrap controller: %v", err)
}
m.GenericAPIServer.AddPostStartHookOrDie(controllerName, func(genericapiserver.PostStartHookContext) error { bootstrapController.Start(); return nil })
m.GenericAPIServer.AddPreShutdownHookOrDie(controllerName, func() error { bootstrapController.Stop(); return nil })
if err := m.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {
return fmt.Errorf("error in registering group versions: %v", err)
}
return nil
}
// newKubernetesServiceControllerConfig returns a configuration for the kubernetes service controller.
func (c completedConfig) newKubernetesServiceControllerConfig(client kubernetes.Interface) (*kubernetesservice.Config, error) {
_, publicServicePort, err := c.GenericConfig.SecureServing.HostPort()
@ -667,7 +626,7 @@ type RESTStorageProvider interface {
// InstallAPIs will install the APIs for the restStorageProviders if they are enabled.
func (m *Instance) InstallAPIs(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter, restStorageProviders ...RESTStorageProvider) error {
apiGroupsInfo := []*genericapiserver.APIGroupInfo{}
nonLegacy := []*genericapiserver.APIGroupInfo{}
// used later in the loop to filter the served resource by those that have expired.
resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluator(*m.GenericAPIServer.Version)
@ -707,10 +666,16 @@ func (m *Instance) InstallAPIs(apiResourceConfigSource serverstorage.APIResource
m.GenericAPIServer.AddPostStartHookOrDie(name, hook)
}
apiGroupsInfo = append(apiGroupsInfo, &apiGroupInfo)
if len(groupName) == 0 {
if err := m.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {
return fmt.Errorf("error in registering legacy API: %w", err)
}
} else {
nonLegacy = append(nonLegacy, &apiGroupInfo)
}
}
if err := m.GenericAPIServer.InstallAPIGroups(apiGroupsInfo...); err != nil {
if err := m.GenericAPIServer.InstallAPIGroups(nonLegacy...); err != nil {
return fmt.Errorf("error in registering group versions: %v", err)
}
return nil

2
pkg/controlplane/instance_test.go
View File

@ -165,7 +165,7 @@ func TestLegacyRestStorageStrategies(t *testing.T) {
ServiceNodePortRange: apiserverCfg.ExtraConfig.ServiceNodePortRange,
}
_, apiGroupInfo, err := storageProvider.NewLegacyRESTStorage(serverstorage.NewResourceConfig(), apiserverCfg.GenericConfig.RESTOptionsGetter)
apiGroupInfo, err := storageProvider.NewRESTStorage(serverstorage.NewResourceConfig(), apiserverCfg.GenericConfig.RESTOptionsGetter)
if err != nil {
t.Errorf("failed to create legacy REST storage: %v", err)
}

71
pkg/registry/core/rest/storage_core.go
View File

@ -36,10 +36,10 @@ import (
networkingv1alpha1client "k8s.io/client-go/kubernetes/typed/networking/v1alpha1"
policyclient "k8s.io/client-go/kubernetes/typed/policy/v1"
restclient "k8s.io/client-go/rest"
"k8s.io/kubernetes/pkg/api/legacyscheme"
api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/cluster/ports"
"k8s.io/kubernetes/pkg/controlplane/controller/kubernetesservice"
"k8s.io/kubernetes/pkg/features"
kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
"k8s.io/kubernetes/pkg/registry/core/componentstatus"
@ -53,6 +53,7 @@ import (
pvcstore "k8s.io/kubernetes/pkg/registry/core/persistentvolumeclaim/storage"
podstore "k8s.io/kubernetes/pkg/registry/core/pod/storage"
podtemplatestore "k8s.io/kubernetes/pkg/registry/core/podtemplate/storage"
"k8s.io/kubernetes/pkg/registry/core/rangeallocation"
controllerstore "k8s.io/kubernetes/pkg/registry/core/replicationcontroller/storage"
resourcequotastore "k8s.io/kubernetes/pkg/registry/core/resourcequota/storage"
secretstore "k8s.io/kubernetes/pkg/registry/core/secret/storage"
@ -67,8 +68,7 @@ import (
)
// GenericLegacyRESTStorageProvider provides information needed to build RESTStorage
// for generic resources in core, but does NOT implement the "normal"
// RESTStorageProvider (yet!)
// for generic resources in core. It implements the "normal" RESTStorageProvider interface.
type GenericLegacyRESTStorageProvider struct {
StorageFactory serverstorage.StorageFactory
EventTTL time.Duration
@ -100,6 +100,12 @@ type LegacyRESTStorageProvider struct {
ServiceNodePortRange utilnet.PortRange
}
type rangeRegistries struct {
clusterIP rangeallocation.RangeRegistry
secondaryClusterIP rangeallocation.RangeRegistry
nodePort rangeallocation.RangeRegistry
}
func (c GenericLegacyRESTStorageProvider) NewRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (genericapiserver.APIGroupInfo, error) {
apiGroupInfo := genericapiserver.APIGroupInfo{
PrioritizedVersions: legacyscheme.Scheme.PrioritizedVersionsForGroup(""),
@ -180,44 +186,45 @@ func (c GenericLegacyRESTStorageProvider) NewRESTStorage(apiResourceConfigSource
return apiGroupInfo, nil
}
func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (kubernetesservice.RangeRegistries, genericapiserver.APIGroupInfo, error) {
func (c LegacyRESTStorageProvider) NewRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (genericapiserver.APIGroupInfo, error) {
apiGroupInfo, err := c.GenericLegacyRESTStorageProvider.NewRESTStorage(apiResourceConfigSource, restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
podDisruptionClient, err := policyclient.NewForConfig(c.LoopbackClientConfig)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
podTemplateStorage, err := podtemplatestore.NewREST(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
limitRangeStorage, err := limitrangestore.NewREST(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
persistentVolumeStorage, persistentVolumeStatusStorage, err := pvstore.NewREST(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
persistentVolumeClaimStorage, persistentVolumeClaimStatusStorage, err := pvcstore.NewREST(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
endpointsStorage, err := endpointsstore.NewREST(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
nodeStorage, err := nodestore.NewStorage(restOptionsGetter, c.KubeletClientConfig, c.ProxyTransport)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
podStorage, err := podstore.NewStorage(
@ -227,12 +234,12 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource
podDisruptionClient,
)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
rangeRegistries, primaryServiceClusterIPAllocator, serviceClusterIPAllocators, serviceNodePortAllocator, err := c.newServiceIPAllocators()
_, primaryServiceClusterIPAllocator, serviceClusterIPAllocators, serviceNodePortAllocator, err := c.newServiceIPAllocators()
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
serviceRESTStorage, serviceStatusStorage, serviceRESTProxy, err := servicestore.NewREST(
restOptionsGetter,
@ -243,7 +250,7 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource
podStorage.Pod,
c.ProxyTransport)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
storage := apiGroupInfo.VersionedResourcesStorageMap["v1"]
@ -256,7 +263,7 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource
if c.ServiceAccountIssuer != nil {
serviceAccountStorage, err = serviceaccountstore.NewREST(restOptionsGetter, c.ServiceAccountIssuer, c.APIAudiences, c.ServiceAccountMaxExpiration, podStorage.Pod.Store, storage["secrets"].(rest.Getter), c.ExtendExpiration)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
}
@ -285,7 +292,7 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource
if resource := "replicationcontrollers"; apiResourceConfigSource.ResourceEnabled(corev1.SchemeGroupVersion.WithResource(resource)) {
controllerStorage, err := controllerstore.NewStorage(restOptionsGetter)
if err != nil {
return kubernetesservice.RangeRegistries{}, genericapiserver.APIGroupInfo{}, err
return genericapiserver.APIGroupInfo{}, err
}
storage[resource] = controllerStorage.Controller
@ -347,20 +354,20 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(apiResourceConfigSource
apiGroupInfo.VersionedResourcesStorageMap["v1"] = storage
}
return rangeRegistries, apiGroupInfo, nil
return apiGroupInfo, nil
}
func (c LegacyRESTStorageProvider) newServiceIPAllocators() (registries kubernetesservice.RangeRegistries, primaryClusterIPAllocator ipallocator.Interface, clusterIPAllocators map[api.IPFamily]ipallocator.Interface, nodePortAllocator *portallocator.PortAllocator, err error) {
func (c LegacyRESTStorageProvider) newServiceIPAllocators() (registries rangeRegistries, primaryClusterIPAllocator ipallocator.Interface, clusterIPAllocators map[api.IPFamily]ipallocator.Interface, nodePortAllocator *portallocator.PortAllocator, err error) {
clusterIPAllocators = map[api.IPFamily]ipallocator.Interface{}
serviceStorageConfig, err := c.StorageFactory.NewConfig(api.Resource("services"))
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, err
return rangeRegistries{}, nil, nil, nil, err
}
serviceClusterIPRange := c.ServiceIPRange
if serviceClusterIPRange.IP == nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("service clusterIPRange is missing")
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("service clusterIPRange is missing")
}
if !utilfeature.DefaultFeatureGate.Enabled(features.MultiCIDRServiceAllocator) {
@ -372,20 +379,20 @@ func (c LegacyRESTStorageProvider) newServiceIPAllocators() (registries kubernet
if err != nil {
return nil, err
}
registries.ServiceClusterIPRegistry = etcd
registries.clusterIP = etcd
return etcd, nil
})
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster IP allocator: %v", err)
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster IP allocator: %v", err)
}
} else {
networkingv1alphaClient, err := networkingv1alpha1client.NewForConfig(c.LoopbackClientConfig)
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, err
return rangeRegistries{}, nil, nil, nil, err
}
primaryClusterIPAllocator, err = ipallocator.NewIPAllocator(&serviceClusterIPRange, networkingv1alphaClient, c.Informers.Networking().V1alpha1().IPAddresses())
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster IP allocator: %v", err)
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster IP allocator: %v", err)
}
}
primaryClusterIPAllocator.EnableMetrics()
@ -403,20 +410,20 @@ func (c LegacyRESTStorageProvider) newServiceIPAllocators() (registries kubernet
if err != nil {
return nil, err
}
registries.SecondaryServiceClusterIPRegistry = etcd
registries.secondaryClusterIP = etcd
return etcd, nil
})
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster secondary IP allocator: %v", err)
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster secondary IP allocator: %v", err)
}
} else {
networkingv1alphaClient, err := networkingv1alpha1client.NewForConfig(c.LoopbackClientConfig)
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, err
return rangeRegistries{}, nil, nil, nil, err
}
secondaryClusterIPAllocator, err = ipallocator.NewIPAllocator(&c.SecondaryServiceIPRange, networkingv1alphaClient, c.Informers.Networking().V1alpha1().IPAddresses())
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster secondary IP allocator: %v", err)
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster secondary IP allocator: %v", err)
}
}
secondaryClusterIPAllocator.EnableMetrics()
@ -430,11 +437,11 @@ func (c LegacyRESTStorageProvider) newServiceIPAllocators() (registries kubernet
if err != nil {
return nil, err
}
registries.ServiceNodePortRegistry = etcd
registries.nodePort = etcd
return etcd, nil
})
if err != nil {
return kubernetesservice.RangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster port allocator: %v", err)
return rangeRegistries{}, nil, nil, nil, fmt.Errorf("cannot create cluster port allocator: %v", err)
}
nodePortAllocator.EnableMetrics()