crdHandle

1.服务发现

1.1.apiservice

• kube-apiserver支持CRD和AA两种扩展，AA对增强API的描述基于apiservice实现，aggregatorserver基于apiservice转发请求。

• 

• 补充

  apiservice是请求处理基准，所有资源注册apiservice，aggregatorserver基于apiservice统一注册代理路由及proxyHandler

1.2.aggregatorServer

• kube-aggregatorServer作为apiserver的统一入口，监听apiservice进行请求转发，内置及CRD资源路由会同步proxyHandler。

• func createAggregatorServer(...) (*aggregatorapiserver.APIAggregator, error) {
  	aggregatorServer, err := aggregatorConfig.Complete().NewWithDelegate(delegateAPIServer)
  	...
  
  	return aggregatorServer, nil
  }
  
  // NewWithDelegate returns a new instance of APIAggregator from the given config.
  func (c completedConfig) NewWithDelegate(delegation genericapiserver.DelegationTarget) (*APIAggregator, error) {
  	genericServer, err := c.GenericConfig.New("kube-aggregator", delegationTarget)
  	...
  
  	s := &APIAggregator{...}
  	...
  
      // apiservice路由
  	apisHandler := &apisHandler{...}
  	...
      s.GenericAPIServer.Handler.NonGoRestfulMux.Handle("/apis", apisHandler)
  	s.GenericAPIServer.Handler.NonGoRestfulMux.UnlistedHandle("/apis/", apisHandler)
  
      // 监听apiservice及注册所有apiservice资源的proxyHandler路由
  	apiserviceRegistrationController := NewAPIServiceRegistrationController(informerFactory.Apiregistration().V1().APIServices(), s)
  	...
  
      // 监听可用的apiservice
  	availableController, err := statuscontrollers.NewAvailableConditionController(...)
  	...
  	
  	s.GenericAPIServer.AddPostStartHookOrDie("apiservice-registration-controller", func(context genericapiserver.PostStartHookContext) error {
  		go apiserviceRegistrationController.Run(context.StopCh, apiServiceRegistrationControllerInitiated)
  		...
  		return nil
  	})
  	s.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-available-controller", func(context genericapiserver.PostStartHookContext) error {
  		// if we end up blocking for long periods of time, we may need to increase workers.
  		go availableController.Run(5, context.StopCh)
  		return nil
  	})
  	...
  
  	return s, nil
  }
  
  // NewAPIServiceRegistrationController returns a new APIServiceRegistrationController.
  func NewAPIServiceRegistrationController(...) *APIServiceRegistrationController {
  	c := &APIServiceRegistrationController{
  		apiHandlerManager: apiHandlerManager,
  		apiServiceLister:  apiServiceInformer.Lister(),
  		apiServiceSynced:  apiServiceInformer.Informer().HasSynced,
  		queue:             workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "APIServiceRegistrationController"),
  	}
  
  	apiServiceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc:    c.addAPIService,
  		UpdateFunc: c.updateAPIService,
  		DeleteFunc: c.deleteAPIService,
  	})
  
  	c.syncFn = c.sync
  
  	return c
  }
  
  func (c *APIServiceRegistrationController) sync(key string) error {
  	apiService, err := c.apiServiceLister.Get(key)
  	if apierrors.IsNotFound(err) {
  		c.apiHandlerManager.RemoveAPIService(key)
  		return nil
  	}
  	...
  
  	return c.apiHandlerManager.AddAPIService(apiService)
  }
  
  // AddAPIService adds an API service.  It is not thread-safe, so only call it on one thread at a time please.
  // It's a slow moving API, so its ok to run the controller on a single thread
  func (s *APIAggregator) AddAPIService(apiService *v1.APIService) error {
  	...
      
  	// register the proxy handler
  	proxyHandler := &proxyHandler{
  		localDelegate:              s.delegateHandler,
  		proxyCurrentCertKeyContent: s.proxyCurrentCertKeyContent,
  		proxyTransportDial:         s.proxyTransportDial,
  		serviceResolver:            s.serviceResolver,
  		rejectForwardingRedirects:  s.rejectForwardingRedirects,
  	}
  	...
  
  	s.proxyHandlers[apiService.Name] = proxyHandler
  	s.GenericAPIServer.Handler.NonGoRestfulMux.Handle(proxyPath, proxyHandler)
  	s.GenericAPIServer.Handler.NonGoRestfulMux.UnlistedHandlePrefix(proxyPath+"/", proxyHandler)
  	...
      
  	return nil
  }
  

• 补充

  内置资源及CRD资源均会注册apiservice，apiserviceRegistrationController会根据apiservice注册代理路由

1.3.内置资源

• 内置资源主要是kubernetes开箱即用的pod/deployment...，基于kube-apiserver的路由创建对应的apiservice(没有的情况下)。

• func createAggregatorServer(...) (*aggregatorapiserver.APIAggregator, error) {
      ...
      
      // 创建apiservice资源控制器
  	autoRegistrationController := autoregister.NewAutoRegisterController(aggregatorServer.APIRegistrationInformers.Apiregistration().V1().APIServices(), apiRegistrationClient)
      
      // 基于kube-apiserver路由创建对应apiservices
  	apiServices := apiServicesToRegister(delegateAPIServer, autoRegistrationController)
  	
      // 创建crd资源控制器,新注册的crd资源更新到apiservice
      crdRegistrationController := crdregistration.NewCRDRegistrationController(
  		apiExtensionInformers.Apiextensions().V1().CustomResourceDefinitions(),
  		autoRegistrationController)
  	...
  
      // 注册启动回调
  	err = aggregatorServer.GenericAPIServer.AddPostStartHook("kube-apiserver-autoregistration", func(context genericapiserver.PostStartHookContext) error {
  		// 启动crd资源注册控制器
          go crdRegistrationController.Run(5, context.StopCh)
  		go func() {
              // 等待crd资源同步完成
              crdRegistrationController.WaitForInitialSync()
              
              // 启动apiservice资源注册控制器
  			autoRegistrationController.Run(5, context.StopCh)
  		}()
  		return nil
  	})
  	...
  
  	return aggregatorServer, nil
  }
  
  func apiServicesToRegister(...) []*v1.APIService {
  	apiServices := []*v1.APIService{}
  
      // 遍历kube-apiserver路由
  	for _, curr := range delegateAPIServer.ListedPaths() {
          // 核心资源的路由
  		if curr == "/api/v1" {
              // 创建apiservice
  			apiService := makeAPIService(schema.GroupVersion{Group: "", Version: "v1"})
              // 利用autoRegistrationController更新
  			registration.AddAPIServiceToSyncOnStart(apiService)
  			apiServices = append(apiServices, apiService)
  			continue
  		}
  		...
  
          // 其它核心资源的apiservice
  		apiService := makeAPIService(schema.GroupVersion{Group: tokens[2], Version: tokens[3]})
  		if apiService == nil {
  			continue
  		}
          
          // 利用autoRegistrationController更新
  		registration.AddAPIServiceToSyncOnStart(apiService)
  		apiServices = append(apiServices, apiService)
  	}
  
  	return apiServices
  }
  
  func makeAPIService(gv schema.GroupVersion) *v1.APIService {
      // 取内置资源的GV优先级({Group: "apps", Version: "v1"}: {group: 17800, version: 15})
  	apiServicePriority, ok := apiVersionPriorities[gv]
  	...
      
  	return &v1.APIService{
  		ObjectMeta: metav1.ObjectMeta{Name: gv.Version + "." + gv.Group},
  		Spec: v1.APIServiceSpec{
  			Group:                gv.Group,
  			Version:              gv.Version,
  			GroupPriorityMinimum: apiServicePriority.group,
  			VersionPriority:      apiServicePriority.version,
  		},
  	}
  }
  

• 补充

  目前的核心资源有两类，一类是早期的没有apiGroup资源，另一类是后期完整的带apiGroup资源

1.4.crd资源

• crd资源的注册基于crd监听控制器，基于informer监听crd变化，crd关联的apiservice同步到autoRegistrationController更新。

• // NewCRDRegistrationController returns a controller which will register CRD GroupVersions.
  func NewCRDRegistrationController(crdinformer crdinformers.CustomResourceDefinitionInformer, apiServiceRegistration AutoAPIServiceRegistration) *crdRegistrationController {
      // 构造crd注册控制器对象
  	c := &crdRegistrationController{
  		crdLister:              crdinformer.Lister(),
  		crdSynced:              crdinformer.Informer().HasSynced,
  		apiServiceRegistration: apiServiceRegistration,
  		syncedInitialSet:       make(chan struct{}),
  		queue:                  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "crd_autoregistration_controller"),
  	}
      
      // crd监听后的处理函数
  	c.syncHandler = c.handleVersionUpdate
  
      // crd监听回调事件
  	crdinformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: func(obj interface{}) {
  			cast := obj.(*apiextensionsv1.CustomResourceDefinition)
  			c.enqueueCRD(cast)
  		},
  		UpdateFunc: func(oldObj, newObj interface{}) {
  			c.enqueueCRD(oldObj.(*apiextensionsv1.CustomResourceDefinition))
  			c.enqueueCRD(newObj.(*apiextensionsv1.CustomResourceDefinition))
  		},
  		DeleteFunc: func(obj interface{}) {
  			cast, ok := obj.(*apiextensionsv1.CustomResourceDefinition)
  			...
  			c.enqueueCRD(cast)
  		},
  	})
  
  	return c
  }
  
  func (c *crdRegistrationController) enqueueCRD(crd *apiextensionsv1.CustomResourceDefinition) {
  	for _, version := range crd.Spec.Versions {
  		c.queue.Add(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name})
  	}
  }
  
  func (c *crdRegistrationController) Run(workers int, stopCh <-chan struct{}) {
  	...
  	defer c.queue.ShutDown()
  	...
  
  	// wait for your secondary caches to fill before starting your work
  	if !cache.WaitForNamedCacheSync("crd-autoregister", stopCh, c.crdSynced) {
  		return
  	}
  
  	// 获取已注册的crd
  	crds, err := c.crdLister.List(labels.Everything())
      ...
      
      // 遍历处理
      for _, crd := range crds {
          for _, version := range crd.Spec.Versions {
              c.syncHandler(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name})
              ...
          }
      }
  	close(c.syncedInitialSet)
  
  	// 启动几个worker进行同步监听
  	for i := 0; i < workers; i++ {
  		go wait.Until(c.runWorker, time.Second, stopCh)
  	}
  
  	// wait until we're told to stop
  	<-stopCh
  }
  
  func (c *crdRegistrationController) runWorker() {
  	// 循环执行
  	for c.processNextWorkItem() {
  	}
  }
  
  // processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
  func (c *crdRegistrationController) processNextWorkItem() bool {
  	// 获取缓存队列crd item
  	key, quit := c.queue.Get()
  	...
      
  	// 由队列移除
  	defer c.queue.Done(key)
  
  	// 执行处理
  	err := c.syncHandler(key.(schema.GroupVersion))
      
  	if err == nil {
  		// 清除item的失败/重试信息
  		c.queue.Forget(key)
  		return true
  	}
  	...
      
      // 失败,退避重试
  	c.queue.AddRateLimited(key)
  
  	return true
  }
  
  // syncHandler
  func (c *crdRegistrationController) handleVersionUpdate(groupVersion schema.GroupVersion) error {
      // 构造apiservice名称
  	apiServiceName := groupVersion.Version + "." + groupVersion.Group
  
  	// 列出所有crds
  	crds, err := c.crdLister.List(labels.Everything())
  	...
      
      // 遍历
  	for _, crd := range crds {
          // 对比资源组
  		if crd.Spec.Group != groupVersion.Group {
  			continue
  		}
  		for _, version := range crd.Spec.Versions {
              // 对比版本
  			if version.Name != groupVersion.Version || !version.Served {
  				continue
  			}
  
              // crd存在,进行apiservice注册
  			c.apiServiceRegistration.AddAPIServiceToSync(&v1.APIService{
  				ObjectMeta: metav1.ObjectMeta{Name: apiServiceName},
  				Spec: v1.APIServiceSpec{
  					Group:                groupVersion.Group,
  					Version:              groupVersion.Version,
  					GroupPriorityMinimum: 1000, // CRDs should have relatively low priority
  					VersionPriority:      100,  // CRDs will be sorted by kube-like versions like any other APIService with the same VersionPriority
  				},
  			})
  			return nil
  		}
  	}
  
      // 未找到,执行apiservice移除
  	c.apiServiceRegistration.RemoveAPIServiceToSync(apiServiceName)
  	return nil
  }
  

• 补充

  crdRegistrationController负责监听crd资源变化，基于informer事件回调将crd加入到autoRegisterController注册队列

1.5.autoController

• autoRegisterController负责监听及同步apiservice资源，基于informer同步的apiservice及queue待处理的进行对比检查及更新。

• // NewAutoRegisterController creates a new autoRegisterController.
  func NewAutoRegisterController(...) *autoRegisterController {
      // 构造自动注册控制器
  	c := &autoRegisterController{
  		apiServiceLister:  apiServiceInformer.Lister(),					// apiservice informer
  		...
  		queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "autoregister"),
  	}
      
      // 处理函数
  	c.syncHandler = c.checkAPIService
  
  	apiServiceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: func(obj interface{}) {
  			cast := obj.(*v1.APIService)
  			c.queue.Add(cast.Name)
  		},
  		UpdateFunc: func(_, obj interface{}) {
  			cast := obj.(*v1.APIService)
  			c.queue.Add(cast.Name)
  		},
  		DeleteFunc: func(obj interface{}) {
  			cast, ok := obj.(*v1.APIService)
  			...
  			c.queue.Add(cast.Name)
  		},
  	})
  
  	return c
  }
  
  // AddAPIServiceToSyncOnStart registers an API service to sync only when the controller starts.
  func (c *autoRegisterController) AddAPIServiceToSyncOnStart(in *v1.APIService) {
  	c.addAPIServiceToSync(in, manageOnStart)
  }
  
  // AddAPIServiceToSync registers an API service to sync continuously.
  func (c *autoRegisterController) AddAPIServiceToSync(in *v1.APIService) {
  	c.addAPIServiceToSync(in, manageContinuously)
  }
  
  // RemoveAPIServiceToSync deletes a registered APIService.
  func (c *autoRegisterController) RemoveAPIServiceToSync(name string) {
  	c.apiServicesToSyncLock.Lock()
  	defer c.apiServicesToSyncLock.Unlock()
  
  	delete(c.apiServicesToSync, name)
  	c.queue.Add(name)
  }
  
  func (c *autoRegisterController) addAPIServiceToSync(in *v1.APIService, syncType string) {
  	c.apiServicesToSyncLock.Lock()
  	defer c.apiServicesToSyncLock.Unlock()
  
  	apiService := in.DeepCopy()
  	if apiService.Labels == nil {
  		apiService.Labels = map[string]string{}
  	}
  	apiService.Labels[AutoRegisterManagedLabel] = syncType
  
  	c.apiServicesToSync[apiService.Name] = apiService
  	c.queue.Add(apiService.Name)
  }
  
  // Run starts the autoregister controller in a loop which syncs API services until stopCh is closed.
  func (c *autoRegisterController) Run(workers int, stopCh <-chan struct{}) {
  	...
  	// make sure the work queue is shutdown which will trigger workers to end
  	defer c.queue.ShutDown()
  	...
      
  	// 等待apiservice同步完成
  	if !controllers.WaitForCacheSync("autoregister", stopCh, c.apiServiceSynced) {
  		return
  	}
  
  	// 获取apiservice
  	services := c.apiServiceLister.List(labels.Everything())
      ...
      
      // 记录apiservice状态
      for _, service := range services {
          c.apiServicesAtStart[service.Name] = true
      }
  
  	// start up your worker threads based on workers.  Some controllers have multiple kinds of workers
  	for i := 0; i < workers; i++ {
  		// 异步处理
  		go wait.Until(c.runWorker, time.Second, stopCh)
  	}
  
  	// wait until we're told to stop
  	<-stopCh
  }
  
  func (c *autoRegisterController) runWorker() {
  	// hot loop until we're told to stop.
  	for c.processNextWorkItem() {
  	}
  }
  
  // processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
  func (c *autoRegisterController) processNextWorkItem() bool {
  	// 获取队列apiservice item
  	key, quit := c.queue.Get()
  	...
      
  	// 由队列移除
  	defer c.queue.Done(key)
  
  	// 执行处理
  	err := c.syncHandler(key.(string))
  	if err == nil {
  		// 重试失败及退避信息
  		c.queue.Forget(key)
  		return true
  	}
  	...
      
  	// 失败则退避入队 
  	c.queue.AddRateLimited(key)
  
  	return true
  }
  
  // checkAPIService syncs the current APIService against a list of desired APIService objects
  func (c *autoRegisterController) checkAPIService(name string) (err error) {
      // 获取缓存的期望apiservice
  	desired := c.GetAPIServiceToSync(name)
      // 获取实际的apiservice
  	curr, err := c.apiServiceLister.Get(name)
  	...
  
  	switch {
  	...
  
  	// 期望为空,集群不存在
  	case apierrors.IsNotFound(err) && desired == nil:
  		return nil
  
  	// 期望apiservice是on-start策略且已同步
  	case isAutomanagedOnStart(desired) && hasSynced:
  		return nil
  
  	// 期望有但集群不存在
  	case apierrors.IsNotFound(err) && desired != nil:
          // 创建
  		_, err := c.apiServiceClient.APIServices().Create(context.TODO(), desired, metav1.CreateOptions{})
  		if apierrors.IsAlreadyExists(err) {
  			// created in the meantime, we'll get called again
  			return nil
  		}
  		return err
  
  	// 实际的apiservice不是注册控制器管理的
  	case !isAutomanaged(curr):
  		return nil
  
  	// 当前apiservice是on-start策略,但启动后才出现
  	case isAutomanagedOnStart(curr) && !c.apiServicesAtStart[name]:
  		return nil
  
  	// 当前apiservice是on-start策略且已同步
  	case isAutomanagedOnStart(curr) && hasSynced:
  		return nil
  
  	// 注册控制器管理的,期望没有但现实有
  	case desired == nil:
  		尝试删除
  		err := c.apiServiceClient.APIServices().Delete(context.TODO(), curr.Name, opts)
  		if apierrors.IsNotFound(err) || apierrors.IsConflict(err) {
  			// deleted or changed in the meantime, we'll get called again
  			return nil
  		}
  		return err
  
  	// 期望与实际一致
  	case reflect.DeepEqual(curr.Spec, desired.Spec):
  		return nil
  	}
  	...
      
  	// 更新apiservice数据
  	_, err = c.apiServiceClient.APIServices().Update(context.TODO(), apiService, metav1.UpdateOptions{})
  	...
      
  	return err
  }
  

• 补充

  apiservice处理情况很多，本质上只负责更新apiserver自注册的apiservice，根据apiservice资源情况进行创建、更新或删除

2.CR请求

2.1.crdHandler

• kube-apiextensionserver执行install注册的路由只是CRD对象本身，CR的处理由外部路由注册的crdHandler处理。

• // New returns a new instance of CustomResourceDefinitions from the given config.
  func (c completedConfig) New(delegation genericapiserver.DelegationTarget) (*CustomResourceDefinitions, error) {
  	...
  	crdHandler, err := NewCustomResourceDefinitionHandler(...)
      
  	...
      // 注册cr处理路由
  	s.GenericAPIServer.Handler.NonGoRestfulMux.Handle("/apis", crdHandler)
  	s.GenericAPIServer.Handler.NonGoRestfulMux.HandlePrefix("/apis/", crdHandler)
  	s.GenericAPIServer.RegisterDestroyFunc(crdHandler.destroy)
  	...
  
  	return s, nil
  }
  
  func NewCustomResourceDefinitionHandler(...) (*crdHandler, error) {
  	ret := &crdHandler{
  		versionDiscoveryHandler: versionDiscoveryHandler,
  		groupDiscoveryHandler:   groupDiscoveryHandler,
  		customStorage:           atomic.Value{},
  		crdLister:               crdInformer.Lister(),
  		delegate:                delegate,
  		restOptionsGetter:       restOptionsGetter,
  		admission:               admission,						// 准入控制
  		establishingController:  establishingController,			
  		...
  		authorizer:              authorizer,					// 鉴权相关
  		...
  	}
      
      // CRD监听回调
  	crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc:    ret.createCustomResourceDefinition,
  		UpdateFunc: ret.updateCustomResourceDefinition,
  		DeleteFunc: func(obj interface{}) {
  			ret.removeDeadStorage()
  		},
  	})
      
      // CR转换相关
  	crConverterFactory, err := conversion.NewCRConverterFactory(serviceResolver, authResolverWrapper)
  	...
  	ret.converterFactory = crConverterFactory
  
  	ret.customStorage.Store(crdStorageMap{})
  
  	return ret, nil
  }
  
  // createCustomResourceDefinition removes potentially stale storage so it gets re-created
  func (r *crdHandler) createCustomResourceDefinition(obj interface{}) {
  	crd := obj.(*apiextensionsv1.CustomResourceDefinition)
  	r.customStorageLock.Lock()
  	defer r.customStorageLock.Unlock()
  	// 加载存储数据
  	storageMap := r.customStorage.Load().(crdStorageMap)
      // 未找到该CRD数据,无需清理
  	oldInfo, found := storageMap[crd.UID]
  	if !found {
  		return
  	}
      // spec与AcceptedNames相同,说明创建事件重复,无需清理
  	if DeepEqual(&crd.Spec, oldInfo.spec) && DeepEqual(&crd.Status.AcceptedNames, oldInfo.acceptedNames) {
  		return
  	}
      
      // 清理旧的CRD数据,后续重建
  	r.removeStorage_locked(crd.UID)
  }
  
  // updateCustomResourceDefinition removes potentially stale storage so it gets re-created
  func (r *crdHandler) updateCustomResourceDefinition(oldObj, newObj interface{}) {
  	oldCRD := oldObj.(*apiextensionsv1.CustomResourceDefinition)
  	newCRD := newObj.(*apiextensionsv1.CustomResourceDefinition)
  
  	r.customStorageLock.Lock()
  	defer r.customStorageLock.Unlock()
  
  	// 延迟建立
  	if !IsCRDConditionTrue(newCRD, Established) && IsCRDConditionTrue(newCRD, NamesAccepted) {
  		// 多master集群,延迟5s再更新建立状态
          if r.masterCount > 1 {
  			r.establishingController.QueueCRD(newCRD.Name, 5*time.Second)
  		// 单master立即建立
          } else {
  			r.establishingController.QueueCRD(newCRD.Name, 0)
  		}
  	}
  
      // UID变化,CRD被替换,清理旧的CRD数据,后续重建
  	if oldCRD.UID != newCRD.UID {
  		r.removeStorage_locked(oldCRD.UID)
  	}
  
      // 加载存储数据
  	storageMap := r.customStorage.Load().(crdStorageMap)
      // 未找到该CRD旧数据,无需清理
  	oldInfo, found := storageMap[newCRD.UID]
  	if !found {
  		return
  	}
      // spec及AcceptedNames无变化,无需清理
  	if DeepEqual(&newCRD.Spec, oldInfo.spec) && DeepEqual(&newCRD.Status.AcceptedNames, oldInfo.acceptedNames) {
  		return
  	}
  
      // 清理旧的CRD数据,后续重建
  	r.removeStorage_locked(newCRD.UID)
  }
  
  // removeStorage_locked removes the cached storage with the given uid as key from the storage map.
  func (r *crdHandler) removeStorage_locked(uid types.UID) {
  	// 加载存储数据
      storageMap := r.customStorage.Load().(crdStorageMap)
      // 获取该CRD旧数据
  	if oldInfo, ok := storageMap[uid]; ok {
  		// 拷贝存储数据,避免并发读写
  		storageMap2 := storageMap.clone()
  
  		// 清理该CRD
  		delete(storageMap2, uid)
          // 原子替换更新后的存储数据
  		r.customStorage.Store(storageMap2)
  
          // 异步销毁旧的存储结构(DestroyFunc)
  		go r.tearDown(oldInfo)
  	}
  }
  

• 补充

  crdHandler会监听CRD资源变化，动态维护CRD版本对应storage存储，同时将最新的CRD定义推入其它controller进行检查及状态更新

2.2.servehttp

• crdHandler.serveHttp()是CRD请求路由的核心逻辑，负责根据根据请求信息检查是否属于CRD资源请求，利用CRD动态服务器处理CR资源。

• func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
  	ctx := req.Context()
      // 获取requestInfo
  	requestInfo, ok := apirequest.RequestInfoFrom(ctx)
  	...
      
      // 不是资源请求
  	if !requestInfo.IsResourceRequest {
  		pathParts := splitPath(requestInfo.Path)
          // api-versions请求(kubectl get --raw /apis/apps/v1 | jq .)
  		if len(pathParts) == 3 {
  			r.versionDiscoveryHandler.ServeHTTP(w, req)
  			return
  		}
          
          // api-groups请求(kubectl get --raw /apis/apps | jq .)
  		if len(pathParts) == 2 {
  			r.groupDiscoveryHandler.ServeHTTP(w, req)
  			return
  		}
  
          // 委托给下一级
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  
      // 生成crdName
  	crdName := requestInfo.Resource + "." + requestInfo.APIGroup
      // 获取crd
  	crd, err := r.crdLister.Get(crdName)
      // 未找到,委托给下一级
  	if apierrors.IsNotFound(err) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  	...
  
  	// 检查资源作用域
  	nsdCRD, nsdReq := crd.Spec.Scope == apiextensionsv1.NamespaceScoped,len(requestInfo.Namespace) > 0
      
      // 非CRD的namespace资源,委托给下一级
  	if !namespacedCRD && namespacedReq {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
      
      // 请求CRD是命名空间级别,未指定命名空间,请求操作不允许跨命名空间,委托给下一级
  	if namespacedCRD && !namespacedReq && !possiblyAcrossAllNamespacesVerbs.Has(requestInfo.Verb) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  
      // CRD请求版本不合法(未建立状态/请求版本未声明),委托给下一级
  	if !apiextensionshelpers.HasServedCRDVersion(crd, requestInfo.APIVersion) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  
  	// CRD命名未接受及未建立状态,委托给下一级
  	if !apiextensionshelpers.IsCRDConditionTrue(crd, apiextensionsv1.NamesAccepted) &&
  		!apiextensionshelpers.IsCRDConditionTrue(crd, apiextensionsv1.Established) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  
      // 检查CRD删除状态
  	terminating := apiextensionshelpers.IsCRDConditionTrue(crd, apiextensionsv1.Terminating)
  
      // 根据crd获取对应的crdInfo
  	crdInfo, err := r.getOrCreateServingInfoFor(crd.UID, crd.Name)
      // crdInfo未找到,委托给下一级
  	if apierrors.IsNotFound(err) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  	...
      
      // 请求版本CRD未提供,委托给下一级
  	if !hasServedCRDVersion(crdInfo.spec, requestInfo.APIVersion) {
  		r.delegate.ServeHTTP(w, req)
  		return
  	}
  	...
  
  	// 请求的CRD子资源定义
  	subresources, err := apiextensionshelpers.GetSubresourcesForVersion(crd, requestInfo.APIVersion)
  	...
      
  	switch {
      // status子资源处理
  	case subresource == "status" && subresources != nil && subresources.Status != nil:
  		handlerFunc = r.serveStatus(w, req, requestInfo, crdInfo, terminating, supportedTypes)
      // scale子资源处理
  	case subresource == "scale" && subresources != nil && subresources.Scale != nil:
  		handlerFunc = r.serveScale(w, req, requestInfo, crdInfo, terminating, supportedTypes)
      // 主资源处理
  	case len(subresource) == 0:
  		handlerFunc = r.serveResource(w, req, requestInfo, crdInfo, crd, terminating, supportedTypes)
  	default:
  		...
  	}
  
  	if handlerFunc != nil {
  		...
          // handlerFunc包装一下执行
  		handler.ServeHTTP(w, req)
  		return
  	}
  }
  

• 补充

  crdHandler.ServeHttp()会检查请求合法性，CR相关的资源请求会执行serveResource进行资源或子资源处理

2.3.crdInfo

• crd.getOrCreateServingInfoFor()用于获取或创建CRD的Serving Info对象，用于API Server对CRD的请求分发、验证及序列化。

• // getOrCreateServingInfoFor gets the CRD serving info for the given CRD UID.
  func (r *crdHandler) getOrCreateServingInfoFor(uid types.UID, name string) (*crdInfo, error) {
      // 先从缓存获取
  	storageMap := r.customStorage.Load().(crdStorageMap)
  	if ret, ok := storageMap[uid]; ok {
  		return ret, nil
  	}
  
  	r.customStorageLock.Lock()
  	defer r.customStorageLock.Unlock()
  
  	// 获取同步的crd对象
  	crd, err := r.crdLister.Get(name)
  	...
      
      // UID可能更新,基于最新的CRD.UID再次尝试从缓存获取
  	storageMap = r.customStorage.Load().(crdStorageMap)
  	if ret, ok := storageMap[crd.UID]; ok {
  		return ret, nil
  	}
  
      // 获取crd版本
  	storageVersion, err := apiextensionshelpers.GetCRDStorageVersion(crd)
  	...
  
  	// 解析各版本CrdInfo
  	for _, v := range crd.Spec.Versions {
          // 获取CRD版本对应schema
  		vl, err := apiextensionshelpers.GetSchemaForVersion(crd, v.Name)
  		...
          
  		if vl == nil {
  			continue
  		}
          ...
          
          // 外部版本转为内部版本的校验函数
  	    Convert_v1_CustomResourceValidation_To_apiextensions_CustomResourceValidation(vl,internalValidation,nil)
          ...
          
          // 构造内部通用数据结构(openAPI schema的树形结构)
  		s, err := structuralschema.NewStructural(internalValidation.OpenAPIV3Schema)
  		...
  		structuralSchemas[v.Name] = s
  	}
  
      // 构建openAPI模型
  	openAPIModels, err := buildOpenAPIModelsForApply(r.staticOpenAPISpec, crd)
  	...
      // 构建typeConverter(比较字段树/合并patch/追踪managed fields)
      typeConverter, err = managedfields.NewTypeConverter(openAPIModels, crd.Spec.PreserveUnknownFields)
  
      // 构建converter(前者转换合法的schema定义字段/后者更宽松的转换)
  	safeConverter, unsafeConverter, err := r.converterFactory.NewConverter(crd)
  	...
      
  	for _, v := range crd.Spec.Versions {
          // 解析版本对应subresource
  		subresources, err := apiextensionshelpers.GetSubresourcesForVersion(crd, v.Name)
  		...
          
          // 取出用户定义的specReplicasPath(.spec.replicas)
  		splitReplicasPath := strings.Split(strings.TrimPrefix(subresources.Scale.SpecReplicasPath, "."), ".")
  		// 重新格式化
          for _, element := range splitReplicasPath {
  			s := element
  			path = append(path, fieldpath.PathElement{FieldName: &s})
  		}
          // 记录到副本定义缓存
  		replicasPathInCustomResource[schema.GroupVersion{Group: crd.Spec.Group, Version: v.Name}.String()] = path
  	}
  
      // 构建CRD版本对应storage/scope/子资源处理器
  	for _, v := range crd.Spec.Versions {
          ...
  
          // 获取打印列
  		columns, err := getColumnsForVersion(crd, v.Name)
  		...
  		table, err := tableconvertor.New(columns)
  		...
  
          // 构建存储(CR读写/校验/转换及subresource管理)
          storages[v.Name] = customresource.NewStorage(...)
  		...
  
          // 构建请求作用域(序列化/权限/字段管理/作用域)
  		reqScope := handlers.RequestScope{...}
  		...
  		requestScopes[v.Name] = &reqScope
  		...
  
          // 构建子资源作用域(序列化/校验/转换)
  		scaleScopes[v.Name] = &scaleScope
  		...
  		statusScopes[v.Name] = &statusScope
  		...
  	}
  
      // 构建及缓存crdInfo
  	ret := &crdInfo{...}
  	storageMap2 := storageMap.clone()
  	storageMap2[crd.UID] = ret
  	r.customStorage.Store(storageMap2)
  
  	return ret, nil
  }
  

• 补充

  crdInfo对象主要保存CR资源的请求及定义元数据和存储后端

2.4.newStorage

• NewStorage()用于创建CRD资源相关的存储后端，内部维护CRD资源及子资源的数据接收结构、scheme元数据、CRD REST存储对象。

• func NewStorage(...) CustomResourceStorage {
  	...
      
  	store := &genericregistry.Store{
          // CR实例获取(基于无结构体)
  		NewFunc: func() runtime.Object {
  			ret := &unstructured.Unstructured{}
  			ret.SetGroupVersionKind(kind)
  			return ret
  		},
  		NewListFunc: func() runtime.Object {
  			// lists are never stored, only manufactured, so stomp in the right kind
  			ret := &unstructured.UnstructuredList{}
  			ret.SetGroupVersionKind(listKind)
  			return ret
  		},
  		PredicateFunc:             strategy.MatchCustomResourceDefinitionStorage,
  		DefaultQualifiedResource:  resource,
  		SingularQualifiedResource: singularResource,
  
  		CreateStrategy:      strategy,
  		UpdateStrategy:      strategy,
  		DeleteStrategy:      strategy,
  		ResetFieldsStrategy: strategy,
  
  		TableConvertor: tableConvertor,
  	}
  	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: strategy.GetAttrs}
  	...
  	storage.CustomResource = &REST{store, categories}
  
      // status子资源存储构建
  	if strategy.status != nil {
  		statusStore := *store
  		statusStrategy := NewStatusStrategy(strategy)
  		statusStore.UpdateStrategy = statusStrategy
  		statusStore.ResetFieldsStrategy = statusStrategy
  		storage.Status = &StatusREST{store: &statusStore}
  	}
  
      // scale子资源存储构建
  	if scale := strategy.scale; scale != nil {
  		...
  		storage.Scale = &ScaleREST{
  			store:               store,
  			specReplicasPath:    scale.SpecReplicasPath,
  			statusReplicasPath:  scale.StatusReplicasPath,
  			labelSelectorPath:   labelSelectorPath,
  			parentGV:            kind.GroupVersion(),
  			replicasPathMapping: replicasPathMapping,
  		}
  	}
  
  	return storage
  }
  

• 注意

  crdInfo比较重要的是NewFunc/NewListFunc和optsGetter，前者用于构造接收对象Unstructured，后者作为存储后端

2.5.serveResource

• serveResource()用于结合crdInfo处理CR主资源请求，根据请求方式及crdInfo维护的storage查询或更新CR资源对象。

• func (r *crdHandler) serveResource(...) http.HandlerFunc {
      // 请求作用域
  	requestScope := crdInfo.requestScopes[requestInfo.APIVersion]
      // CR存储
  	storage := crdInfo.storages[requestInfo.APIVersion].CustomResource
  
  	switch requestInfo.Verb {
  	case "get":
  		return handlers.GetResource(storage, requestScope)
  	case "list":
  		return handlers.ListResource(storage, storage, requestScope, false, r.minRequestTimeout)
  	case "watch":
  		return handlers.ListResource(storage, storage, requestScope, true, r.minRequestTimeout)
  	case "create":
  		// 刚创建完成,延迟2s
  		if justCreated {
  			time.Sleep(2 * time.Second)
  		}
  		...
  		return handlers.CreateResource(storage, requestScope, r.admission)
  	case "update":
  		return handlers.UpdateResource(storage, requestScope, r.admission)
  	case "patch":
  		return handlers.PatchResource(storage, requestScope, r.admission, supportedTypes)
  	case "delete":
  		return handlers.DeleteResource(storage, true, requestScope, r.admission)
  	case "deletecollection":
  		return handlers.DeleteCollection(storage, true, requestScope, r.admission)
  	default:
  		...
  		return nil
  	}
  }
  
  func ListResource(...) http.HandlerFunc {
  	return func(w http.ResponseWriter, req *http.Request) {
  		...
  
          // 获取请求namespace
  		namespace, err := scope.Namer.Namespace(req)
  		...
          
          // 获取请求name
  		_, name, err := scope.Namer.Name(req)
  		...
  
          // 响应类型
  		outputMediaType, _, err := negotiation.NegotiateOutputMediaType(req, scope.Serializer, scope)
  		...
          
          // 请求参数
  		metainternalversionscheme.ParameterCodec.DecodeParameters(req.URL.Query(), scope.MetaGroupVersion,&opts)
          ...
  
  		// fieldSelector转换
  		if opts.FieldSelector != nil {
              // 请求的字段名转换为内部字段名
  			fn := func(label, value string) (newLabel, newValue string, err error) {
  				return scope.Convertor.ConvertFieldLabel(scope.Kind, label, value)
  			}
  			opts.FieldSelector, err = opts.FieldSelector.Transform(fn)
              ...
  		}
  
          // 请求到具体name
  		if hasName {
  			nameSelector := fields.OneTermEqualSelector("metadata.name", name)
  			// 校验selector条件
  			if opts.FieldSelector != nil && !opts.FieldSelector.Empty() {
  				selectedName, ok := opts.FieldSelector.RequiresExactMatch("metadata.name")
  				if !ok || name != selectedName {
  					return
  				}
  			} else {
  				opts.FieldSelector = nameSelector
  			}
  		}
  
          // 开启watch监听
  		if opts.Watch || forceWatch {
  			...
              
              // 激活底层watch流
  			watcher, err := rw.Watch(ctx, &opts)
  			...
  
  			// 响应watch流			
              serveWatch(watcher, scope, outputMediaType, req, w, timeout)
  			return
  		}
  		...
          
          // 执行存储查询
  		result, err := r.List(ctx, &opts)
  		...
          
          // 序列化数据及响应
  		transformResponseObject(ctx, scope, req, w, http.StatusOK, outputMediaType, result)
  	}
  }
  
  // List returns a list of items matching labels and field according to the
  // store's PredicateFunc.
  func (e *Store) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
  	...
      
      // label条件
      label = options.LabelSelector
  	...
      
      // field字段筛选条件
      field = options.FieldSelector
      ...
      
      // 执行查询
  	out, err := e.ListPredicate(ctx, e.PredicateFunc(label, field), options)
  	...
      
      // 执行修饰回调
  	if e.Decorator != nil {
  		e.Decorator(out)
  	}
  	return out, nil
  }
  
  // ListPredicate returns a list of all the items matching the given SelectionPredicate.
  func (e *Store) ListPredicate(...) (runtime.Object, error) {
      // 查询条件初始化
  	if options == nil {
  		// By default we should serve the request from etcd.
  		options = &metainternalversion.ListOptions{ResourceVersion: ""}
  	}
  	p.Limit = options.Limit
  	p.Continue = options.Continue
      
      // 创建列表对象
  	list := e.NewListFunc()
  	// 资源GVR信息
      qualifiedResource := e.qualifiedResourceFromContext(ctx)
      // 构造list查询参数
  	storageOpts := storage.ListOptions{
  		ResourceVersion:      options.ResourceVersion,
  		ResourceVersionMatch: options.ResourceVersionMatch,
  		Predicate:            p,
  		Recursive:            true,
  	}
  
  	// 请求上下文的namespace为空
  	if requestNamespace, _ := genericapirequest.NamespaceFrom(ctx); len(requestNamespace) == 0 {
          // fieldSelector筛选了namespace
  		if selectorNamespace, ok := p.MatchesSingleNamespace(); ok {
              // 校验namespace名称合法性
  			if len(validation.ValidateNamespaceName(selectorNamespace, false)) == 0 {
                  // 设置到请求上下文
  				ctx = genericapirequest.WithNamespace(ctx, selectorNamespace)
  			}
  		}
  	}
  
      // fieldSelector筛选了name
  	if name, ok := p.MatchesSingle(); ok {
          // 基于请求上下文生成查询索引(/registry/pods/<namespace>/<name>)
          if key, err := e.KeyFunc(ctx, name); err == nil {
              // 关闭递归查询
  			storageOpts.Recursive = false
              // 执行查询
  			err := e.Storage.GetList(ctx, key, storageOpts, list)
  			return list, storeerr.InterpretListError(err, qualifiedResource)
  		}
  		// if we cannot extract a key based on the current context, the optimization is skipped
  	}
  
      // list查询
  	err := e.Storage.GetList(ctx, e.KeyRootFunc(ctx), storageOpts, list)
  	return list, storeerr.InterpretListError(err, qualifiedResource)
  }
  
  func (s *DryRunnableStorage) GetList(...) error {
      // 这里会有两层实现,分别是storage和cacher,先以storage分析,list-watch机制再讲cacher
  	return s.Storage.GetList(ctx, key, opts, listObj)
  }
  
  // GetList implements storage.Interface.
  func (s *store) GetList(ctx context.Context, key string,opts storage.ListOptions,listObj runtime.Object) error {
      // key标准化(name/namespace)
      preparedKey, err := s.prepareKey(key)
  	...
      
      // 获取列表items指针
  	listPtr, err := meta.GetItemsPtr(listObj)
  	...
  	v, err := conversion.EnforcePtr(listPtr)
  	...
  
  	// 递归查询,key以/结尾
  	if recursive && !strings.HasSuffix(preparedKey, "/") {
  		preparedKey += "/"
  	}
  	keyPrefix := preparedKey
  	...
  
      // 对象初始化回调
  	newItemFunc := getNewItemFunc(listObj, v)
  	...
      // 分页条件及resourceVersion范围构造
      ...
  
  	
  	for {
  		...
          // 查询etcd数据
  		getResp, err = s.client.KV.Get(ctx, preparedKey, options...)
  		...
  
          // resourceVersion校验
  		s.validateMinimumResourceVersion(resourceVersion, uint64(getResp.Header.Revision))
          ...
          
  		// slice提前扩容
  		if pred.Empty() {
  			growSlice(v, len(getResp.Kvs))
  		} else {
  			growSlice(v, 2048, len(getResp.Kvs))
  		}
  
  		// KV结果处理
  		for i, kv := range getResp.Kvs {
  			...
  
              // 解码原始数据(解密)
  			data, _, err := s.transformer.TransformFromStorage(ctx, kv.Value, authenticatedDataString(kv.Key))
  			...
  
              // 追加到列表
  			appendListItem(v, data, uint64(kv.ModRevision), pred, s.codec, s.versioner, newItemFunc)
              ...
  
  			// 释放
  			getResp.Kvs[i] = nil
  		}
  		...
  	}
  	...
  
  	// 数据未查完
  	if hasMore {
  		// 下一次的起始索引
  		next, err := storage.EncodeContinue(string(lastKey)+"\x00", keyPrefix, returnedRV)
  		...
          
          // 向list对象打上metadata.resourceVersion和metadata.continue
  		return s.versioner.UpdateList(listObj, uint64(returnedRV), next, remainingItemCount)
  	}
  
  	// 向list对象打上metadata.resourceVersion
  	return s.versioner.UpdateList(listObj, uint64(returnedRV), "", nil)
  }
  

• 补充

  List查询涉及翻页会麻烦一些，其它修改操作和之前一样会结合admit准入检查