【kubernetes/k8s源码分析】 kube-apiserver admission controller 源码分析

喜欢ヅ旅行 2021-12-15 00:19 213阅读 0赞

kubernetes git version: v1.14

**初始化阶段** admission 代码流程

main

-->  NewAPIServerCommand

-->  NewServerRunOptions

-->  kubeoptions.NewAdmissionOptions

## 结构体 AdmissionOptions  ##

路径 k8s.io/apiserver/pkg/server/options/admission.go

*  RecommendedPluginOrder：代表了有序的推荐插件列表集合
 *  DefaultOffPlugins：代表了默认禁止的插件
 *  EnablePlugins:：开启的插件列表，通过kube-apiserver 启动参数设置 --enable-admission-plugins 选项
 *  DisablePlugins:：禁止的插件列表，通过kube-apiserver 启动参数设置 --disable-admission-plugins 选项
 *  Plugins：代表了所有已经注册的插件

>     // AdmissionOptions holds the admission options
>     type AdmissionOptions struct {
>     	// RecommendedPluginOrder holds an ordered list of plugin names we recommend to use by default
>     	RecommendedPluginOrder []string
>     	// DefaultOffPlugins is a set of plugin names that is disabled by default
>     	DefaultOffPlugins sets.String
>     
>     	// EnablePlugins indicates plugins to be enabled passed through `--enable-admission-plugins`.
>     	EnablePlugins []string
>     	// DisablePlugins indicates plugins to be disabled passed through `--disable-admission-plugins`.
>     	DisablePlugins []string
>     	// ConfigFile is the file path with admission control configuration.
>     	ConfigFile string
>     	// Plugins contains all registered plugins.
>     	Plugins *admission.Plugins
>     	// Decorators is a list of admission decorator to wrap around the admission plugins
>     	Decorators admission.Decorators
>     }

# 1. NewAdmissionOptions #

实例化 AdmissionOptions，包括基本通用的 admission，注册这些admission 插件，AllOrderedPlugins 定义了插件的推荐顺序集合，DefaultOffAdmissionPlugins 定义了默认关闭的 admission 插件

// NewAdmissionOptions creates a new instance of AdmissionOptions
    // Note:
    //  In addition it calls RegisterAllAdmissionPlugins to register
    //  all kube-apiserver admission plugins.
    //
    //  Provides the list of RecommendedPluginOrder that holds sane values
    //  that can be used by servers that don't care about admission chain.
    //  Servers that do care can overwrite/append that field after creation.
    func NewAdmissionOptions() *AdmissionOptions {
    	options := genericoptions.NewAdmissionOptions()
    	// register all admission plugins
    	RegisterAllAdmissionPlugins(options.Plugins)
    	// set RecommendedPluginOrder
    	options.RecommendedPluginOrder = AllOrderedPlugins
    	// set DefaultOffPlugins
    	options.DefaultOffPlugins = DefaultOffAdmissionPlugins()
    
    	return &AdmissionOptions{
    		GenericAdmission: options,
    	}
    }

###     1.1 NewAdmissionOptions ###

路径 vendor/k8s.io/apiserver/pkg/server/options/admission.go

实例化 AdmissionOptions，具体参数详看上述 '结构体 AdmissionOptions'

RecommendPluginOrder 推荐有序插件包括NamespaceLifecycle，MutatingAdmissionWebhook，ValidatingAdmissionWebhook

分别调用 Register 函数注册这三个推荐的有序插件

// NewAdmissionOptions creates a new instance of AdmissionOptions
    // Note:
    //  In addition it calls RegisterAllAdmissionPlugins to register
    //  all generic admission plugins.
    //
    //  Provides the list of RecommendedPluginOrder that holds sane values
    //  that can be used by servers that don't care about admission chain.
    //  Servers that do care can overwrite/append that field after creation.
    func NewAdmissionOptions() *AdmissionOptions {
    	options := &AdmissionOptions{
    		Plugins: admission.NewPlugins(),
    		// This list is mix of mutating admission plugins and validating
    		// admission plugins. The apiserver always runs the validating ones
    		// after all the mutating ones, so their relative order in this list
    		// doesn't matter.
    		RecommendedPluginOrder: []string{lifecycle.PluginName, initialization.PluginName, mutatingwebhook.PluginName, validatingwebhook.PluginName},
    		DefaultOffPlugins:      sets.NewString(initialization.PluginName),
    	}
    	server.RegisterAllAdmissionPlugins(options.Plugins)
    	return options
    }

**    1.1.1 NewPlugins 实例化 Plugins**

**     registry 记录了插件以及方法**

// Factory is a function that returns an Interface for admission decisions.
    // The config parameter provides an io.Reader handler to the factory in
    // order to load specific configurations. If no configuration is provided
    // the parameter is nil.
    type Factory func(config io.Reader) (Interface, error)
    
    type Plugins struct {
    	lock     sync.Mutex
    	registry map[string]Factory
    }
    
    func NewPlugins() *Plugins {
    	return &Plugins{}
    }

**    1.1.2 RegisterAllAdmissionPlugins**

**     **  分别调用 Register 函数注册这三个推荐的有序插件，包括NamespaceLifecycle，MutatingAdmissionWebhook，ValidatingAdmissionWebhook

// RegisterAllAdmissionPlugins registers all admission plugins
    func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
    	lifecycle.Register(plugins)
    	validatingwebhook.Register(plugins)
    	mutatingwebhook.Register(plugins)
    }

**   1.1.3 Register 函数**

比如 lifecycle 这个插件注册NamespaceLifecycle 以及方法

// Register registers a plugin
    func Register(plugins *admission.Plugins) {
    	plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
    		return NewLifecycle(sets.NewString(metav1.NamespaceDefault, metav1.NamespaceSystem, metav1.NamespacePublic))
    	})
    }

###     1.2 RegisterAllAdmissionPlugins ###

注册了一大堆 admission 插件，与 1.1.3 类同

// RegisterAllAdmissionPlugins registers all admission plugins and
    // sets the recommended plugins order.
    func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
    	admit.Register(plugins) // DEPRECATED as no real meaning
    	alwayspullimages.Register(plugins)
    	antiaffinity.Register(plugins)
    	defaulttolerationseconds.Register(plugins)
    	deny.Register(plugins) // DEPRECATED as no real meaning
    	eventratelimit.Register(plugins)
    	exec.Register(plugins)
    	extendedresourcetoleration.Register(plugins)
    	gc.Register(plugins)
    	imagepolicy.Register(plugins)
    	limitranger.Register(plugins)
    	autoprovision.Register(plugins)
    	exists.Register(plugins)
    	noderestriction.Register(plugins)
    	nodetaint.Register(plugins)
    	label.Register(plugins) // DEPRECATED, future PVs should not rely on labels for zone topology
    	podnodeselector.Register(plugins)
    	podpreset.Register(plugins)
    	podtolerationrestriction.Register(plugins)
    	resourcequota.Register(plugins)
    	podsecuritypolicy.Register(plugins)
    	podpriority.Register(plugins)
    	scdeny.Register(plugins)
    	serviceaccount.Register(plugins)
    	setdefault.Register(plugins)
    	resize.Register(plugins)
    	storageobjectinuseprotection.Register(plugins)
    }

###    1.3 AllOrderedPlugins 定义了有序的 admission 插件列表 ###

>     // AllOrderedPlugins is the list of all the plugins in order.
>     var AllOrderedPlugins = []string{
>     	admit.PluginName,                        // AlwaysAdmit
>     	autoprovision.PluginName,                // NamespaceAutoProvision
>     	lifecycle.PluginName,                    // NamespaceLifecycle
>     	exists.PluginName,                       // NamespaceExists
>     	scdeny.PluginName,                       // SecurityContextDeny
>     	antiaffinity.PluginName,                 // LimitPodHardAntiAffinityTopology
>     	podpreset.PluginName,                    // PodPreset
>     	limitranger.PluginName,                  // LimitRanger
>     	serviceaccount.PluginName,               // ServiceAccount
>     	noderestriction.PluginName,              // NodeRestriction
>     	nodetaint.PluginName,                    // TaintNodesByCondition
>     	alwayspullimages.PluginName,             // AlwaysPullImages
>     	imagepolicy.PluginName,                  // ImagePolicyWebhook
>     	podsecuritypolicy.PluginName,            // PodSecurityPolicy
>     	podnodeselector.PluginName,              // PodNodeSelector
>     	podpriority.PluginName,                  // Priority
>     	defaulttolerationseconds.PluginName,     // DefaultTolerationSeconds
>     	podtolerationrestriction.PluginName,     // PodTolerationRestriction
>     	exec.DenyEscalatingExec,                 // DenyEscalatingExec
>     	exec.DenyExecOnPrivileged,               // DenyExecOnPrivileged
>     	eventratelimit.PluginName,               // EventRateLimit
>     	extendedresourcetoleration.PluginName,   // ExtendedResourceToleration
>     	label.PluginName,                        // PersistentVolumeLabel
>     	setdefault.PluginName,                   // DefaultStorageClass
>     	storageobjectinuseprotection.PluginName, // StorageObjectInUseProtection
>     	gc.PluginName,                           // OwnerReferencesPermissionEnforcement
>     	resize.PluginName,                       // PersistentVolumeClaimResize
>     	mutatingwebhook.PluginName,              // MutatingAdmissionWebhook
>     	validatingwebhook.PluginName,            // ValidatingAdmissionWebhook
>     	resourcequota.PluginName,                // ResourceQuota
>     	deny.PluginName,                         // AlwaysDeny
>     }

###     1.4 DefaultOffAdmissionPlugins 定义了禁止的 admission plugin 列表 ###

默认开启的 admission plugin 有

*  NamespaceLifecycle
 *  LimitRanger
 *  ServiceAccount
 *  DefaultStorageClass
 *  PersistentVolumeClaimResize
 *  DefaultTolerationSeconds
 *  MutatingAdmissionWebhook
 *  ValidatingAdmissionWebhook
 *  ResourceQuota

>     // DefaultOffAdmissionPlugins get admission plugins off by default for kube-apiserver.
>     func DefaultOffAdmissionPlugins() sets.String {
>     	defaultOnPlugins := sets.NewString(
>     		lifecycle.PluginName,                //NamespaceLifecycle
>     		limitranger.PluginName,              //LimitRanger
>     		serviceaccount.PluginName,           //ServiceAccount
>     		setdefault.PluginName,               //DefaultStorageClass
>     		resize.PluginName,                   //PersistentVolumeClaimResize
>     		defaulttolerationseconds.PluginName, //DefaultTolerationSeconds
>     		mutatingwebhook.PluginName,          //MutatingAdmissionWebhook
>     		validatingwebhook.PluginName,        //ValidatingAdmissionWebhook
>     		resourcequota.PluginName,            //ResourceQuota
>     	)
>     
>     	if utilfeature.DefaultFeatureGate.Enabled(features.PodPriority) {
>     		defaultOnPlugins.Insert(podpriority.PluginName) //PodPriority
>     	}
>     
>     	if utilfeature.DefaultFeatureGate.Enabled(features.TaintNodesByCondition) {
>     		defaultOnPlugins.Insert(nodetaint.PluginName) //TaintNodesByCondition
>     	}
>     
>     	return sets.NewString(AllOrderedPlugins...).Difference(defaultOnPlugins)
>     }

buildGenericConfig

-->  s.Admission.ApplyTo

-->  a.GenericAdmission.ApplyTo

-->  a.Plugins.NewFromPlugins

-->  InitPlugin

# 2. buildGenericConfig #

Config 结构主要用来初始化 admission 插件，路径实现 pkg/kubeapiserver/admission/config.go

admissionConfig := &kubeapiserveradmission.Config{
    	ExternalInformers:    versionedInformers,
    	LoopbackClientConfig: genericConfig.LoopbackClientConfig,
    	CloudConfigFile:      s.CloudProvider.CloudConfigFile,
    }

admissionConfig.New 函数主要用来为 admission 建立插件以及 start hook，具体如下 2.1 讲解

pluginInitializers, admissionPostStartHook, err = admissionConfig.New(proxyTransport, serviceResolver)
    if err != nil {
    	lastErr = fmt.Errorf("failed to create admission plugin initializer: %v", err)
    	return
    }

###     2.1 New 函数 ###

NewPluginInitializer 实例化 PluginInitializer，这个结构用来初始化 admission plugin

// New sets up the plugins and admission start hooks needed for admission
    func (c *Config) New(proxyTransport *http.Transport, serviceResolver webhook.ServiceResolver) ([]admission.PluginInitializer, server.PostStartHookFunc, error) {
    	webhookAuthResolverWrapper := webhook.NewDefaultAuthenticationInfoResolverWrapper(proxyTransport, c.LoopbackClientConfig)
    	webhookPluginInitializer := webhookinit.NewPluginInitializer(webhookAuthResolverWrapper, serviceResolver)
    
    	var cloudConfig []byte
    	if c.CloudConfigFile != "" {
    		var err error
    		cloudConfig, err = ioutil.ReadFile(c.CloudConfigFile)
    		if err != nil {
    			klog.Fatalf("Error reading from cloud configuration file %s: %#v", c.CloudConfigFile, err)
    		}
    	}
    	internalClient, err := internalclientset.NewForConfig(c.LoopbackClientConfig)
    	if err != nil {
    		return nil, nil, err
    	}

###     2.2 NewPluginInitializer 实例化 PluginInitializer ###

具体实现路径为 pkg/kubeapiserver/admission/initializer.go

discoveryClient := cacheddiscovery.NewMemCacheClient(internalClient.Discovery())
    discoveryRESTMapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryClient)
    kubePluginInitializer := NewPluginInitializer(
    	cloudConfig,
    	discoveryRESTMapper,
    	quotainstall.NewQuotaConfigurationForAdmission(),
    )

###     2.3 admissionPostStartHook 函数是重置 cache 操作 ###

admissionPostStartHook := func(context genericapiserver.PostStartHookContext) error {
    	discoveryRESTMapper.Reset()
    	go utilwait.Until(discoveryRESTMapper.Reset, 30*time.Second, context.StopCh)
    	return nil
    }
    
    return []admission.PluginInitializer{webhookPluginInitializer, kubePluginInitializer}, admissionPostStartHook, nil

# 3. ApplyTo #

路径 k8s.io/apiserver/pkg/server/options/admission.go

// ApplyTo adds the admission chain to the server configuration.
    // In case admission plugin names were not provided by a custer-admin they will be prepared from the recommended/default values.
    // In addition the method lazily initializes a generic plugin that is appended to the list of pluginInitializers
    // note this method uses:
    //  genericconfig.Authorizer
    func (a *AdmissionOptions) ApplyTo(
    	c *server.Config,
    	informers informers.SharedInformerFactory,
    	kubeAPIServerClientConfig *rest.Config,
    	pluginInitializers ...admission.PluginInitializer,
    ) error {
    	if a == nil {
    		return nil
    	}

###     3.1 NewFromPlugins ###

InitPlugin 主要对插件进行初始化工作，分别调用 Initialize 初始化，在调用 ValidateInitialization 验证是否实现了接口方法 ValidateInitialization

最后包裹 handlers，chainAdmissionHandler 包含了插件的 handler，实现了 Admit 和 Validate 方法，就是对所有插件 handler 调用 Admit 和 Validate 方法

*  func (admissionHandler chainAdmissionHandler) Admit(a Attributes, o ObjectInterfaces) error
 *  func (admissionHandler chainAdmissionHandler) Validate(a Attributes, o ObjectInterfaces) error

// NewFromPlugins returns an admission.Interface that will enforce admission control decisions of all
    // the given plugins.
    func (ps *Plugins) NewFromPlugins(pluginNames []string, configProvider ConfigProvider, pluginInitializer PluginInitializer, decorator Decorator) (Interface, error) {
    	handlers := []Interface{}
    	mutationPlugins := []string{}
    	validationPlugins := []string{}
    	for _, pluginName := range pluginNames {
    		pluginConfig, err := configProvider.ConfigFor(pluginName)
    		if err != nil {
    			return nil, err
    		}
    
    		plugin, err := ps.InitPlugin(pluginName, pluginConfig, pluginInitializer)
    		if err != nil {
    			return nil, err
    		}
    
    	return chainAdmissionHandler(handlers), nil
    }

###     3.2 赋值 AdmissionControl ###

具体操作就是又包裹了一下 pluginHandlerWithMetrics

*  func (p pluginHandlerWithMetrics) Admit(a admission.Attributes, o admission.ObjectInterfaces) error
 *  func (p pluginHandlerWithMetrics) Validate(a admission.Attributes, o admission.ObjectInterfaces) error

// WithStepMetrics is a decorator for a whole admission phase, i.e. admit or validation.admission step.
    func WithStepMetrics(i admission.Interface) admission.Interface {
    	return WithMetrics(i, Metrics.ObserveAdmissionStep)
    }
    
    // WithMetrics is a decorator for admission handlers with a generic observer func.
    func WithMetrics(i admission.Interface, observer ObserverFunc, extraLabels ...string) admission.Interface {
    	return &pluginHandlerWithMetrics{
    		Interface:   i,
    		observer:    observer,
    		extraLabels: extraLabels,
    	}
    }

初始化工作完成了，那究竟哪里会使用 admission 呢？ 请看下文，具体 kube-apiserver 启动源码分析不再赘述，直接关键函数开始

registerResourceHandlers

-->  metrics.InstrumentRouteFunc

-->  restfulUpdateResource              PUT

-->  restfulPatchResource                 PATCH

-->  restfulCreateNamedResource    POST

-->  restfulCreateResource               POST

-->  restfulDeleteResource               DELETE

-->  restfulDeleteCollection               DELETECOLLECTION

-->  restfulConnectResource            CONNECT

# 4. registerResourceHandlers #

可以得到 PUT PATCH POST DELETE DELETECOLLECTION CONNECT 这些 method 用到了 admission

switch action.Verb {
    case "PUT": // Update a resource.
    	doc := "replace the specified " + kind
    	if isSubresource {
    		doc = "replace " + subresource + " of the specified " + kind
    	}
    	handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulUpdateResource(updater, reqScope, admit))
    	
    case "PATCH": // Partially update a resource
    	handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulPatchResource(patcher, reqScope, admit, supportedTypes))
    	
    case "POST": // Create a resource.
    	var handler restful.RouteFunction
    	if isNamedCreater {
    		handler = restfulCreateNamedResource(namedCreater, reqScope, admit)
    	} else {
    		handler = restfulCreateResource(creater, reqScope, admit)
    	}
    	handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler)
    	
    case "DELETE": // Delete a resource.
    	handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulDeleteResource(gracefulDeleter, isGracefulDeleter, reqScope, admit))
    	
    case "DELETECOLLECTION":
    	handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulDeleteCollection(collectionDeleter, isCollectionDeleter, reqScope, admit))
    	
    case "CONNECT":
    	for _, method := range connecter.ConnectMethods() {
    		handler := metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, restfulConnectResource(connecter, reqScope, admit, path, isSubresource))
    
    }

###     4.1 InstrumentRouteFunc 函数 ###

包裹了 type RouteFunction func(\*Request, \*Response) ,主要关心 routeFunc(request,response),也就是传入的 restfulXXXXXResource ,如上列表所示

// InstrumentRouteFunc works like Prometheus' InstrumentHandlerFunc but wraps
    // the go-restful RouteFunction instead of a HandlerFunc plus some Kubernetes endpoint specific information.
    func InstrumentRouteFunc(verb, group, version, resource, subresource, scope, component string, routeFunc restful.RouteFunction) restful.RouteFunction {
    	return restful.RouteFunction(func(request *restful.Request, response *restful.Response) {
    		now := time.Now()
    
    		delegate := &ResponseWriterDelegator{ResponseWriter: response.ResponseWriter}
    
    		_, cn := response.ResponseWriter.(http.CloseNotifier)
    		_, fl := response.ResponseWriter.(http.Flusher)
    		_, hj := response.ResponseWriter.(http.Hijacker)
    		var rw http.ResponseWriter
    		if cn && fl && hj {
    			rw = &fancyResponseWriterDelegator{delegate}
    		} else {
    			rw = delegate
    		}
    		response.ResponseWriter = rw
    
    		routeFunc(request, response)
    
    		MonitorRequest(request.Request, verb, group, version, resource, subresource, scope, component, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Since(now))
    	})
    }

###     4.2 restfulCreateResource 函数 ###

简单,假设不知道干麽的,继续看代码

func restfulCreateResource(r rest.Creater, scope handlers.RequestScope, admit admission.Interface) restful.RouteFunction {
    	return func(req *restful.Request, res *restful.Response) {
    		handlers.CreateResource(r, scope, admit)(res.ResponseWriter, req.Request)
    	}
    }
    
    // CreateResource returns a function that will handle a resource creation.
    func CreateResource(r rest.Creater, scope RequestScope, admission admission.Interface) http.HandlerFunc {
    	return createHandler(&namedCreaterAdapter{r}, scope, admission, false)
    }

###     4.3 createHandler 函数 ###

路径 k8s.io/apiserver/pkg/endpoints/handlers/create.go,终于找到主要的执行函数了,这个对于 CREATE 的action

func createHandler(r rest.NamedCreater, scope RequestScope, admit admission.Interface, includeName bool) http.HandlerFunc {
    	return func(w http.ResponseWriter, req *http.Request) {
    		// For performance tracking purposes.
    		trace := utiltrace.New("Create " + req.URL.Path)
    		defer trace.LogIfLong(500 * time.Millisecond)
    
    		if isDryRun(req.URL) && !utilfeature.DefaultFeatureGate.Enabled(features.DryRun) {
    			scope.err(errors.NewBadRequest("the dryRun alpha feature is disabled"), w, req)
    			return
    		}

此去省略几千字,略过一堆解析 body 等操作,只关心 admission 的处理

admissionAttributes := admission.NewAttributesRecord(obj, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Create, dryrun.IsDryRun(options.DryRun), userInfo)

###     4.4 结构体 attributeRecord  ###

实现了接口 Attributes

type attributesRecord struct {
    	kind        schema.GroupVersionKind
    	namespace   string
    	name        string
    	resource    schema.GroupVersionResource
    	subresource string
    	operation   Operation
    	dryRun      bool
    	object      runtime.Object
    	oldObject   runtime.Object
    	userInfo    user.Info
    
    	// other elements are always accessed in single goroutine.
    	// But ValidatingAdmissionWebhook add annotations concurrently.
    	annotations     map[string]string
    	annotationsLock sync.RWMutex
    }

###     4.5 这个已经是包裹的 pluginHandlerWithMetrics 在层层拨开,具体到每个 admissin 插件的 Admit 方法 ###

本文只讲解具体的几个 admission plugin,其他雷同

if mutatingAdmission, ok := admit.(admission.MutationInterface); ok && mutatingAdmission.Handles(admission.Create) {
    	err = mutatingAdmission.Admit(admissionAttributes, &scope)
    	if err != nil {
    		scope.err(err, w, req)
    		return
    	}
    }

# 6. MutatingAdmissionWebhook #

group: admissionregistration.k8s.io kind: MutatingWebhookConfiguration

###     6.1  NewMutatingWebhook 函数 ###

实例化 Plugin, 允许的包括 CONNECT CREATE DELETE UPDATE

> webhooks:  
>   - name: kube-webhook.mutating.me  
>     clientConfig:  
>       service:  
>         name: kube-webhook-svc  
>         namespace: default  
>         path: "/webhook"  
>       caBundle: $\{CA\_BUNDLE\}  
>     rules:  
>       - **operations: \[ "CREATE" \]**  
>         apiGroups: \["apps"\]  
>         apiVersions: \["v1"\]  
>         resources: \["deployments"\]

// NewMutatingWebhook returns a generic admission webhook plugin.
    func NewMutatingWebhook(configFile io.Reader) (*Plugin, error) {
    	handler := admission.NewHandler(admission.Connect, admission.Create, admission.Delete, admission.Update)
    	p := &Plugin{}
    	var err error
    	p.Webhook, err = generic.NewWebhook(handler, configFile, configuration.NewMutatingWebhookConfigurationManager, newMutatingDispatcher(p))
    	if err != nil {
    		return nil, err
    	}
    
    	return p, nil
    }

###     6.2 Admit 函数 ###

// Admit makes an admission decision based on the request attributes.
    func (a *Plugin) Admit(attr admission.Attributes, o admission.ObjectInterfaces) error {
    	return a.Webhook.Dispatch(attr, o)
    }

###     6.3 Dispatch 函数 ###

中间略过,直奔 Dispatch 这个函数,有原来实例化时 newMutatingDispatcher 就是实现了 Dispatcher 接口

// Dispatch is called by the downstream Validate or Admit methods.
    func (a *Webhook) Dispatch(attr admission.Attributes, o admission.ObjectInterfaces) error {
    	if rules.IsWebhookConfigurationResource(attr) {
    		return nil
    	}
    	if !a.WaitForReady() {
    		return admission.NewForbidden(attr, fmt.Errorf("not yet ready to handle request"))
    	}
    	hooks := a.hookSource.Webhooks()
    	// TODO: Figure out if adding one second timeout make sense here.
    	ctx := context.TODO()
    
        return a.dispatcher.Dispatch(ctx, &versionedAttr, o, relevantHooks)

###     6.4 Dispatch 函数 ###

路径 k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/dispatcher.go

对每个 hook 调用 callAttrMutatingHook 进行处理

func (a *mutatingDispatcher) Dispatch(ctx context.Context, attr *generic.VersionedAttributes, o admission.ObjectInterfaces, relevantHooks []*v1beta1.Webhook) error {
    	for _, hook := range relevantHooks {
    		t := time.Now()
    		err := a.callAttrMutatingHook(ctx, hook, attr, o)
    		admissionmetrics.Metrics.ObserveWebhook(time.Since(t), err != nil, attr.Attributes, "admit", hook.Name)
    		if err == nil {
    			continue
    		}

###     6.5 callAttrMutatingHook 函数 ###

// note that callAttrMutatingHook updates attr
    func (a *mutatingDispatcher) callAttrMutatingHook(ctx context.Context, h *v1beta1.Webhook, attr *generic.VersionedAttributes, o admission.ObjectInterfaces) error {
    	if attr.IsDryRun() {
    		if h.SideEffects == nil {
    			return &webhook.ErrCallingWebhook{WebhookName: h.Name, Reason: fmt.Errorf("Webhook SideEffects is nil")}
    		}
    		if !(*h.SideEffects == v1beta1.SideEffectClassNone || *h.SideEffects == v1beta1.SideEffectClassNoneOnDryRun) {
    			return webhookerrors.NewDryRunUnsupportedErr(h.Name)
    		}
    	}
    
    	// Currently dispatcher only supports `v1beta1` AdmissionReview
    	// TODO: Make the dispatcher capable of sending multiple AdmissionReview versions
    	if !util.HasAdmissionReviewVersion(v1beta1.SchemeGroupVersion.Version, h) {
    		return &webhook.ErrCallingWebhook{WebhookName: h.Name, Reason: fmt.Errorf("webhook does not accept v1beta1 AdmissionReview")}
    	}

6.5.1 其他略过,这里是不是很熟悉

创建 AdmissionReview 请求,发送 POST 请求,所以自己定义的 admission server 就可以接受到请求了

>     // AdmissionReview describes an admission review request/response.
>     type AdmissionReview struct {
>        metav1.TypeMeta `json:",inline"`
>        // Request describes the attributes for the admission request.
>        // +optional
>        Request *AdmissionRequest `json:"request,omitempty" protobuf:"bytes,1,opt,name=request"`
>        // Response describes the attributes for the admission response.
>        // +optional
>        Response *AdmissionResponse `json:"response,omitempty" protobuf:"bytes,2,opt,name=response"`
>     }

// Make the webhook request
    request := request.CreateAdmissionReview(attr)
    client, err := a.cm.HookClient(util.HookClientConfigForWebhook(h))
    if err != nil {
    	return &webhook.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
    }
    response := &admissionv1beta1.AdmissionReview{}
    r := client.Post().Context(ctx).Body(&request)
    if h.TimeoutSeconds != nil {
    	r = r.Timeout(time.Duration(*h.TimeoutSeconds) * time.Second)
    }
    if err := r.Do().Into(response); err != nil {
    	return &webhook.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
    }
    
    if response.Response == nil {
    	return &webhook.ErrCallingWebhook{WebhookName: h.Name, Reason: fmt.Errorf("Webhook response was absent")}
    }

# 7. AlwaysPullImages admission 插件 #

###     7.1 实例化 AlwaysPullImages ###

// AlwaysPullImages is an implementation of admission.Interface.
    // It looks at all new pods and overrides each container's image pull policy to Always.
    type AlwaysPullImages struct {
    	*admission.Handler
    }
    
    // NewAlwaysPullImages creates a new always pull images admission control handler
    func NewAlwaysPullImages() *AlwaysPullImages {
    	return &AlwaysPullImages{
    		Handler: admission.NewHandler(admission.Create, admission.Update),
    	}
    }

###     7.2 Admit 函数 ###

比较简单就是修改了 pod 属性, 将 imagePullPolicy 修改为 Always

// Admit makes an admission decision based on the request attributes
    func (a *AlwaysPullImages) Admit(attributes admission.Attributes, o admission.ObjectInterfaces) (err error) {
    	// Ignore all calls to subresources or resources other than pods.
    	if shouldIgnore(attributes) {
    		return nil
    	}
    	pod, ok := attributes.GetObject().(*api.Pod)
    	if !ok {
    		return apierrors.NewBadRequest("Resource was marked with kind Pod but was unable to be converted")
    	}
    
    	for i := range pod.Spec.InitContainers {
    		pod.Spec.InitContainers[i].ImagePullPolicy = api.PullAlways
    	}
    
    	for i := range pod.Spec.Containers {
    		pod.Spec.Containers[i].ImagePullPolicy = api.PullAlways
    	}
    
    	return nil
    }

###     7.3 Validate 函数 ###

就是拿到 pod 信息, 验证是否  imagePullPolicy 为 Always

// Validate makes sure that all containers are set to always pull images
    func (*AlwaysPullImages) Validate(attributes admission.Attributes, o admission.ObjectInterfaces) (err error) {
    	if shouldIgnore(attributes) {
    		return nil
    	}
    
    	pod, ok := attributes.GetObject().(*api.Pod)
    	if !ok {
    		return apierrors.NewBadRequest("Resource was marked with kind Pod but was unable to be converted")
    	}
    
    	for i := range pod.Spec.InitContainers {
    		if pod.Spec.InitContainers[i].ImagePullPolicy != api.PullAlways {
    			return admission.NewForbidden(attributes,
    				field.NotSupported(field.NewPath("spec", "initContainers").Index(i).Child("imagePullPolicy"),
    					pod.Spec.InitContainers[i].ImagePullPolicy, []string{string(api.PullAlways)},
    				),
    			)
    		}
    	}
    	for i := range pod.Spec.Containers {
    		if pod.Spec.Containers[i].ImagePullPolicy != api.PullAlways {
    			return admission.NewForbidden(attributes,
    				field.NotSupported(field.NewPath("spec", "containers").Index(i).Child("imagePullPolicy"),
    					pod.Spec.Containers[i].ImagePullPolicy, []string{string(api.PullAlways)},
    				),
    			)
    		}
    	}
    
    	return nil
    }

## 总结: ##

分析了 admission 初始化以及插件注册流程

分析了 MutatingAdmissionWebhook 以及 AlwaysPullImages 插件实现