接著 Kubernetes:kube-apiserver 之啟動流程(一) 加以介紹。
建立完通用 APIServer 後繼續建立 APIExtensions Server。
func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget) (*CustomResourceDefinitions, error) {
genericServer, err := c.GenericConfig.New("apiextensions-apiserver", delegationTarget)
s := &CustomResourceDefinitions{
GenericAPIServer: genericServer,
}
// 儲存建立 REST API 到資源實體的資訊
apiGroupInfo := genericapiserver.NewDefaultAPIGroupInfo(apiextensions.GroupName, Scheme, metav1.ParameterCodec, Codecs)
// 資源實體
storage := map[string]rest.Storage{}
// customresourcedefinitions
if resource := "customresourcedefinitions"; apiResourceConfig.ResourceEnabled(v1.SchemeGroupVersion.WithResource(resource)) {
// 建立資源實體
customResourceDefinitionStorage, err := customresourcedefinition.NewREST(Scheme, c.GenericConfig.RESTOptionsGetter)
if err != nil {
return nil, err
}
storage[resource] = customResourceDefinitionStorage
storage[resource+"/status"] = customresourcedefinition.NewStatusREST(Scheme, customResourceDefinitionStorage)
}
if len(storage) > 0 {
apiGroupInfo.VersionedResourcesStorageMap[v1.SchemeGroupVersion.Version] = storage
}
if err := s.GenericAPIServer.InstallAPIGroup(&apiGroupInfo); err != nil {
return nil, err
}
APIGroupInfo 物件用於描述資源組資訊，storage 儲存資源到資源實體的對應關係。
資源實體，通過 NewREST() 函數建立。
# kubernetes/vendor/k8s.io/apiextensions-apiserver/pkg/registry/customresourcedefinition/etcd.go
package customresourcedefinition
// NewREST returns a RESTStorage object that will work against API services.
func NewREST(scheme *runtime.Scheme, optsGetter generic.RESTOptionsGetter) (*REST, error) {
strategy := NewStrategy(scheme)
store := &genericregistry.Store{
NewFunc: func() runtime.Object { return &apiextensions.CustomResourceDefinition{} },
NewListFunc: func() runtime.Object { return &apiextensions.CustomResourceDefinitionList{} },
PredicateFunc: MatchCustomResourceDefinition,
DefaultQualifiedResource: apiextensions.Resource("customresourcedefinitions"),
SingularQualifiedResource: apiextensions.Resource("customresourcedefinition"),
CreateStrategy: strategy,
UpdateStrategy: strategy,
DeleteStrategy: strategy,
ResetFieldsStrategy: strategy,
// TODO: define table converter that exposes more than name/creation timestamp
TableConvertor: rest.NewDefaultTableConvertor(apiextensions.Resource("customresourcedefinitions")),
}
options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: GetAttrs}
if err := store.CompleteWithOptions(options); err != nil {
return nil, err
}
return &REST{store}, nil
}
可以看到，資源實體是在資源包 customresourcedefinition 的 etcd.go 中建立的，建立的資源實體負責和 etcd 互動。
(關於 etcd 互動的部分先不講，後續會專門介紹。)
建立完資源實體後，通過 apiGroupInfo.VersionedResourcesStorageMap[v1.SchemeGroupVersion.Version] = storage 將資源實體儲存到 apiGroupInfo。
繼續呼叫 InstallAPIGroup(apiGroupInfo *APIGroupInfo) 安裝 REST API。
# kubernetes/vendor/k8s.io/apiserver/pkg/server/genericapiserver.go
func (s *GenericAPIServer) InstallAPIGroup(apiGroupInfo *APIGroupInfo) error {
return s.InstallAPIGroups(apiGroupInfo)
}
func (s *GenericAPIServer) InstallAPIGroups(apiGroupInfos ...*APIGroupInfo) error {
for _, apiGroupInfo := range apiGroupInfos {
// 呼叫 installAPIResources
if err := s.installAPIResources(APIGroupPrefix, apiGroupInfo, openAPIModels); err != nil {
return fmt.Errorf("unable to install api resources: %v", err)
}
}
s.DiscoveryGroupManager.AddGroup(apiGroup)
s.Handler.GoRestfulContainer.Add(discovery.NewAPIGroupHandler(s.Serializer, apiGroup).WebService())
}
// installAPIResources is a private method for installing the REST storage backing each api groupversionresource
func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *APIGroupInfo, typeConverter managedfields.TypeConverter) error {
for _, groupVersion := range apiGroupInfo.PrioritizedVersions {
apiGroupVersion, err := s.getAPIGroupVersion(apiGroupInfo, groupVersion, apiPrefix)
if err != nil {
return err
}
// 呼叫 InstallREST
discoveryAPIResources, r, err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer)
}
}
# kubernetes/vendor/k8s.io/apiserver/pkg/endpoints/groupversion.go
func (g *APIGroupVersion) InstallREST(container *restful.Container) ([]apidiscoveryv2beta1.APIResourceDiscovery, []*storageversion.ResourceInfo, error) {
prefix := path.Join(g.Root, g.GroupVersion.Group, g.GroupVersion.Version)
installer := &APIInstaller{
group: g,
prefix: prefix,
minRequestTimeout: g.MinRequestTimeout,
}
// 呼叫 Install
apiResources, resourceInfos, ws, registrationErrors := installer.Install()
container.Add(ws)
return aggregatedDiscoveryResources, removeNonPersistedResources(resourceInfos), utilerrors.NewAggregate(registrationErrors)
}
# kubernetes/vendor/k8s.io/apiserver/pkg/endpoints/installer.go
// Install handlers for API resources.
func (a *APIInstaller) Install() ([]metav1.APIResource, []*storageversion.ResourceInfo, *restful.WebService, []error) {
for _, path := range paths {
// 註冊資源 Handler
apiResource, resourceInfo, 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)
}
if resourceInfo != nil {
resourceInfos = append(resourceInfos, resourceInfo)
}
}
return apiResources, resourceInfos, ws, errors
}
如上例所示，註冊資源 REST API 的呼叫鏈很長，通過逐層呼叫，走到 registerResourceHandlers 註冊資源 handler。
registerResourceHandlers 函數非常的長，主要抓一點：註冊 RESTful API 的資源 handler 需要什麼？回答好這個問題基本上就能抓住 registerResourceHandlers 的精髓了。
註冊資源 handler 需要知道資源的 API path 和資源實體（指和 etcd 互動的資源範例）的對應關係，接著需要知道哪些 action 可以存取 API path。
圍繞這兩塊看資源 handler 的註冊過程。
func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage, ws *restful.WebService) (*metav1.APIResource, *storageversion.ResourceInfo, error) {
// what verbs are supported by the storage, used to know what verbs we support per path
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)
gvAcceptor, _ := storage.(rest.GroupVersionAcceptor)
// Get the list of actions for the given scope.
switch {
case !namespaceScoped:
...
default:
// Handler for standard REST verbs (GET, PUT, POST and DELETE).
actions := []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)
}
for _, action := range actions {
switch action.Verb {
case "GET": // Get a resource.
var handler restful.RouteFunction
if isGetterWithOptions {
handler = restfulGetResourceWithOptions(getterWithOptions, reqScope, isSubresource)
} else {
handler = restfulGetResource(getter, reqScope)
}
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)
}
}
}
可以看到，通過資源實體 storage 支援的介面型別可以反射出資源支援的方法。接著將支援的方法加入 actions。actions 儲存的是 action.Verb 和支援的資源 API path 資訊。
拿到 actions 後，通過匹配 actions.Verb 建立 action.Verb -> action.Path -> handler 的路由。其中，建立 handler 需要用到 storage，因為 handler 是通過 storage 和 etcd 互動的。
通過 go-restful 庫建立上述路由，接著 ws.Route(route) 將 route 加入到 restful.WebService 中。
回頭看 InstallREST。
func (g *APIGroupVersion) InstallREST(container *restful.Container) ([]apidiscoveryv2beta1.APIResourceDiscovery, []*storageversion.ResourceInfo, error) {
apiResources, resourceInfos, ws, registrationErrors := installer.Install()
container.Add(ws)
return aggregatedDiscoveryResources, removeNonPersistedResources(resourceInfos), utilerrors.NewAggregate(registrationErrors)
}
func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *APIGroupInfo, typeConverter managedfields.TypeConverter) error {
discoveryAPIResources, r, err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer)
}
通過 container.Add(ws) 將 ws 新增到 go-restful 的 container 中，該 container 即為 APIExtensions Server 的 Handler 所提供的 GoRestfulContainer。
這裡介紹了 APIExtensions Server 的 REST API 建立過程。對於 KubeAPIServer 和 AggregatorServer 的建立過程與之類似，不過多介紹。
REST API 建立好以後，下一步就到如何執行 kube-apiserver了。
kube-apiserver 作為提供 RESTful API 的元件，其執行主要是監聽埠和啟動服務。理清了這點，就能在複雜的執行程式碼中找出頭緒。
呼叫 APIAggregator.PrepareRun 和 preparedAPIAggregator.Run 執行 kube-apiserver。
func Run(opts options.CompletedOptions, stopCh <-chan struct{}) error {
prepared, err := server.PrepareRun()
if err != nil {
return err
}
return prepared.Run(stopCh)
}
啟動過程在 PrepareRun 中。
func (s *APIAggregator) PrepareRun() (preparedAPIAggregator, error) {
prepared := s.GenericAPIServer.PrepareRun()
return preparedAPIAggregator{APIAggregator: s, runnable: prepared}, nil
}
執行 prepared.Run(stopCh) 實際呼叫的是 preparedGenericAPIServer.Run 方法。
func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
// 呼叫 preparedGenericAPIServer.NonBlockingRun
stoppedCh, listenerStoppedCh, err := s.NonBlockingRun(stopHttpServerCh, shutdownTimeout)
if err != nil {
return err
}
}
func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}, shutdownTimeout time.Duration) (<-chan struct{}, <-chan struct{}, error) {
if s.SecureServingInfo != nil && s.Handler != nil {
var err error
// 呼叫 SecureServingInfo.Serve
stoppedCh, listenerStoppedCh, err = s.SecureServingInfo.Serve(s.Handler, shutdownTimeout, internalStopCh)
if err != nil {
close(internalStopCh)
return nil, nil, err
}
}
}
func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Duration, stopCh <-chan struct{}) (<-chan struct{}, <-chan struct{}, error) {
// 組 http.Server
secureServer := &http.Server{
Addr: s.Listener.Addr().String(),
Handler: handler,
MaxHeaderBytes: 1 << 20,
TLSConfig: tlsConfig,
IdleTimeout: 90 * time.Second, // matches http.DefaultTransport keep-alive timeout
ReadHeaderTimeout: 32 * time.Second, // just shy of requestTimeoutUpperBound
}
return RunServer(secureServer, s.Listener, shutdownTimeout, stopCh)
}
func RunServer(
server *http.Server,
ln net.Listener,
shutDownTimeout time.Duration,
stopCh <-chan struct{},
) (<-chan struct{}, <-chan struct{}, error) {
go func() {
// 呼叫 http 的 Server.Serve 提供 `RESTful API` 服務
err := server.Serve(listener)
msg := fmt.Sprintf("Stopped listening on %s", ln.Addr().String())
}
}
可以看到，最終呼叫 http 包的 Server.Serve 提供 RESTful API 服務。
至此，已介紹完 kube-apiserver 從啟動到執行的核心邏輯。下一篇，將重點介紹 kube-apiserver 是怎麼和 etcd 進行互動的。