Kubernetes源碼分析之kube-apiserver

本節全部的代碼基於最新的1.13.4版本。html

啓動分析

同Kubernetes全部的組件啓動代碼一致,apiserver啓動使用的是cobra的命令行方式 node

如圖,啓動主要完成三個步驟:
一、完成參數的配置;
二、判斷配置是否合法;
三、執行最終的 Run方法。
Run方法比較簡單
如圖,主要執行兩個步驟:
一、建立server端;
二、啓動server。
由於apiserver本質上就是一個server服務器,全部代碼核心就是如何配置server,包括路由、訪問權限以及同數據庫(etcd)的交互等。先看一下server端是如何建立起來的

Server端建立

Server端的建立集中在CreateServerChain方法。方法代碼以下:git

// CreateServerChain creates the apiservers connected via delegation.
// CreateServerChain建立經過委託鏈接的apiservers,建立一系列的server
func CreateServerChain(completedOptions completedServerRunOptions, stopCh <-chan struct{}) (*genericapiserver.GenericAPIServer, error) {
	nodeTunneler, proxyTransport, err := CreateNodeDialer(completedOptions)
	if err != nil {
		return nil, err
	}

	// 1.建立kubeAPIServerConfig配置
	kubeAPIServerConfig, insecureServingInfo, serviceResolver, pluginInitializer, admissionPostStartHook, err := CreateKubeAPIServerConfig(completedOptions, nodeTunneler, proxyTransport)
	if err != nil {
		return nil, err
	}

	// If additional API servers are added, they should be gated.
	// 2.判斷是否配置了擴展API server,建立apiExtensionsConfig配置
	apiExtensionsConfig, err := createAPIExtensionsConfig(*kubeAPIServerConfig.GenericConfig, kubeAPIServerConfig.ExtraConfig.VersionedInformers, pluginInitializer, completedOptions.ServerRunOptions, completedOptions.MasterCount,
		serviceResolver, webhook.NewDefaultAuthenticationInfoResolverWrapper(proxyTransport, kubeAPIServerConfig.GenericConfig.LoopbackClientConfig))
	if err != nil {
		return nil, err
	}

	// apiExtensionsServer,可擴展的API server
	// 3.啓動擴展的API server
	apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegate())
	if err != nil {
		return nil, err
	}

	// 4.啓動最核心的kubeAPIServer
	kubeAPIServer, err := CreateKubeAPIServer(kubeAPIServerConfig, apiExtensionsServer.GenericAPIServer, admissionPostStartHook)
	if err != nil {
		return nil, err
	}

	// otherwise go down the normal path of standing the aggregator up in front of the API server
	// this wires up openapi
	kubeAPIServer.GenericAPIServer.PrepareRun()

	// This will wire up openapi for extension api server
	apiExtensionsServer.GenericAPIServer.PrepareRun()

	// aggregator comes last in the chain
	// 5.聚合層的配置aggregatorConfig
	aggregatorConfig, err := createAggregatorConfig(*kubeAPIServerConfig.GenericConfig, completedOptions.ServerRunOptions, kubeAPIServerConfig.ExtraConfig.VersionedInformers, serviceResolver, proxyTransport, pluginInitializer)
	if err != nil {
		return nil, err
	}
	// 6.aggregatorServer,聚合服務器,對全部的服務器訪問的整合
	aggregatorServer, err := createAggregatorServer(aggregatorConfig, kubeAPIServer.GenericAPIServer, apiExtensionsServer.Informers)
	if err != nil {
		// we don't need special handling for innerStopCh because the aggregator server doesn't create any go routines
		return nil, err
	}

	// 7.啓動非安全端口的server
	if insecureServingInfo != nil {
		insecureHandlerChain := kubeserver.BuildInsecureHandlerChain(aggregatorServer.GenericAPIServer.UnprotectedHandler(), kubeAPIServerConfig.GenericConfig)
		if err := insecureServingInfo.Serve(insecureHandlerChain, kubeAPIServerConfig.GenericConfig.RequestTimeout, stopCh); err != nil {
			return nil, err
		}
	}

	// 8.返回GenericAPIServer,後續啓動安全端口的server
	return aggregatorServer.GenericAPIServer, nil
}
複製代碼

建立過程主要有如下步驟:
一、根據配置構造apiserver的配置,調用方法CreateKubeAPIServerConfig
二、根據配置構造擴展的apiserver的配置,調用方法爲createAPIExtensionsConfig
三、建立server,包括擴展的apiserver和原生的apiserver,調用方法爲createAPIExtensionsServerCreateKubeAPIServer。主要就是將各個handler的路由方法註冊到Container中去,徹底遵循go-restful的設計模式,即將處理方法註冊到Route中去,同一個根路徑下的Route註冊到WebService中去,WebService註冊到Container中,Container負責分發。訪問的過程爲Container-->WebService-->Route。更加詳細的go-restful使用能夠參考其代碼;
四、聚合server的配置和和建立。主要就是將原生的apiserver和擴展的apiserver的訪問進行整合,添加後續的一些處理接口。調用方法爲createAggregatorConfigcreateAggregatorServer
五、建立完成,返回配置的server信息。
以上幾個步驟,最核心的就是apiserver如何建立,即如何按照go-restful的模式,添加路由和相應的處理方法,以CreateKubeAPIServer方法爲例,createAPIExtensionsServer相似。github

建立

CreateKubeAPIServer方法以下web

func CreateKubeAPIServer(kubeAPIServerConfig *master.Config, delegateAPIServer genericapiserver.DelegationTarget, admissionPostStartHook genericapiserver.PostStartHookFunc) (*master.Master, error) {
	kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer)
	if err != nil {
		return nil, err
	}

	kubeAPIServer.GenericAPIServer.AddPostStartHookOrDie("start-kube-apiserver-admission-initializer", admissionPostStartHook)

	return kubeAPIServer, nil
}
複製代碼

經過Complete方法完成配置的最終合法化,New方法生成kubeAPIServer的配置,進入New方法,數據庫

// New returns a new instance of Master from the given config.
// Certain config fields will be set to a default value if unset.
// Certain config fields must be specified, including:
// KubeletClientConfig
// 經過給定的配置,返回一個新的Master實例。對於部分未配置的選項,可使用默認配置;可是對於KubeletClientConfig這樣的配置,必須手動指定
func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget) (*Master, error) {
	if reflect.DeepEqual(c.ExtraConfig.KubeletClientConfig, kubeletclient.KubeletClientConfig{}) {
		return nil, fmt.Errorf("Master.New() called with empty config.KubeletClientConfig")
	}

	// 1.初始化,建立go-restful的Container,初始化apiServerHandler
	s, err := c.GenericConfig.New("kube-apiserver", delegationTarget)
	if err != nil {
		return nil, err
	}

	if c.ExtraConfig.EnableLogsSupport {
		routes.Logs{}.Install(s.Handler.GoRestfulContainer)
	}

	m := &Master{
		GenericAPIServer: s,
	}

	// install legacy rest storage
	// /api開頭的版本api註冊到Container中去,如Pod、Namespace等資源
	if c.ExtraConfig.APIResourceConfigSource.VersionEnabled(apiv1.SchemeGroupVersion) {
		legacyRESTStorageProvider := corerest.LegacyRESTStorageProvider{
			StorageFactory:              c.ExtraConfig.StorageFactory,
			ProxyTransport:              c.ExtraConfig.ProxyTransport,
			KubeletClientConfig:         c.ExtraConfig.KubeletClientConfig,
			EventTTL:                    c.ExtraConfig.EventTTL,
			ServiceIPRange:              c.ExtraConfig.ServiceIPRange,
			ServiceNodePortRange:        c.ExtraConfig.ServiceNodePortRange,
			LoopbackClientConfig:        c.GenericConfig.LoopbackClientConfig,
			ServiceAccountIssuer:        c.ExtraConfig.ServiceAccountIssuer,
			ServiceAccountMaxExpiration: c.ExtraConfig.ServiceAccountMaxExpiration,
			APIAudiences:                c.GenericConfig.Authentication.APIAudiences,
		}
		m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter, legacyRESTStorageProvider)
	}

	// The order here is preserved in discovery.
	// If resources with identical names exist in more than one of these groups (e.g. "deployments.apps"" and "deployments.extensions"),
	// the order of this list determines which group an unqualified resource name (e.g. "deployments") should prefer.
	// This priority order is used for local discovery, but it ends up aggregated in `k8s.io/kubernetes/cmd/kube-apiserver/app/aggregator.go
	// with specific priorities.
	// TODO: describe the priority all the way down in the RESTStorageProviders and plumb it back through the various discovery
	// handlers that we have.
	// /apis開頭版本的api註冊到Container中
	restStorageProviders := []RESTStorageProvider{
		auditregistrationrest.RESTStorageProvider{},
		authenticationrest.RESTStorageProvider{Authenticator: c.GenericConfig.Authentication.Authenticator, APIAudiences: c.GenericConfig.Authentication.APIAudiences},
		authorizationrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorization.Authorizer, RuleResolver: c.GenericConfig.RuleResolver},
		autoscalingrest.RESTStorageProvider{},
		batchrest.RESTStorageProvider{},
		certificatesrest.RESTStorageProvider{},
		coordinationrest.RESTStorageProvider{},
		extensionsrest.RESTStorageProvider{},
		networkingrest.RESTStorageProvider{},
		policyrest.RESTStorageProvider{},
		rbacrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorization.Authorizer},
		schedulingrest.RESTStorageProvider{},
		settingsrest.RESTStorageProvider{},
		storagerest.RESTStorageProvider{},
		// keep apps after extensions so legacy clients resolve the extensions versions of shared resource names.
		// See https://github.com/kubernetes/kubernetes/issues/42392
		appsrest.RESTStorageProvider{},
		admissionregistrationrest.RESTStorageProvider{},
		eventsrest.RESTStorageProvider{TTL: c.ExtraConfig.EventTTL},
	}
	m.InstallAPIs(c.ExtraConfig.APIResourceConfigSource, c.GenericConfig.RESTOptionsGetter, restStorageProviders...)

	if c.ExtraConfig.Tunneler != nil {
		m.installTunneler(c.ExtraConfig.Tunneler, corev1client.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig).Nodes())
	}

	m.GenericAPIServer.AddPostStartHookOrDie("ca-registration", c.ExtraConfig.ClientCARegistrationHook.PostStartHook)

	return m, nil
}
複製代碼

包含如下步驟:
一、按照go-restful的模式,調用c.GenericConfig.New方法初始化化Container,即gorestfulContainer,初始方法爲NewAPIServerHandler。初始化以後,添加路由。bootstrap

func installAPI(s *GenericAPIServer, c *Config) {

	// 添加"/"與"/index.html"路由
	if c.EnableIndex {
		routes.Index{}.Install(s.listedPathProvider, s.Handler.NonGoRestfulMux)
	}
	// 添加"/swagger-ui/"路由
	if c.SwaggerConfig != nil && c.EnableSwaggerUI {
		routes.SwaggerUI{}.Install(s.Handler.NonGoRestfulMux)
	}
	// 添加"/debug"相關路由
	if c.EnableProfiling {
		routes.Profiling{}.Install(s.Handler.NonGoRestfulMux)
		if c.EnableContentionProfiling {
			goruntime.SetBlockProfileRate(1)
		}
		// so far, only logging related endpoints are considered valid to add for these debug flags.
		routes.DebugFlags{}.Install(s.Handler.NonGoRestfulMux, "v", routes.StringFlagPutHandler(logs.GlogSetter))
	}
	// 添加"/metrics"路由
	if c.EnableMetrics {
		if c.EnableProfiling {
			routes.MetricsWithReset{}.Install(s.Handler.NonGoRestfulMux)
		} else {
			routes.DefaultMetrics{}.Install(s.Handler.NonGoRestfulMux)
		}
	}

	// 添加"/version"路由
	routes.Version{Version: c.Version}.Install(s.Handler.GoRestfulContainer)

	if c.EnableDiscovery {
		s.Handler.GoRestfulContainer.Add(s.DiscoveryGroupManager.WebService())
	}
}
複製代碼

該方法中添加了包括/、/swagger-ui、/debug/*、/metrics、/version幾條路由,經過訪問apiserver便可看到相關的信息 後端

二、判斷是否支持logs相關的路由,若是支持,則添加/logs路由;
三、添加以 api開頭的路由
在集羣中對應的路由有
即/api和/api/v1,比較經常使用的資源像Pods就是該路由對應的資源;
四、添加以 apis開頭的路由
在集羣中對應的路由有
能夠看到apis開頭的路由明顯較多。應該是因爲kubernetes設計之初的版本都是以api/v1開頭,後續擴展的版本以apis開頭命名。如今更多的是經過CRD與自定義Controller的方法擴展API,再也不進行api版本的擴展。基本上代碼中的名稱均可以在實際集羣中找到對應的API。

路由添加(api開頭)

api開頭的路由經過InstallLegacyAPI方法添加。進入InstallLegacyAPI方法,以下:設計模式

func (m *Master) InstallLegacyAPI(c *completedConfig, restOptionsGetter generic.RESTOptionsGetter, legacyRESTStorageProvider corerest.LegacyRESTStorageProvider) {
	legacyRESTStorage, apiGroupInfo, err := legacyRESTStorageProvider.NewLegacyRESTStorage(restOptionsGetter)
	if err != nil {
		klog.Fatalf("Error building core storage: %v", err)
	}

	controllerName := "bootstrap-controller"
	coreClient := corev1client.NewForConfigOrDie(c.GenericConfig.LoopbackClientConfig)
	bootstrapController := c.NewBootstrapController(legacyRESTStorage, coreClient, coreClient, coreClient)
	m.GenericAPIServer.AddPostStartHookOrDie(controllerName, bootstrapController.PostStartHook)
	m.GenericAPIServer.AddPreShutdownHookOrDie(controllerName, bootstrapController.PreShutdownHook)

	if err := m.GenericAPIServer.InstallLegacyAPIGroup(genericapiserver.DefaultLegacyAPIPrefix, &apiGroupInfo); err != nil {
		klog.Fatalf("Error in registering group versions: %v", err)
	}
}
複製代碼

經過NewLegacyRESTStorage方法建立各個資源的RESTStorage。RESTStorage是一個結構體,具體的定義在vendor/k8s.io/apiserver/pkg/registry/generic/registry/store.go下,結構體內主要包含NewFunc返回特定資源信息、NewListFunc返回特定資源列表、CreateStrategy特定資源建立時的策略、UpdateStrategy更新時的策略以及DeleteStrategy刪除時的策略等重要方法。
NewLegacyRESTStorage內部,能夠看到建立了多種資源的RESTStorage api

常見的像event、secret、namespace、endpoints等,統一調用 NewREST方法構造相應的資源。待全部資源的store建立完成以後,使用 restStorageMap的Map類型將每一個資源的路由和對應的store對應起來,方便後續去作路由的統一規劃,代碼以下:

restStorageMap := map[string]rest.Storage{
		"pods":             podStorage.Pod,
		"pods/attach":      podStorage.Attach,
		"pods/status":      podStorage.Status,
		"pods/log":         podStorage.Log,
		"pods/exec":        podStorage.Exec,
		"pods/portforward": podStorage.PortForward,
		"pods/proxy":       podStorage.Proxy,
		"pods/binding":     podStorage.Binding,
		"bindings":         podStorage.Binding,

		"podTemplates": podTemplateStorage,

		"replicationControllers":        controllerStorage.Controller,
		"replicationControllers/status": controllerStorage.Status,

		"services":        serviceRest,
		"services/proxy":  serviceRestProxy,
		"services/status": serviceStatusStorage,

		"endpoints": endpointsStorage,

		"nodes":        nodeStorage.Node,
		"nodes/status": nodeStorage.Status,
		"nodes/proxy":  nodeStorage.Proxy,

		"events": eventStorage,

		"limitRanges":                   limitRangeStorage,
		"resourceQuotas":                resourceQuotaStorage,
		"resourceQuotas/status":         resourceQuotaStatusStorage,
		"namespaces":                    namespaceStorage,
		"namespaces/status":             namespaceStatusStorage,
		"namespaces/finalize":           namespaceFinalizeStorage,
		"secrets":                       secretStorage,
		"serviceAccounts":               serviceAccountStorage,
		"persistentVolumes":             persistentVolumeStorage,
		"persistentVolumes/status":      persistentVolumeStatusStorage,
		"persistentVolumeClaims":        persistentVolumeClaimStorage,
		"persistentVolumeClaims/status": persistentVolumeClaimStatusStorage,
		"configMaps":                    configMapStorage,

		"componentStatuses": componentstatus.NewStorage(componentStatusStorage{c.StorageFactory}.serversToValidate),
	}
	if legacyscheme.Scheme.IsVersionRegistered(schema.GroupVersion{Group: "autoscaling", Version: "v1"}) {
		restStorageMap["replicationControllers/scale"] = controllerStorage.Scale
	}
	if legacyscheme.Scheme.IsVersionRegistered(schema.GroupVersion{Group: "policy", Version: "v1beta1"}) {
		restStorageMap["pods/eviction"] = podStorage.Eviction
	}
	if serviceAccountStorage.Token != nil {
		restStorageMap["serviceaccounts/token"] = serviceAccountStorage.Token
	}
	apiGroupInfo.VersionedResourcesStorageMap["v1"] = restStorageMap
複製代碼

最終完成以api開頭的全部資源的RESTStorage操做。
建立完以後,則開始進行路由的安裝,執行InstallLegacyAPIGroup方法,主要調用鏈爲InstallLegacyAPIGroup-->installAPIResources-->InstallREST-->Install-->registerResourceHandlers,最終核心的路由構造在registerResourceHandlers方法內。這是一個很是複雜的方法,整個方法的代碼在700行左右。方法的主要功能是經過上一步驟構造的RESTStorage判斷該資源能夠執行哪些操做(如create、update等),將其對應的操做存入到action,每個action對應一個標準的rest操做,如create對應的action操做爲POST、update對應的action操做爲PUT。最終根據actions數組依次遍歷,對每個操做添加一個handler方法,註冊到route中去,route註冊到webservice中去,完美匹配go-restful的設計模式。

路由添加(apis開頭)

api開頭的路由主要是對基礎資源的路由實現,而對於其餘附加的資源,如認證相關、網絡相關等各類擴展的api資源,統一以apis開頭命名,實現入口爲InstallAPIs
InstallAPIsInstallLegacyAPIGroup主要的區別是獲取RESTStorage的方式。對於api開頭的路由來講,都是/api/v1這種統一的格式;而對於apis開頭路由則不同,它包含了多種不一樣的格式(Kubernetes代碼內叫groupName),如/apis/apps、/apis/certificates.k8s.io等各類無規律的groupName。爲此,kubernetes提供了一種RESTStorageProvider的工廠模式的接口

// RESTStorageProvider is a factory type for REST storage.
type RESTStorageProvider interface {
	GroupName() string
	NewRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (genericapiserver.APIGroupInfo, bool)
}
複製代碼

全部以apis開頭的路由的資源都須要實現該接口。GroupName()方法獲取到的就是相似於/apis/apps、/apis/certificates.k8s.io這樣的groupName,NewRESTStorage方法獲取到的是相對應的RESTStorage封裝後的信息。各類資源的NewRESTStorage接口實現如圖:

後續的操做和以前的步驟相似,經過構造go-restful格式的路由信息,完成建立,此處不作贅述。

Server端啓動

經過CreateServerChain建立完server後,繼續調用GenericAPIServer的Run方法完成最終的啓動工做。首先經過PrepareRun方法完成啓動前的路由收尾工做,該方法主要完成了SwaggerOpenAPI路由的註冊工做(SwaggerOpenAPI主要包含了Kubernetes API的全部細節與規範),並完成/healthz路由的註冊工做。完成後,開始最終的server啓動工做。
Run方法裏經過NonBlockingRun方法啓動安全的http server(非安全方式的啓動在CreateServerChain方法已經完成)

// Run spawns the secure http server. It only returns if stopCh is closed
// or the secure port cannot be listened on initially.
// Run方法會建立一個安全的http server。只有在stopCh關閉或最初沒法監聽安全端口時返回
func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
	// NonBlockingRun建立一個安全的http server
	err := s.NonBlockingRun(stopCh)
	if err != nil {
		return err
	}

	<-stopCh

	// 接收到stopCh以後的處理動做
	err = s.RunPreShutdownHooks()
	if err != nil {
		return err
	}

	// Wait for all requests to finish, which are bounded by the RequestTimeout variable.
	s.HandlerChainWaitGroup.Wait()

	return nil
}
複製代碼

啓動主要工做包括配置各類證書認證、時間參數、報文大小參數之類,以後經過調用net/http庫的啓動方式啓動,代碼比較簡潔,不一一列出了。

權限相關

ApiServer中與權限相關的主要有三種機制,即經常使用的認證鑑權准入控制。對apiserver來講,主要提供的就是rest風格的接口,因此各類權限最終仍是集中到對接口的權限判斷上。
以最核心的kubeAPIServerConfig舉例,在CreateServerChain方法中,調用了CreateKubeAPIServerConfig的方法,該方法主要的做用是建立kubeAPIServer的配置。進入該方法,調用了buildGenericConfig建立一些通用的配置,在NewConfig下,返回了DefaultBuildHandlerChain,該方法主要就是用來對apiserver rest接口的鏈式判斷,即俗稱的filter操做,先記錄下,後續分析。

func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
	handler := genericapifilters.WithAuthorization(apiHandler, c.Authorization.Authorizer, c.Serializer)
	handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)
	handler = genericapifilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
	handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyChecker, c.LongRunningFunc)
	failedHandler := genericapifilters.Unauthorized(c.Serializer, c.Authentication.SupportsBasicAuth)
	failedHandler = genericapifilters.WithFailedAuthenticationAudit(failedHandler, c.AuditBackend, c.AuditPolicyChecker)
	handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences)
	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc, c.RequestTimeout)
	handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.HandlerChainWaitGroup)
	handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver)
	handler = genericfilters.WithPanicRecovery(handler)
	return handler
}
複製代碼

配置文件建立完成後,再進行建立工做,進入到CreateKubeAPIServer方法,在初始化go-restful的Container的方法內,能夠看到

handlerChainBuilder方法就是對返回的 DefaultBuildHandlerChain方法的一種封裝,並做爲參數傳入到 NewAPIServerHandler方法內。進入 NewAPIServerHandler方法,以下:

func NewAPIServerHandler(name string, s runtime.NegotiatedSerializer, handlerChainBuilder HandlerChainBuilderFn, notFoundHandler http.Handler) *APIServerHandler {
	nonGoRestfulMux := mux.NewPathRecorderMux(name)
	if notFoundHandler != nil {
		nonGoRestfulMux.NotFoundHandler(notFoundHandler)
	}

	gorestfulContainer := restful.NewContainer()
	gorestfulContainer.ServeMux = http.NewServeMux()
	gorestfulContainer.Router(restful.CurlyRouter{}) // e.g. for proxy/{kind}/{name}/{*}
	gorestfulContainer.RecoverHandler(func(panicReason interface{}, httpWriter http.ResponseWriter) {
		logStackOnRecover(s, panicReason, httpWriter)
	})
	gorestfulContainer.ServiceErrorHandler(func(serviceErr restful.ServiceError, request *restful.Request, response *restful.Response) {
		serviceErrorHandler(s, serviceErr, request, response)
	})

	director := director{
		name:               name,
		goRestfulContainer: gorestfulContainer,
		nonGoRestfulMux:    nonGoRestfulMux,
	}

	return &APIServerHandler{
		FullHandlerChain:   handlerChainBuilder(director),
		GoRestfulContainer: gorestfulContainer,
		NonGoRestfulMux:    nonGoRestfulMux,
		Director:           director,
	}
}
複製代碼

配置中經過將director做爲參數傳到handlerChainBuilder的回調方法內,完成對gorestfulContainer的handler的註冊工做。其實director就是一個實現了http.Handler的變量。因此,整個的處理邏輯就是將類型爲http.Handler的director做爲參數,傳遞到鏈式filterDefaultBuildHandlerChain方法內。經過DefaultBuildHandlerChain對每個步驟的filter操做,完成權限控制等之類的操做。如何經過net/http包實現filter的功能,能夠參考這篇文章。完成相似於filter的功能以後,後續就是作啓動工做,包括證書驗證、TLS認證之類的工做,不作過多贅述。主要看下filterDefaultBuildHandlerChain方法是如何處理接口的鑑權操做。

RBAC啓動

Kubernetes中比較重要的用的比較多的可能就是RBAC了。在DefaultBuildHandlerChain方法內,經過調用genericapifilters.WithAuthorization方法,實現對每一個接口的權限的filter操做。WithAuthorization方法以下

func WithAuthorization(handler http.Handler, a authorizer.Authorizer, s runtime.NegotiatedSerializer) http.Handler {
	if a == nil {
		klog.Warningf("Authorization is disabled")
		return handler
	}
	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		ctx := req.Context()
		ae := request.AuditEventFrom(ctx)

		attributes, err := GetAuthorizerAttributes(ctx)
		if err != nil {
			responsewriters.InternalError(w, req, err)
			return
		}
		authorized, reason, err := a.Authorize(attributes)
		// an authorizer like RBAC could encounter evaluation errors and still allow the request, so authorizer decision is checked before error here.
		if authorized == authorizer.DecisionAllow {
			audit.LogAnnotation(ae, decisionAnnotationKey, decisionAllow)
			audit.LogAnnotation(ae, reasonAnnotationKey, reason)
			handler.ServeHTTP(w, req)
			return
		}
		if err != nil {
			audit.LogAnnotation(ae, reasonAnnotationKey, reasonError)
			responsewriters.InternalError(w, req, err)
			return
		}

		klog.V(4).Infof("Forbidden: %#v, Reason: %q", req.RequestURI, reason)
		audit.LogAnnotation(ae, decisionAnnotationKey, decisionForbid)
		audit.LogAnnotation(ae, reasonAnnotationKey, reason)
		responsewriters.Forbidden(ctx, attributes, w, req, reason, s)
	})
}
複製代碼

一、調用GetAuthorizerAttributes方法獲取配置的各類屬性值;
二、調用Authorize方法判斷權限是否經過,不一樣的權限實現其接口,完成鑑權任務;

能夠看到,就包含有RBAC單獨的處理;
三、若是鑑權成功經過,則調用 handler.ServeHTTP方法繼續下一步的 filter操做;不然,直接返回錯誤信息。
以RBAC爲例, Authorize方法最終調用 VisitRulesFor方法實現權限的判斷,方法在 kubernetes/pkg/registry/rbac/validation/rule.go文件內。 VisitRulesFor主要代碼以下

func (r *DefaultRuleResolver) VisitRulesFor(user user.Info, namespace string, visitor func(source fmt.Stringer, rule *rbacv1.PolicyRule, err error) bool) {
	if clusterRoleBindings, err := r.clusterRoleBindingLister.ListClusterRoleBindings(); err != nil {
		if !visitor(nil, nil, err) {
			return
		}
	} else {
		sourceDescriber := &clusterRoleBindingDescriber{}
		for _, clusterRoleBinding := range clusterRoleBindings {
			subjectIndex, applies := appliesTo(user, clusterRoleBinding.Subjects, "")
			if !applies {
				continue
			}
			rules, err := r.GetRoleReferenceRules(clusterRoleBinding.RoleRef, "")
			if err != nil {
				if !visitor(nil, nil, err) {
					return
				}
				continue
			}
			sourceDescriber.binding = clusterRoleBinding
			sourceDescriber.subject = &clusterRoleBinding.Subjects[subjectIndex]
			for i := range rules {
				if !visitor(sourceDescriber, &rules[i], nil) {
					return
				}
			}
		}
	}

	if len(namespace) > 0 {
		if roleBindings, err := r.roleBindingLister.ListRoleBindings(namespace); err != nil {
			if !visitor(nil, nil, err) {
				return
			}
		} else {
			sourceDescriber := &roleBindingDescriber{}
			for _, roleBinding := range roleBindings {
				subjectIndex, applies := appliesTo(user, roleBinding.Subjects, namespace)
				if !applies {
					continue
				}
				rules, err := r.GetRoleReferenceRules(roleBinding.RoleRef, namespace)
				if err != nil {
					if !visitor(nil, nil, err) {
						return
					}
					continue
				}
				sourceDescriber.binding = roleBinding
				sourceDescriber.subject = &roleBinding.Subjects[subjectIndex]
				for i := range rules {
					if !visitor(sourceDescriber, &rules[i], nil) {
						return
					}
				}
			}
		}
	}
}
複製代碼

主要工做就是對clusterRoleBinding以及roleBinding與配置的資源進行判斷,比較清晰明瞭,這與咱們使用RBAC的思路基本一致。

數據庫操做

ApiServer與數據庫的交互主要指的是與etcd的交互。Kubernetes全部的組件不直接與etcd交互,都是經過請求apiserver,apiserver與etcd進行交互完成數據的最終落盤。
在以前的路由實現已經說過,apiserver最終實現的handler對應的後端數據是以Store的結構保存的。這裏以api開頭的路由舉例,在NewLegacyRESTStorage方法中,經過NewREST或者NewStorage會生成各類資源對應的Storage,以endpoints爲例,生成的方法以下

func NewREST(optsGetter generic.RESTOptionsGetter) *REST {
	store := &genericregistry.Store{
		NewFunc:                  func() runtime.Object { return &api.Endpoints{} },
		NewListFunc:              func() runtime.Object { return &api.EndpointsList{} },
		DefaultQualifiedResource: api.Resource("endpoints"),

		CreateStrategy: endpoint.Strategy,
		UpdateStrategy: endpoint.Strategy,
		DeleteStrategy: endpoint.Strategy,

		TableConvertor: printerstorage.TableConvertor{TablePrinter: printers.NewTablePrinter().With(printersinternal.AddHandlers)},
	}
	options := &generic.StoreOptions{RESTOptions: optsGetter}
	if err := store.CompleteWithOptions(options); err != nil {
		panic(err) // TODO: Propagate error up
	}
	return &REST{store}
}
複製代碼

主要看CompleteWithOptions方法,在CompleteWithOptions方法內,調用了RESTOptions的GetRESTOptions方法,依次調用StorageWithCacher-->NewRawStorage-->Create方法建立最終依賴的後端存儲:

// Create creates a storage backend based on given config.
func Create(c storagebackend.Config) (storage.Interface, DestroyFunc, error) {
	switch c.Type {
	case "etcd2":
		return nil, nil, fmt.Errorf("%v is no longer a supported storage backend", c.Type)
	case storagebackend.StorageTypeUnset, storagebackend.StorageTypeETCD3:
		return newETCD3Storage(c)
	default:
		return nil, nil, fmt.Errorf("unknown storage type: %s", c.Type)
	}
}
複製代碼

能夠看到,經過Create方法判斷是建立etcd2或是etcd3的後端etcd版本,目前版本默認的是etcd3。
建立完成對應的存儲以後,接下來要作的工做就是將對應的handler方法和最終的後臺存儲實現綁定起來(handler方法處理最終的數據須要落盤)。
還記着以前說的有個比較長的方法registerResourceHandlers,用來處理具體的handler路由。再次回到該方法,

如圖,每一個case語句代碼一種請求類型和相對應的handler方法,以 POST方法爲例,對應的是Create操做。
handler參數的調用最終都會走到 createHandler方法處,位於 kubernetes/vendor/k8s.io/apiserver/pkg/endpoints/handlers/crete.go下。最核心的步驟即調用了Create方法
一步步走下去,最終調用的是位於 kubernetes/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go下的 Create方法。該方法主要包含 BeforeCreateStorage.CreateAfterCreate以及 Decorator等主要方法。對應於 POST操做,則最主要的方法爲 Storage.Create。因爲目前使用基本都是etcd3,因此實現的方法以下

func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
	if version, err := s.versioner.ObjectResourceVersion(obj); err == nil && version != 0 {
		return errors.New("resourceVersion should not be set on objects to be created")
	}
	if err := s.versioner.PrepareObjectForStorage(obj); err != nil {
		return fmt.Errorf("PrepareObjectForStorage failed: %v", err)
	}
	data, err := runtime.Encode(s.codec, obj)
	if err != nil {
		return err
	}
	key = path.Join(s.pathPrefix, key)

	opts, err := s.ttlOpts(ctx, int64(ttl))
	if err != nil {
		return err
	}

	newData, err := s.transformer.TransformToStorage(data, authenticatedDataString(key))
	if err != nil {
		return storage.NewInternalError(err.Error())
	}

	txnResp, err := s.client.KV.Txn(ctx).If(
		notFound(key),
	).Then(
		clientv3.OpPut(key, string(newData), opts...),
	).Commit()
	if err != nil {
		return err
	}
	if !txnResp.Succeeded {
		return storage.NewKeyExistsError(key, 0)
	}

	if out != nil {
		putResp := txnResp.Responses[0].GetResponsePut()
		return decode(s.codec, s.versioner, data, out, putResp.Header.Revision)
	}
	return nil
}
複製代碼

主要操做爲:
一、調用Encode方法序列化;
二、調用path.Join解析Key;
三、調用TransformToStorage將數據類型進行轉換;
四、調用客戶端方法進行etcd的寫入操做。
至此,完成handler處理與對應的etcd數據庫操做的綁定,即完成整個路由後端的操做步驟。
對etcd操做更具體的能夠參考這篇文章

以上均爲我的學習總結,若是錯誤歡迎指正!

相關文章
相關標籤/搜索