kubernetes源码阅读笔记——API Server（之二）

本篇将着重分析InstallLegacyAPIGroup方法。

vendor/k8s.io/apiserver/pkg/server/genericapiserver.go

func (s *GenericAPIServer) InstallLegacyAPIGroup(apiPrefix string, apiGroupInfo *APIGroupInfo) error {
    if !s.legacyAPIGroupPrefixes.Has(apiPrefix) {
        return fmt.Errorf("%q is not in the allowed legacy API prefixes: %v", apiPrefix, s.legacyAPIGroupPrefixes.List())
    }
    if err := s.installAPIResources(apiPrefix, apiGroupInfo); err != nil {
        return err
    }

    // Install the version handler.
    // Add a handler at /<apiPrefix> to enumerate the supported api versions.
    s.Handler.GoRestfulContainer.Add(discovery.NewLegacyRootAPIHandler(s.discoveryAddresses, s.Serializer, apiPrefix).WebService())

    return nil
}

首先，判断传入的前缀是否合法。

其次，调用installAPIResources方法。

最后，在/api路径下生成一个WebService，并添加进Container。

进入installAPIResources方法：

vendor/k8s.io/apiserver/pkg/server/genericapiserver.go

func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *APIGroupInfo) error {
	openAPIGroupModels, err := s.getOpenAPIModelsForGroup(apiPrefix, apiGroupInfo)
	if err != nil {
		return fmt.Errorf("unable to get openapi models for group %v: %v", apiPrefix, err)
	}
	for _, groupVersion := range apiGroupInfo.PrioritizedVersions {
		if len(apiGroupInfo.VersionedResourcesStorageMap[groupVersion.Version]) == 0 {
			klog.Warningf("Skipping API %v because it has no resources.", groupVersion)
			continue
		}

		apiGroupVersion := s.getAPIGroupVersion(apiGroupInfo, groupVersion, apiPrefix)
		if apiGroupInfo.OptionsExternalVersion != nil {
			apiGroupVersion.OptionsExternalVersion = apiGroupInfo.OptionsExternalVersion
		}
		apiGroupVersion.OpenAPIModels = openAPIGroupModels

		if err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer); err != nil {
			return fmt.Errorf("unable to setup API %v: %v", apiGroupInfo, err)
		}
	}

	return nil
}

前面的逻辑类似于对传入的apiGroupInfo进行标准格式化，并循环地遍历所有可用version，包装成为APIGroupVersion对象，并对其调用InstallREST方法，为对象注册Handler。核心在于InstallREST方法。

一、InstallREST

进入InstallREST方法：

vendor/k8s.io/apiserver/pkg/endpoints/groupversion.go

func (g *APIGroupVersion) InstallREST(container *restful.Container) error {
	prefix := path.Join(g.Root, g.GroupVersion.Group, g.GroupVersion.Version)
	installer := &APIInstaller{
		group:                        g,
		prefix:                       prefix,
		minRequestTimeout:            g.MinRequestTimeout,
		enableAPIResponseCompression: g.EnableAPIResponseCompression,
	}

	apiResources, ws, registrationErrors := installer.Install()
	versionDiscoveryHandler := discovery.NewAPIVersionHandler(g.Serializer, g.GroupVersion, staticLister{apiResources})
	versionDiscoveryHandler.AddToWebService(ws)
	container.Add(ws)
	return utilerrors.NewAggregate(registrationErrors)
}

方法逻辑是，基于这个GroupVersion对象的元素新建一个前缀字符串，然后建立一个APIInstaller结构体，调用Install方法，然后将返回的WebService添加进Container中。

进入Install方法：

vendor/k8s.io/apiserver/pkg/endpoints/installer.go

// Install handlers for API resources.
func (a *APIInstaller) Install() ([]metav1.APIResource, *restful.WebService, []error) {
    var apiResources []metav1.APIResource
    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
    for path := range a.group.Storage {
        paths[i] = path
        i++
    }
    sort.Strings(paths)
    for _, path := range paths {
        apiResource, err := a.registerResourceHandlers(path, a.group.Storage[path], ws)
        if err != nil {
            errors = append(errors, fmt.Errorf("error in registering resource: %s, %v", path, err))
        }
        if apiResource != nil {
            apiResources = append(apiResources, *apiResource)
        }
    }
    return apiResources, ws, errors
}

这个方法的逻辑是，先创建一个WebService，然后将APIInstaller结构体中保存的所有API路径存入数组paths，对数组排序，遍历数组所有元素并调用registerResourceHandlers方法，将路径、对应的storage和WebService建立对应关系。最后，将这些对应关系存入apiResources数组中，并返回APIResource和WebService。

二、registerResourceHandlers

registerResourceHandlers方法长达数百行，是创建API的核心方法。节选最重要的部分如下：

vendor/k8s.io/apiserver/pkg/endpoints/installer.go

func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService) (*metav1.APIResource, error) {
    
　　　　...

　　　　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)

　　　　if !isMetadata {
   　　　　　　storageMeta = defaultStorageMetadata{}
　　　　}
　　　　...


　　　　// Handler for standard REST verbs (GET, PUT, POST and DELETE).
　　　　// Add actions at the resource path: /api/apiVersion/resource
　　　　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)
　　　　
　　　　// Add actions at the item path: /api/apiVersion/resource/{name}
　　　　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)

　　　　actions = appendIf(actions, action{"CONNECT", itemPath, nameParams, namer, false}, isConnecter)
　　　　actions = appendIf(actions, action{"CONNECT", itemPath + "/{path:*}", proxyParams, namer, false}, isConnecter && connectSubpath)

　　　　...
　　　　
　　　　routes := []*restful.RouteBuilder{}

　　　　case "GET": // Get a resource.
　　　　　　　　var handler restful.RouteFunction
　　　　　　　　if isGetterWithOptions {
   　　　　　　　　　　handler = restfulGetResourceWithOptions(getterWithOptions, reqScope, isSubresource)
　　　　　　　　} else {
   　　　　　　　　　　handler = restfulGetResource(getter, exporter, reqScope)
　　　　　　　　}

　　　　　　　　if needOverride {
   　　　　　　　　　　// need change the reported verb
　　　　　　　　　　　　handler = metrics.InstrumentRouteFunc(verbOverrider.OverrideMetricsVerb(action.Verb), group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler)
　　　　　　　　} else {
　　　　　　　　　　　　handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, handler)
　　　　　　　　}

　　　　　　　　if a.enableAPIResponseCompression {
   　　　　　　　　　　handler = genericfilters.RestfulWithCompression(handler)
　　　　　　　　}
　　　　　　　　doc := "read the specified " + kind
　　　　　　　　if isSubresource {
   　　　　　　　　　　doc = "read " + subresource + " of the specified " + kind
　　　　　　　　}
　　　　　　　　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)
　　　　　　　　...
　　　　　　　　routes = append(routes, route)

　　　　...

　　　　
　　　　for _, route := range routes {
   　　　　　　route.Metadata(ROUTE_META_GVK, metav1.GroupVersionKind{
      　　　　　　　　Group:   reqScope.Kind.Group,
      　　　　　　　　Version: reqScope.Kind.Version,
      　　　　　　　　Kind:    reqScope.Kind.Kind,
   　　　　　　})
   　　　　　　route.Metadata(ROUTE_META_ACTION, strings.ToLower(action.Verb))
   　　　　　　ws.Route(route)
　　　　}
　　　　...
　　　　return &apiResource, nil
}

代码虽长，逻辑很清晰。大致的执行逻辑是：

1.判断storage是否支持create、list、get等方法，并对所有支持的方法进行进一步的处理，如if !isMetadata这一块一样，内容过多不一一贴出；

2.将所有支持的方法存入actions数组中；

3.遍历actions数组，在一个switch语句中，为所有元素定义路由。如贴出的case "GET"这一块，首先创建并包装一个handler对象，然后调用WebService的一系列方法，创建一个route对象，将handler绑定到这个route上。后面还有case "PUT"、case "DELETE"等一系列case，不一一贴出。最后，将route加入routes数组中。

4.遍历routes数组，将route加入WebService中。

5.最后，返回一个APIResource结构体。

这样，Install方法就通过调用registerResourceHandlers方法，完成了WebService与APIResource的绑定。

至此，InstallLegacyAPI方法的逻辑就分析完了。总的来说，这个方法遵循了go-restful的设计模式，在/api路径下注册了WebService，并将WebService加入Container中。

三、Container.Add

回到InstallREST方法，下一步调用container包的Add方法，将WebService添加到container中。

该方法位于外部包中，功能很简单，就是检查错误后，将传入的WebService加入传进来的restful Container中。

vendor/github.com/emicklei/go-restful/container.go

// Add a WebService to the Container. It will detect duplicate root paths and exit in that case.
func (c *Container) Add(service *WebService) *Container {
    c.webServicesLock.Lock()
    defer c.webServicesLock.Unlock()

    // if rootPath was not set then lazy initialize it
    if len(service.rootPath) == 0 {
        service.Path("/")
    }

    // cannot have duplicate root paths
    for _, each := range c.webServices {
        if each.RootPath() == service.RootPath() {
            log.Printf("[restful] WebService with duplicate root path detected:['%v']", each)
            os.Exit(1)
        }
    }

    // If not registered on root then add specific mapping
    if !c.isRegisteredOnRoot {
        c.isRegisteredOnRoot = c.addHandler(service, c.ServeMux)
    }
    c.webServices = append(c.webServices, service)
    return c
}

至此，InstallLegacyAPIGroup方法的核心“installAPIResources”就分析完了。后面还有一行s.Handler.GoRestfulContainer.Add(...)，同样调用了Container包里的Add方法，作用是为/<apiPrefix>路径添加一个可枚举出所有可用version的Handler。

总结一下，InstallLegacyAPIGroup方法虽然复杂，但是作用就是一条，即为传入的路径注册Handler。