码迷,mamicode.com
首页 > Windows程序 > 详细

kubernetes kube-apiserver启动流程分析

时间:2021-02-02 11:01:00      阅读:0      评论:0      收藏:0      [点我收藏+]

标签:注册   priority   命令行   reader   add   状态   实例化   版本   gate   

kubernetes代码版本:v1.20

看这篇文章的前提:

  • 有 golang 的基础

  • 对于 kubernetes 有基本的了解

kube-apiserver的启动过程可以分为以下几个步骤

  • 1.资源注册
  • 2.解析命令行参数
  • 3.创建 apiserver 通用配置
  • 4.创建 APIExtensionsServer
  • 5.创建 KubeAPIServer
  • 6.创建 AggregatorServer
  • 7.启动服务

并不会把每一步都仔细分析。我只挑主要的步骤进行研究一下。

1.资源注册

首先要把 apiserver 支持的资源注册到资源表中,资源注册这一过程并不是通过函数调用实现的,而是使用 golang 的 import 导包机制实现资源注册(golang 的导入机制我就不赘述了)。

整个组件的入口函数是这个文件:

k8s.io/kubernetes/cmd/kube-apiserver/apiserver.go

首先就是初始化 command 对象,执行了 app.NewAPIServerCommand() 函数。函数在 k8s.io/kubernetes/cmd/kube-apiserver/app/server.go 文件中。

可以发现在 import 的时候导入了这个包

k8s.io/kubernetes/pkg/api/legacyscheme

k8s.io/kubernetes/vendor/k8s.io/kube-aggregator/pkg/apiserver/scheme/scheme.go 文件中,在 scheme 包里面,定义了三个全局变量

var (
    // 资源注册表
	Scheme = runtime.NewScheme()
    // 编、解码器
	Codecs = serializer.NewCodecFactory(Scheme)
    // 参数编、解码器
	ParameterCodec = runtime.NewParameterCodec(Scheme)
)

这三个变量可以在 apiserver 组件的代码中任何地方使用。

kube-apiserver 启动时还导入了 controlplane 包,包中的 import_known_versions.go 文件调用了 kubernetes 支持资源的 install 包,代码如下:
k8s.io/kubernetes/pkg/controlplane/import_known_versions.go

import (
	// These imports are the API groups the API server will support.
	_ "k8s.io/kubernetes/pkg/apis/admission/install"
	_ "k8s.io/kubernetes/pkg/apis/admissionregistration/install"
	_ "k8s.io/kubernetes/pkg/apis/apiserverinternal/install"
	_ "k8s.io/kubernetes/pkg/apis/apps/install"
	_ "k8s.io/kubernetes/pkg/apis/authentication/install"
	_ "k8s.io/kubernetes/pkg/apis/authorization/install"
	_ "k8s.io/kubernetes/pkg/apis/autoscaling/install"
	_ "k8s.io/kubernetes/pkg/apis/batch/install"
	_ "k8s.io/kubernetes/pkg/apis/certificates/install"
	_ "k8s.io/kubernetes/pkg/apis/coordination/install"
	_ "k8s.io/kubernetes/pkg/apis/core/install"
	_ "k8s.io/kubernetes/pkg/apis/discovery/install"
	_ "k8s.io/kubernetes/pkg/apis/events/install"
	_ "k8s.io/kubernetes/pkg/apis/extensions/install"
	_ "k8s.io/kubernetes/pkg/apis/flowcontrol/install"
	_ "k8s.io/kubernetes/pkg/apis/imagepolicy/install"
	_ "k8s.io/kubernetes/pkg/apis/networking/install"
	_ "k8s.io/kubernetes/pkg/apis/node/install"
	_ "k8s.io/kubernetes/pkg/apis/policy/install"
	_ "k8s.io/kubernetes/pkg/apis/rbac/install"
	_ "k8s.io/kubernetes/pkg/apis/scheduling/install"
	_ "k8s.io/kubernetes/pkg/apis/storage/install"
)

可以看到里面注册的所有资源都是在 k8s.io/kubernetes/pkg/apis 文件下,都在本资源下面的 install 包里面 定义了自己的 init 方法。

以 admission 为例:

func Install(scheme *runtime.Scheme) {
    // 首先注册 admission 的内部版本
	utilruntime.Must(admission.AddToScheme(scheme))
    // 其他外部版本
	utilruntime.Must(v1beta1.AddToScheme(scheme))
	utilruntime.Must(v1.AddToScheme(scheme))
    // 版本顺序
	utilruntime.Must(scheme.SetVersionPriority(v1.SchemeGroupVersion, v1beta1.SchemeGroupVersion))
}
***
简单说一下什么叫做内部版本,什么叫做外部版本:
kubernetes 中每个资源都有自己的不同版本,例如 v1, v1/beat1, v1/alpha1 等等,当资源需要在不同版本之间转换时,我们不能为每两个不同的版本之间都写一个转换方法,如果后续有新的版本添加进来,就意味着需要为新的版本对应已经存在的每个版本都编写转换方法,复杂度会指数级上升。在kubernetes中则通过一个内部版本的设计来进行解决,内部版本是一个稳定的版本,所有的版本都只针对目标版本来进行转换的实现,而不关注其他版本。可以理解为,这个所谓的’内部版本‘包含其他所有版本的所有字段,我们在进行转换的时候先把对象转换成内部版本,然后再从内部版本转换为目标版本即可。
***

2.解析命令行参数

kubernetes 使用的是开源的命令行库 Cobra,所有组件均使用这个解析库,此库在 github 上面是开源的,有兴趣的可以自行了解一下。

我只简单叙述一下:
Cobra既是一个用来创建强大的现代CLI命令行的golang库,也是一个生成程序应用和命令行文件的程序。使用方法如下:

创建 Cmd 主命令对象,并在对象中定义各种 Run 方法(此处只是定义,并不是执行),执行顺序是 PersistentPreRun --> PerRun --> Run --> PostRun --> PersistentPostRun.然后添加命令行参数(Flag),比如我们使用的 kubectl get pod -n kube-system 后面的 -n 就是 Flag,最后执行 command 对象的 Execute 方法回调我们此前定义的各种函数。

现在,让我们在回到最初的 NewAPIServerCommand 函数中,看一看 kubernetes 的命令行解析代码(不太重要的代码我会省略):

k8s.io/kubernetes/cmd/kube-apiserver/app/server.go

func NewAPIServerCommand() *cobra.Command {
    // 初始化各个模块的默认配置
	s := options.NewServerRunOptions()
    // 生成 cmd 对象
	cmd := &cobra.Command{
		...
        // 定义方法
		RunE: func(cmd *cobra.Command, args []string) error {
			...
            // 填充成完整的参数对象
			completedOptions, err := Complete(s)
			...
            // 验证参数的合法性
			if errs := completedOptions.Validate(); len(errs) != 0 {
				return utilerrors.NewAggregate(errs)
			}
            // 将完全的参数对象传入 Run 函数。Run 里面完成了 apiserver 组件的启动逻辑,这是一个常驻进程。
			return Run(completedOptions, genericapiserver.SetupSignalHandler())
		},
        ...
	}
	...
}

顺便回到 apiserver 的入口文件中,看看 main 文件中的启动代码:

func main() {
	rand.Seed(time.Now().UnixNano())

	command := app.NewAPIServerCommand()

	logs.InitLogs()
	defer logs.FlushLogs()

	if err := command.Execute(); err != nil {
		os.Exit(1)
	}
}

就是按照我上面说的顺序定义的,kubernetes 中所有组件的启动流程大体上都是这个样子。


进入上面说的 Run 函数里面我们接着看,接下来的所有服务项的配置和创建都是在 CreateServerChain 函数中完成的。

CreateServerChain 是完成 server 初始化的方法,里面包含 APIExtensionsServer、KubeAPIServer、AggregatorServer 初始化的所有流程,最终返回 aggregatorapiserver.APIAggregator 实例,初始化流程主要有:http filter chain 的配置、API Group 的注册、http path 与 handler 的关联以及 handler 后端存储 etcd 的配置。


3.创建 apiserver 通用配置

在 CreateServerChain 函数中,基本可以从函数名猜出来每一步在干什么,通用配置的创建就是在函数 CreateKubeAPIServerConfig() 中完成的。进入到以下函数中看一下详细实现:CreateKubeAPIServerConfig() --> buildGenericConfig()
创建通用配置流程主要有以下几步:

  • 1.GenericConfig 实例化

代码如下:

// 为 genericConfig 设置默认值。
genericConfig = genericapiserver.NewConfig(legacyscheme.Codecs)
// 启动/禁止 GV 及 resource
genericConfig.MergedResourceConfig = controlplane.DefaultAPIResourceConfigSource()
...
// openAPI 规范
genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(generatedopenapi.GetOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))
...
genericConfig.Version = &kubeVersion
  • 2.StorageFactoryConfig

apiserver 组件使用 etcd 作为集群的存储,系统中所使用的所有资源、集群状态、配置等都在这上面保存。代码部分如下:

// 初始化 storageFactoryConfig 配置对象。
storageFactoryConfig := kubeapiserver.NewStorageFactoryConfig()
storageFactoryConfig.APIResourceConfig = genericConfig.MergedResourceConfig
// 初始化 etcd 相关的配置信息,补全配置对象,返回 completedStorageFactoryConfig
completedStorageFactoryConfig, err := storageFactoryConfig.Complete(s.Etcd)
if err != nil {
    lastErr = err
    return
}
// 根据上面补完的配置信息,创建 storageFactory 对象
storageFactory, lastErr = completedStorageFactoryConfig.New()
if lastErr != nil {
    return
}
  • 3.Authorizer认证、授权配置

作为整个系统的存储对象交互入口,每个系统的请求都需要经过认证、授权、准入控制器这些阶段,准入控制器下面再说,涉及到的代码如下:

genericConfig.Authorization.Authorizer, genericConfig.RuleResolver, err = BuildAuthorizer(s, genericConfig.EgressSelector, versionedInformers)

这个有必要进入到函数内部在看一看具体做了些什么,按照下面的路径点进去:

BuildAuthorizer() --> authorizationConfig.New()

New 函数主要就是在 for 循环中根据 config.AuthorizationModes 配置了 authorizers 和 ruleResolvers 两个变量,这个 config.AuthorizationModes 是在最初在初始化配置对象执行 options.NewServerRunOptions() 的时候赋值的,具体路径及代码如下:
k8s.io/kubernetes/pkg/kubeapiserver/options/authorization.go

func NewBuiltInAuthorizationOptions() *BuiltInAuthorizationOptions {
	return &BuiltInAuthorizationOptions{
        // 初始赋值就一个 "AlwaysAllow" 字符串,这是默认的配置。
		Modes:                       []string{authzmodes.ModeAlwaysAllow},
		WebhookVersion:              "v1beta1",
		WebhookCacheAuthorizedTTL:   5 * time.Minute,
		WebhookCacheUnauthorizedTTL: 30 * time.Second,
		WebhookRetryBackoff:         genericoptions.DefaultAuthWebhookRetryBackoff(),
	}
}

在回到New()中,在函数最后返回了认证和授权对象。

return union.New(authorizers...), union.NewRuleResolvers(ruleResolvers...), nil

authorizers 是已启用的认证器列表,union.New将它合并成一个认证器。

ruleResolvers 是已启用的规则解析器,union.NewRuleResolvers 也是合并了一下

可以看到默认的授权是 AlwaysAllow,具体其它类型可以在启动的时候在配置里面设置,只要配置了,就会实例化该授权对象,认证的时候会遍历每一个授权器,有一个认证成功就ok。

  • 4.Admission准入控制器配置

准入控制(Admission Control)在授权后对请求做进一步的验证或添加默认参数,在对kubernetes api服务器的请求过程中,先经过认证、授权后,执行准入操作,再对目标对象进行操作

在对集群进行请求时,每个准入控制插件都按顺序运行,只有全部插件都通过的请求才会进入系统,如果序列中的任何插件拒绝请求,则整个请求将被拒绝,并返回错误信息。

准入控制器是在初始化 ServerRunOptions 的时候 New 的,得再回到最开始的 NewAPIServerCommand() 函数中,
k8s.io/kubernetes/cmd/kube-apiserver/app/server.go

s := options.NewServerRunOptions()

函数中执行的操作就是给哥哥配置赋默认值,找到 Admission 字段,他执行的赋值函数是

kubeoptions.NewAdmissionOptions()

这个函数中主要执行了两个 RegisterAllAdmissionPlugins() 函数,来看代码:

func NewAdmissionOptions() *AdmissionOptions {
	// 在这个函数里面执行了一个 RegisterAllAdmissionPlugins 函数,进到函数里面能看到。
	options := genericoptions.NewAdmissionOptions()
	// 第二个 RegisterAllAdmissionPlugins 函数。
	RegisterAllAdmissionPlugins(options.Plugins)
	options.RecommendedPluginOrder = AllOrderedPlugins
	options.DefaultOffPlugins = DefaultOffAdmissionPlugins()

	return &AdmissionOptions{
		GenericAdmission: options,
	}
}

这两个 RegisterAllAdmissionPlugins 所执行的注册的插件不太一样,但是能看到都是执行了不同组件中的 Register() 函数,其实这个操作就是把控制插件存放进配置对象中,上面的代码最开始赋值的变量 options,是一个 AdmissionOptions 对象,

type AdmissionOptions struct {
	RecommendedPluginOrder []string
	DefaultOffPlugins sets.String

	EnablePlugins []string
	DisablePlugins []string
	ConfigFile string
	Plugins *admission.Plugins
	Decorators admission.Decorators
}

对象中的 Plugins 字段就是存放插件的,是一个 admission.Plugins 对象:

type Factory func(config io.Reader) (Interface, error)

type Plugins struct {
	// 并发保护
	lock     sync.Mutex
	// 以键值对的形式存放插件,key 就是插件的名称,value 是插件的实现函数,是上面的 Factory 对象
	registry map[string]Factory
}

可以随便点开一个插件的注册代码看一下,例如按照下面递进的进入到函数里面:
NewAdmissionOptions() --> RegisterAllAdmissionPlugins() --> admit.Register() --> plugins.Register(name, func)

// 安全的把 name 和 Factory 对象保存到 registry 字段里。
func (ps *Plugins) Register(name string, plugin Factory) {
	ps.lock.Lock()
	defer ps.lock.Unlock()
	if ps.registry != nil {
		_, found := ps.registry[name]
		if found {
			klog.Fatalf("Admission plugin %q was registered twice", name)
		}
	} else {
		ps.registry = map[string]Factory{}
	}

	klog.V(1).Infof("Registered admission plugin %q", name)
	ps.registry[name] = plugin
}

4.创建 APIExtensionsServer

5.创建 KubeAPIServer

6.创建 AggregatorServer

7.启动 服务

kubernetes kube-apiserver启动流程分析

标签:注册   priority   命令行   reader   add   状态   实例化   版本   gate   

原文地址:https://www.cnblogs.com/codenoob/p/14356269.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!