Kubernetes v1.5.0java
go-restful是用於構建REST-style web服務的golang包。
它是出現時由於一個javaer在golang中沒找到順手的REST-based服務構建包,因此就按照他在java裏經常使用的JAX-RS的設計,在golang中造了一個輪子。node
1.Route:
路由包含兩種,一種是標準JSR311接口規範的實現RouterJSR311,一種是快速路由CurlyRouter。
CurlyRouter支持正則表達式和動態參數,相比RouterJSR11更加輕量級,apiserver中使用的就是這種路由。
一種Route的設定包含:請求方法(http Method),請求路徑(URL Path),輸入輸出類型(JSON/YAML)以及對應的回掉函數restful.RouteFunction,響應內容類型(Accept)等。git
2.WebService:
WebService邏輯上是Route的集合,功能上主要是爲一組Route統一設置包括root path,請求響應的數據類型等一些通用的屬性。
須要注意的是,WebService必須加入到Container中才能生效。github
func (g *APIGroupVersion) InstallREST(container *restful.Container) error { installer := g.newInstaller() ws := installer.NewWebService() 。。。 container.Add(ws) return utilerrors.NewAggregate(registrationErrors) }
上面是k8s的REST註冊接口,也調用了Container.Add(ws),才能讓這個ws生效。golang
3.Container:
Container邏輯上是WebService的集合,功能上能夠實現多終端的效果。
它包括一組restful.WebService和一個http.ServeMux對象,使用RouteSelector進行請求派發。
例如,下面代碼中建立了兩個Container,分別在不一樣的port上提供服務。
該代碼是go-restful的example:web
package main import ( "github.com/emicklei/go-restful" "io" "log" "net/http" ) func main() { ws := new(restful.WebService) ws.Route(ws.GET("/hello").To(hello)) // ws被添加到默認的container restful.DefaultContainer中 restful.Add(ws) go func() { // restful.DefaultContainer監聽在端口8080上 http.ListenAndServe(":8080", nil) }() container2 := restful.NewContainer() ws2 := new(restful.WebService) ws2.Route(ws2.GET("/hello").To(hello2)) // ws2被添加到container2中 container2.Add(ws2) // container2中監聽端口8081 server := &http.Server{Addr: ":8081", Handler: container2} log.Fatal(server.ListenAndServe()) } func hello(req *restful.Request, resp *restful.Response) { io.WriteString(resp, "default world") } func hello2(req *restful.Request, resp *restful.Response) { io.WriteString(resp, "second world") }
4.Filter:
Filter用於動態的攔截請求和響應,相似於放置在相應組件前的鉤子,在相應組件功能運行前捕獲請求或者響應,主要用於記錄log,驗證,重定向等功能。
go-restful中有三種類型的Filter:正則表達式
Container Filter:
運行在Container中全部的WebService執行以前。json
// install a (global) filter for the default container (processed before any webservice) restful.Filter(globalLogging)
WebService Filter:
運行在WebService中全部的Route執行以前。api
// install a webservice filter (processed before any route) ws.Filter(webserviceLogging).Filter(measureTime)
Route Filter:
運行在調用Route綁定的方法以前。restful
// install 2 chained route filters (processed before calling findUser) ws.Route(ws.GET("/{user-id}").Filter(routeLogging).Filter(NewCountFilter().routeCounter).To(findUser))
拿用官方提供的例子:
package main import ( "github.com/emicklei/go-restful" "log" "net/http" ) type User struct { Id, Name string } type UserResource struct { // normally one would use DAO (data access object) users map[string]User } func (u UserResource) Register(container *restful.Container) { // 建立新的WebService ws := new(restful.WebService) // 設定WebService對應的路徑("/users")和支持的MIME類型(restful.MIME_XML/ restful.MIME_JSON) ws. Path("/users"). Consumes(restful.MIME_XML, restful.MIME_JSON). Produces(restful.MIME_JSON, restful.MIME_XML) // you can specify this per route as well // 添加路由: GET /{user-id} --> u.findUser ws.Route(ws.GET("/{user-id}").To(u.findUser)) // 添加路由: POST / --> u.updateUser ws.Route(ws.POST("").To(u.updateUser)) // 添加路由: PUT /{user-id} --> u.createUser ws.Route(ws.PUT("/{user-id}").To(u.createUser)) // 添加路由: DELETE /{user-id} --> u.removeUser ws.Route(ws.DELETE("/{user-id}").To(u.removeUser)) // 將初始化好的WebService添加到Container中 container.Add(ws) } // GET http://localhost:8080/users/1 // func (u UserResource) findUser(request *restful.Request, response *restful.Response) { id := request.PathParameter("user-id") usr := u.users[id] if len(usr.Id) == 0 { response.AddHeader("Content-Type", "text/plain") response.WriteErrorString(http.StatusNotFound, "User could not be found.") } else { response.WriteEntity(usr) } } // POST http://localhost:8080/users // <User><Id>1</Id><Name>Melissa Raspberry</Name></User> // func (u *UserResource) updateUser(request *restful.Request, response *restful.Response) { usr := new(User) err := request.ReadEntity(&usr) if err == nil { u.users[usr.Id] = *usr response.WriteEntity(usr) } else { response.AddHeader("Content-Type", "text/plain") response.WriteErrorString(http.StatusInternalServerError, err.Error()) } } // PUT http://localhost:8080/users/1 // <User><Id>1</Id><Name>Melissa</Name></User> // func (u *UserResource) createUser(request *restful.Request, response *restful.Response) { usr := User{Id: request.PathParameter("user-id")} err := request.ReadEntity(&usr) if err == nil { u.users[usr.Id] = usr response.WriteHeader(http.StatusCreated) response.WriteEntity(usr) } else { response.AddHeader("Content-Type", "text/plain") response.WriteErrorString(http.StatusInternalServerError, err.Error()) } } // DELETE http://localhost:8080/users/1 // func (u *UserResource) removeUser(request *restful.Request, response *restful.Response) { id := request.PathParameter("user-id") delete(u.users, id) } func main() { // 建立一個空的Container wsContainer := restful.NewContainer() // 設定路由爲CurlyRouter(快速路由) wsContainer.Router(restful.CurlyRouter{}) // 建立自定義的Resource Handle(此處爲UserResource) u := UserResource{map[string]User{}} // 建立WebService,並將WebService加入到Container中 u.Register(wsContainer) log.Printf("start listening on localhost:8080") server := &http.Server{Addr: ":8080", Handler: wsContainer} // 啓動服務 log.Fatal(server.ListenAndServe()) }
上面的示例代碼構建RESTful服務,分爲幾個步驟,apiServer也是相似:
1.建立Container
2.配置Container屬性:ServeMux/Router type等
3.建立自定義的Resource Handle,實現Resource相關的處理方式。
4.建立對應Resource的WebService,在WebService中添加響應Route,並將WebService加入到Container中。
5.啓動監聽服務。
apiServer的Container相關的結構是APIContainer。
路徑:pkg/genericapiserver/mux/container.go
type APIContainer struct { *restful.Container NonSwaggerRoutes PathRecorderMux SecretRoutes Mux }
而該結構是在GenericAPIServer中被使用,分析過apiServer的啓動過程的話,應該對該結構比較熟悉。
type GenericAPIServer struct { discoveryAddresses DiscoveryAddresses LoopbackClientConfig *restclient.Config minRequestTimeout time.Duration ... requestContextMapper api.RequestContextMapper // 這裏使用到了restful.Container HandlerContainer *genericmux.APIContainer SecureServingInfo *SecureServingInfo InsecureServingInfo *ServingInfo effectiveSecurePort, effectiveInsecurePort int ExternalAddress string storage map[string]rest.Storage Serializer runtime.NegotiatedSerializer Handler http.Handler InsecureHandler http.Handler apiGroupsForDiscoveryLock sync.RWMutex apiGroupsForDiscovery map[string]unversioned.APIGroup ... }
而該結構的初始化是在master的初始化過程當中進行的。
調用過程: main --> App.Run --> master.Complete.New --> c.Config.GenericConfig.SkipComplete().New()
路徑: pkg/genericapiserver/config.go
func (c completedConfig) New() (*GenericAPIServer, error) { if c.Serializer == nil { return nil, fmt.Errorf("Genericapiserver.New() called with config.Serializer == nil") } s := &GenericAPIServer{ discoveryAddresses: c.DiscoveryAddresses, LoopbackClientConfig: c.LoopbackClientConfig, legacyAPIGroupPrefixes: c.LegacyAPIGroupPrefixes, admissionControl: c.AdmissionControl, requestContextMapper: c.RequestContextMapper, Serializer: c.Serializer, minRequestTimeout: time.Duration(c.MinRequestTimeout) * time.Second, enableSwaggerSupport: c.EnableSwaggerSupport, SecureServingInfo: c.SecureServingInfo, InsecureServingInfo: c.InsecureServingInfo, ExternalAddress: c.ExternalAddress, apiGroupsForDiscovery: map[string]unversioned.APIGroup{}, enableOpenAPISupport: c.EnableOpenAPISupport, openAPIConfig: c.OpenAPIConfig, postStartHooks: map[string]postStartHookEntry{}, } // 這裏進行了Contianer的初始化 s.HandlerContainer = mux.NewAPIContainer(http.NewServeMux(), c.Serializer) // 添加了DynamicApisDiscovery的 s.installAPI(c.Config) s.Handler, s.InsecureHandler = c.BuildHandlerChainsFunc(s.HandlerContainer.ServeMux, c.Config) return s, nil }
繼續調用mux.NewAPIContainer()接口建立,該接口的兩個參數:新建了一個http的ServeMux; 另外一個是實現了編解碼序列化反序列化的對象
func NewAPIContainer(mux *http.ServeMux, s runtime.NegotiatedSerializer) *APIContainer { c := APIContainer{ // New一個Container Container: restful.NewContainer(), NonSwaggerRoutes: PathRecorderMux{ mux: mux, }, SecretRoutes: mux, } // 配置http.ServeMux c.Container.ServeMux = mux // 配置該Container的路由方式:CurlyRouter 即快速路由 c.Container.Router(restful.CurlyRouter{}) // e.g. for proxy/{kind}/{name}/{*} // 配置panic產生以後的恢復處理函數 apiserver.InstallRecoverHandler(s, c.Container) apiserver.InstallServiceErrorHandler(s, c.Container) return &c }
看下apiserver.InstallRecoverHandler()實現:
func InstallRecoverHandler(s runtime.NegotiatedSerializer, container *restful.Container) { container.RecoverHandler(func(panicReason interface{}, httpWriter http.ResponseWriter) { logStackOnRecover(s, panicReason, httpWriter) }) } // RecoverHandler changes the default function (logStackOnRecover) to be called // when a panic is detected. DoNotRecover must be have its default value (=false). func (c *Container) RecoverHandler(handler RecoverHandleFunction) { c.recoverHandleFunc = handler }
根據英文註釋能夠看明白,該RecoverHandler就是在產生panic後會調用的恢復處理函數,默認的調用函數是logStackOnRecover,調用Container.RecoverHandler()後會修改該默認函數,而且Container.DoNotRecover的bool值必須是false才能生效。
apiserver.InstallServiceErrorHandler()接口就不看了,其實就是修改Service Error產生後的錯誤處理函數,默認是調用writeServiceError()。
到這裏Container的初始化基本OK了。
Container已建立而且也進行了初始化。該輪到WebService了,這節會介紹k8s的WebService的建立及添加。
接續上文的Container初始化入口,繼續往下看s.installAPI(c.Config):
func (s *GenericAPIServer) installAPI(c *Config) { // 這裏本來還有不少routes.Install()函數 // 這些install()貌似和mux有關。 // 而mux就是一個http的多分器,用於派發某個Request路徑到對應的http.Handler進行處理 。。。 // 往HandlerContainer中的Container裏添加WebService // 該WebService的建立在s.DynamicApisDiscovery()中進行 // 實際上建立的WebService是用於list 該group下的全部versions s.HandlerContainer.Add(s.DynamicApisDiscovery()) }
先看下WebService的建立接口s.DynamicApisDiscovery():
路徑:pkg/genericapiserver/genericapiserver.go
// DynamicApisDiscovery returns a webservice serving api group discovery. // Note: during the server runtime apiGroupsForDiscovery might change. func (s *GenericAPIServer) DynamicApisDiscovery() *restful.WebService { return apiserver.NewApisWebService(s.Serializer, APIGroupPrefix, func(req *restful.Request) []unversioned.APIGroup { // 須要加鎖 // 接口註釋也有說明。由於k8s能夠動態加載第三方apiGroups s.apiGroupsForDiscoveryLock.RLock() defer s.apiGroupsForDiscoveryLock.RUnlock() // 將apiGroupsForDiscovery中全部的APIGroup按照其名字進行升序排序 sortedGroups := []unversioned.APIGroup{} groupNames := make([]string, 0, len(s.apiGroupsForDiscovery)) for groupName := range s.apiGroupsForDiscovery { groupNames = append(groupNames, groupName) } sort.Strings(groupNames) for _, groupName := range groupNames { sortedGroups = append(sortedGroups, s.apiGroupsForDiscovery[groupName]) } // 建立切片,並填充各個APIGroup的ServerAddressByClientCIDRs信息 clientIP := utilnet.GetClientIP(req.Request) serverCIDR := s.discoveryAddresses.ServerAddressByClientCIDRs(clientIP) groups := make([]unversioned.APIGroup, len(sortedGroups)) for i := range sortedGroups { groups[i] = sortedGroups[i] groups[i].ServerAddressByClientCIDRs = serverCIDR } return groups }) }
繼續深刻看apiserver.NewApisWebService(),該接口傳入了編解碼對象,APIGroup的Prefix,還有一個function。
func NewApisWebService(s runtime.NegotiatedSerializer, apiPrefix string, f func(req *restful.Request) []unversioned.APIGroup) *restful.WebService { // 用於向後兼容v1.1版本,返回一個空的APIGroup ss := StripVersionNegotiatedSerializer{s} // 獲取支持的媒體類型,好比:application/json,application/yaml mediaTypes, _ := mediaTypesForSerializer(s) // 構建go-restful的Route處理方法 rootAPIHandler := RootAPIHandler(ss, f) // 建立WebService ws := new(restful.WebService) // 添加Path ws.Path(apiPrefix) // API 說明 ws.Doc("get available API versions") // 配置GET("/") 轉到rootAPIHandler()接口 ws.Route(ws.GET("/").To(rootAPIHandler). Doc("get available API versions"). Operation("getAPIVersions"). Produces(mediaTypes...). Consumes(mediaTypes...). Writes(unversioned.APIGroupList{})) return ws }
到這裏list某個Group下全部的versions的API已經註冊完成了。
這些都不是關鍵的RESTful API的註冊,關鍵的註冊都會在pkg/apiserver/apiserver.go中的InstallREST()接口中進行。
琢磨過apiServer啓動流程的同窗,應該會知道/api和/apis的註冊接口最後都會調用到該接口。
/api的註冊接口是pkg/genericapiserver/genericapiserver.go中的InstallLegacyAPIGroup()接口
/apis的註冊接口是InstallAPIGroup()。
這兩個接口都會調用s.installAPIResources(),最後再調用apiGroupVersion.InstallREST()進行API註冊。
流程基本就是這樣,接着咱們直接進入InstallREST()接口看實現:
func (g *APIGroupVersion) InstallREST(container *restful.Container) error { // 拼裝path: "Prefix/Group/Version" // 而後填充並返回一個APIInstaller對象 installer := g.newInstaller() // 建立一個WebService ws := installer.NewWebService() // 這個是關鍵,會對各類URL進行註冊 apiResources, registrationErrors := installer.Install(ws) lister := g.ResourceLister if lister == nil { lister = staticLister{apiResources} } // 增長一個list的API AddSupportedResourcesWebService(g.Serializer, ws, g.GroupVersion, lister) // 將該WebService加入到Container container.Add(ws) return utilerrors.NewAggregate(registrationErrors) }
前兩個調用函數比較簡單,這裏不進行介紹了。直接進入關鍵函數installer.Install(ws):
func (a *APIInstaller) Install(ws *restful.WebService) (apiResources []unversioned.APIResource, errors []error) { errors = make([]error, 0) proxyHandler := (&ProxyHandler{ prefix: a.prefix + "/proxy/", storage: a.group.Storage, serializer: a.group.Serializer, mapper: a.group.Context, }) // 將全部的path合成一個切片,並按照升序從新排序 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 { // 註冊各個URL,關鍵接口 // 傳入的參數:path,rest.Storage,WebService,Handler apiResource, err := a.registerResourceHandlers(path, a.group.Storage[path], ws, proxyHandler) if err != nil { errors = append(errors, fmt.Errorf("error in registering resource: %s, %v", path, err)) } // 將全部註冊成功的Resource合成一個切片 // 將該切片做爲返回值,便於以後的接口註冊list Resources的API if apiResource != nil { apiResources = append(apiResources, *apiResource) } } return apiResources, errors }
該接口先是遍歷全部的path,並升序從新排列,而後循環調用接口註冊各個URL的API,並將這些註冊成功的APIResource加入到同一個切片中。
咱們繼續看a.registerResourceHandlers()接口:
func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService, proxyHandler http.Handler) (*unversioned.APIResource, error) { ... // 構建creater、lister、deleter、updater、watcher等,其實就是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) deleter, isDeleter := storage.(rest.Deleter) 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) _, isRedirector := storage.(rest.Redirector) connecter, isConnecter := storage.(rest.Connecter) storageMeta, isMetadata := storage.(rest.StorageMetadata) ... var apiResource unversioned.APIResource // k8s資源分爲兩類:無namespace的RESTScopeNameRoot; 有namespace的RESTScopeNameNamespace // 在對應的path上添加各種actions,並指定對應的handler。 switch scope.Name() { case meta.RESTScopeNameRoot: // Handle non-namespace scoped resources like nodes. resourcePath := resource resourceParams := params itemPath := resourcePath + "/{name}" nameParams := append(params, nameParam) proxyParams := append(nameParams, pathParam) suffix := "" if hasSubresource { suffix = "/" + subresource itemPath = itemPath + suffix resourcePath = itemPath resourceParams = nameParams } apiResource.Name = path apiResource.Namespaced = false apiResource.Kind = resourceKind namer := rootScopeNaming{scope, a.group.Linker, gpath.Join(a.prefix, resourcePath, "/"), suffix} // 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) // DEPRECATED actions = appendIf(actions, action{"WATCHLIST", "watch/" + resourcePath, resourceParams, namer, false}, allowWatchList) // 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}, isDeleter) actions = appendIf(actions, action{"WATCH", "watch/" + itemPath, nameParams, namer, false}, isWatcher) // We add "proxy" subresource to remove the need for the generic top level prefix proxy. // The generic top level prefix proxy is deprecated in v1.2, and will be removed in 1.3, or 1.4 at the latest. // TODO: DEPRECATED in v1.2. actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath + "/{path:*}", proxyParams, namer, false}, isRedirector) // TODO: DEPRECATED in v1.2. actions = appendIf(actions, action{"PROXY", "proxy/" + itemPath, nameParams, namer, false}, isRedirector) actions = appendIf(actions, action{"CONNECT", itemPath, nameParams, namer, false}, isConnecter) actions = appendIf(actions, action{"CONNECT", itemPath + "/{path:*}", proxyParams, namer, false}, isConnecter && connectSubpath) break case meta.RESTScopeNameNamespace: ... break default: return nil, fmt.Errorf("unsupported restscope: %s", scope.Name()) } ... // 根據以前生成的actions,進行遍歷 // 而後在WebService中添加指定的route for _, action := range actions { versionedObject := storageMeta.ProducesObject(action.Verb) if versionedObject == nil { versionedObject = defaultVersionedObject } reqScope.Namer = action.Namer namespaced := "" if apiResource.Namespaced { namespaced = "Namespaced" } operationSuffix := "" if strings.HasSuffix(action.Path, "/{path:*}") { operationSuffix = operationSuffix + "WithPath" } if action.AllNamespaces { operationSuffix = operationSuffix + "ForAllNamespaces" namespaced = "" } // 判斷action的動做類型 // 生成響應的handler,建立route添加到WebService中 switch action.Verb { case "GET": // Get a resource. var handler restful.RouteFunction // 判斷是否有參數 if isGetterWithOptions { handler = GetResourceWithOptions(getterWithOptions, reqScope) } else { handler = GetResource(getter, exporter, reqScope) } // 生成處理函數 handler = metrics.InstrumentRouteFunc(action.Verb, resource, handler) doc := "read the specified " + kind if hasSubresource { 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", versionedObject). Writes(versionedObject) if isGetterWithOptions { if err := addObjectParams(ws, route, versionedGetOptions); err != nil { return nil, err } } if isExporter { if err := addObjectParams(ws, route, versionedExportOptions); err != nil { return nil, err } } addParams(route, action.Params) ws.Route(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. ... case "DELETECOLLECTION": ... case "WATCH": // Watch a resource. ... case "WATCHLIST": // Watch all resources of a kind. ... case "PROXY": // Proxy requests to a resource. ... case "CONNECT": ... } default: return nil, fmt.Errorf("unrecognized action verb: %s", action.Verb) } // Note: update GetAttribs() when adding a custom handler. } return &apiResource, nil }
首先構建creater、lister、getter、deleter、updater、patcher、watcher,其實他們都是storage,只是對應着對etcd的不一樣操做。
而後針對全部的action,構建響應的handler。建立對應的route,最後把route添加到service裏面。這樣就完成了api的註冊。
關鍵的REST API註冊基本就這樣結束了,除此以外還會有不少別的API的註冊:
好比APIGroupVersion.InstallREST()接口中的AddSupportedResourcesWebService(g.Serializer, ws, g.GroupVersion, lister);
GenericAPIServer.InstallLegacyAPIGroup()接口中的apiserver.AddApiWebService()的調用;
等等。。
其實上面也註冊了各類REST API,好比像PodList,ServiceList,ReplicationControllerList等。這些就不深刻了,都是大同小異。
1.go-restful example: http://ernestmicklei.com/2012...
2.go-restful api desgin: http://ernestmicklei.com/2012...
3.go-restful github code: https://github.com/emicklei/g...
4.go-restful GoDoc: https://godoc.org/github.com/...