k8s apiserver源码

restStorageProviders := []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{},
              discoveryrest.StorageProvider{},
              extensionsrest.RESTStorageProvider{},
              networkingrest.RESTStorageProvider{},
              noderest.RESTStorageProvider{},
              policyrest.RESTStorageProvider{},
              rbacrest.RESTStorageProvider{Authorizer: c.GenericConfig.Authorization.Authorizer},
              schedulingrest.RESTStorageProvider{},
              settingsrest.RESTStorageProvider{},
              storagerest.RESTStorageProvider{},
              flowcontrolrest.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.StorageProvider{},
              admissionregistrationrest.RESTStorageProvider{},
              eventsrest.RESTStorageProvider{TTL: c.ExtraConfig.EventTTL},
      }

// NewRESTStorage returns APIGroupInfo object.
func (p StorageProvider) NewRESTStorage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (genericapiserver.APIGroupInfo, bool, error) {
        apiGroupInfo := genericapiserver.NewDefaultAPIGroupInfo(apps.GroupName, legacyscheme.Scheme, legacyscheme.ParameterCodec, legacyscheme.Codecs)
        // 必须得启用某几个版本, 否则VersionedResourcesStorageMap里是空的
        if apiResourceConfigSource.VersionEnabled(appsapiv1.SchemeGroupVersion) {
                storageMap, err := p.v1Storage(apiResourceConfigSource, restOptionsGetter)
                if err != nil {
                        return genericapiserver.APIGroupInfo{}, false, err
                }
                apiGroupInfo.VersionedResourcesStorageMap[appsapiv1.SchemeGroupVersion.Version] = storageMap
        }

        return apiGroupInfo, true, nil
}

func (p StorageProvider) v1Storage(apiResourceConfigSource serverstorage.APIResourceConfigSource, restOptionsGetter generic.RESTOptionsGetter) (map[string]rest.Storage, error) {
        storage := map[string]rest.Storage{}

        // deployments
        deploymentStorage, err := deploymentstore.NewStorage(restOptionsGetter)
        if err != nil {
                return storage, err
        }
        storage["deployments"] = deploymentStorage.Deployment
        storage["deployments/status"] = deploymentStorage.Status
        storage["deployments/scale"] = deploymentStorage.Scale

        // ...
}

// kubernetes/pkg/registry/apps/deployment/storage/storage.go
// NewStorage returns new instance of DeploymentStorage.
func NewStorage(optsGetter generic.RESTOptionsGetter) (DeploymentStorage, error) {
        deploymentRest, deploymentStatusRest, deploymentRollbackRest, err := NewREST(optsGetter)
        if err != nil {
                return DeploymentStorage{}, err
        }

        return DeploymentStorage{
                Deployment: deploymentRest,
                Status:     deploymentStatusRest,
                Scale:      &ScaleREST{store: deploymentRest.Store},
                Rollback:   deploymentRollbackRest,
        }, nil
}

// REST implements a RESTStorage for Deployments.
type REST struct {
        *genericregistry.Store
        categories []string
}

// NewREST returns a RESTStorage object that will work against deployments.
func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST, *RollbackREST, error) {
        store := &genericregistry.Store{
                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,

                TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)},
        }
        options := &generic.StoreOptions{RESTOptions: optsGetter}
        // 根据配置文件初始化store
        if err := store.CompleteWithOptions(options); err != nil {
                return nil, nil, nil, err
        }

        statusStore := *store
        statusStore.UpdateStrategy = deployment.StatusStrategy
        return &REST{store, []string{"all"}}, &StatusREST{store: &statusStore}, &RollbackREST{store: store}, nil
}

// 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 in 1.11
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

// registerResourceHandlers函数
switch action.Verb {
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, deprecated, removedRelease, handler)
        } else {
                handler = metrics.InstrumentRouteFunc(action.Verb, group, version, resource, subresource, requestScope, metrics.APIServerComponent, deprecated, removedRelease, handler)
        }
        if enableWarningHeaders {
                handler = utilwarning.AddWarningsHandler(handler, warnings)
        }

        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)
        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)
        routes = append(routes, route)

// pkg/registry/apps/deployment/storage/storage.go
store := &genericregistry.Store{
        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,

        TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)},
}

// 然后提供options完成初始化
options := &generic.StoreOptions{RESTOptions: optsGetter}
if err := store.CompleteWithOptions(options); err != nil {
        return nil, nil, nil, err
}

if e.Storage.Storage == nil {
        e.Storage.Codec = opts.StorageConfig.Codec
        var err error
        e.Storage.Storage, e.DestroyFunc, err = opts.Decorator(
                opts.StorageConfig,
                prefix,
                keyFunc,
                e.NewFunc,
                e.NewListFunc,
                attrFunc,
                options.TriggerFunc,
                options.Indexers,
        )
        if err != nil {
                return err
        }
        e.StorageVersioner = opts.StorageConfig.EncodeVersioner

        if opts.CountMetricPollPeriod > 0 {
                stopFunc := e.startObservingCount(opts.CountMetricPollPeriod)
                previousDestroy := e.DestroyFunc
                e.DestroyFunc = func() {
                        stopFunc()
                        if previousDestroy != nil {
                                previousDestroy()
                        }
                }
        }
}

func (s *EtcdOptions) ApplyWithStorageFactoryTo(factory serverstorage.StorageFactory, c *server.Config) error {
        if err := s.addEtcdHealthEndpoint(c); err != nil {
                return err
        }
        c.RESTOptionsGetter = &StorageFactoryRestOptionsFactory{Options: *s, StorageFactory: factory}
        return nil
}

type StorageFactoryRestOptionsFactory struct {
        Options        EtcdOptions
        StorageFactory serverstorage.StorageFactory
}

func (f *StorageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
        storageConfig, err := f.StorageFactory.NewConfig(resource)
        if err != nil {
                return generic.RESTOptions{}, fmt.Errorf("unable to find storage destination for %v, due to %v", resource, err.Error())
        }

        ret := generic.RESTOptions{
                StorageConfig:           storageConfig,
                Decorator:               generic.UndecoratedStorage,
                DeleteCollectionWorkers: f.Options.DeleteCollectionWorkers,
                EnableGarbageCollection: f.Options.EnableGarbageCollection,
                ResourcePrefix:          f.StorageFactory.ResourcePrefix(resource),
                CountMetricPollPeriod:   f.Options.StorageConfig.CountMetricPollPeriod,
        }
        // 如果配置了EnableWatchCache, 设置Decorator = genericregistry.StorageWithCacher()
        if f.Options.EnableWatchCache {
                sizes, err := ParseWatchCacheSizes(f.Options.WatchCacheSizes)
                if err != nil {
                        return generic.RESTOptions{}, err
                }
                size, ok := sizes[resource]
                if ok && size > 0 {
                        klog.Warningf("Dropping watch-cache-size for %v - watchCache size is now dynamic", resource)
                }
                if ok && size <= 0 {
                        ret.Decorator = generic.UndecoratedStorage
                } else {
                        ret.Decorator = genericregistry.StorageWithCacher()
                }
        }

        return ret, nil
}

import (
        ...
        cacherstorage "k8s.io/apiserver/pkg/storage/cacher"
        ...
)

// Creates a cacher based given storageConfig.
func StorageWithCacher() generic.StorageDecorator {
        return func(
                storageConfig *storagebackend.Config,
                resourcePrefix string,
                keyFunc func(obj runtime.Object) (string, error),
                newFunc func() runtime.Object,
                newListFunc func() runtime.Object,
                getAttrsFunc storage.AttrFunc,
                triggerFuncs storage.IndexerFuncs,
                indexers *cache.Indexers) (storage.Interface, factory.DestroyFunc, error) {

                s, d, err := generic.NewRawStorage(storageConfig)
                if err != nil {
                        return s, d, err
                }
                if klog.V(5).Enabled() {
                        klog.Infof("Storage caching is enabled for %s", objectTypeToString(newFunc()))
                }

                cacherConfig := cacherstorage.Config{
                        Storage:        s,
                        Versioner:      etcd3.APIObjectVersioner{},
                        ResourcePrefix: resourcePrefix,
                        KeyFunc:        keyFunc,
                        NewFunc:        newFunc,
                        NewListFunc:    newListFunc,
                        GetAttrsFunc:   getAttrsFunc,
                        IndexerFuncs:   triggerFuncs,
                        Indexers:       indexers,
                        Codec:          storageConfig.Codec,
                }
                cacher, err := cacherstorage.NewCacherFromConfig(cacherConfig)
                if err != nil {
                        return nil, func() {}, err
                }
                destroyFunc := func() {
                        cacher.Stop()
                        d()
                }

                // TODO : Remove RegisterStorageCleanup below when PR
                // https://github.com/kubernetes/kubernetes/pull/50690
                // merges as that shuts down storage properly
                RegisterStorageCleanup(destroyFunc)

                return cacher, destroyFunc, nil
        }
}

// NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from
// its internal cache and updating its cache in the background based on the
// given configuration.
func NewCacherFromConfig(config Config) (*Cacher, error) {
        // ...
        objType := reflect.TypeOf(obj)
        cacher := &Cacher{
                ready:          newReady(),
                storage:        config.Storage,
                objectType:     objType,
                versioner:      config.Versioner,
                newFunc:        config.NewFunc,
                indexedTrigger: indexedTrigger,
                watcherIdx:     0,
                watchers: indexedWatchers{
                        allWatchers:   make(map[int]*cacheWatcher),
                        valueWatchers: make(map[string]watchersMap),
                },
                // TODO: Figure out the correct value for the buffer size.
                incoming:              make(chan watchCacheEvent, 100),
                dispatchTimeoutBudget: newTimeBudget(stopCh),
                // We need to (potentially) stop both:
                // - wait.Until go-routine
                // - reflector.ListAndWatch
                // and there are no guarantees on the order that they will stop.
                // So we will be simply closing the channel, and synchronizing on the WaitGroup.
                stopCh:           stopCh,
                clock:            config.Clock,
                timer:            time.NewTimer(time.Duration(0)),
                bookmarkWatchers: newTimeBucketWatchers(config.Clock, defaultBookmarkFrequency),
        }

        // ...

        // 缓存对象, 实现了Store接口(Add/Delete等), 处理对应的watch事件
        watchCache := newWatchCache(
                config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers, config.Clock, objType)
        // 实现了List/Watch接口, 本质上是一个etcd的client
        listerWatcher := NewCacherListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc)
        reflectorName := "storage/cacher.go:" + config.ResourcePrefix

        // 包装了watchCache和listerWater
        reflector := cache.NewNamedReflector(reflectorName, listerWatcher, obj, watchCache, 0)
        // Configure reflector's pager to for an appropriate pagination chunk size for fetching data from
        // storage. The pager falls back to full list if paginated list calls fail due to an "Expired" error.
        reflector.WatchListPageSize = storageWatchListPageSize

        cacher.watchCache = watchCache
        cacher.reflector = reflector

        // 启动事件分发器, 该函数会不断从cacher.incoming这个chan中读取事件
        go cacher.dispatchEvents()

        cacher.stopWg.Add(1)
        go func() {
                defer cacher.stopWg.Done()
                defer cacher.terminateAllWatchers()
                wait.Until(
                        func() {
                                if !cacher.isStopped() {
                                        cacher.startCaching(stopCh)
                                }
                        }, time.Second, stopCh,
                        // 每隔一秒调用一次startCache
                )
        }()

        return cacher, nil
}

func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
        // ...

        // Note that since onReplace may be not called due to errors, we explicitly
        // need to retry it on errors under lock.
        // Also note that startCaching is called in a loop, so there's no need
        // to have another loop here.
        if err := c.reflector.ListAndWatch(stopChannel); err != nil {
                klog.Errorf("cacher (%v): unexpected ListAndWatch error: %v; reinitializing...", c.objectType.String(), err)
        }
}

// Assumes that lock is already held for write.
func (w *watchCache) updateCache(event *watchCacheEvent) {
        w.resizeCacheLocked(event.RecordTime)
        if w.isCacheFullLocked() {
                // Cache is full - remove the oldest element.
                w.startIndex++
        }
        w.cache[w.endIndex%w.capacity] = event
        w.endIndex++
}

// resizeCacheLocked resizes the cache if necessary:
// - increases capacity by 2x if cache is full and all cached events occurred within last eventFreshDuration.
// - decreases capacity by 2x when recent quarter of events occurred outside of eventFreshDuration(protect watchCache from flapping).
func (w *watchCache) resizeCacheLocked(eventTime time.Time) {
        if w.isCacheFullLocked() && eventTime.Sub(w.cache[w.startIndex%w.capacity].RecordTime) < eventFreshDuration {
                capacity := min(w.capacity*2, w.upperBoundCapacity)
                if capacity > w.capacity {
                        w.doCacheResizeLocked(capacity)
                }
                return
        }
        if w.isCacheFullLocked() && eventTime.Sub(w.cache[(w.endIndex-w.capacity/4)%w.capacity].RecordTime) > eventFreshDuration {
                capacity := max(w.capacity/2, w.lowerBoundCapacity)
                if capacity < w.capacity {
                        w.doCacheResizeLocked(capacity)
                }
                return
        }
}

// Watch implements storage.Interface.
func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
        watchRV, err := c.versioner.ParseResourceVersion(opts.ResourceVersion)
        // ...

        // 等待cache加载完成
        c.ready.wait()

        // ...

        // Determine watch timeout('0' means deadline is not set, ignore checking)
        deadline, _ := ctx.Deadline()
        // Create a watcher here to reduce memory allocations under lock,
        // given that memory allocation may trigger GC and block the thread.
        // Also note that emptyFunc is a placeholder, until we will be able
        // to compute watcher.forget function (which has to happen under lock).
        watcher := newCacheWatcher(chanSize, filterWithAttrsFunction(key, pred), emptyFunc, c.versioner, deadline, pred.AllowWatchBookmarks, c.objectType)


        // We explicitly use thread unsafe version and do locking ourself to ensure that
        // no new events will be processed in the meantime. The watchCache will be unlocked
        // on return from this function.
        // Note that we cannot do it under Cacher lock, to avoid a deadlock, since the
        // underlying watchCache is calling processEvent under its lock.
        // GetAllEventsSinceThreadUnsafe会将缓存内的所有数据(指定的ResourceVersion之后的数据)以Add事件的形式聚合
        c.watchCache.RLock()
        defer c.watchCache.RUnlock()
        initEvents, err := c.watchCache.GetAllEventsSinceThreadUnsafe(watchRV)
        // ...

        // With some events already sent, update resourceVersion so that
        // events that were buffered and not yet processed won't be delivered
        // to this watcher second time causing going back in time.
        // initEvents[len(initEvents)-1]是缓存中最新的数据
        if len(initEvents) > 0 {
                watchRV = initEvents[len(initEvents)-1].ResourceVersion
        }

        func() {
                c.Lock()
                defer c.Unlock()
                // Update watcher.forget function once we can compute it.
                watcher.forget = forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported)
                c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported)

                // Add it to the queue only when the client support watch bookmarks.
                if watcher.allowWatchBookmarks {
                        c.bookmarkWatchers.addWatcher(watcher)
                }
                c.watcherIdx++
        }()

        go watcher.process(ctx, initEvents, watchRV)
        return watcher, nil
}

// List implements storage.Interface.
func (c *Cacher) List(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
        resourceVersion := opts.ResourceVersion
        pred := opts.Predicate
        pagingEnabled := utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
        hasContinuation := pagingEnabled && len(pred.Continue) > 0
        hasLimit := pagingEnabled && pred.Limit > 0 && resourceVersion != "0"
        if resourceVersion == "" || hasContinuation || hasLimit || opts.ResourceVersionMatch == metav1.ResourceVersionMatchExact {
                // If resourceVersion is not specified, serve it from underlying
                // storage (for backward compatibility). If a continuation is
                // requested, serve it from the underlying storage as well.
                // Limits are only sent to storage when resourceVersion is non-zero
                // since the watch cache isn't able to perform continuations, and
                // limits are ignored when resource version is zero.
                return c.storage.List(ctx, key, opts, listObj)
        }

        // ...

        // 缓存没有ready
        if listRV == 0 && !c.ready.check() {
                // If Cacher is not yet initialized and we don't require any specific
                // minimal resource version, simply forward the request to storage.
                return c.storage.List(ctx, key, opts, listObj)
        }

        // ...

        // 获取数据
        objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV, pred.MatcherIndex(), trace)
        if err != nil {
                return err
        }

        // ...
        // 过滤数据, 可以看到是先从缓存获取全部数据, 在一个个过滤
        for _, obj := range objs {
                elem, ok := obj.(*storeElement)
                if !ok {
                        return fmt.Errorf("non *storeElement returned from storage: %v", obj)
                }
                if filter(elem.Key, elem.Labels, elem.Fields) {
                        listVal.Set(reflect.Append(listVal, reflect.ValueOf(elem.Object).Elem()))
                }
        }
}