分布式 blog

记录技术点滴

0%

Kubernetes ApiServer 代码走读

发表于 更新于 分类于 ， ，

Kubernetes ApiServer 代码走读#

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
cmd/kube-apiserver/app/server.go:NewAPIServerCommand()-->
completedOptions, err := Complete(s)-->
	s.Etcd.WatchCacheSizes, err = serveroptions.WriteWatchCacheSizes(sizes)
Run(completedOptions, genericapiserver.SetupSignalHandler())-->CreateServerChain()-->
	CreateServerChain()-->
		CreateKubeAPIServerConfig-->
			buildGenericConfig(s.ServerRunOptions, proxyTransport)-->
				genericapiserver.NewConfig(legacyscheme.Codecs) // create codec factory for encoding/decoding
				controlplane.DefaultAPIResourceConfigSource() // group version: enabled/disabled
				storageFactoryConfig.Complete(s.Etcd)
				completedStorageFactoryConfig.New()--> // register access path in etcd for all k8s objects
					storageFactory.AddCohabitatingResources(networking.Resource("networkpolicies"), extensions.Resource("networkpolicies"))
				s.Etcd.ApplyWithStorageFactoryTo(storageFactory, genericConfig)-->
					c.AddHealthChecks()
					c.RESTOptionsGetter = &StorageFactoryRestOptionsFactory{Options: *s, StorageFactory: factory}
// 认证
				s.Authentication.ApplyTo()--> // clientcert, serviceaccount, bootstrap token, 
					authenticatorConfig.New()-->
						newWebhookTokenAuthenticator(config) // webhook
// 鉴权
				BuildAuthorizer(s, genericConfig.EgressSelector, versionedInformers)-->
					authorizationConfig.New()-->
						rbacAuthorizer := rbac.New()--> // if authorizer type is rbac
// 准入
				buildServiceResolver(s.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers)
				admissionConfig.New(proxyTransport, genericConfig.EgressSelector, serviceResolver)-->
					admission.PluginInitializer{webhookPluginInitializer, kubePluginInitializer}

			net.SplitHostPort(s.Etcd.StorageConfig.Transport.ServerList[0])
			utilwait.PollImmediate(etcdRetryInterval, etcdRetryLimit*etcdRetryInterval, preflight.EtcdConnection{ServerList: s.Etcd.StorageConfig.Transport.ServerList}.CheckEtcdServers)
			capabilities.Initialize() // allow privillage?
			config := &controlplane.Config{}
		createAPIExtensionsConfig()
		createAPIExtensionsServer()-->
			apiextensionsConfig.Complete().New(delegateAPIServer)-->
				s.AddHealthChecks(delegateCheck)
// 注册通用handler
				installAPI(s, c.Config) // register generic api handler e.g. index, profiling, metrics, flow control
		CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer)
			kubeAPIServerConfig.Complete().New(delegateAPIServer)
				m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter, legacyRESTStorageProvider)-->
					m.GenericAPIServer.AddPostStartHookOrDie(controllerName, bootstrapController.PostStartHook)-->
						controlplane.controller.Start()-->
							async.NewRunner(c.RunKubernetesNamespaces, c.RunKubernetesService, repairClusterIPs.RunUntil, repairNodePorts.RunUntil)
					m.GenericAPIServer.AddPreShutdownHookOrDie(controllerName, bootstrapController.PreShutdownHook)
// 注册core group API handler
					m.GenericAPIServer.InstallLegacyAPIGroup() // register handler for /api
					restStorageProviders := []RESTStorageProvider{appsrest.StorageProvider{}}
				m.InstallAPIs(c.ExtraConfig.APIResourceConfigSource, c.GenericConfig.RESTOptionsGetter, restStorageProviders...)-->
// 初始化对应group中对象的watch cache
					restStorageBuilder.NewRESTStorage(apiResourceConfigSource, restOptionsGetter)--> // trigger appsrest.StorageProvider
						p.v1Storage(apiResourceConfigSource, restOptionsGetter)-->
							daemonsetstore.NewREST(restOptionsGetter)-->
								store.CompleteWithOptions(options)-->
									opts, err := options.RESTOptions.GetRESTOptions(e.DefaultQualifiedResource)--> // etcd.go
										ret.Decorator = genericregistry.StorageWithCacher()-->
											cacherstorage.NewCacherFromConfig(cacherConfig)-->
												watchCache := newWatchCache()-->
// 注册API handler
					m.GenericAPIServer.InstallAPIGroups(apiGroupsInfo...)-->  // register handler for /apis
						s.installAPIResources(APIGroupPrefix, apiGroupInfo, openAPIModels)-->
							apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer)-->
								discovery.NewAPIVersionHandler(g.Serializer, g.GroupVersion, staticLister{apiResources})
		createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)-->
			apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController)
	server.PrepareRun()-->
		s.GenericAPIServer.PrepareRun()-->
			s.installHealthz()
			s.installLivez()
			s.installReadyz()
	prepared.Run(stopCh)-->
		s.runnable.Run(stopCh)--> // preparedGenericAPIServer.Run()
			s.NonBlockingRun(delayedStopCh)-->
				s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh)-->
					RunServer(secureServer, s.Listener, shutdownTimeout, stopCh)