
This change updates the API server code to load the encryption config once at start up instead of multiple times. Previously the code would set up the storage transformers and the etcd healthz checks in separate parse steps. This is problematic for KMS v2 key ID based staleness checks which need to be able to assert that the API server has a single view into the KMS plugin's current key ID. Signed-off-by: Monis Khan <mok@microsoft.com>
715 lines
29 KiB
Go
715 lines
29 KiB
Go
/*
|
|
Copyright 2014 The Kubernetes Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
// Package app does all of the work necessary to create a Kubernetes
|
|
// APIServer by binding together the API, master and APIServer infrastructure.
|
|
// It can be configured and called directly or via the hyperkube framework.
|
|
package app
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"fmt"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
|
|
oteltrace "go.opentelemetry.io/otel/trace"
|
|
|
|
extensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
|
|
utilerrors "k8s.io/apimachinery/pkg/util/errors"
|
|
utilnet "k8s.io/apimachinery/pkg/util/net"
|
|
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
|
"k8s.io/apimachinery/pkg/util/sets"
|
|
"k8s.io/apiserver/pkg/admission"
|
|
"k8s.io/apiserver/pkg/authorization/authorizer"
|
|
genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
|
|
openapinamer "k8s.io/apiserver/pkg/endpoints/openapi"
|
|
genericfeatures "k8s.io/apiserver/pkg/features"
|
|
genericapiserver "k8s.io/apiserver/pkg/server"
|
|
"k8s.io/apiserver/pkg/server/egressselector"
|
|
"k8s.io/apiserver/pkg/server/filters"
|
|
serveroptions "k8s.io/apiserver/pkg/server/options"
|
|
serverstorage "k8s.io/apiserver/pkg/server/storage"
|
|
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
|
utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
|
|
"k8s.io/apiserver/pkg/util/notfoundhandler"
|
|
"k8s.io/apiserver/pkg/util/openapi"
|
|
"k8s.io/apiserver/pkg/util/webhook"
|
|
clientgoinformers "k8s.io/client-go/informers"
|
|
clientgoclientset "k8s.io/client-go/kubernetes"
|
|
"k8s.io/client-go/rest"
|
|
"k8s.io/client-go/util/keyutil"
|
|
cliflag "k8s.io/component-base/cli/flag"
|
|
"k8s.io/component-base/cli/globalflag"
|
|
"k8s.io/component-base/logs"
|
|
logsapi "k8s.io/component-base/logs/api/v1"
|
|
_ "k8s.io/component-base/metrics/prometheus/workqueue" // for workqueue metric registration
|
|
"k8s.io/component-base/term"
|
|
"k8s.io/component-base/version"
|
|
"k8s.io/component-base/version/verflag"
|
|
"k8s.io/klog/v2"
|
|
aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
|
|
aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme"
|
|
netutils "k8s.io/utils/net"
|
|
|
|
"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
|
|
"k8s.io/kubernetes/pkg/api/legacyscheme"
|
|
"k8s.io/kubernetes/pkg/capabilities"
|
|
"k8s.io/kubernetes/pkg/controlplane"
|
|
"k8s.io/kubernetes/pkg/controlplane/reconcilers"
|
|
generatedopenapi "k8s.io/kubernetes/pkg/generated/openapi"
|
|
"k8s.io/kubernetes/pkg/kubeapiserver"
|
|
kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
|
|
kubeauthenticator "k8s.io/kubernetes/pkg/kubeapiserver/authenticator"
|
|
"k8s.io/kubernetes/pkg/kubeapiserver/authorizer/modes"
|
|
rbacrest "k8s.io/kubernetes/pkg/registry/rbac/rest"
|
|
"k8s.io/kubernetes/pkg/serviceaccount"
|
|
)
|
|
|
|
func init() {
|
|
utilruntime.Must(logsapi.AddFeatureGates(utilfeature.DefaultMutableFeatureGate))
|
|
}
|
|
|
|
// NewAPIServerCommand creates a *cobra.Command object with default parameters
|
|
func NewAPIServerCommand() *cobra.Command {
|
|
s := options.NewServerRunOptions()
|
|
cmd := &cobra.Command{
|
|
Use: "kube-apiserver",
|
|
Long: `The Kubernetes API server validates and configures data
|
|
for the api objects which include pods, services, replicationcontrollers, and
|
|
others. The API Server services REST operations and provides the frontend to the
|
|
cluster's shared state through which all other components interact.`,
|
|
|
|
// stop printing usage when the command errors
|
|
SilenceUsage: true,
|
|
PersistentPreRunE: func(*cobra.Command, []string) error {
|
|
// silence client-go warnings.
|
|
// kube-apiserver loopback clients should not log self-issued warnings.
|
|
rest.SetDefaultWarningHandler(rest.NoWarnings{})
|
|
return nil
|
|
},
|
|
RunE: func(cmd *cobra.Command, args []string) error {
|
|
verflag.PrintAndExitIfRequested()
|
|
fs := cmd.Flags()
|
|
|
|
// Activate logging as soon as possible, after that
|
|
// show flags with the final logging configuration.
|
|
if err := logsapi.ValidateAndApply(s.Logs, utilfeature.DefaultFeatureGate); err != nil {
|
|
return err
|
|
}
|
|
cliflag.PrintFlags(fs)
|
|
|
|
// set default options
|
|
completedOptions, err := Complete(s)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// validate options
|
|
if errs := completedOptions.Validate(); len(errs) != 0 {
|
|
return utilerrors.NewAggregate(errs)
|
|
}
|
|
// add feature enablement metrics
|
|
utilfeature.DefaultMutableFeatureGate.AddMetrics()
|
|
return Run(completedOptions, genericapiserver.SetupSignalHandler())
|
|
},
|
|
Args: func(cmd *cobra.Command, args []string) error {
|
|
for _, arg := range args {
|
|
if len(arg) > 0 {
|
|
return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), args)
|
|
}
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
|
|
fs := cmd.Flags()
|
|
namedFlagSets := s.Flags()
|
|
verflag.AddFlags(namedFlagSets.FlagSet("global"))
|
|
globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
|
|
options.AddCustomGlobalFlags(namedFlagSets.FlagSet("generic"))
|
|
for _, f := range namedFlagSets.FlagSets {
|
|
fs.AddFlagSet(f)
|
|
}
|
|
|
|
cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
|
|
cliflag.SetUsageAndHelpFunc(cmd, namedFlagSets, cols)
|
|
|
|
return cmd
|
|
}
|
|
|
|
// Run runs the specified APIServer. This should never exit.
|
|
func Run(completeOptions completedServerRunOptions, stopCh <-chan struct{}) error {
|
|
// To help debugging, immediately log version
|
|
klog.Infof("Version: %+v", version.Get())
|
|
|
|
klog.InfoS("Golang settings", "GOGC", os.Getenv("GOGC"), "GOMAXPROCS", os.Getenv("GOMAXPROCS"), "GOTRACEBACK", os.Getenv("GOTRACEBACK"))
|
|
|
|
server, err := CreateServerChain(completeOptions)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
prepared, err := server.PrepareRun()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return prepared.Run(stopCh)
|
|
}
|
|
|
|
// CreateServerChain creates the apiservers connected via delegation.
|
|
func CreateServerChain(completedOptions completedServerRunOptions) (*aggregatorapiserver.APIAggregator, error) {
|
|
kubeAPIServerConfig, serviceResolver, pluginInitializer, err := CreateKubeAPIServerConfig(completedOptions)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// If additional API servers are added, they should be gated.
|
|
apiExtensionsConfig, err := createAPIExtensionsConfig(*kubeAPIServerConfig.GenericConfig, kubeAPIServerConfig.ExtraConfig.VersionedInformers, pluginInitializer, completedOptions.ServerRunOptions, completedOptions.MasterCount,
|
|
serviceResolver, webhook.NewDefaultAuthenticationInfoResolverWrapper(kubeAPIServerConfig.ExtraConfig.ProxyTransport, kubeAPIServerConfig.GenericConfig.EgressSelector, kubeAPIServerConfig.GenericConfig.LoopbackClientConfig, kubeAPIServerConfig.GenericConfig.TracerProvider))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
notFoundHandler := notfoundhandler.New(kubeAPIServerConfig.GenericConfig.Serializer, genericapifilters.NoMuxAndDiscoveryIncompleteKey)
|
|
apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegateWithCustomHandler(notFoundHandler))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// aggregator comes last in the chain
|
|
aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, completedOptions.ServerRunOptions, kubeAPIServerConfig.ExtraConfig.VersionedInformers, serviceResolver, kubeAPIServerConfig.ExtraConfig.ProxyTransport, pluginInitializer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
|
|
if err != nil {
|
|
// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
|
|
return nil, err
|
|
}
|
|
|
|
return aggregatorServer, nil
|
|
}
|
|
|
|
// CreateKubeAPIServer creates and wires a workable kube-apiserver
|
|
func CreateKubeAPIServer(kubeAPIServerConfig *controlplane.Config, delegateAPIServer genericapiserver.DelegationTarget) (*controlplane.Instance, error) {
|
|
kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return kubeAPIServer, nil
|
|
}
|
|
|
|
// CreateProxyTransport creates the dialer infrastructure to connect to the nodes.
|
|
func CreateProxyTransport() *http.Transport {
|
|
var proxyDialerFn utilnet.DialFunc
|
|
// Proxying to pods and services is IP-based... don't expect to be able to verify the hostname
|
|
proxyTLSClientConfig := &tls.Config{InsecureSkipVerify: true}
|
|
proxyTransport := utilnet.SetTransportDefaults(&http.Transport{
|
|
DialContext: proxyDialerFn,
|
|
TLSClientConfig: proxyTLSClientConfig,
|
|
})
|
|
return proxyTransport
|
|
}
|
|
|
|
// CreateKubeAPIServerConfig creates all the resources for running the API server, but runs none of them
|
|
func CreateKubeAPIServerConfig(s completedServerRunOptions) (
|
|
*controlplane.Config,
|
|
aggregatorapiserver.ServiceResolver,
|
|
[]admission.PluginInitializer,
|
|
error,
|
|
) {
|
|
proxyTransport := CreateProxyTransport()
|
|
|
|
genericConfig, versionedInformers, serviceResolver, pluginInitializers, admissionPostStartHook, storageFactory, err := buildGenericConfig(s.ServerRunOptions, proxyTransport)
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
|
|
capabilities.Setup(s.AllowPrivileged, s.MaxConnectionBytesPerSec)
|
|
|
|
s.Metrics.Apply()
|
|
serviceaccount.RegisterMetrics()
|
|
|
|
config := &controlplane.Config{
|
|
GenericConfig: genericConfig,
|
|
ExtraConfig: controlplane.ExtraConfig{
|
|
APIResourceConfigSource: storageFactory.APIResourceConfigSource,
|
|
StorageFactory: storageFactory,
|
|
EventTTL: s.EventTTL,
|
|
KubeletClientConfig: s.KubeletConfig,
|
|
EnableLogsSupport: s.EnableLogsHandler,
|
|
ProxyTransport: proxyTransport,
|
|
|
|
ServiceIPRange: s.PrimaryServiceClusterIPRange,
|
|
APIServerServiceIP: s.APIServerServiceIP,
|
|
SecondaryServiceIPRange: s.SecondaryServiceClusterIPRange,
|
|
|
|
APIServerServicePort: 443,
|
|
|
|
ServiceNodePortRange: s.ServiceNodePortRange,
|
|
KubernetesServiceNodePort: s.KubernetesServiceNodePort,
|
|
|
|
EndpointReconcilerType: reconcilers.Type(s.EndpointReconcilerType),
|
|
MasterCount: s.MasterCount,
|
|
|
|
ServiceAccountIssuer: s.ServiceAccountIssuer,
|
|
ServiceAccountMaxExpiration: s.ServiceAccountTokenMaxExpiration,
|
|
ExtendExpiration: s.Authentication.ServiceAccounts.ExtendExpiration,
|
|
|
|
VersionedInformers: versionedInformers,
|
|
|
|
IdentityLeaseDurationSeconds: s.IdentityLeaseDurationSeconds,
|
|
IdentityLeaseRenewIntervalSeconds: s.IdentityLeaseRenewIntervalSeconds,
|
|
},
|
|
}
|
|
|
|
clientCAProvider, err := s.Authentication.ClientCert.GetClientCAContentProvider()
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
config.ExtraConfig.ClusterAuthenticationInfo.ClientCA = clientCAProvider
|
|
|
|
requestHeaderConfig, err := s.Authentication.RequestHeader.ToAuthenticationRequestHeaderConfig()
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
if requestHeaderConfig != nil {
|
|
config.ExtraConfig.ClusterAuthenticationInfo.RequestHeaderCA = requestHeaderConfig.CAContentProvider
|
|
config.ExtraConfig.ClusterAuthenticationInfo.RequestHeaderAllowedNames = requestHeaderConfig.AllowedClientNames
|
|
config.ExtraConfig.ClusterAuthenticationInfo.RequestHeaderExtraHeaderPrefixes = requestHeaderConfig.ExtraHeaderPrefixes
|
|
config.ExtraConfig.ClusterAuthenticationInfo.RequestHeaderGroupHeaders = requestHeaderConfig.GroupHeaders
|
|
config.ExtraConfig.ClusterAuthenticationInfo.RequestHeaderUsernameHeaders = requestHeaderConfig.UsernameHeaders
|
|
}
|
|
|
|
if err := config.GenericConfig.AddPostStartHook("start-kube-apiserver-admission-initializer", admissionPostStartHook); err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
|
|
if config.GenericConfig.EgressSelector != nil {
|
|
// Use the config.GenericConfig.EgressSelector lookup to find the dialer to connect to the kubelet
|
|
config.ExtraConfig.KubeletClientConfig.Lookup = config.GenericConfig.EgressSelector.Lookup
|
|
|
|
// Use the config.GenericConfig.EgressSelector lookup as the transport used by the "proxy" subresources.
|
|
networkContext := egressselector.Cluster.AsNetworkContext()
|
|
dialer, err := config.GenericConfig.EgressSelector.Lookup(networkContext)
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
c := proxyTransport.Clone()
|
|
c.DialContext = dialer
|
|
config.ExtraConfig.ProxyTransport = c
|
|
}
|
|
|
|
// Load the public keys.
|
|
var pubKeys []interface{}
|
|
for _, f := range s.Authentication.ServiceAccounts.KeyFiles {
|
|
keys, err := keyutil.PublicKeysFromFile(f)
|
|
if err != nil {
|
|
return nil, nil, nil, fmt.Errorf("failed to parse key file %q: %v", f, err)
|
|
}
|
|
pubKeys = append(pubKeys, keys...)
|
|
}
|
|
// Plumb the required metadata through ExtraConfig.
|
|
config.ExtraConfig.ServiceAccountIssuerURL = s.Authentication.ServiceAccounts.Issuers[0]
|
|
config.ExtraConfig.ServiceAccountJWKSURI = s.Authentication.ServiceAccounts.JWKSURI
|
|
config.ExtraConfig.ServiceAccountPublicKeys = pubKeys
|
|
|
|
return config, serviceResolver, pluginInitializers, nil
|
|
}
|
|
|
|
// BuildGenericConfig takes the master server options and produces the genericapiserver.Config associated with it
|
|
func buildGenericConfig(
|
|
s *options.ServerRunOptions,
|
|
proxyTransport *http.Transport,
|
|
) (
|
|
genericConfig *genericapiserver.Config,
|
|
versionedInformers clientgoinformers.SharedInformerFactory,
|
|
serviceResolver aggregatorapiserver.ServiceResolver,
|
|
pluginInitializers []admission.PluginInitializer,
|
|
admissionPostStartHook genericapiserver.PostStartHookFunc,
|
|
storageFactory *serverstorage.DefaultStorageFactory,
|
|
lastErr error,
|
|
) {
|
|
genericConfig = genericapiserver.NewConfig(legacyscheme.Codecs)
|
|
genericConfig.MergedResourceConfig = controlplane.DefaultAPIResourceConfigSource()
|
|
|
|
if lastErr = s.GenericServerRunOptions.ApplyTo(genericConfig); lastErr != nil {
|
|
return
|
|
}
|
|
|
|
if lastErr = s.SecureServing.ApplyTo(&genericConfig.SecureServing, &genericConfig.LoopbackClientConfig); lastErr != nil {
|
|
return
|
|
}
|
|
if lastErr = s.Features.ApplyTo(genericConfig); lastErr != nil {
|
|
return
|
|
}
|
|
if lastErr = s.APIEnablement.ApplyTo(genericConfig, controlplane.DefaultAPIResourceConfigSource(), legacyscheme.Scheme); lastErr != nil {
|
|
return
|
|
}
|
|
if lastErr = s.EgressSelector.ApplyTo(genericConfig); lastErr != nil {
|
|
return
|
|
}
|
|
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerTracing) {
|
|
if lastErr = s.Traces.ApplyTo(genericConfig.EgressSelector, genericConfig); lastErr != nil {
|
|
return
|
|
}
|
|
}
|
|
// wrap the definitions to revert any changes from disabled features
|
|
getOpenAPIDefinitions := openapi.GetOpenAPIDefinitionsWithoutDisabledFeatures(generatedopenapi.GetOpenAPIDefinitions)
|
|
genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(getOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))
|
|
genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
|
|
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.OpenAPIV3) {
|
|
genericConfig.OpenAPIV3Config = genericapiserver.DefaultOpenAPIV3Config(getOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))
|
|
genericConfig.OpenAPIV3Config.Info.Title = "Kubernetes"
|
|
}
|
|
|
|
genericConfig.LongRunningFunc = filters.BasicLongRunningRequestCheck(
|
|
sets.NewString("watch", "proxy"),
|
|
sets.NewString("attach", "exec", "proxy", "log", "portforward"),
|
|
)
|
|
|
|
kubeVersion := version.Get()
|
|
genericConfig.Version = &kubeVersion
|
|
|
|
if genericConfig.EgressSelector != nil {
|
|
s.Etcd.StorageConfig.Transport.EgressLookup = genericConfig.EgressSelector.Lookup
|
|
}
|
|
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerTracing) {
|
|
s.Etcd.StorageConfig.Transport.TracerProvider = genericConfig.TracerProvider
|
|
} else {
|
|
s.Etcd.StorageConfig.Transport.TracerProvider = oteltrace.NewNoopTracerProvider()
|
|
}
|
|
if lastErr = s.Etcd.Complete(genericConfig.StorageObjectCountTracker, genericConfig.DrainedNotify()); lastErr != nil {
|
|
return
|
|
}
|
|
|
|
storageFactoryConfig := kubeapiserver.NewStorageFactoryConfig()
|
|
storageFactoryConfig.APIResourceConfig = genericConfig.MergedResourceConfig
|
|
storageFactory, lastErr = storageFactoryConfig.Complete(s.Etcd).New()
|
|
if lastErr != nil {
|
|
return
|
|
}
|
|
if lastErr = s.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig); lastErr != nil {
|
|
return
|
|
}
|
|
|
|
// Use protobufs for self-communication.
|
|
// Since not every generic apiserver has to support protobufs, we
|
|
// cannot default to it in generic apiserver and need to explicitly
|
|
// set it in kube-apiserver.
|
|
genericConfig.LoopbackClientConfig.ContentConfig.ContentType = "application/vnd.kubernetes.protobuf"
|
|
// Disable compression for self-communication, since we are going to be
|
|
// on a fast local network
|
|
genericConfig.LoopbackClientConfig.DisableCompression = true
|
|
|
|
kubeClientConfig := genericConfig.LoopbackClientConfig
|
|
clientgoExternalClient, err := clientgoclientset.NewForConfig(kubeClientConfig)
|
|
if err != nil {
|
|
lastErr = fmt.Errorf("failed to create real external clientset: %v", err)
|
|
return
|
|
}
|
|
versionedInformers = clientgoinformers.NewSharedInformerFactory(clientgoExternalClient, 10*time.Minute)
|
|
|
|
// Authentication.ApplyTo requires already applied OpenAPIConfig and EgressSelector if present
|
|
if lastErr = s.Authentication.ApplyTo(&genericConfig.Authentication, genericConfig.SecureServing, genericConfig.EgressSelector, genericConfig.OpenAPIConfig, genericConfig.OpenAPIV3Config, clientgoExternalClient, versionedInformers); lastErr != nil {
|
|
return
|
|
}
|
|
|
|
genericConfig.Authorization.Authorizer, genericConfig.RuleResolver, err = BuildAuthorizer(s, genericConfig.EgressSelector, versionedInformers)
|
|
if err != nil {
|
|
lastErr = fmt.Errorf("invalid authorization config: %v", err)
|
|
return
|
|
}
|
|
if !sets.NewString(s.Authorization.Modes...).Has(modes.ModeRBAC) {
|
|
genericConfig.DisabledPostStartHooks.Insert(rbacrest.PostStartHookName)
|
|
}
|
|
|
|
lastErr = s.Audit.ApplyTo(genericConfig)
|
|
if lastErr != nil {
|
|
return
|
|
}
|
|
|
|
admissionConfig := &kubeapiserveradmission.Config{
|
|
ExternalInformers: versionedInformers,
|
|
LoopbackClientConfig: genericConfig.LoopbackClientConfig,
|
|
CloudConfigFile: s.CloudProvider.CloudConfigFile,
|
|
}
|
|
serviceResolver = buildServiceResolver(s.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers)
|
|
pluginInitializers, admissionPostStartHook, err = admissionConfig.New(proxyTransport, genericConfig.EgressSelector, serviceResolver, genericConfig.TracerProvider)
|
|
if err != nil {
|
|
lastErr = fmt.Errorf("failed to create admission plugin initializer: %v", err)
|
|
return
|
|
}
|
|
|
|
err = s.Admission.ApplyTo(
|
|
genericConfig,
|
|
versionedInformers,
|
|
kubeClientConfig,
|
|
utilfeature.DefaultFeatureGate,
|
|
pluginInitializers...)
|
|
if err != nil {
|
|
lastErr = fmt.Errorf("failed to initialize admission: %v", err)
|
|
return
|
|
}
|
|
|
|
if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIPriorityAndFairness) && s.GenericServerRunOptions.EnablePriorityAndFairness {
|
|
genericConfig.FlowControl, lastErr = BuildPriorityAndFairness(s, clientgoExternalClient, versionedInformers)
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// BuildAuthorizer constructs the authorizer
|
|
func BuildAuthorizer(s *options.ServerRunOptions, EgressSelector *egressselector.EgressSelector, versionedInformers clientgoinformers.SharedInformerFactory) (authorizer.Authorizer, authorizer.RuleResolver, error) {
|
|
authorizationConfig := s.Authorization.ToAuthorizationConfig(versionedInformers)
|
|
|
|
if EgressSelector != nil {
|
|
egressDialer, err := EgressSelector.Lookup(egressselector.ControlPlane.AsNetworkContext())
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
authorizationConfig.CustomDial = egressDialer
|
|
}
|
|
|
|
return authorizationConfig.New()
|
|
}
|
|
|
|
// BuildPriorityAndFairness constructs the guts of the API Priority and Fairness filter
|
|
func BuildPriorityAndFairness(s *options.ServerRunOptions, extclient clientgoclientset.Interface, versionedInformer clientgoinformers.SharedInformerFactory) (utilflowcontrol.Interface, error) {
|
|
if s.GenericServerRunOptions.MaxRequestsInFlight+s.GenericServerRunOptions.MaxMutatingRequestsInFlight <= 0 {
|
|
return nil, fmt.Errorf("invalid configuration: MaxRequestsInFlight=%d and MaxMutatingRequestsInFlight=%d; they must add up to something positive", s.GenericServerRunOptions.MaxRequestsInFlight, s.GenericServerRunOptions.MaxMutatingRequestsInFlight)
|
|
}
|
|
return utilflowcontrol.New(
|
|
versionedInformer,
|
|
extclient.FlowcontrolV1beta3(),
|
|
s.GenericServerRunOptions.MaxRequestsInFlight+s.GenericServerRunOptions.MaxMutatingRequestsInFlight,
|
|
s.GenericServerRunOptions.RequestTimeout/4,
|
|
), nil
|
|
}
|
|
|
|
// completedServerRunOptions is a private wrapper that enforces a call of Complete() before Run can be invoked.
|
|
type completedServerRunOptions struct {
|
|
*options.ServerRunOptions
|
|
}
|
|
|
|
// Complete set default ServerRunOptions.
|
|
// Should be called after kube-apiserver flags parsed.
|
|
func Complete(s *options.ServerRunOptions) (completedServerRunOptions, error) {
|
|
var options completedServerRunOptions
|
|
// set defaults
|
|
if err := s.GenericServerRunOptions.DefaultAdvertiseAddress(s.SecureServing.SecureServingOptions); err != nil {
|
|
return options, err
|
|
}
|
|
|
|
// process s.ServiceClusterIPRange from list to Primary and Secondary
|
|
// we process secondary only if provided by user
|
|
apiServerServiceIP, primaryServiceIPRange, secondaryServiceIPRange, err := getServiceIPAndRanges(s.ServiceClusterIPRanges)
|
|
if err != nil {
|
|
return options, err
|
|
}
|
|
s.PrimaryServiceClusterIPRange = primaryServiceIPRange
|
|
s.SecondaryServiceClusterIPRange = secondaryServiceIPRange
|
|
s.APIServerServiceIP = apiServerServiceIP
|
|
|
|
if err := s.SecureServing.MaybeDefaultWithSelfSignedCerts(s.GenericServerRunOptions.AdvertiseAddress.String(), []string{"kubernetes.default.svc", "kubernetes.default", "kubernetes"}, []net.IP{apiServerServiceIP}); err != nil {
|
|
return options, fmt.Errorf("error creating self-signed certificates: %v", err)
|
|
}
|
|
|
|
if len(s.GenericServerRunOptions.ExternalHost) == 0 {
|
|
if len(s.GenericServerRunOptions.AdvertiseAddress) > 0 {
|
|
s.GenericServerRunOptions.ExternalHost = s.GenericServerRunOptions.AdvertiseAddress.String()
|
|
} else {
|
|
if hostname, err := os.Hostname(); err == nil {
|
|
s.GenericServerRunOptions.ExternalHost = hostname
|
|
} else {
|
|
return options, fmt.Errorf("error finding host name: %v", err)
|
|
}
|
|
}
|
|
klog.Infof("external host was not specified, using %v", s.GenericServerRunOptions.ExternalHost)
|
|
}
|
|
|
|
s.Authentication.ApplyAuthorization(s.Authorization)
|
|
|
|
// Use (ServiceAccountSigningKeyFile != "") as a proxy to the user enabling
|
|
// TokenRequest functionality. This defaulting was convenient, but messed up
|
|
// a lot of people when they rotated their serving cert with no idea it was
|
|
// connected to their service account keys. We are taking this opportunity to
|
|
// remove this problematic defaulting.
|
|
if s.ServiceAccountSigningKeyFile == "" {
|
|
// Default to the private server key for service account token signing
|
|
if len(s.Authentication.ServiceAccounts.KeyFiles) == 0 && s.SecureServing.ServerCert.CertKey.KeyFile != "" {
|
|
if kubeauthenticator.IsValidServiceAccountKeyFile(s.SecureServing.ServerCert.CertKey.KeyFile) {
|
|
s.Authentication.ServiceAccounts.KeyFiles = []string{s.SecureServing.ServerCert.CertKey.KeyFile}
|
|
} else {
|
|
klog.Warning("No TLS key provided, service account token authentication disabled")
|
|
}
|
|
}
|
|
}
|
|
|
|
if s.ServiceAccountSigningKeyFile != "" && len(s.Authentication.ServiceAccounts.Issuers) != 0 && s.Authentication.ServiceAccounts.Issuers[0] != "" {
|
|
sk, err := keyutil.PrivateKeyFromFile(s.ServiceAccountSigningKeyFile)
|
|
if err != nil {
|
|
return options, fmt.Errorf("failed to parse service-account-issuer-key-file: %v", err)
|
|
}
|
|
if s.Authentication.ServiceAccounts.MaxExpiration != 0 {
|
|
lowBound := time.Hour
|
|
upBound := time.Duration(1<<32) * time.Second
|
|
if s.Authentication.ServiceAccounts.MaxExpiration < lowBound ||
|
|
s.Authentication.ServiceAccounts.MaxExpiration > upBound {
|
|
return options, fmt.Errorf("the service-account-max-token-expiration must be between 1 hour and 2^32 seconds")
|
|
}
|
|
if s.Authentication.ServiceAccounts.ExtendExpiration {
|
|
if s.Authentication.ServiceAccounts.MaxExpiration < serviceaccount.WarnOnlyBoundTokenExpirationSeconds*time.Second {
|
|
klog.Warningf("service-account-extend-token-expiration is true, in order to correctly trigger safe transition logic, service-account-max-token-expiration must be set longer than %d seconds (currently %s)", serviceaccount.WarnOnlyBoundTokenExpirationSeconds, s.Authentication.ServiceAccounts.MaxExpiration)
|
|
}
|
|
if s.Authentication.ServiceAccounts.MaxExpiration < serviceaccount.ExpirationExtensionSeconds*time.Second {
|
|
klog.Warningf("service-account-extend-token-expiration is true, enabling tokens valid up to %d seconds, which is longer than service-account-max-token-expiration set to %s seconds", serviceaccount.ExpirationExtensionSeconds, s.Authentication.ServiceAccounts.MaxExpiration)
|
|
}
|
|
}
|
|
}
|
|
|
|
s.ServiceAccountIssuer, err = serviceaccount.JWTTokenGenerator(s.Authentication.ServiceAccounts.Issuers[0], sk)
|
|
if err != nil {
|
|
return options, fmt.Errorf("failed to build token generator: %v", err)
|
|
}
|
|
s.ServiceAccountTokenMaxExpiration = s.Authentication.ServiceAccounts.MaxExpiration
|
|
}
|
|
|
|
if s.Etcd.EnableWatchCache {
|
|
sizes := kubeapiserver.DefaultWatchCacheSizes()
|
|
// Ensure that overrides parse correctly.
|
|
userSpecified, err := serveroptions.ParseWatchCacheSizes(s.Etcd.WatchCacheSizes)
|
|
if err != nil {
|
|
return options, err
|
|
}
|
|
for resource, size := range userSpecified {
|
|
sizes[resource] = size
|
|
}
|
|
s.Etcd.WatchCacheSizes, err = serveroptions.WriteWatchCacheSizes(sizes)
|
|
if err != nil {
|
|
return options, err
|
|
}
|
|
}
|
|
|
|
for key, value := range s.APIEnablement.RuntimeConfig {
|
|
if key == "v1" || strings.HasPrefix(key, "v1/") ||
|
|
key == "api/v1" || strings.HasPrefix(key, "api/v1/") {
|
|
delete(s.APIEnablement.RuntimeConfig, key)
|
|
s.APIEnablement.RuntimeConfig["/v1"] = value
|
|
}
|
|
if key == "api/legacy" {
|
|
delete(s.APIEnablement.RuntimeConfig, key)
|
|
}
|
|
}
|
|
|
|
options.ServerRunOptions = s
|
|
return options, nil
|
|
}
|
|
|
|
var testServiceResolver webhook.ServiceResolver
|
|
|
|
// SetServiceResolverForTests allows the service resolver to be overridden during tests.
|
|
// Tests using this function must run serially as this function is not safe to call concurrently with server start.
|
|
func SetServiceResolverForTests(resolver webhook.ServiceResolver) func() {
|
|
if testServiceResolver != nil {
|
|
panic("test service resolver is set: tests are either running concurrently or clean up was skipped")
|
|
}
|
|
|
|
testServiceResolver = resolver
|
|
|
|
return func() {
|
|
testServiceResolver = nil
|
|
}
|
|
}
|
|
|
|
func buildServiceResolver(enabledAggregatorRouting bool, hostname string, informer clientgoinformers.SharedInformerFactory) webhook.ServiceResolver {
|
|
if testServiceResolver != nil {
|
|
return testServiceResolver
|
|
}
|
|
|
|
var serviceResolver webhook.ServiceResolver
|
|
if enabledAggregatorRouting {
|
|
serviceResolver = aggregatorapiserver.NewEndpointServiceResolver(
|
|
informer.Core().V1().Services().Lister(),
|
|
informer.Core().V1().Endpoints().Lister(),
|
|
)
|
|
} else {
|
|
serviceResolver = aggregatorapiserver.NewClusterIPServiceResolver(
|
|
informer.Core().V1().Services().Lister(),
|
|
)
|
|
}
|
|
// resolve kubernetes.default.svc locally
|
|
if localHost, err := url.Parse(hostname); err == nil {
|
|
serviceResolver = aggregatorapiserver.NewLoopbackServiceResolver(serviceResolver, localHost)
|
|
}
|
|
return serviceResolver
|
|
}
|
|
|
|
func getServiceIPAndRanges(serviceClusterIPRanges string) (net.IP, net.IPNet, net.IPNet, error) {
|
|
serviceClusterIPRangeList := []string{}
|
|
if serviceClusterIPRanges != "" {
|
|
serviceClusterIPRangeList = strings.Split(serviceClusterIPRanges, ",")
|
|
}
|
|
|
|
var apiServerServiceIP net.IP
|
|
var primaryServiceIPRange net.IPNet
|
|
var secondaryServiceIPRange net.IPNet
|
|
var err error
|
|
// nothing provided by user, use default range (only applies to the Primary)
|
|
if len(serviceClusterIPRangeList) == 0 {
|
|
var primaryServiceClusterCIDR net.IPNet
|
|
primaryServiceIPRange, apiServerServiceIP, err = controlplane.ServiceIPRange(primaryServiceClusterCIDR)
|
|
if err != nil {
|
|
return net.IP{}, net.IPNet{}, net.IPNet{}, fmt.Errorf("error determining service IP ranges: %v", err)
|
|
}
|
|
return apiServerServiceIP, primaryServiceIPRange, net.IPNet{}, nil
|
|
}
|
|
|
|
_, primaryServiceClusterCIDR, err := netutils.ParseCIDRSloppy(serviceClusterIPRangeList[0])
|
|
if err != nil {
|
|
return net.IP{}, net.IPNet{}, net.IPNet{}, fmt.Errorf("service-cluster-ip-range[0] is not a valid cidr")
|
|
}
|
|
|
|
primaryServiceIPRange, apiServerServiceIP, err = controlplane.ServiceIPRange(*primaryServiceClusterCIDR)
|
|
if err != nil {
|
|
return net.IP{}, net.IPNet{}, net.IPNet{}, fmt.Errorf("error determining service IP ranges for primary service cidr: %v", err)
|
|
}
|
|
|
|
// user provided at least two entries
|
|
// note: validation asserts that the list is max of two dual stack entries
|
|
if len(serviceClusterIPRangeList) > 1 {
|
|
_, secondaryServiceClusterCIDR, err := netutils.ParseCIDRSloppy(serviceClusterIPRangeList[1])
|
|
if err != nil {
|
|
return net.IP{}, net.IPNet{}, net.IPNet{}, fmt.Errorf("service-cluster-ip-range[1] is not an ip net")
|
|
}
|
|
secondaryServiceIPRange = *secondaryServiceClusterCIDR
|
|
}
|
|
return apiServerServiceIP, primaryServiceIPRange, secondaryServiceIPRange, nil
|
|
}
|