Kubernetes API Server handler 注册过程分析

复制// k8s.io/apiserver/pkg/server/config.go func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler { handler := filterlatency.TrackCompleted(apiHandler) // 构造权限检查filter handler = genericapifilters.WithAuthorization(handler, c.Authorization.Authorizer, c.Serializer) ... // 构造认证filter handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences) ... // 构造请求超时filter, LongRunningFunc会判断该请求是否是需要LongRunning的，比如watch的请求，如果是，该filter不会对这类请求生效 // WithTimeoutForNonLongRunningRequests will call the rest of the request handling in a go-routine with the // context with deadline. The go-routine can keep running, while the timeout logic will return a timeout to the client. handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc) handler = genericapifilters.WithRequestDeadline(handler, c.AuditBackend, c.AuditPolicyRuleEvaluator, c.LongRunningFunc, c.Serializer, c.RequestTimeout) handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.HandlerChainWaitGroup) ... // 初始化RequestInfo的filter并将其放入context中，后续的处理逻辑可以从context直接获取RequestInfo handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver) .... return handler }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.

复制//pkg/controlplane/instance.go func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget) (*Instance, error) { ... // 构造内置资源的RESTStorageProvider restStorageProviders := []RESTStorageProvider{ apiserverinternalrest.StorageProvider{}, 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{}, discoveryrest.StorageProvider{}, networkingrest.RESTStorageProvider{}, noderest.RESTStorageProvider{}, policyrest.RESTStorageProvider{}, rbacrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorization.Authorizer}, schedulingrest.RESTStorageProvider{}, storagerest.RESTStorageProvider{}, flowcontrolrest.RESTStorageProvider{InformerFactory: c.GenericConfig.SharedInformerFactory}, // keep apps after extensions so legacy clients resolve the extensions versions of shared resource names. // See https://github.com/kubernetes/kubernetes/issues/42392 appsrest.StorageProvider{}, admissionregistrationrest.RESTStorageProvider{}, eventsrest.RESTStorageProvider{TTL: c.ExtraConfig.EventTTL}, } if err := m.InstallAPIs(c.ExtraConfig.APIResourceConfigSource, c.GenericConfig.RESTOptionsGetter, restStorageProviders...); err != nil { return nil, err } ... }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.

复制//k8s.io/apiserver/pkg/server/genericapiserver.go // Info about an API group. type APIGroupInfo struct { PrioritizedVersions []schema.GroupVersion // Info about the resources in this group. Its a map from version to resource to the storage. VersionedResourcesStorageMap map[string]map[string]rest.Storage ... // NegotiatedSerializer controls how this group encodes and decodes data NegotiatedSerializer runtime.NegotiatedSerializer // ParameterCodec performs conversions for query parameters passed to API calls ParameterCodec runtime.ParameterCodec ... }1.2.3.4.5.6.7.8.9.10.11.12.13.14.

复制// k8s.io/apiserver/pkg/registry/rest/rest.go // Storage is a generic interface for RESTful storage services. // Resources which are exported to the RESTful API of apiserver need to implement this interface. It is expected // that objects may implement any of the below interfaces. type Storage interface { // New returns an empty object that can be used with Create and Update after request data has been put into it. // This object must be a pointer type for use with Codec.DecodeInto([]byte, runtime.Object) New() runtime.Object // Destroy cleans up its resources on shutdown. // Destroy has to be implemented in thread-safe way and be prepared // for being called more than once. Destroy() }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.

复制// k8s.io/apiserver/pkg/registry/rest/rest.go // 资源对象支持POST请求，例入通过kubectl create一个资源对象。 // Creater is an object that can create an instance of a RESTful object. type Creater interface { // New returns an empty object that can be used with Create after request data has been put into it. // This object must be a pointer type for use with Codec.DecodeInto([]byte, runtime.Object) New() runtime.Object // Create creates a new version of a resource. Create(ctx context.Context, obj runtime.Object, createValidation ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) } // 资源对象支持GET请求，例如通过kubectl get 一个资源对象。 // Getter is an object that can retrieve a named RESTful resource. type Getter interface { // Get finds a resource in the storage by name and returns it. // Although it can return an arbitrary error value, IsNotFound(err) is true for the // returned error value err when the specified resource is not found. Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) } // 支持对资源对象进行watch操作 例如通过kubectl get 资源对象 -w。 type Watcher interface { // label selects on labels; field selects on the objects fields. Not all fields // are supported; an error should be returned if field tries to select on a field that // isnt supported. resourceVersion allows for continuing/starting a watch at a // particular version. Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.

复制// NewREST returns a RESTStorage object that will work against deployments. func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST, *RollbackREST, error) { // 创建一个deployments的genericregistry.Store store := &genericregistry.Store{ // 初始化一个空资源对象，这里使用的是internal的版本，下面定义的各种strategy操作的对象也是internal版本，这样就不用为每一种版本编写一个strategy策略 NewFunc: func() runtime.Object { return &apps.Deployment{} }, // 初始化一个空资源对象列表 NewListFunc: func() runtime.Object { return &apps.DeploymentList{} }, DefaultQualifiedResource: apps.Resource("deployments"), // 创建更新删除策略 主要是做校验及控制那些字段不能被用户覆盖用 CreateStrategy: deployment.Strategy, UpdateStrategy: deployment.Strategy, DeleteStrategy: deployment.Strategy, ResetFieldsStrategy: deployment.Strategy, TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)}, } options := &generic.StoreOptions{RESTOptions: optsGetter} // 继续完成store其他属性的初始化，比如初始化store.Storage属性。Storage主要用于和底层存储层交互 if err := store.CompleteWithOptions(options); err != nil { return nil, nil, nil, err } statusStore := *store // deployment的status子资源也是使用store, 区别是更新策略不一样, 即在update时会用旧对象的spec和lable覆盖新对象的，防止非status字段被用户意外覆盖 statusStore.UpdateStrategy = deployment.StatusStrategy statusStore.ResetFieldsStrategy = deployment.StatusStrategy return &REST{store}, &StatusREST{store: &statusStore}, &RollbackREST{store: store}, nil }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.

复制// k8s.io/apiserver/pkg/endpoints/groupversion.go func (g *APIGroupVersion) InstallREST(container *restful.Container) ([]*storageversion.ResourceInfo, error) { // 从InstallAPI调用链下来这里的g.Root为/apis，这样就可以确定handler的前缀为/apis/{goup}/{version} prefix := path.Join(g.Root, g.GroupVersion.Group, g.GroupVersion.Version) installer := &APIInstaller{ group: g, prefix: prefix, minRequestTimeout: g.MinRequestTimeout, } apiResources, resourceInfos, ws, registrationErrors := installer.Install() versionDiscoveryHandler := discovery.NewAPIVersionHandler(g.Serializer, g.GroupVersion, staticLister{apiResources}) versionDiscoveryHandler.AddToWebService(ws) container.Add(ws) return removeNonPersistedResources(resourceInfos), utilerrors.NewAggregate(registrationErrors) }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.

复制// Install handlers for API resources. func (a *APIInstaller) Install() ([]metav1.APIResource, []*storageversion.ResourceInfo, *restful.WebService, []error) { var apiResources []metav1.APIResource var resourceInfos []*storageversion.ResourceInfo var errors []error ws := a.newWebService() // Register the paths in a deterministic (sorted) order to get a deterministic swagger spec. paths := make([]string, len(a.group.Storage)) var i int = 0 // a.goup.Storage的签名是 map[string]Storage, for循环的path是map的key，即资源名称 for path := range a.group.Storage { paths[i] = path i++ } sort.Strings(paths) for _, path := range paths { apiResource, resourceInfo, err := a.registerResourceHandlers(path, a.group.Storage[path], ws) ... }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.

复制// k8s.io/apiserver/pkg/endpoints/installer.go func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService) (*metav1.APIResource, *storageversion.ResourceInfo, error) { ... // 初始化rest容器，根目录是APIInstaller的prefix属性，从InstallAPI调用链下来值为/apis/{goup}/{version} ws := a.newWebService() ... // 进行类型转换判断当前的storage支持哪些类型的操作 creater, isCreater := storage.(rest.Creater) namedCreater, isNamedCreater := storage.(rest.NamedCreater) lister, isLister := storage.(rest.Lister) getter, isGetter := storage.(rest.Getter) getterWithOptions, isGetterWithOptions := storage.(rest.GetterWithOptions) gracefulDeleter, isGracefulDeleter := storage.(rest.GracefulDeleter) collectionDeleter, isCollectionDeleter := storage.(rest.CollectionDeleter) updater, isUpdater := storage.(rest.Updater) patcher, isPatcher := storage.(rest.Patcher) watcher, isWatcher := storage.(rest.Watcher) connecter, isConnecter := storage.(rest.Connecter) storageMeta, isMetadata := storage.(rest.StorageMetadata) storageVersionProvider, isStorageVersionProvider := storage.(rest.StorageVersionProvider) // Get the list of actions for the given scope. switch { case !namespaceScoped: // 构造有无namespace资源的action // Handle non-namespace scoped resources like nodes. ... default: // 构造有namespace资源的action // 构造handler的注册路径 namespaceParamName := "namespaces" // Handler for standard REST verbs (GET, PUT, POST and DELETE). namespaceParam := ws.PathParameter("namespace", "object name and auth scope, such as for teams and projects").DataType("string") namespacedPath := namespaceParamName + "/{namespace}/" + resource namespaceParams := []*restful.Parameter{namespaceParam} //resourcePath的值为 /namespaces/{namespace}/{resource} resourcePath := namespacedPath resourceParams := namespaceParams // itemPath： /namespaces/{namespace}/{resource}/{name} // name是请求资源对象的名字 itemPath := namespacedPath + "/{name}" nameParams := append(namespaceParams, nameParam) proxyParams := append(nameParams, pathParam) itemPathSuffix := "" if isSubresource { itemPathSuffix = "/" + subresource // 有子资源等情况下 resourcePath被定义为：/namespaces/{namespace}/{resource}/{name}/{subResource} itemPath = itemPath + itemPathSuffix // itemPath与resourcePath的值一样 resourcePath = itemPath resourceParams = nameParams } apiResource.Name = path apiResource.Namespaced = true apiResource.Kind = resourceKind namer := handlers.ContextBasedNaming{ Namer: a.group.Namer, ClusterScoped: false, } // 根据storage实现的接口添加添加相关的action actions = appendIf(actions, action{"LIST", resourcePath, resourceParams, namer, false}, isLister) actions = appendIf(actions, action{"POST", resourcePath, resourceParams, namer, false}, isCreater) actions = appendIf(actions, action{"DELETECOLLECTION", resourcePath, resourceParams, namer, false}, isCollectionDeleter) // DEPRECATED in 1.11 actions = appendIf(actions, action{"WATCHLIST", "watch/" + resourcePath, resourceParams, namer, false}, allowWatchList) actions = appendIf(actions, action{"GET", itemPath, nameParams, namer, false}, isGetter) if getSubpath { actions = appendIf(actions, action{"GET", itemPath + "/{path:*}", proxyParams, namer, false}, isGetter) } actions = appendIf(actions, action{"PUT", itemPath, nameParams, namer, false}, isUpdater) actions = appendIf(actions, action{"PATCH", itemPath, nameParams, namer, false}, isPatcher) actions = appendIf(actions, action{"DELETE", itemPath, nameParams, namer, false}, isGracefulDeleter) // DEPRECATED in 1.11 actions = appendIf(actions, action{"WATCH", "watch/" + itemPath, nameParams, namer, false}, isWatcher) actions = appendIf(actions, action{"CONNECT", itemPath, nameParams, namer, false}, isConnecter) actions = appendIf(actions, action{"CONNECT", itemPath + "/{path:*}", proxyParams, namer, false}, isConnecter && connectSubpath) // list or post across namespace. // For ex: LIST all pods in all namespaces by sending a LIST request at /api/apiVersion/pods. // TODO: more strongly type whether a resource allows these actions on "all namespaces" (bulk delete) if !isSubresource { actions = appendIf(actions, action{"LIST", resource, params, namer, true}, isLister) // DEPRECATED in 1.11 actions = appendIf(actions, action{"WATCHLIST", "watch/" + resource, params, namer, true}, allowWatchList) } } ... for _, action := range actions { ... switch action.Verb { case "GET": // Get a resource. var handler restful.RouteFunction // 构造get请求的handler // restfulGetResourceWithOptions和restfulGetResource将handlers.GetResource函数转换成restful.RouteFunction，即handler的函数签名 if isGetterWithOptions { handler = restfulGetResourceWithOptions(getterWithOptions, reqScope, isSubresource) } else { handler = restfulGetResource(getter, reqScope) } ... // 将handler注册到rest容器中 // action.Path是上面定义的itemPath或resourcePath，对于GET来说是itemPath // 当前注册的handler的路径是ws的根路径加上ation.Path. 完整的路径为：/apis/{goup}/{version}/namespaces/{namespace}/{resource}/{name} route := ws.GET(action.Path).To(handler). Doc(doc). Param(ws.QueryParameter("pretty", "If true, then the output is pretty printed.")). Operation("read"+namespaced+kind+strings.Title(subresource)+operationSuffix). Produces(append(storageMeta.ProducesMIMETypes(action.Verb), mediaTypes...)...). Returns(http.StatusOK, "OK", producedObject). Writes(producedObject) if isGetterWithOptions { if err := AddObjectParams(ws, route, versionedGetOptions); err != nil { return nil, nil, err } } addParams(route, action.Params) routes = append(routes, route) } case "LIST": // List all resources of a kind. ... case "PUT": // Update a resource. ... case "PATCH": // Partially update a resource ... case "POST": // Create a resource. ... case "DELETE": // Delete a resource. .... } ... }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.48.49.50.51.52.53.54.55.56.57.58.59.60.61.62.63.64.65.66.67.68.69.70.71.72.73.74.75.76.77.78.79.80.81.82.83.84.85.86.87.88.89.90.91.92.93.94.95.96.97.98.99.100.101.102.103.104.105.106.107.108.109.110.111.112.113.114.115.116.117.118.119.120.121.122.123.124.125.126.127.128.129.130.131.132.133.134.

复制// k8s.io/apiserver/pkg/endpoints/handlers/get.go // GetResource returns a function that handles retrieving a single resource from a rest.Storage object. func GetResource(r rest.Getter, scope *RequestScope) http.HandlerFunc { return getResourceHandler(scope, func(ctx context.Context, name string, req *http.Request, trace *utiltrace.Trace) (runtime.Object, error) { // check for export options := metav1.GetOptions{} // 获取查询串 if values := req.URL.Query(); len(values) > 0 { ... // 将查询串解码成metav1.GetOptions if err := metainternalversionscheme.ParameterCodec.DecodeParameters(values, scope.MetaGroupVersion, &options); err != nil { err = errors.NewBadRequest(err.Error()) return nil, err } } if trace != nil { trace.Step("About to Get from storage") } // 交给Getter接口处理 return r.Get(ctx, name, &options) }) } // getResourceHandler is an HTTP handler function for get requests. It delegates to the // passed-in getterFunc to perform the actual get. func getResourceHandler(scope *RequestScope, getter getterFunc) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { ... namespace, name, err := scope.Namer.Name(req) ... ctx := req.Context() ctx = request.WithNamespace(ctx, namespace) ... result, err := getter(ctx, name, req, trace) ... // 对处理结果进行转化为用户期望的格式并写入到response中返回给用户 transformResponseObject(ctx, scope, trace, req, w, http.StatusOK, outputMediaType, result) } }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.

复制// k8s.io/apiserver/pkg/endpoints/handlers/create.go // CreateNamedResource returns a function that will handle a resource creation with name. func CreateNamedResource(r rest.NamedCreater, scope *RequestScope, admission admission.Interface) http.HandlerFunc { return createHandler(r, scope, admission, true) } func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Interface, includeName bool) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { ... // 从request中取出请求body body, err := limitedReadBody(req, scope.MaxRequestBodyBytes) ... // 对查询传进行解码生成metav1.CreateOptions options := &metav1.CreateOptions{} values := req.URL.Query() if err := metainternalversionscheme.ParameterCodec.DecodeParameters(values, scope.MetaGroupVersion, options); err != nil { ... } // 将请求body解码成资源对象, defaultGVK是用户请求的版本，这里decoder解码出来的对象是internal版本的对象 obj, gvk, err := decoder.Decode(body, &defaultGVK, original) ... admissionAttributes := admission.NewAttributesRecord(obj, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Create, options, dryrun.IsDryRun(options.DryRun), userInfo) // 构建调用create方法的函数 requestFunc := func() (runtime.Object, error) { return r.Create( ctx, name, obj, rest.AdmissionToValidateObjectFunc(admit, admissionAttributes, scope), options, ) } // Dedup owner references before updating managed fields dedupOwnerReferencesAndAddWarning(obj, req.Context(), false) result, err := finisher.FinishRequest(ctx, func() (runtime.Object, error) { ... // 执行mutation的admission操作，即在创建时对象进行修改操作。 // admin在buildGenericConfig中初始化，通过config传递给genericsever，然后传递到此处 if mutatingAdmission, ok := admit.(admission.MutationInterface); ok && mutatingAdmission.Handles(admission.Create) { if err := mutatingAdmission.Admit(ctx, admissionAttributes, scope); err != nil { return nil, err } } // Dedup owner references again after mutating admission happens dedupOwnerReferencesAndAddWarning(obj, req.Context(), true) // 调用创建方法 result, err := requestFunc() ... return result, err }) ... // resutl也是internal版本的对象，transformResponseObject会转换为用户请求的版本并输出 transformResponseObject(ctx, scope, trace, req, w, code, outputMediaType, result) }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.48.49.50.51.52.53.54.55.