startSrv
南风未起 2025-07-05 19:39:22 apiserver
# 1.初始化
# 1.1.runsever
apiserver启动基于cobra参数解析，初始化的apiserver command及注册启动命令会由cobra回调，核心入口就是RunE注册的函数。func main() { // 核心入口 command := app.NewAPIServerCommand() code := cli.Run(command) os.Exit(code) } // NewAPIServerCommand creates a *cobra.Command object with default parameters func NewAPIServerCommand() *cobra.Command { s := options.NewServerRunOptions() cmd := &cobra.Command{ ... RunE: func(cmd *cobra.Command, args []string) error { ... // 填充默认配置 // serviceIP/jwtToken签发器及过期时间/对外暴露地址/etcd缓存 completedOptions, err := Complete(s) ... // 验证配置 completedOptions.Validate() ... // 启动 return Run(completedOptions, genericapiserver.SetupSignalHandler()) }, ... } ... return cmd } // Run runs the specified APIServer. This should never exit. func Run(completeOptions completedServerRunOptions, stopCh <-chan struct{}) error { ... // 创建server chain server, err := CreateServerChain(completeOptions) ... // 准备工作(钩子及handler注册) prepared, err := server.PrepareRun() ... // 开始执行 return prepared.Run(stopCh) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
注意
CreateServerChain()完成三个server的初始化工作，PrepareRun()补充一些handler注册，最终调用prepared.Run()启动
# 1.2.serverchain
CreateServerChain()会依次完成核心server初始化，构建aggregatorServer-->apiserver-->apiextensionserver链式委托服务。// CreateServerChain creates the apiservers connected via delegation. func CreateServerChain(completedOptions completedServerRunOptions) (*aggregatorapiserver.APIAggregator, error) { // kubeApiServer配置 kubeAPIServerConfig, serviceResolver, pluginInitializer, err := CreateKubeAPIServerConfig(completedOptions) ... // kubeapiExtensionServer配置 apiExtensionsConfig, err := createAPIExtensionsConfig(...) ... // notFound处理器(委托的最后一级) notFoundHandler := notfoundhandler.New(...) // 创建apiExtensionServer apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegateWithCustomHandler(notFoundHandler)) ... // 创建kubeApiServer kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer) ... // kubeAggregatorServer配置 aggregatorConfig, err := createAggregatorConfig(...) ... // 创建aggregatorServer aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers, crdAPIEnabled) ... return aggregatorServer, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
注意
kubernetes将扩展和核心部分分离，分别是apiserver和apiextensionserver，两者的请求分配和服务发现由aggregatorserver实现
# 1.3.genericServer
genericAPIServer是apiserver非常重要的对象，负责请求处理及接受资源注册，三个组件都会基于genericAPIServer作为HTTP Server。// CreateServerChain creates the apiservers connected via delegation. func CreateServerChain(completedOptions completedServerRunOptions) (*aggregatorapiserver.APIAggregator, error) { ... // 初始化apiExtensionServer apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegateWithCustomHandler(notFoundHandler)) ... // 初始化kubeApiServer kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer) ... // 初始化aggregatorServer aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers, crdAPIEnabled) ... return aggregatorServer, nil } func createAPIExtensionsServer(...) (*apiextensionsapiserver.CustomResourceDefinitions, error) { return apiextensionsConfig.Complete().New(delegateAPIServer) } // New returns a new instance of CustomResourceDefinitions from the given config. func (c completedConfig) New(delegation genericapiserver.DelegationTarget) (*CustomResourceDefinitions, error) { // 初始化genericAPIServer genericServer, err := c.GenericConfig.New("apiextensions-apiserver", delegation) ... return s, nil } // New creates a new server which logically combines the handling chain with the passed server. func (c completedConfig) New(name string, delegation DelegationTarget) (*GenericAPIServer, error) { ... // 处理链构造(认证/鉴权/审计) handlerChainBuilder := func(handler http.Handler) http.Handler { return c.BuildHandlerChainFunc(handler, c.Config) } ... // 初始化apiServerHandler handler := NewAPIServerHandler(name,c.Serializer,handlerChainBuilder,delegation.UnprotectedHandler()) // 构造genericAPIServer s := &GenericAPIServer{ ... Handler: handler, ... listedPathProvider: handler, ... } ... return s, nil } func NewAPIServerHandler(name string, s runtime.NegotiatedSerializer, handlerChainBuilder HandlerChainBuilderFn, notFoundHandler http.Handler) *APIServerHandler { // 构造nonGoRestfulMux(注册非restful路由) nonGoRestfulMux := mux.NewPathRecorderMux(name) if notFoundHandler != nil { nonGoRestfulMux.NotFoundHandler(notFoundHandler) } // 构造container(注册api/err/recover handler) gorestfulContainer := restful.NewContainer() gorestfulContainer.ServeMux = http.NewServeMux() gorestfulContainer.Router(restful.CurlyRouter{}) // e.g. for proxy/{kind}/{name}/{*} gorestfulContainer.RecoverHandler(func(panicReason interface{}, httpWriter http.ResponseWriter) { logStackOnRecover(s, panicReason, httpWriter) }) gorestfulContainer.ServiceErrorHandler(func(serviceErr restful.ServiceError, request *restful.Request, response *restful.Response) { serviceErrorHandler(s, serviceErr, request, response) }) // 构造director,负责请求路由到nonGoRestfulMux/container director := director{ name: name, goRestfulContainer: gorestfulContainer, nonGoRestfulMux: nonGoRestfulMux, } return &APIServerHandler{ FullHandlerChain: handlerChainBuilder(director), GoRestfulContainer: gorestfulContainer, NonGoRestfulMux: nonGoRestfulMux, Director: director, } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
注意
三个组件通过委托关联，
aggregator-->apiserver-->apiextensionserver-->notFoundServer
# 1.4.handlerChain
director会被包一层实现认证、鉴权及审计，包装的buildHandlerChainFunc其实就是通用配置中设置的defaultBuildHandlerChain。// CreateServerChain creates the apiservers connected via delegation. func CreateServerChain(completedOptions completedServerRunOptions) (*aggregatorapiserver.APIAggregator, error) { kubeAPIServerConfig, serviceResolver, pluginInitializer, err := CreateKubeAPIServerConfig(completedOptions) ... return aggregatorServer, nil } // CreateKubeAPIServerConfig creates all the resources for running the API server, but runs none of them func CreateKubeAPIServerConfig(s completedServerRunOptions) (...) { ... genericConfig... := buildGenericConfig(s.ServerRunOptions, proxyTransport) ... return config, serviceResolver, pluginInitializers, nil } // BuildGenericConfig takes the master server options and produces the genericapiserver.Config. func buildGenericConfig(...) (...) { genericConfig = genericapiserver.NewConfig(legacyscheme.Codecs) ... return } // NewConfig returns a Config struct with the default values func NewConfig(codecs serializer.CodecFactory) *Config { ... return &Config{ Serializer: codecs, BuildHandlerChainFunc: DefaultBuildHandlerChain, // 这里就是handler chain包装器 ... } } // DefaultBuildHandlerChain wrapper director. func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler { ... // 鉴权处理 handler = genericapifilters.WithAuthorization(handler, c.Authorization.Authorizer, c.Serializer) ... // APF限流 if c.FlowControl != nil { ... handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl, requestWorkEstimator) ... // 静态限流 } else { handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc) } ... // 审计处理 handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyRuleEvaluator, c.LongRunningFunc) ... // 认证处理 handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences, c.Authentication.RequestHeaderConfig) ... // 跨域处理 handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true") // 设置普通请求的超时 handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc) // 设置长连接的最大处理时长 handler = genericapifilters.WithRequestDeadline(handler, c.AuditBackend, c.AuditPolicyRuleEvaluator, c.LongRunningFunc, c.Serializer, c.RequestTimeout) ... // 添加cache control头,防止敏感API响应被缓存 handler = genericapifilters.WithCacheControl(handler) // 添加HSTS响应头,强制使用https访问 handler = genericfilters.WithHSTS(handler, c.HSTSDirectives) ... // http请求解析成结构化requestInfo handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver) ... // 初始化审计上下文handler handler = genericapifilters.WithAuditInit(handler) return handler }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
注意
director传入会一层层包装修饰器，构造一个反方向的调用链，由最后一个修饰handler依次准入
# 1.5.requestInfo
withRequestInfo()是处理链中比较通用的部分，用于解析restful请求数据，向后续handler提供一个通用的上下文对象requestInfo。// WithRequestInfo attaches a RequestInfo to the context. func WithRequestInfo(handler http.Handler, resolver request.RequestInfoResolver) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { ctx := req.Context() // 解析请求 info, err := resolver.NewRequestInfo(req) ... // 保存到ctx req = req.WithContext(request.WithRequestInfo(ctx, info)) // 委托到下一级 handler.ServeHTTP(w, req) }) } // CreateKubeAPIServer creates and wires a workable kube-apiserver func CreateKubeAPIServer(kubeAPIServerConfig *controlplane.Config, delegateAPIServer genericapiserver.DelegationTarget) (*controlplane.Instance, error) { return kubeAPIServerConfig.Complete().New(delegateAPIServer) } // Complete fills in any fields not set that are required to have valid data. It's mutating the receiver. func (c *Config) Complete() CompletedConfig { cfg := completedConfig{ c.GenericConfig.Complete(c.ExtraConfig.VersionedInformers), &c.ExtraConfig, } ... return CompletedConfig{&cfg} } // resolver会在genericConfig对象的cpmplete阶段完成初始化 func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedConfig { ... if c.RequestInfoResolver == nil { c.RequestInfoResolver = NewRequestInfoResolver(c) } ... return CompletedConfig{&completedConfig{c, informers}} } func NewRequestInfoResolver(c *Config) *apirequest.RequestInfoFactory { ... return &apirequest.RequestInfoFactory{ APIPrefixes: apiPrefixes, // apis和legacyAPI资源前缀 GrouplessAPIPrefixes: legacyAPIPrefixes, // 无group的legacy资源声明 } }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
注意
requestInfo是http.Request的抽象，将http请求转换为更友好的理解形式，提取请求的资源属性用于后续的路由解析
# 2.启动
# 2.1.handler
apiServerHandler是genericAPIServer的核心处理模块，kube-apiserver启动其实会注册到http server用于真正的请求处理。// GenericAPIServer contains state for a Kubernetes cluster api server. type GenericAPIServer struct { ... // Handler holds the handlers being used by this API server Handler *APIServerHandler ... } // APIServerHandlers holds the different http.Handlers used by the API server. type APIServerHandler struct { FullHandlerChain http.Handler // 前面提到的handlerChainBuilder GoRestfulContainer *restful.Container // container(注册api资源路由) NonGoRestfulMux *mux.PathRecorderMux // nonGoRestfulMux(注册非restful路由) Director http.Handler // 分发器 } // ServeHTTP makes it an http.Handler func (a *APIServerHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { a.FullHandlerChain.ServeHTTP(w, r) } // 最终会执行到这里,director会被handlerChain包裹,执行认证/鉴权/审计后才执行真正的请求分发 func (d director) ServeHTTP(w http.ResponseWriter, req *http.Request) { path := req.URL.Path // 优先处理restful资源请求——/apis、/apis/、/apis/<group>/<version> for _, ws := range d.goRestfulContainer.RegisteredWebServices() { switch { // /apis根路径 case ws.RootPath() == "/apis": if path == "/apis" || path == "/apis/" { // 分发到webservice d.goRestfulContainer.Dispatch(w, req) return } // 子路径匹配 case strings.HasPrefix(path, ws.RootPath()): if len(path) == len(ws.RootPath()) || path[len(ws.RootPath())] == '/' { // 分发到webservice d.goRestfulContainer.Dispatch(w, req) return } } } // 无法匹配,转到nonGoRestfulMux d.nonGoRestfulMux.ServeHTTP(w, req) } // ServeHTTP makes it an http.Handler func (m *PathRecorderMux) ServeHTTP(w http.ResponseWriter, r *http.Request) { m.mux.Load().(*pathHandler).ServeHTTP(w, r) } // ServeHTTP makes it an http.Handler func (h *pathHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // 匹配pathHandler if exactHandler, ok := h.pathToHandler[r.URL.Path]; ok { exactHandler.ServeHTTP(w, r) return } // 匹配prefixHandler for _, prefixHandler := range h.prefixHandlers { if strings.HasPrefix(r.URL.Path, prefixHandler.prefix) { prefixHandler.handler.ServeHTTP(w, r) return } } // 前两者一般是aggregatorserver注册的apiHandler或apiextensionserver注册的crdHandler // 均无法匹配到,委托给下一级 h.notFoundHandler.ServeHTTP(w, r) }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
注意
1.
director外面用handlerChainBuilder包一层，结合通用的处理方法(认证/鉴权/审计)构造为handler处理链2.
director真正处理会优先匹配restful资源请求，无法找到进一步匹配非restful请求，找不到就委托给下一级的director3.
kube-apiserver三个组件发挥作用的其实就是director + aggregatorserver.handler
# 2.2.prepareRun
server.PrepareRun()主要调用genericService.PrepareRun()完成健康检查、存活检查及openAPI路由的注册工作。// PrepareRun prepares the aggregator to run. func (s *APIAggregator) PrepareRun() (preparedAPIAggregator, error) { ... // 初始化healthz/livez/readyz handler prepared := s.GenericAPIServer.PrepareRun() ... // 这里的runnable就是genericAPIServer return preparedAPIAggregator{APIAggregator: s, runnable: prepared}, nil } // PrepareRun does post API installation setup steps. It calls recursively the same function of the delegates. func (s *GenericAPIServer) PrepareRun() preparedGenericAPIServer { // 继续递归执行组件初始化,顺序为apiserver-->apiextensionserver-->noufoundserver s.delegationTarget.PrepareRun() ... // 初始化healthz handler s.installHealthz() // 初始化livez handler s.installLivez() // 初始化readyz handler readinessStopCh := s.lifecycleSignals.ShutdownInitiated.Signaled() err := s.addReadyzShutdownCheck(readinessStopCh) ... s.installReadyz() return preparedGenericAPIServer{s} }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
注意
delegationTarget就是serverChain各组件的genericServer，会依次传递到apiServer及apiextensionserver
# 2.2.runServer
prepared.Run()是kube-apiserver启动的核心模块，会运行各种钩子函数，构造及启动http server，以监听处理外部请求及实现优雅退出。func (s preparedAPIAggregator) Run(stopCh <-chan struct{}) error { // 执行的是genericAPIServer.Run() return s.runnable.Run(stopCh) } func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error { ... // http server启动入口 stoppedCh, listenerStoppedCh, err := s.NonBlockingRun(stopHttpServerCh, shutdownTimeout) // run shutdown hooks directly. func() { ... // 收到关闭信息,执行preShutdownHook钩子函数 s.RunPreShutdownHooks() }() ... // 关闭审计服务 if s.AuditBackend != nil { s.AuditBackend.Shutdown() } // wait for stoppedCh that is closed when the graceful termination (server.Shutdown) is finished. <-listenerStoppedCh <-stoppedCh return nil } // NonBlockingRun spawns the secure http server. func (s preparedGenericAPIServer) NonBlockingRun(...) (<-chan struct{}, <-chan struct{}, error) { ... // 启动服务 if s.SecureServingInfo != nil && s.Handler != nil { ... stoppedCh, listenerStoppedCh, err = s.SecureServingInfo.Serve(s.Handler, shutdownTimeout,internalStopCh) ... } ... // 执行postStartHook钩子函数 s.RunPostStartHooks(stopCh) // 通知systemd服务ready systemd.SdNotify(true, "READY=1\n") ... return stoppedCh, listenerStoppedCh, nil } // Serve runs the secure http server. func (s *SecureServingInfo) Serve(...) (<-chan struct{}, <-chan struct{}, error) { ... // 构造http server secureServer := &http.Server{ Addr: s.Listener.Addr().String(), Handler: handler, // apiServerHandler MaxHeaderBytes: 1 << 20, TLSConfig: tlsConfig, IdleTimeout: 90 * time.Second, // matches http.DefaultTransport keep-alive timeout ReadHeaderTimeout: 32 * time.Second, // just shy of requestTimeoutUpperBound } ... return RunServer(secureServer, s.Listener, shutdownTimeout, stopCh) } // RunServer spawns a go-routine continuously serving until the stopCh is closed. func RunServer(server *http.Server, ln net.Listener, shutDownTimeout time.Duration, stopCh <-chan struct{},) (<-chan struct{}, <-chan struct{}, error) { ... go func() { ... // 构造listener listener = tcpKeepAliveListener{ln} if server.TLSConfig != nil { listener = tls.NewListener(listener, server.TLSConfig) } // 启动服务 err := server.Serve(listener) ... }() return serverShutdownCh, listenerStoppedCh, nil }1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
注意
genericAPIServer启动分为两部分，一部分是自己业务相关的设置和钩子函数执行，一部分是http标准库相关设置