k8s apiserver启动执行流程之aggregatorServer

启动执行流程之aggregatorServer

本文主要分析kubernetes在启动kube-apiserver时，创建聚合服务的过程。总览中概括性标记了流程。本节会详细分析，主要有以下几点：

1. createAPIExtensionsConfig
2. createAPIExtensionsServer

createAPIExtensionsConfig

构建apiextensionsapiserver.Config扩展配置 – 其实就是包装了通用apiserver配置和其他额外的配置

• webhook.NewDefaultAuthenticationInfoResolverWrapper
  这里不做具体分析，后续会渗入分析apiserver服务。目前我们只需要大致知道这是干什么的
  1. 构建一个认证解析器包装器，参数proxyTransport和参数egressSelector只能使用其中一个
  2. 参数proxyTransport： 表示传输器，用来获取拨号函数
  3. 参数egressSelector： 表示流量出选择器，用来获取拨号函数
  4. 参数kubeapiserverClientConfig： 表示apiserver restclient配置，表示没有经过代理proxy和egress的配置
  5. 参数tp： 用于rest请求的trace
• apiextensions.go
  • createAPIExtensionsConfig

 // 构建apiextensionsapiserver.Config扩展配置 -- 其实就是包装了通用apiserver配置和其他额外的配置
func createAPIExtensionsConfig(
   kubeAPIServerConfig genericapiserver.Config,
   externalInformers kubeexternalinformers.SharedInformerFactory,
   pluginInitializers []admission.PluginInitializer,
   commandOptions *options.ServerRunOptions,
   masterCount int,
   serviceResolver webhook.ServiceResolver,
   authResolverWrapper webhook.AuthenticationInfoResolverWrapper,
) (*apiextensionsapiserver.Config, error) {
   // 做一个浅拷贝让我们更改一些配置，大多数配置实际上保持不变。我们只需要处理一些与 apiextensions 的细节相关的配置
   genericConfig := kubeAPIServerConfig
   genericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{}
   genericConfig.RESTOptionsGetter = nil

   // 用 apiextensions 的方案覆盖 genericConfig.AdmissionControl，因为 apiextensions apiserver 应该使用自己的方案来转换资源。
   err := commandOptions.Admission.ApplyTo(
   	&genericConfig,
   	externalInformers,
   	genericConfig.LoopbackClientConfig,
   	feature.DefaultFeatureGate,
   	pluginInitializers...)
   if err != nil {
   	return nil, err
   }

   // 使用备份（不能直接修改原始数据）的Etcd选项来做一些修改
   etcdOptions := *commandOptions.Etcd
   // 获取是否支持分页
   etcdOptions.StorageConfig.Paging = utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
   // 构建编解码器  --  解码为v1beta或者v1，编码为internal
   etcdOptions.StorageConfig.Codec = apiextensionsapiserver.Codecs.LegacyCodec(v1beta1.SchemeGroupVersion, v1.SchemeGroupVersion)

   // 更喜欢更紧凑的序列化（v1beta1）进行存储，直到 http://issue.k8s.io/82292 解决了 v1 序列化太大但可以存储 v1beta1 序列化的对象的问题
   // 根据传入参数和对应方法逻辑，其实采用的是v1beta1版本
   etcdOptions.StorageConfig.EncodeVersioner = runtime.NewMultiGroupVersioner(v1beta1.SchemeGroupVersion, schema.GroupKind{Group: v1beta1.GroupName})
   // 构建一个生成etcdOptions的工厂
   genericConfig.RESTOptionsGetter = &genericoptions.SimpleRestOptionsFactory{Options: etcdOptions}

   // 使用 apiextensions 默认值和注册表覆盖 MergedResourceConfig
   if err := commandOptions.APIEnablement.ApplyTo(
   	&genericConfig,
   	apiextensionsapiserver.DefaultAPIResourceConfigSource(),
   	apiextensionsapiserver.Scheme); err != nil {
   	return nil, err
   }

   apiextensionsConfig := &apiextensionsapiserver.Config{
   	GenericConfig: &genericapiserver.RecommendedConfig{
   		Config:                genericConfig,
   		SharedInformerFactory: externalInformers,
   	},
   	ExtraConfig: apiextensionsapiserver.ExtraConfig{
   		CRDRESTOptionsGetter: apiextensionsoptions.NewCRDRESTOptionsGetter(etcdOptions),
   		MasterCount:          masterCount,
   		AuthResolverWrapper:  authResolverWrapper,
   		ServiceResolver:      serviceResolver,
   	},
   }

   // 因为在之前的 CreateKubeAPIServerConfig 函数已经执行过了AddPostStartHook，所以我们需要清除 poststarthooks，这样我们就不会将它们多次（失败时）添加到所有服务器.
   apiextensionsConfig.GenericConfig.PostStartHooks = map[string]genericapiserver.PostStartHookConfigEntry{}

   return apiextensionsConfig, 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

• createAPIExtensionsConfig细分某些重点方法
  • commandOptions.Admission.ApplyTo
    用 apiextensions 的方案覆盖 genericConfig.AdmissionControl，因为 apiextensions apiserver 应该使用自己的方案来转换资源。

  // 将准入链选项添加到服务器配置中
  func (a *AdmissionOptions) ApplyTo(
   c *server.Config,
   informers informers.SharedInformerFactory,
   kubeAPIServerClientConfig *rest.Config,
   features featuregate.FeatureGate,
   pluginInitializers ...admission.PluginInitializer,
   ) error {
   // 表示没有需要配置的准入链选项
   if a == nil {
   	return nil
   }
  
   // 准入插件不为空
   if a.PluginNames != nil {
   	// 用来设置启用的准入插件和禁用的准入插件
   	a.GenericAdmission.EnablePlugins, a.GenericAdmission.DisablePlugins = computePluginNames(a.PluginNames, a.GenericAdmission.RecommendedPluginOrder)
   }
  
   return a.GenericAdmission.ApplyTo(c, informers, kubeAPIServerClientConfig, features, pluginInitializers...)
  }
  // 明确禁用所有不在启用列表中的插件
  func computePluginNames(explicitlyEnabled []string, all []string) (enabled []string, disabled []string) {
   return explicitlyEnabled, sets.NewString(all...).Difference(sets.NewString(explicitlyEnabled...)).List()
  }
  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

createAPIExtensionsServer

1. 完成apiextensionsConfig的完全配置
2. 船舰一个apiserver（暴露group为"apiextensions.k8s.io"的api，支持crd等操作）

// 这里不做深入分析，后续apiserver服务会做具体分析
func createAPIExtensionsServer(apiextensionsConfig *apiextensionsapiserver.Config, delegateAPIServer genericapiserver.DelegationTarget) (*apiextensionsapiserver.CustomResourceDefinitions, error) {
   return apiextensionsConfig.Complete().New(delegateAPIServer)
}
1
2
3
4

createAggregatorConfig

创建聚合服务需要的配置
1

// 1.完成apiextensionsConfig的完全配置 2.船舰一个apiserver（暴露group为"apiextensions.k8s.io"的api，支持crd等操作）
// 这里不做深入分析，后续apiserver服务会做具体分析
func createAPIExtensionsServer(apiextensionsConfig *apiextensionsapiserver.Config, delegateAPIServer genericapiserver.DelegationTarget) (*apiextensionsapiserver.CustomResourceDefinitions, error) {
   return apiextensionsConfig.Complete().New(delegateAPIServer)
}
1
2
3
4
5