k8s的apiserver_k8s源码分析

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE使用 1年只要46元 售后保障 童叟无欺

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拓展服务 主要针对CRD
	apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegate())
	#API核心服务
	kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer)
	#API聚合服务 聚合前面的server
	aggregatorServer, 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){ 
   
	...
	#实例化Config
	authInfo.Authenticator, openAPIConfig.SecurityDefinitions, err = authenticatorConfig.New()
	...
}

#authenticatorConfig.New
func (config Config) New() (authenticator.Request, *spec.SecurityDefinitions, error) { 
   
	#重点关注authenticators 和 tokenAuthenticators两个变量
	var authenticators []authenticator.Request
	var 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)
		
	}
	#...其他各种认证方式
	#如果没有认证方式 则启动anonymous
	if len(authenticators) == 0 { 
   
		if config.Anonymous { 
   
			return anonymous.NewAuthenticator(), &securityDefinitions, nil
		}
		return nil, &securityDefinitions, nil
	}
	#整合两种认证方式authenticators 和 tokenAuthenticators
	authenticator := 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,
	}
	#注册LegacyAPI
	if 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) { 
   
	#实例化一个APIserver
	s := &GenericAPIServer
	#启动之后的钩子函数
	for k, v := range delegationTarget.PostStartHooks() { 
   
		s.postStartHooks[k] = v
	}
	#关闭之前的钩子函数
	for k, v := range delegationTarget.PreShutdownHooks() { 
   
		s.preShutdownHooks[k] = v
	}
	...
	#安装相应的APIserver
	installAPI(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,
		...
	}
}