k8s APIserver源码入门解读

APIserver

func main() {...#核心command := app.NewAPIServerCommand()#日志logs.InitLogs()defer logs.FlushLogs()...
}

#NewAPIServerCommand
#核心是调用run函数  运行AIPserver 永远不会退出
return Run(completedOptions, genericapiserver.SetupSignalHandler())

#Run
#创建server链路
server, err := CreateServerChain(completeOptions, stopCh)

func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan struct{}) (*aggregatorapiserver.APIAggregator, error) {#创建APIserver的配置kubeAPIServerConfig, serviceResolver, pluginInitializer, err := CreateKubeAPIServerConfig(completedOptions)#API拓展服务 主要针对CRDapiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegate())#API核心服务kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer)#API聚合服务 聚合前面的serveraggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
}

#CreateKubeAPIServerConfig
func CreateKubeAPIServerConfig(s completedServerRunOptions) (*controlplane.Config,aggregatorapiserver.ServiceResolver,[]admission.PluginInitializer,error,
) {#建立通用配置genericConfig, versionedInformers, serviceResolver, pluginInitializers, admissionPostStartHook, storageFactory, err := buildGenericConfig(s.ServerRunOptions, proxyTransport)if err != nil {}

#buildGenericConfig
func buildGenericConfig(...#默认生成api文档genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(generatedopenapi.GetOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))...#采用etcd作为存储方案completedStorageFactoryConfig, err := storageFactoryConfig.Complete(s.Etcd)#认证机制if lastErr = s.Authentication.ApplyTo#授权机制genericConfig.Authorization.Authorizer, genericConfig.RuleResolver, err = BuildAuthorizer(s, genericConfig.EgressSelector, versionedInformers)#准入机制err = s.Admission.ApplyTo()
}

#认证机制
#s.Authentication.ApplyTo
func (o *BuiltInAuthenticationOptions) ApplyTo(authInfo *genericapiserver.AuthenticationInfo, secureServing *genericapiserver.SecureServingInfo, egressSelector *egressselector.EgressSelector, openAPIConfig *openapicommon.Config, extclient kubernetes.Interface, versionedInformer informers.SharedInformerFactory){...#实例化ConfigauthInfo.Authenticator, openAPIConfig.SecurityDefinitions, err = authenticatorConfig.New()...
}

#authenticatorConfig.New
func (config Config) New() (authenticator.Request, *spec.SecurityDefinitions, error) {#重点关注authenticators 和 tokenAuthenticators两个变量var authenticators []authenticator.Requestvar tokenAuthenticators []authenticator.Token#添加requestHeader认证方式if config.RequestHeaderConfig != nil {requestHeaderAuthenticator := headerrequest.NewDynamicVerifyOptionsSecure(...)#追加认证方式authenticators = append(authenticators, authenticator.WrapAudienceAgnosticRequest(config.APIAudiences, requestHeaderAuthenticator))}#添加CLientCA认证方式if config.ClientCAContentProvider != nil {...}#添加Token认证方式if len(config.TokenAuthFile) > 0 {tokenAuth, err := newAuthenticatorFromTokenFile(config.TokenAuthFile)}#...其他各种认证方式#如果没有认证方式 则启动anonymousif len(authenticators) == 0 {if config.Anonymous {return anonymous.NewAuthenticator(), &securityDefinitions, nil}return nil, &securityDefinitions, nil}#整合两种认证方式authenticators 和 tokenAuthenticatorsauthenticator := union.New(authenticators...)}return authenticator, &securityDefinitions, nil
}

#授权机制
#BuildAuthorizer
#调用此函数
return authorizationConfig.New()

#New
func (config Config) New() (authorizer.Authorizer, authorizer.RuleResolver, error) {...case modes.ModeNode:...const (...#对生产来说最有用出的模式RBAC 角色-用户模式ModeRBAC string = "RBAC"...
)

#CreateKubeAPIServer
func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget) (*Instance, error) {#GenericServer实例化s, err := c.GenericConfig.New("kube-apiserver", delegationTarget)...#masterserver实例化m := &Instance{GenericAPIServer:          s,ClusterAuthenticationInfo: c.ExtraConfig.ClusterAuthenticationInfo,}#注册LegacyAPIif c.ExtraConfig.APIResourceConfigSource.VersionEnabled(apiv1.SchemeGroupVersion) {...}if err := m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter, legacyRESTStorageProvider); err != nil {return nil, err}}#REST接口的存储定义 可以看到很多k8s上的常见定义 比如node节点/storage存储/event事件等等restStorageProviders := []RESTStorageProvider{...}#安装 API if err := m.InstallAPIs(c.ExtraConfig.APIResourceConfigSource, c.GenericConfig.RESTOptionsGetter, restStorageProviders...); err != nil {return nil, err}#添加钩子m.GenericAPIServer.AddPostStartHookOrDie...return m, nil
}

#GenericConfig.New
func (c completedConfig) New(name string, delegationTarget DelegationTarget) {#实例化一个APIservers := &GenericAPIServer#启动之后的钩子函数for k, v := range delegationTarget.PostStartHooks() {s.postStartHooks[k] = v}#关闭之前的钩子函数for k, v := range delegationTarget.PreShutdownHooks() {s.preShutdownHooks[k] = v}...#安装相应的APIserverinstallAPI(s, c.Config)
}

# installAPI
func installAPI(s *GenericAPIServer, c *Config) {#添加/index.html路由规则if c.EnableIndex {}#添加pprof的路由规则if c.EnableProfiling {}#添加监控相关的/metrics的指标路由规则if c.EnableMetrics {}#添加版本相关的路由规则routes.Version{Version: c.Version}.Install(s.Handler.GoRestfulContainer)#看起服务发现if c.EnableDiscovery {}...
}

func (m *Master) InstallLegacyAPI(c *completedConfig, restOptionsGetter generic.RESTOptionsGetter, legacyRESTStorageProvider corerest.LegacyRESTStorageProvider) error {# RESTStorage的初始化legacyRESTStorage, apiGroupInfo, err := legacyRESTStorageProvider.NewLegacyRESTStorage(restOptionsGetter)# 前缀为 /api，注册上对应的Version和Resource# Pod作为核心资源，没有Group的概念if err := m.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {return fmt.Errorf("error in registering group versions: %v", err)}return nil
}

#NewLegacyRESTStorage
#RESTStorage的初始化
func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(restOptionsGetter generic.RESTOptionsGetter) (LegacyRESTStorage, genericapiserver.APIGroupInfo, error) {# pod 模板podTemplateStorage, err := podtemplatestore.NewREST(restOptionsGetter)# event事件eventStorage, err := eventstore.NewREST(restOptionsGetter, uint64(c.EventTTL.Seconds()))# limitRange资源限制limitRangeStorage, err := limitrangestore.NewREST(restOptionsGetter)# resourceQuota资源配额resourceQuotaStorage, resourceQuotaStatusStorage, err := resourcequotastore.NewREST(restOptionsGetter)# secret加密secretStorage, err := secretstore.NewREST(restOptionsGetter)# PV 存储persistentVolumeStorage, persistentVolumeStatusStorage, err := pvstore.NewREST(restOptionsGetter)# PVC 存储persistentVolumeClaimStorage, persistentVolumeClaimStatusStorage, err := pvcstore.NewREST(restOptionsGetter)# ConfigMap 配置configMapStorage, err := configmapstore.NewREST(restOptionsGetter)...# pod模板podStorage, err := podstore.NewStorage()# 保存storage的对应关系restStorageMap := map[string]rest.Storage{"pods":             podStorage.Pod,...}
}

#podstore.NewStorage
#Pod初始化
func NewStorage(optsGetter generic.RESTOptionsGetter, k client.ConnectionInfoGetter, proxyTransport http.RoundTripper, podDisruptionBudgetClient policyclient.PodDisruptionBudgetsGetter) (PodStorage, error) {store := &genericregistry.Store{// 增改删的策略CreateStrategy:      registrypod.Strategy,UpdateStrategy:      registrypod.Strategy,DeleteStrategy:      registrypod.Strategy,...}
}