APIServer源码分析之入口点解析
2023-03-17 k8s技术圈
加入收藏
Kube.NETes（K8s）集群中最关键的组件之一是 API Server，它是所有集群管理活动的入口点。从本文开始，我们将对 K8s API Server 的代码进行详细分析，并探讨其应用入口点、框架以及与 etcd 的通信。
应用入口点
K8s API Server 的主要入口点位于 cmd/kube-apiserver/apiserver.go 文件的。
// cmd/kube-apiserver/apiserver.go
// apiserver is the main api server and master for the cluster.
// it is responsible for serving the cluster management API.
package main
import (
"os"
"k8s.io/component-base/cli"
_ "k8s.io/component-base/logs/json/register" // 用于JSON日志格式注册
_ "k8s.io/component-base/metrics/prometheus/clientgo" // 加载所有的 prometheus client-go 插件
_ "k8s.io/component-base/metrics/prometheus/version" // 用于版本指标注册
"k8s.io/kubernetes/cmd/kube-apiserver/App"
)
func main() {
command := app.NewAPIServerCommand()
code := cli.Run(command)
os.Exit(code)
}
其中的 app.NewAPIServerCommand() 是构建的一个 cobra 的命令对象，cli.Run 然后执行该命令即可，所以我们直接查看 NewAPIServerCommand 函数是如果构造 cobra.Command 对象的：
// cmd/kube-apiserver/app/server.go
// NewAPIServerCommand 使用默认参数创建一个 *cobra.Command 对象
func NewAPIServerCommand() *cobra.Command {
// NewServerRunOptions 使用默认参数创建一个新的 ServerRunOptions 对象。
// ServerRunOption 对象是运行 apiserver 需要的对象
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.`,
// ......
RunE: func(cmd *cobra.Command, args []string) error {
verflag.PrintAndExitIfRequested()
fs := cmd.Flags()
if err := s.Logs.ValidateAndApply(); err != nil {
return err
}
cliflag.PrintFlags(fs)
err := checkNonZeroInsecurePort(fs)
if err != nil {
return err
}
// 设置默认选项
completedOptions, err := Complete(s)
if err != nil {
return err
}
// 校验选项
if errs := completedOptions.Validate(); len(errs) != 0 {
return utilerrors.NewAggregate(errs)
}
return Run(completedOptions, genericapiserver.SetupSignalHandler())
},
}
// ......
return cmd
}
该函数最核心的功能就是使用 Complete(s) 函数来生成 apiserver 启动需要的默认参数，然后将默认参数传递给 Run 函数进行启动。
// cmd/kube-apiserver/app/server.go
// Run 运行指定的 APIServer，不能退出.
func Run(completeOptions completedServerRunOptions, stopCh <-chan struct{}) error {
// 创建服务链（包含的3个server组件）
server, err := CreateServerChain(completeOptions, stopCh)
// 服务启动前的准备工作，包括健康检查、存活检查、OpenAPI路由注册等
prepared, err := server.PrepareRun()
// 正式启动运行
return prepared.Run(stopCh)
}
在 Run 函数中首先会通过 CreateServerChain 函数通过委托创建连接的 APIServer 对象。
// cmd/kube-apiserver/app/server.go
// CreateServerChain 通过委托创建连接的APIServer
func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan struct{}) (*aggregatorapiserver.APIAggregator, error) {
// CreateKubeAPIServerConfig 创建用于运行 APIServer 的所有配置资源，但不运行任何资源
kubeAPIServerConfig, serviceResolver, pluginInitializer, err := CreateKubeAPIServerConfig(completedOptions)
// // 创建 APIExtensionsServer 配置
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))
// 创建APIExtensionsServer并注册路由
apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegateWithCustomHandler(notFoundHandler))
// 创建KubeAPIServer并注册路由
kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer)
// // 创建 aggregatorServer 配置
aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, completedOptions.ServerRunOptions, kubeAPIServerConfig.ExtraConfig.VersionedInformers, serviceResolver, kubeAPIServerConfig.ExtraConfig.ProxyTransport, pluginInitializer)
// 创建aggregatorServer并注册路由
aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
return aggregatorServer, nil
}
上面的函数中可以看到 CreateServerChain 会创建3个 server：APIExtensionServer、KubeAPIServer、AggregratorServer，APIServer 就是依靠这3个组件来对不同类型的请求进行处理的：
- APIExtensionServer: 主要负责处理 CustomResourceDefinition（CRD）方面的请求。
- KubeAPIServer: 主要负责处理 K8s 内置资源的请求，此外还会包括通用处理、认证、鉴权等。
- AggregratorServer: 主要负责聚合器方面的处理，它充当一个代理服务器，将请求转发到聚合进来的 K8s service 中。
创建每个 server 都有对应的 config，可以看出上面函数中的 apiExtensionServer 和 aggregatorServer 的 Config 需要依赖 kubeAPIServerConfig，而这几个 ServerConfig 都需要依赖 GenericConfig，CreateKubeAPIServerConfig 函数创建 kubeAPIServerConfig ，在该函数中通过调用 buildGenericConfig 来创建 GenericConfig 对象，如下代码所示。
// cmd/kube-apiserver/app/server.go
// CreateKubeAPIServerConfig 创建用于运行 APIServer 的所有配置资源
func CreateKubeAPIServerConfig(s completedServerRunOptions) (
*controlplane.Config,
aggregatorapiserver.ServiceResolver,
[]admission.PluginInitializer,
error,
) {
proxyTransport := CreateProxyTransport()
// 构建通用配置
genericConfig, versionedInformers, serviceResolver, pluginInitializers, admissionPostStartHook, storageFactory, err := buildGenericConfig(s.ServerRunOptions, proxyTransport)
// ......
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,
},
}
// ......
return config, serviceResolver, pluginInitializers, nil
}
func buildGenericConfig(
s *options.ServerRunOptions,
proxyTransport *http.Transport,
)(...){
//创建一个通用配置对象
genericConfig = genericapiserver.NewConfig(legacyscheme.Codecs)
// ......
//创建认证实例
if lastErr = s.Authentication.ApplyTo(&genericConfig.Authentication, genericConfig.SecureServing, genericConfig.EgressSelector, genericConfig.OpenAPIConfig, clientgoExternalClient, versionedInformers); lastErr != nil {
return
}
// ...
// openapi/swagger配置，OpenAPIConfig 用于生成 OpenAPI 规范
getOpenAPIDefinitions := openapi.GetOpenAPIDefinitionsWithoutDisabledFeatures(generatedopenapi.GetOpenAPIDefinitions)
genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(getOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))
genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
genericConfig.LongRunningFunc = filters.BasicLongRunningRequestCheck(
sets.NewString("watch", "proxy"),
sets.NewString("attach", "exec", "proxy", "log", "portforward"),
)
// storageFactoryConfig 对象定义了 kube-apiserver 与 etcd 的交互方式，如：etcd认证、地址、存储前缀等
// 该对象也定义了资源存储方式，如：资源信息、资源编码信息、资源状态等
storageFactoryConfig := kubeapiserver.NewStorageFactoryConfig()
storageFactoryConfig.APIResourceConfig = genericConfig.MergedResourceConfig
completedStorageFactoryConfig, err := storageFactoryConfig.Complete(s.Etcd)
storageFactory, lastErr = completedStorageFactoryConfig.New()
if lastErr = s.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig); lastErr != nil {
return
}
// ......
// 初始化 SharedInformerFactory
kubeClientConfig := genericConfig.LoopbackClientConfig
clientgoExternalClient, err := clientgoclientset.NewForConfig(kubeClientConfig)
versionedInformers = clientgoinformers.NewSharedInformerFactory(clientgoExternalClient, 10*time.Minute)
// 认证配置，内部调用 authenticatorConfig.New()
// K8s提供了9种认证机制，每种认证机制被实例化后都成为认证器
if lastErr = s.Authentication.ApplyTo(&genericConfig.Authentication, genericConfig.SecureServing, genericConfig.EgressSelector, genericConfig.OpenAPIConfig, clientgoExternalClient, versionedInformers); lastErr != nil {
return
}
// 创建鉴权实例，K8s也提供了6种授权机制，每种授权机制被实例化后都成为授权器
genericConfig.Authorization.Authorizer, genericConfig.RuleResolver, err = BuildAuthorizer(s, genericConfig.EgressSelector, versionedInformers)
// ...
// 审计
lastErr = s.Audit.ApplyTo(genericConfig)
// 准入控制器
// k8s资源在认证和授权通过，被持久化到etcd之前进入准入控制逻辑
// 准入控制包括：对请求的资源进行自定义操作（校验、修改、拒绝）
// 准入控制器通过 Plugins 数据结构统一注册、存放、管理
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)
err = s.Admission.ApplyTo(
genericConfig,
versionedInformers,
kubeClientConfig,
feature.DefaultFeatureGate,
pluginInitializers...)
// ...
}
然后我们再来分别看看这3个 Server 是如何构建的。
go-restful框架
这里我们就不得不先了解下 go-restful 这个框架了，因为 APIServer 就使用的这个框架。下面的代码是 go-restful 官方的一个示例，这个 demo 了解后基本上就知道 go-restful 框架是如何使用的了：
package main
import (
"log"
"net/http"
restfulspec "Github.com/emicklei/go-restful-openapi/v2"
restful "github.com/emicklei/go-restful/v3"
"github.com/go-openapi/spec"
)
// UserResource is the REST layer to the User domain
type UserResource struct {
// normally one would use DAO (data access object)
users map[string]User
}
// WebService creates a new service that can handle REST requests for User resources.
func (u UserResource) WebService() *restful.WebService {
ws := new(restful.WebService)
ws.
Path("/users").
Consumes(restful.MIME_XML, restful.MIME_JSON).
Produces(restful.MIME_JSON, restful.MIME_XML) // you can specify this per route as well
tags := []string{"users"}
ws.Route(ws.GET("/").To(u.findAllUsers).
// docs
Doc("get all users").
Metadata(restfulspec.KeyOpenAPITags, tags).
Writes([]User{}).
Returns(200, "OK", []User{}))
ws.Route(ws.GET("/{user-id}").To(u.findUser).
// docs
Doc("get a user").
Param(ws.PathParameter("user-id", "identifier of the user").DataType("integer").DefaultValue("1")).
Metadata(restfulspec.KeyOpenAPITags, tags).
Writes(User{}). // on the response
Returns(200, "OK", User{}).
Returns(404, "Not Found", nil))
ws.Route(ws.PUT("/{user-id}").To(u.updateUser).
// docs
Doc("update a user").
Param(ws.PathParameter("user-id", "identifier of the user").DataType("string")).
Metadata(restfulspec.KeyOpenAPITags, tags).
Reads(User{})) // from the request
ws.Route(ws.PUT("").To(u.createUser).
// docs
Doc("create a user").
Metadata(restfulspec.KeyOpenAPITags, tags).
Reads(User{})) // from the request
ws.Route(ws.DELETE("/{user-id}").To(u.removeUser).
// docs
Doc("delete a user").
Metadata(restfulspec.KeyOpenAPITags, tags).
Param(ws.PathParameter("user-id", "identifier of the user").DataType("string")))
return ws
}
// GET http://localhost:8080/users
//
func (u UserResource) findAllUsers(request *restful.Request, response *restful.Response) {
list := []User{}
for _, each := range u.users {
list = append(list, each)
}
response.WriteEntity(list)
}
// GET http://localhost:8080/users/1
//
func (u UserResource) findUser(request *restful.Request, response *restful.Response) {
id := request.PathParameter("user-id")
usr := u.users[id]
if len(usr.ID) == 0 {
response.WriteErrorString(http.StatusNotFound, "User could not be found.")
} else {
response.WriteEntity(usr)
}
}
// PUT http://localhost:8080/users/1
// <User><Id>1</Id><Name>Melissa Raspberry</Name></User>
//
func (u *UserResource) updateUser(request *restful.Request, response *restful.Response) {
usr := new(User)
err := request.ReadEntity(&usr)
if err == nil {
u.users[usr.ID] = *usr
response.WriteEntity(usr)
} else {
response.WriteError(http.StatusInternalServerError, err)
}
}
// PUT http://localhost:8080/users/1
// <User><Id>1</Id><Name>Melissa</Name></User>
//
func (u *UserResource) createUser(request *restful.Request, response *restful.Response) {
usr := User{ID: request.PathParameter("user-id")}
err := request.ReadEntity(&usr)
if err == nil {
u.users[usr.ID] = usr
response.WriteHeaderAndEntity(http.StatusCreated, usr)
} else {
response.WriteError(http.StatusInternalServerError, err)
}
}
// DELETE http://localhost:8080/users/1
//
func (u *UserResource) removeUser(request *restful.Request, response *restful.Response) {
id := request.PathParameter("user-id")
delete(u.users, id)
}
func main() {
u := UserResource{map[string]User{}}
restful.DefaultContainer.Add(u.WebService())
config := restfulspec.Config{
WebServices: restful.RegisteredWebServices(), // you control what services are visible
APIPath: "/apidocs.json",
PostBuildSwaggerObjectHandler: enrichSwaggerObject}
restful.DefaultContainer.Add(restfulspec.NewOpenAPIService(config))
// Optionally, you can install the Swagger Service which provides a nice Web UI on your REST API
// You need to download the Swagger html5 assets and change the FilePath location in the config below.
// Open http://localhost:8080/apidocs/?url=http://localhost:8080/apidocs.json
http.Handle("/apidocs/", http.StripPrefix("/apidocs/", http.FileServer(http.Dir("/Users/emicklei/Projects/swagger-ui/dist"))))
log.Printf("start listening on localhost:8080")
log.Fatal(http.ListenAndServe(":8080", nil))
}
func enrichSwaggerObject(swo *spec.Swagger) {
swo.Info = &spec.Info{
InfoProps: spec.InfoProps{
Title: "UserService",
Description: "Resource for managing Users",
Contact: &spec.ContactInfo{
ContactInfoProps: spec.ContactInfoProps{
Name: "john",
Email: "john@doe.rp",
URL: "http://johndoe.org",
},
},
License: &spec.License{
LicenseProps: spec.LicenseProps{
Name: "MIT",
URL: "http://mit.org",
},
},
Version: "1.0.0",
},
}
swo.Tags = []spec.Tag{spec.Tag{TagProps: spec.TagProps{
Name: "users",
Description: "Managing users"}}}
}
// User is just a sample type
type User struct {
ID string `json:"id" description:"identifier of the user"`
Name string `json:"name" description:"name of the user" default:"john"`
Age int `json:"age" description:"age of the user" default:"21"`
}
这个示例代码，就是使用 go-restful 的核心功能实现了一个简单的 RESTful 的 API，实现了对 User 的增删查改，其中有这么几个核心概念：Container、WebService、Route。
- Container：服务器容器，包含多个 WebService 和一个 http.ServerMux。
- WebService：服务，由多个 Route 组成，一个 WebService 其实代表某一个对象相关的服务，如上例中的 /users，针对该 /users 要实现RESTful API，那么需要向其添加增删查改的路由，即 Route，它是 Route 的集合。
- Route：路由，包含了 url，http 方法，接收和响应的媒体类型以及处理函数。每一个 Route，根据 Method 和 Path，映射到对应的方法中，即是 Method/Path 到 Function 映射关系的抽象，如上例中的 ws.Route(ws.GET("/{user-id}").To(u.findUser))，就是针对 /users/{user-id}该路径的GET请求，则被路由到 findUser 方法中进行处理。
- Container 是 WebService 的集合，可以向 Container 中添加多个 WebService，而 Container 因为实现了 ServeHTTP() 方法，其本质上还是一个http Handler，可以直接用在 http Server 中。
Kubernetes 中对 go-restful 的使用比较基础，就使用到了其最基础的路由功能，由于 K8s 有很多内置的资源对象，也包括 CRD 这种自定义资源对象，所以一开始并不是直接将这些资源对应对应的接口硬编码的，而是通过一系列代码动态注册的，所以接下来我们分析的其实就是想办法让 APIServer 能够提供如下所示的路由处理出来：
GET /apis/apps/v1/namespaces/{namespace}/deployments/{name}
POST /apis/apps/v1/namespaces/{namespace}/deployments
GET /apis/apps/v1/namespaces/{namespace}/daemonsets/{name}
POST /apis/apps/v1/namespaces/{namespace}/daemonsets
对 go-restful 有一个基础了解后，后面就可以去了解下这3个 Server 具体是如何实例化的了。
