GenericAPIServer

GenericAPIServer可以理解为Kubernetes中提供API服务的结构体。定义在/pkg/genericapiserver/genericapiserver.go中：

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
// GenericAPIServer contains state for a Kubernetes cluster api server.
type GenericAPIServer struct {
	// discoveryAddresses is used to build cluster IPs for discovery.
	discoveryAddresses DiscoveryAddresses
	// LoopbackClientConfig is a config for a privileged loopback connection to the API server
	LoopbackClientConfig *restclient.Config
	// minRequestTimeout is how short the request timeout can be.  This is used to build the RESTHandler
	minRequestTimeout time.Duration
	// enableSwaggerSupport indicates that swagger should be served.  This is currently separate because
	// the API group routes are created *after* initialization and you can't generate the swagger routes until
	// after those are available.
	// TODO eventually we should be able to factor this out to take place during initialization.
	enableSwaggerSupport bool
	// legacyAPIGroupPrefixes is used to set up URL parsing for authorization and for validating requests
	// to InstallLegacyAPIGroup
	legacyAPIGroupPrefixes sets.String
	// admissionControl is used to build the RESTStorage that backs an API Group.
	admissionControl admission.Interface
	// requestContextMapper provides a way to get the context for a request.  It may be nil.
	requestContextMapper api.RequestContextMapper
	// The registered APIs
	//***含有REST中的container***//
	HandlerContainer *genericmux.APIContainer
	SecureServingInfo   *SecureServingInfo
	InsecureServingInfo *ServingInfo
	// numerical ports, set after listening
	effectiveSecurePort, effectiveInsecurePort int
	// ExternalAddress is the address (hostname or IP and port) that should be used in
	// external (public internet) URLs for this GenericAPIServer.
	ExternalAddress string
	// storage contains the RESTful endpoints exposed by this GenericAPIServer
	storage map[string]rest.Storage
	// Serializer controls how common API objects not in a group/version prefix are serialized for this server.
	// Individual APIGroups may define their own serializers.
	Serializer runtime.NegotiatedSerializer
	// "Outputs"
	Handler         http.Handler
	InsecureHandler http.Handler
	// Map storing information about all groups to be exposed in discovery response.
	// The map is from name to the group.
	apiGroupsForDiscoveryLock sync.RWMutex
	apiGroupsForDiscovery     map[string]unversioned.APIGroup
	// See Config.$name for documentation of these flags
	enableOpenAPISupport bool
	openAPIConfig        *common.Config
	// PostStartHooks are each called after the server has started listening, in a separate go func for each
	// with no guaranteee of ordering between them.  The map key is a name used for error reporting.
	// It may kill the process with a panic if it wishes to by returning an error
	postStartHookLock    sync.Mutex
	postStartHooks       map[string]postStartHookEntry
	postStartHooksCalled bool
	// healthz checks
	healthzLock    sync.Mutex
	healthzChecks  []healthz.HealthzChecker
	healthzCreated bool
}

现在来看下如何生成GenericAPIServer。GenericAPIServer通过Config的New()方法来生成，定义在/pkg/genericapiserver/config.go中：

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
//***生成GenericAPIServer***//
func (c completedConfig) New() (*GenericAPIServer, error) {
	if c.Serializer == nil {
		return nil, fmt.Errorf("Genericapiserver.New() called with config.Serializer == nil")
	}
	s := &GenericAPIServer{
		discoveryAddresses:     c.DiscoveryAddresses,
		LoopbackClientConfig:   c.LoopbackClientConfig,
		legacyAPIGroupPrefixes: c.LegacyAPIGroupPrefixes,
		admissionControl:       c.AdmissionControl,
		requestContextMapper:   c.RequestContextMapper,
		Serializer:             c.Serializer,
		minRequestTimeout:    time.Duration(c.MinRequestTimeout) * time.Second,
		enableSwaggerSupport: c.EnableSwaggerSupport,
		SecureServingInfo:   c.SecureServingInfo,
		InsecureServingInfo: c.InsecureServingInfo,
		ExternalAddress:     c.ExternalAddress,
		apiGroupsForDiscovery: map[string]unversioned.APIGroup{},
		enableOpenAPISupport: c.EnableOpenAPISupport,
		openAPIConfig:        c.OpenAPIConfig,
		postStartHooks: map[string]postStartHookEntry{},
	}
	//***生成restful的container***//
	s.HandlerContainer = mux.NewAPIContainer(http.NewServeMux(), c.Serializer)
	//***安装特殊功能的路径***//
	s.installAPI(c.Config)
	//***调用BuildHandlerChainsFunc()生成Handler***//
	//***把HandlerContainer进一步封装，以进行认证，授权等动作***//
	s.Handler, s.InsecureHandler = c.BuildHandlerChainsFunc(s.HandlerContainer.ServeMux, c.Config)
	return s, nil
}

所以现在的问题转变成如何构造Config。在/cmd/kube-apiserver/app/server.go中的Run()函数中有：

1
2
3
genericConfig := genericapiserver.NewConfig(). // create the new config
				ApplyOptions(s.GenericServerRunOptions). // apply the options selected
				Complete()                               // set default values based on the known values

其中Newconfig()，ApplyOptions()和Complete()都定义在/pkg/genericapiserver/config.go中：Newconfig()生成一个Config结构体值；ApplyOption()可以把apiserver的参数填到Config中；Complete()把Config的信息填充完整。

那么options是怎么来的呢？
来看/cmd/kube-apiserver/app/options/options.go中的NewServerRunOptions()函数：

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
//***生成apiserver的run option***//
func NewServerRunOptions() *ServerRunOptions {
	s := ServerRunOptions{
		//***生成GenericServerRunOptions***//
		GenericServerRunOptions: genericoptions.NewServerRunOptions().WithEtcdOptions(),
		EventTTL:                1 * time.Hour,
		KubeletConfig: kubeletclient.KubeletClientConfig{
			Port: ports.KubeletPort,
			PreferredAddressTypes: []string{
				string(api.NodeHostName),
				string(api.NodeInternalIP),
				string(api.NodeExternalIP),
				string(api.NodeLegacyHostIP),
			},
			EnableHttps: true,
			HTTPTimeout: time.Duration(5) * time.Second,
		},
		WebhookTokenAuthnCacheTTL: 2 * time.Minute,
	}
	return &s
}
可以看出，GenericAPIServer的options使用genericoptions.NewServerRunOptions().WithEtcdOptions()生成的，到/pkg/genericapiserver/options/server_run_options.go中看下：
``` Go
func NewServerRunOptions() *ServerRunOptions {
	return &ServerRunOptions{
		AdmissionControl: "AlwaysAdmit",
		AnonymousAuth:    false,
		//***默认为AlwaysAllow***//
		AuthorizationMode:                        "AlwaysAllow",
		AuthorizationWebhookCacheAuthorizedTTL:   5 * time.Minute,
		AuthorizationWebhookCacheUnauthorizedTTL: 30 * time.Second,
		BindAddress:                              net.ParseIP("0.0.0.0"),
		CertDirectory:                            "/var/run/kubernetes",
		DefaultStorageMediaType:                  "application/json",
		DefaultStorageVersions:                   registered.AllPreferredGroupVersions(),
		DeleteCollectionWorkers:                  1,
		EnableGarbageCollection:                  true,
		EnableProfiling:                          true,
		EnableContentionProfiling:                false,
		EnableWatchCache:                         true,
		InsecureBindAddress:                      net.ParseIP("127.0.0.1"),
		InsecurePort:                             8080,
		LongRunningRequestRE:                     DefaultLongRunningRequestRE,
		MasterCount:                              1,
		MasterServiceNamespace:                   api.NamespaceDefault,
		MaxRequestsInFlight:                      400,
		MinRequestTimeout:                        1800,
		RuntimeConfig:                            make(config.ConfigurationMap),
		SecurePort:                               6443,
		ServiceNodePortRange:                     DefaultServiceNodePortRange,
		StorageVersions:                          registered.AllPreferredGroupVersions(),
	}
}
//***设置StorageConfig***//
func (o *ServerRunOptions) WithEtcdOptions() *ServerRunOptions {
	o.StorageConfig = storagebackend.Config{
		Prefix: DefaultEtcdPathPrefix,
		// Default cache size to 0 - if unset, its size will be set based on target
		// memory usage.
		DeserializationCacheSize: 0,
	}
	return o
}

所以，经过options, Config然后才生成GenericAPIServer的。

有了GenericAPIServer，那么如何跑起来呢？来看GenericAPIServer的Run()方法，定义在/pkg/genericapiserver/genericapiserver.go中：

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
// Run spawns the http servers (secure and insecure). It only returns if stopCh is closed
// or one of the ports cannot be listened on initially.
func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) {
	//***监听安全端口***//
	if s.SecureServingInfo != nil && s.Handler != nil {
		if err := s.serveSecurely(stopCh); err != nil {
			glog.Fatal(err)
		}
	}
	//***监听非安全端口***//
	if s.InsecureServingInfo != nil && s.InsecureHandler != nil {
		if err := s.serveInsecurely(stopCh); err != nil {
			glog.Fatal(err)
		}
	}
	s.RunPostStartHooks()
	// err == systemd.SdNotifyNoSocket when not running on a systemd system
	if err := systemd.SdNotify("READY=1\n"); err != nil && err != systemd.SdNotifyNoSocket {
		glog.Errorf("Unable to send systemd daemon successful start message: %v\n", err)
	}
	//***监听stpCh channel，如果有信号，则退出***//
	<-stopCh
}

很容易理解，GenericAPIServer的Run()方法就是开始监听安全端口和非安全端口。serveSecurely()和serveInsecurely()定义在/pkg/genericapiserver/serve.go，具体分析略。

Master

Master的本质就是一个GenericAPIServer，定义在/pkg/master/master.go中：

1
2
3
type Master struct {
	GenericAPIServer *genericapiserver.GenericAPIServer
}

所以Master是对GenericAPIServer的封装。
Master也是从Config中来，这个Config定义在/pkg/master/master.go中：

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
type Config struct {
	GenericConfig *genericapiserver.Config
	StorageFactory           genericapiserver.StorageFactory
	EnableWatchCache         bool
	EnableCoreControllers    bool
	EndpointReconcilerConfig EndpointReconcilerConfig
	DeleteCollectionWorkers  int
	EventTTL                 time.Duration
	KubeletClientConfig      kubeletclient.KubeletClientConfig
	// Used to start and monitor tunneling
	Tunneler          genericapiserver.Tunneler
	EnableUISupport   bool
	EnableLogsSupport bool
	ProxyTransport    http.RoundTripper
	// Values to build the IP addresses used by discovery
	// The range of IPs to be assigned to services with type=ClusterIP or greater
	ServiceIPRange net.IPNet
	// The IP address for the GenericAPIServer service (must be inside ServiceIPRange)
	APIServerServiceIP net.IP
	// Port for the apiserver service.
	APIServerServicePort int
	// TODO, we can probably group service related items into a substruct to make it easier to configure
	// the API server items and `Extra*` fields likely fit nicely together.
	// The range of ports to be assigned to services with type=NodePort or greater
	ServiceNodePortRange utilnet.PortRange
	// Additional ports to be exposed on the GenericAPIServer service
	// extraServicePorts is injectable in the event that more ports
	// (other than the default 443/tcp) are exposed on the GenericAPIServer
	// and those ports need to be load balanced by the GenericAPIServer
	// service because this pkg is linked by out-of-tree projects
	// like openshift which want to use the GenericAPIServer but also do
	// more stuff.
	ExtraServicePorts []api.ServicePort
	// Additional ports to be exposed on the GenericAPIServer endpoints
	// Port names should align with ports defined in ExtraServicePorts
	ExtraEndpointPorts []api.EndpointPort
	// If non-zero, the "kubernetes" services uses this port as NodePort.
	// TODO(sttts): move into master
	KubernetesServiceNodePort int
	// Number of masters running; all masters must be started with the
	// same value for this field. (Numbers > 1 currently untested.)
	MasterCount int
}

可以看出，Master的Config包含GenericConfig。
Master的Config可以经过Complete()填充后成为completedConfig。completedConfig和Config一样，就是字段全部都填有值了。再来看下Master的生成函数：

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
//***从config中生成master***//
func (c completedConfig) New() (*Master, error) {
	if reflect.DeepEqual(c.KubeletClientConfig, kubeletclient.KubeletClientConfig{}) {
		return nil, fmt.Errorf("Master.New() called with empty config.KubeletClientConfig")
	}
	//***通过generic config生成GenericAPIServer***//
	s, err := c.Config.GenericConfig.SkipComplete().New() // completion is done in Complete, no need for a second time
	if err != nil {
		return nil, err
	}
	//***UI redirect功能***//
	if c.EnableUISupport {
		routes.UIRedirect{}.Install(s.HandlerContainer)
	}
	//***log相关***//
	if c.EnableLogsSupport {
		routes.Logs{}.Install(s.HandlerContainer)
	}
	m := &Master{
		GenericAPIServer: s,
	}
	//***生成restOptionsFactory***//
	restOptionsFactory := restOptionsFactory{
		deleteCollectionWorkers: c.DeleteCollectionWorkers,
		enableGarbageCollection: c.GenericConfig.EnableGarbageCollection,
		storageFactory:          c.StorageFactory,
	}
	//***赋值storageDecorator***//
	if c.EnableWatchCache {
		restOptionsFactory.storageDecorator = registry.StorageWithCacher
	} else {
		restOptionsFactory.storageDecorator = generic.UndecoratedStorage
	}
	// install legacy rest storage
	if c.GenericConfig.APIResourceConfigSource.AnyResourcesForVersionEnabled(apiv1.SchemeGroupVersion) {
		legacyRESTStorageProvider := corerest.LegacyRESTStorageProvider{
			StorageFactory:       c.StorageFactory,
			ProxyTransport:       c.ProxyTransport,
			KubeletClientConfig:  c.KubeletClientConfig,
			EventTTL:             c.EventTTL,
			ServiceIPRange:       c.ServiceIPRange,
			ServiceNodePortRange: c.ServiceNodePortRange,
			LoopbackClientConfig: c.GenericConfig.LoopbackClientConfig,
		}
		//***Fankang***//
		//***调用InstallLegacyAPI()***//
		m.InstallLegacyAPI(c.Config, restOptionsFactory.NewFor, legacyRESTStorageProvider)
	}
	//***构造restStorageProvider数组***//
	restStorageProviders := []genericapiserver.RESTStorageProvider{
		appsrest.RESTStorageProvider{},
		authenticationrest.RESTStorageProvider{Authenticator: c.GenericConfig.Authenticator},
		authorizationrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorizer},
		autoscalingrest.RESTStorageProvider{},
		batchrest.RESTStorageProvider{},
		certificatesrest.RESTStorageProvider{},
		//***Fankang***//
		//***extension包***//
		extensionsrest.RESTStorageProvider{ResourceInterface: thirdparty.NewThirdPartyResourceServer(s, c.StorageFactory)},
		policyrest.RESTStorageProvider{},
		rbacrest.RESTStorageProvider{AuthorizerRBACSuperUser: c.GenericConfig.AuthorizerRBACSuperUser},
		storagerest.RESTStorageProvider{},
	}
	//***调用InstallAPIs()***//
	m.InstallAPIs(c.Config.GenericConfig.APIResourceConfigSource, restOptionsFactory.NewFor, restStorageProviders...)
	if c.Tunneler != nil {
		m.installTunneler(c.Tunneler, coreclient.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig).Nodes())
	}
	return m, nil
}

可以看出在生成Master的同时，系统会调用InstallLegacyAPI()及InstallAPIs()来安装各种API，这就是我们下次分析要讲的内容。

kube-apiserver

之前说过Master是对GenericAPIServer的封装，再来看kube-apiserver是如何生成Master的。
先来看kube-apiserver的main()，定义在/cmd/kube-apiserver/apiserver.go中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func main() {
	rand.Seed(time.Now().UTC().UnixNano())
	s := options.NewServerRunOptions()
	s.AddFlags(pflag.CommandLine)
	flag.InitFlags()
	logs.InitLogs()
	defer logs.FlushLogs()
	verflag.PrintAndExitIfRequested()
	if err := app.Run(s); err != nil {
		fmt.Fprintf(os.Stderr, "%v\n", err)
		os.Exit(1)
	}
}

main()函数先调用options.NewServerRunOptions()生成options，然后通过调用AddFlags()把参数添加到options中，最后通过app.Run( )

再来看/cmd/kube-apiserver/app/server.go中的Run()函数：

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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
// Run runs the specified APIServer.  This should never exit.
func Run(s *options.ServerRunOptions) error {
	//***Fankang***//
	//***验证ETCD Server的配置，只是看ETCD列表是否为0而已***//
	genericvalidation.VerifyEtcdServersList(s.GenericServerRunOptions)
	genericapiserver.DefaultAndValidateRunOptions(s.GenericServerRunOptions)
	genericConfig := genericapiserver.NewConfig(). // create the new config
							ApplyOptions(s.GenericServerRunOptions). // apply the options selected
							Complete()                               // set default values based on the known values
	serviceIPRange, apiServerServiceIP, err := genericapiserver.DefaultServiceIPRange(s.GenericServerRunOptions.ServiceClusterIPRange)
	if err != nil {
		glog.Fatalf("Error determining service IP ranges: %v", err)
	}
	if err := genericConfig.MaybeGenerateServingCerts(apiServerServiceIP); err != nil {
		glog.Fatalf("Failed to generate service certificate: %v", err)
	}
	capabilities.Initialize(capabilities.Capabilities{
		AllowPrivileged: s.AllowPrivileged,
		// TODO(vmarmol): Implement support for HostNetworkSources.
		PrivilegedSources: capabilities.PrivilegedSources{
			HostNetworkSources: []string{},
			HostPIDSources:     []string{},
			HostIPCSources:     []string{},
		},
		PerConnectionBandwidthLimitBytesPerSec: s.MaxConnectionBytesPerSec,
	})
	// Setup tunneler if needed
	var tunneler genericapiserver.Tunneler
	var proxyDialerFn apiserver.ProxyDialerFunc
	if len(s.SSHUser) > 0 {
		// Get ssh key distribution func, if supported
		var installSSH genericapiserver.InstallSSHKey
		cloud, err := cloudprovider.InitCloudProvider(s.GenericServerRunOptions.CloudProvider, s.GenericServerRunOptions.CloudConfigFile)
		if err != nil {
			glog.Fatalf("Cloud provider could not be initialized: %v", err)
		}
		if cloud != nil {
			if instances, supported := cloud.Instances(); supported {
				installSSH = instances.AddSSHKeyToAllInstances
			}
		}
		if s.KubeletConfig.Port == 0 {
			glog.Fatalf("Must enable kubelet port if proxy ssh-tunneling is specified.")
		}
		// Set up the tunneler
		// TODO(cjcullen): If we want this to handle per-kubelet ports or other
		// kubelet listen-addresses, we need to plumb through options.
		healthCheckPath := &url.URL{
			Scheme: "https",
			Host:   net.JoinHostPort("127.0.0.1", strconv.FormatUint(uint64(s.KubeletConfig.Port), 10)),
			Path:   "healthz",
		}
		tunneler = genericapiserver.NewSSHTunneler(s.SSHUser, s.SSHKeyfile, healthCheckPath, installSSH)
		// Use the tunneler's dialer to connect to the kubelet
		s.KubeletConfig.Dial = tunneler.Dial
		// Use the tunneler's dialer when proxying to pods, services, and nodes
		proxyDialerFn = tunneler.Dial
	}
	// Proxying to pods and services is IP-based... don't expect to be able to verify the hostname
	proxyTLSClientConfig := &tls.Config{InsecureSkipVerify: true}
	if s.GenericServerRunOptions.StorageConfig.DeserializationCacheSize == 0 {
		// When size of cache is not explicitly set, estimate its size based on
		// target memory usage.
		glog.V(2).Infof("Initalizing deserialization cache size based on %dMB limit", s.GenericServerRunOptions.TargetRAMMB)
		// This is the heuristics that from memory capacity is trying to infer
		// the maximum number of nodes in the cluster and set cache sizes based
		// on that value.
		// From our documentation, we officially recomment 120GB machines for
		// 2000 nodes, and we scale from that point. Thus we assume ~60MB of
		// capacity per node.
		// TODO: We may consider deciding that some percentage of memory will
		// be used for the deserialization cache and divide it by the max object
		// size to compute its size. We may even go further and measure
		// collective sizes of the objects in the cache.
		clusterSize := s.GenericServerRunOptions.TargetRAMMB / 60
		s.GenericServerRunOptions.StorageConfig.DeserializationCacheSize = 25 * clusterSize
		if s.GenericServerRunOptions.StorageConfig.DeserializationCacheSize < 1000 {
			s.GenericServerRunOptions.StorageConfig.DeserializationCacheSize = 1000
		}
	}
	storageGroupsToEncodingVersion, err := s.GenericServerRunOptions.StorageGroupsToEncodingVersion()
	if err != nil {
		glog.Fatalf("error generating storage version map: %s", err)
	}
	//***生成DefaultStorageFactory***//
	storageFactory, err := genericapiserver.BuildDefaultStorageFactory(
		s.GenericServerRunOptions.StorageConfig, s.GenericServerRunOptions.DefaultStorageMediaType, api.Codecs,
		genericapiserver.NewDefaultResourceEncodingConfig(), storageGroupsToEncodingVersion,
		// FIXME: this GroupVersionResource override should be configurable
		[]unversioned.GroupVersionResource{batch.Resource("cronjobs").WithVersion("v2alpha1")},
		master.DefaultAPIResourceConfigSource(), s.GenericServerRunOptions.RuntimeConfig)
	if err != nil {
		glog.Fatalf("error in initializing storage factory: %s", err)
	}
	storageFactory.AddCohabitatingResources(batch.Resource("jobs"), extensions.Resource("jobs"))
	storageFactory.AddCohabitatingResources(autoscaling.Resource("horizontalpodautoscalers"), extensions.Resource("horizontalpodautoscalers"))
	for _, override := range s.GenericServerRunOptions.EtcdServersOverrides {
		tokens := strings.Split(override, "#")
		if len(tokens) != 2 {
			glog.Errorf("invalid value of etcd server overrides: %s", override)
			continue
		}
		apiresource := strings.Split(tokens[0], "/")
		if len(apiresource) != 2 {
			glog.Errorf("invalid resource definition: %s", tokens[0])
			continue
		}
		group := apiresource[0]
		resource := apiresource[1]
		groupResource := unversioned.GroupResource{Group: group, Resource: resource}
		servers := strings.Split(tokens[1], ";")
		storageFactory.SetEtcdLocation(groupResource, servers)
	}
	// Default to the private server key for service account token signing
	if len(s.ServiceAccountKeyFiles) == 0 && s.GenericServerRunOptions.TLSPrivateKeyFile != "" {
		if authenticator.IsValidServiceAccountKeyFile(s.GenericServerRunOptions.TLSPrivateKeyFile) {
			s.ServiceAccountKeyFiles = []string{s.GenericServerRunOptions.TLSPrivateKeyFile}
		} else {
			glog.Warning("No TLS key provided, service account token authentication disabled")
		}
	}
	var serviceAccountGetter serviceaccount.ServiceAccountTokenGetter
	if s.ServiceAccountLookup {
		// If we need to look up service accounts and tokens,
		// go directly to etcd to avoid recursive auth insanity
		storageConfig, err := storageFactory.NewConfig(api.Resource("serviceaccounts"))
		if err != nil {
			glog.Fatalf("Unable to get serviceaccounts storage: %v", err)
		}
		serviceAccountGetter = serviceaccountcontroller.NewGetterFromStorageInterface(storageConfig, storageFactory.ResourcePrefix(api.Resource("serviceaccounts")), storageFactory.ResourcePrefix(api.Resource("secrets")))
	}
	//***认证器***//
	apiAuthenticator, securityDefinitions, err := authenticator.New(authenticator.AuthenticatorConfig{
		Anonymous:                   s.GenericServerRunOptions.AnonymousAuth,
		AnyToken:                    s.GenericServerRunOptions.EnableAnyToken,
		BasicAuthFile:               s.GenericServerRunOptions.BasicAuthFile,
		ClientCAFile:                s.GenericServerRunOptions.ClientCAFile,
		TokenAuthFile:               s.GenericServerRunOptions.TokenAuthFile,
		OIDCIssuerURL:               s.GenericServerRunOptions.OIDCIssuerURL,
		OIDCClientID:                s.GenericServerRunOptions.OIDCClientID,
		OIDCCAFile:                  s.GenericServerRunOptions.OIDCCAFile,
		OIDCUsernameClaim:           s.GenericServerRunOptions.OIDCUsernameClaim,
		OIDCGroupsClaim:             s.GenericServerRunOptions.OIDCGroupsClaim,
		ServiceAccountKeyFiles:      s.ServiceAccountKeyFiles,
		ServiceAccountLookup:        s.ServiceAccountLookup,
		ServiceAccountTokenGetter:   serviceAccountGetter,
		KeystoneURL:                 s.GenericServerRunOptions.KeystoneURL,
		KeystoneCAFile:              s.GenericServerRunOptions.KeystoneCAFile,
		WebhookTokenAuthnConfigFile: s.WebhookTokenAuthnConfigFile,
		WebhookTokenAuthnCacheTTL:   s.WebhookTokenAuthnCacheTTL,
		RequestHeaderConfig:         s.GenericServerRunOptions.AuthenticationRequestHeaderConfig(),
	})
	if err != nil {
		glog.Fatalf("Invalid Authentication Config: %v", err)
	}
	privilegedLoopbackToken := uuid.NewRandom().String()
	selfClientConfig, err := s.GenericServerRunOptions.NewSelfClientConfig(privilegedLoopbackToken)
	if err != nil {
		glog.Fatalf("Failed to create clientset: %v", err)
	}
	client, err := s.GenericServerRunOptions.NewSelfClient(privilegedLoopbackToken)
	if err != nil {
		glog.Errorf("Failed to create clientset: %v", err)
	}
	sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)
	//***授权***//
	authorizationConfig := authorizer.AuthorizationConfig{
		PolicyFile:                  s.GenericServerRunOptions.AuthorizationPolicyFile,
		WebhookConfigFile:           s.GenericServerRunOptions.AuthorizationWebhookConfigFile,
		WebhookCacheAuthorizedTTL:   s.GenericServerRunOptions.AuthorizationWebhookCacheAuthorizedTTL,
		WebhookCacheUnauthorizedTTL: s.GenericServerRunOptions.AuthorizationWebhookCacheUnauthorizedTTL,
		RBACSuperUser:               s.GenericServerRunOptions.AuthorizationRBACSuperUser,
		InformerFactory:             sharedInformers,
	}
	authorizationModeNames := strings.Split(s.GenericServerRunOptions.AuthorizationMode, ",")
	apiAuthorizer, err := authorizer.NewAuthorizerFromAuthorizationConfig(authorizationModeNames, authorizationConfig)
	if err != nil {
		glog.Fatalf("Invalid Authorization Config: %v", err)
	}
	//***admission controller***//
	admissionControlPluginNames := strings.Split(s.GenericServerRunOptions.AdmissionControl, ",")
	// TODO(dims): We probably need to add an option "EnableLoopbackToken"
	if apiAuthenticator != nil {
		var uid = uuid.NewRandom().String()
		tokens := make(map[string]*user.DefaultInfo)
		tokens[privilegedLoopbackToken] = &user.DefaultInfo{
			Name:   user.APIServerUser,
			UID:    uid,
			Groups: []string{user.SystemPrivilegedGroup},
		}
		tokenAuthenticator := authenticator.NewAuthenticatorFromTokens(tokens)
		apiAuthenticator = authenticatorunion.New(tokenAuthenticator, apiAuthenticator)
		tokenAuthorizer := authorizer.NewPrivilegedGroups(user.SystemPrivilegedGroup)
		apiAuthorizer = authorizerunion.New(tokenAuthorizer, apiAuthorizer)
	}
	pluginInitializer := admission.NewPluginInitializer(sharedInformers, apiAuthorizer)
	admissionController, err := admission.NewFromPlugins(client, admissionControlPluginNames, s.GenericServerRunOptions.AdmissionControlConfigFile, pluginInitializer)
	if err != nil {
		glog.Fatalf("Failed to initialize plugins: %v", err)
	}
	proxyTransport := utilnet.SetTransportDefaults(&http.Transport{
		Dial:            proxyDialerFn,
		TLSClientConfig: proxyTLSClientConfig,
	})
	kubeVersion := version.Get()
	genericConfig.Version = &kubeVersion
	genericConfig.LoopbackClientConfig = selfClientConfig
	genericConfig.Authenticator = apiAuthenticator
	genericConfig.Authorizer = apiAuthorizer
	genericConfig.AdmissionControl = admissionController
	genericConfig.APIResourceConfigSource = storageFactory.APIResourceConfigSource
	genericConfig.OpenAPIConfig.Info.Title = "Kubernetes"
	genericConfig.OpenAPIConfig.Definitions = generatedopenapi.OpenAPIDefinitions
	genericConfig.EnableOpenAPISupport = true
	genericConfig.EnableMetrics = true
	genericConfig.OpenAPIConfig.SecurityDefinitions = securityDefinitions
	//***master的config***//
	config := &master.Config{
		GenericConfig: genericConfig.Config,
		StorageFactory:          storageFactory,
		EnableWatchCache:        s.GenericServerRunOptions.EnableWatchCache,
		EnableCoreControllers:   true,
		DeleteCollectionWorkers: s.GenericServerRunOptions.DeleteCollectionWorkers,
		EventTTL:                s.EventTTL,
		KubeletClientConfig:     s.KubeletConfig,
		EnableUISupport:         true,
		EnableLogsSupport:       true,
		ProxyTransport:          proxyTransport,
		Tunneler: tunneler,
		ServiceIPRange:       serviceIPRange,
		APIServerServiceIP:   apiServerServiceIP,
		APIServerServicePort: 443,
		ServiceNodePortRange:      s.GenericServerRunOptions.ServiceNodePortRange,
		KubernetesServiceNodePort: s.GenericServerRunOptions.KubernetesServiceNodePort,
		MasterCount: s.GenericServerRunOptions.MasterCount,
	}
	if s.GenericServerRunOptions.EnableWatchCache {
		glog.V(2).Infof("Initalizing cache sizes based on %dMB limit", s.GenericServerRunOptions.TargetRAMMB)
		cachesize.InitializeWatchCacheSizes(s.GenericServerRunOptions.TargetRAMMB)
		cachesize.SetWatchCacheSizes(s.GenericServerRunOptions.WatchCacheSizes)
	}
	//***生成master***//
	m, err := config.Complete().New()
	if err != nil {
		return err
	}
	sharedInformers.Start(wait.NeverStop)
	//***启动master***//
	m.GenericAPIServer.PrepareRun().Run(wait.NeverStop)
	return nil
}

Run()的流程如下：

  • 生成genericConfig:

    1
    2
    3
    genericConfig := genericapiserver.NewConfig(). // create the new config
    							ApplyOptions(s.GenericServerRunOptions). // apply the options selected
    							Complete()                               // set default values based on the known values

  • 把apiAuthenticator, apiAuthorizer待回填到genericConfig:

    1
    2
    genericConfig.Authenticator = apiAuthenticator
    genericConfig.Authorizer = apiAuthorizer

  • 生成master Config:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    config := &master.Config{
    		GenericConfig: genericConfig.Config,
    		StorageFactory:          storageFactory,
    		EnableWatchCache:        s.GenericServerRunOptions.EnableWatchCache,
    		EnableCoreControllers:   true,
    		DeleteCollectionWorkers: s.GenericServerRunOptions.DeleteCollectionWorkers,
    		EventTTL:                s.EventTTL,
    		KubeletClientConfig:     s.KubeletConfig,
    		EnableUISupport:         true,
    		EnableLogsSupport:       true,
    		ProxyTransport:          proxyTransport,
    		Tunneler: tunneler,
    		ServiceIPRange:       serviceIPRange,
    		APIServerServiceIP:   apiServerServiceIP,
    		APIServerServicePort: 443,
    		ServiceNodePortRange:      s.GenericServerRunOptions.ServiceNodePortRange,
    		KubernetesServiceNodePort: s.GenericServerRunOptions.KubernetesServiceNodePort,
    		MasterCount: s.GenericServerRunOptions.MasterCount,
    	}

  • 生成master:

    1
    2
    3
    4
    5
    //***生成master***//
    m, err := config.Complete().New()
    if err != nil {
    	return err
    }

  • 启动master

    1
    m.GenericAPIServer.PrepareRun().Run(wait.NeverStop)