admission机制分析-v1.5.2

什么是admission

在kube-apiserver的参数中，有如下参数：
--admission-control=NamespaceLifecycle,NamespaceExists,LimitRanger,ResourceQuota
这个参数表示的就是admission。来看下官方文档对admission的定义：”An admission control plug-in is a piece of code that intercepts requests to the Kubernetes API server prior to persistence of the object, but after the request is authenticated and authorized.”，即admission的plugin会在认证和授权后对请求进行检查。admission会使用每一个plugin对请求进行检查，一旦有一个不通过，则拒绝该请求。目前admission主要有如下plugin:

• AlwaysAdmit: 允许所有的请求通过；
• AlwaysDeny: 拒绝所有的请求；
• DenyEscalatingExec: 拒绝执行”系统命令”的请求；
• ImagePolicyWebhook: 镜像下载相关，没有进行详细研究；
• ServiceAccount: ServiceAccount会在生成的容器中注入安全相关的信息以访问Kubernetes集群；
• SecurityContextDeny: PodSecurityPolicy相关，PodSecurityPolicy可以对容器进行各种限制；
• ResourceQuota: ResourceQuota可以对用户命名空间下的配额进行检查；
• LimitRanger: 对Pods申请的资源的大小进行限制；
• NamespaceLifecycle: NamespaceLifecycle会拒绝在正在删除中的Namespace下创建资源的请求。

所以，本次分析将着重于admission框架的实现。

注册

来看下plugin是如何注册的，在/cmd/kube-apiserver/app/plugins.go中有：

import (
	// Cloud providers
	_ "k8s.io/kubernetes/pkg/cloudprovider/providers"

	// Admission policies
	_ "k8s.io/kubernetes/plugin/pkg/admission/admit"
	_ "k8s.io/kubernetes/plugin/pkg/admission/alwayspullimages"
	_ "k8s.io/kubernetes/plugin/pkg/admission/antiaffinity"
	_ "k8s.io/kubernetes/plugin/pkg/admission/deny"
	_ "k8s.io/kubernetes/plugin/pkg/admission/exec"
	_ "k8s.io/kubernetes/plugin/pkg/admission/gc"
	_ "k8s.io/kubernetes/plugin/pkg/admission/imagepolicy"
	_ "k8s.io/kubernetes/plugin/pkg/admission/initialresources"
	_ "k8s.io/kubernetes/plugin/pkg/admission/limitranger"
	_ "k8s.io/kubernetes/plugin/pkg/admission/namespace/autoprovision"
	_ "k8s.io/kubernetes/plugin/pkg/admission/namespace/exists"
	_ "k8s.io/kubernetes/plugin/pkg/admission/namespace/lifecycle"
	_ "k8s.io/kubernetes/plugin/pkg/admission/persistentvolume/label"
	_ "k8s.io/kubernetes/plugin/pkg/admission/podnodeselector"
	_ "k8s.io/kubernetes/plugin/pkg/admission/resourcequota"
	_ "k8s.io/kubernetes/plugin/pkg/admission/security/podsecuritypolicy"
	_ "k8s.io/kubernetes/plugin/pkg/admission/securitycontext/scdeny"
	_ "k8s.io/kubernetes/plugin/pkg/admission/serviceaccount"
	_ "k8s.io/kubernetes/plugin/pkg/admission/storageclass/default"
)

Go语言在引入包时会自动执行包的init()方法，以resourcequota为例来看下init()方法，定义在/plugin/pkg/admission/resourcequota/admission.go中：

func init() {
	//***向admission中的plugins中注册resourcequota的创建函数***//
	admission.RegisterPlugin("ResourceQuota",
		func(client clientset.Interface, config io.Reader) (admission.Interface, error) {
			// NOTE: we do not provide informers to the registry because admission level decisions
			// does not require us to open watches for all items tracked by quota.
			registry := install.NewRegistry(client, nil)
			return NewResourceQuota(client, registry, 5, make(chan struct{}))
		})
}

func NewResourceQuota(client clientset.Interface, registry quota.Registry, numEvaluators int, stopCh <-chan struct{}) (admission.Interface, error) {
	quotaAccessor, err := newQuotaAccessor(client)
	if err != nil {
		return nil, err
	}
	go quotaAccessor.Run(stopCh)

	evaluator := NewQuotaEvaluator(quotaAccessor, registry, nil, numEvaluators, stopCh)

	return &quotaAdmission{
		Handler:   admission.NewHandler(admission.Create, admission.Update),
		evaluator: evaluator,
	}, nil
}

其中，还需要注意的是，每个plugin都只针对一定的操作进行检查，如resourcequota只会检查Create和Update的请求。操作的管理通过Handler进行。

所以，plugin初始化的过程就是向admission的plugins变量中注册plugin的名称和初始化函数。现在，只需要记着admission的plugins注册着plugin名称及其初始始化函数即可。

plugins

我们来看admission的plugins，定义在/pkg/admission/plugins.go中：

// All registered admission options.
var (
	pluginsMutex sync.Mutex
	//***所有的plugin创建函数都会注册在plugins中***//
	plugins = make(map[string]Factory)

	// PluginEnabledFn checks whether a plugin is enabled.  By default, if you ask about it, it's enabled.
	PluginEnabledFn = func(name string, config io.Reader) bool {
		return true
	}
)

plugins.go中还定义有如下函数：

GetPlugins()

GetPlugins()中可以返回在plugins中注册的所有plugin的名称。

// GetPlugins enumerates the names of all registered plugins.
func GetPlugins() []string {
	pluginsMutex.Lock()
	defer pluginsMutex.Unlock()
	keys := []string{}
	for k := range plugins {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

RegisterPlugin()

RegisterPlugin()提供了向plugins注册的方法，即plugin通过调用RegisterPlugins()向plugins注册。

func RegisterPlugin(name string, plugin Factory) {
	pluginsMutex.Lock()
	defer pluginsMutex.Unlock()
	_, found := plugins[name]
	if found {
		glog.Fatalf("Admission plugin %q was registered twice", name)
	}
	glog.V(1).Infof("Registered admission plugin %q", name)
	plugins[name] = plugin
}

getPlugin()

getPlugin()可以从plugins中获取指定plugin的初始化函数，并创建该plugin。

func getPlugin(name string, client clientset.Interface, config io.Reader) (Interface, bool, error) {
	pluginsMutex.Lock()
	defer pluginsMutex.Unlock()
	f, found := plugins[name]
	if !found {
		return nil, false, nil
	}

	config1, config2, err := splitStream(config)
	if err != nil {
		return nil, true, err
	}
	if !PluginEnabledFn(name, config1) {
		return nil, true, nil
	}

	ret, err := f(client, config2)
	return ret, true, err
}

InitPlugin()

InitPlugin()通过getPlugin()获取plugin，并返回。

//***创建指定plugin***//
func InitPlugin(name string, client clientset.Interface, configFilePath string) Interface {
	var (
		config *os.File
		err    error
	)

	if name == "" {
		glog.Info("No admission plugin specified.")
		return nil
	}

	if configFilePath != "" {
		config, err = os.Open(configFilePath)
		if err != nil {
			glog.Fatalf("Couldn't open admission plugin configuration %s: %#v",
				configFilePath, err)
		}

		defer config.Close()
	}

	plugin, found, err := getPlugin(name, client, config)
	if err != nil {
		glog.Fatalf("Couldn't init admission plugin %q: %v", name, err)
	}
	if !found {
		glog.Fatalf("Unknown admission plugin: %s", name)
	}

	return plugin
}

handler

handler可以判断该plugin是否需要处理某请求。admission的handler定义在/pkg/admission/handler.go中：

type Handler struct {
	operations sets.String
	readyFunc  ReadyFunc
}

其中：

• operations: 为操作的集合；
• readyFunc: 标识该plugin是否做好处理请求的准备。

Handles()

Handles()可以判断请求的动作是否在plugin的处理动作列表中。

func (h *Handler) Handles(operation Operation) bool {
	return h.operations.Has(string(operation))
}

NewHandler()

NewHandler()可以生成一个新的handler，但只能设置operations。

func NewHandler(ops ...Operation) *Handler {
	operations := sets.NewString()
	for _, op := range ops {
		operations.Insert(string(op))
	}
	return &Handler{
		operations: operations,
	}
}

SetReadyFunc()

SetReadyFunc()可以设置handler的readyFunc字段。

func (h *Handler) SetReadyFunc(readyFunc ReadyFunc) {
	h.readyFunc = readyFunc
}

WaitForReady()

WaitForReady()可以判断该plugin是否已经完成初始化。

func (h *Handler) WaitForReady() bool {
	// there is no ready func configured, so we return immediately
	if h.readyFunc == nil {
		return true
	}
	return h.waitForReadyInternal(time.After(timeToWaitForReady))
}

func (h *Handler) waitForReadyInternal(timeout <-chan time.Time) bool {
	// there is no configured ready func, so return immediately
	if h.readyFunc == nil {
		return true
	}
	for !h.readyFunc() {
		select {
		case <-time.After(100 * time.Millisecond):
		case <-timeout:
			return h.readyFunc()
		}
	}
	return true
}

chainAdmissionHandler

chainAdmissionHandler定义在/pkg/admission/chain.go中，是admission的入口：

// chainAdmissionHandler is an instance of admission.Interface that performs admission control using a chain of admission handlers
type chainAdmissionHandler []Interface

NewFromPlugins()

NewFromPlugins()可以依据plugins生成chainAdmissionHandler。

//***把admission中的plugin加入到chainAdmissionHandler中***//
func NewFromPlugins(client clientset.Interface, pluginNames []string, configFilePath string, plugInit PluginInitializer) (Interface, error) {
	plugins := []Interface{}
	for _, pluginName := range pluginNames {
		plugin := InitPlugin(pluginName, client, configFilePath)
		if plugin != nil {
			plugins = append(plugins, plugin)
		}
	}
	plugInit.Initialize(plugins)
	// ensure that plugins have been properly initialized
	if err := Validate(plugins); err != nil {
		return nil, err
	}
	//***此处是类型转换***//
	return chainAdmissionHandler(plugins), nil
}

Admit()

Admit()使用每个plugin对请求进行检查，如果某plugin不能处理该请求的动作，则直接略过。每个plugin都实现有Admit()方法。

//***查看请求是否符合每一个admission***//
func (admissionHandler chainAdmissionHandler) Admit(a Attributes) error {
	for _, handler := range admissionHandler {
		if !handler.Handles(a.GetOperation()) {
			continue
		}
		err := handler.Admit(a)
		if err != nil {
			return err
		}
	}
	return nil
}

Handles()

Handles()判断chainAdmissionHandler中是否存在plugin可以处理该请求。

// Handles will return true if any of the handlers handles the given operation
func (admissionHandler chainAdmissionHandler) Handles(operation Operation) bool {
	for _, handler := range admissionHandler {
		if handler.Handles(operation) {
			return true
		}
	}
	return false
}

调用

Kubernetes在/cmd/kube-apiserver/app/server.go中生成admisionController:

admissionController, err := admission.NewFromPlugins(client, admissionControlPluginNames, s.GenericServerRunOptions.AdmissionControlConfigFile, pluginInitializer)
if err != nil {
	glog.Fatalf("Failed to initialize plugins: %v", err)
}

生成admissionController之后，在/pkg/admission/chain.go中，使用

if admit != nil && admit.Handles(admission.Delete) {
	userInfo, _ := api.UserFrom(ctx)

	err = admit.Admit(admission.NewAttributesRecord(nil, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Delete, userInfo))
	if err != nil {
		scope.err(err, res.ResponseWriter, req.Request)
		return
	}
}

来处理请求。