Kubernetes API Server—请求处理链路

基于 Kubernetes v1.37 源码（staging/src/k8s.io/apiserver/）

一、API Server 的职责 @

API Server 是 Kubernetes 集群的唯一入口，所有组件都通过它与集群交互：

kubectl → API Server → etcd
kubelet → API Server → etcd
controller-manager → API Server → etcd
scheduler → API Server → etcd

核心职责：

REST API 暴露：提供 CRUD 接口
认证 (Authentication)：确认请求者身份
授权 (Authorization)：确认请求者权限
准入 (Admission)：修改/验证请求
存储抽象：封装 etcd 操作

二、整体架构 @

┌─────────────────────────────────────────────────────────────┐
│                     HTTP Server                             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │                  Filter Chain                         │  │
│  │  RequestInfo → Timeout → CORS → AuthN → Audit →     │  │
│  │  Impersonation → MaxInFlight/APF → AuthZ → Handler   │  │
│  └──────────────────────────────────────────────────────┘  │
│                            ↓                                │
│  ┌──────────────────────────────────────────────────────┐  │
│  │            Director (路由分发)                        │  │
│  │  /api/v1/pods → GoRestfulContainer                   │  │
│  │  /healthz → NonGoRestfulMux                          │  │
│  └──────────────────────────────────────────────────────┘  │
│                            ↓                                │
│  ┌──────────────────────────────────────────────────────┐  │
│  │           REST Handlers                               │  │
│  │  Create / Get / Update / Delete / Watch / List       │  │
│  └──────────────────────────────────────────────────────┘  │
│                            ↓                                │
│  ┌──────────────────────────────────────────────────────┐  │
│  │           Storage Layer                               │  │
│  │  Cacher (本地缓存) → etcd3 (持久化存储)              │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

三、启动流程 @

// cmd/kube-apiserver/app/server.go
func Run(ctx context.Context, opts options.CompletedOptions) error {
    // 1. 构建配置并补全默认值
    config, err := NewConfig(opts)
    completed, err := config.Complete()
    
    // 2. 创建 API Server 链（kube-apiserver → apiextensions → aggregator）
    server, err := CreateServerChain(completed)
    
    // 3. 准备运行（初始化存储、安装 API）
    prepared, err := server.PrepareRun()
    
    // 4. 启动服务
    return prepared.Run(ctx)
}

// staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go
func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
    // 1. 非阻塞启动 HTTPS 服务器
    //    内部调用 s.SecureServingInfo.Serve(s.Handler, ...)
    stoppedCh, err := s.NonBlockingRun(stopCh, shutdownTimeout)
    
    // 2. 执行 PostStartHooks
    
    // 3. 阻塞等待关闭信号，收到后优雅关停
    <-stopCh
    ...
}

四、Filter Chain（过滤器链） @

API Server 使用责任链模式处理请求，每个 Filter 负责一个职责：

HTTP Request
    ↓
WithRequestInfo (请求信息解析，最前端)
    ↓
WithWaitGroup (跟踪在途请求，配合优雅关停)
    ↓
WithRequestDeadline / WithTimeoutForNonLongRunningRequests (超时控制)
    ↓
WithWarningRecorder (告警头)
    ↓
WithCORS (跨域处理)
    ↓
WithAuthentication (认证)
    ↓
WithTracing (链路追踪)
    ↓
WithAudit (审计日志)
    ↓
WithImpersonation (用户伪装)
    ↓
WithMaxInFlightLimit / WithPriorityAndFairness (限流)
    ↓
WithAuthorization (授权)
    ↓
Director (路由分发)
    ↓
REST Handler（准入 Admission 在 Handler 内部执行，见 4.3 节）

注意：准入（Admission）不在 HTTP 过滤器链中，而是在 Create/Update 等 Handler 内部调用——Mutating 由 Handler 直接调用 MutationInterface.Admit，Validating 经 rest.AdmissionToValidateObjectFunc 包装后传入存储层。

4.1 认证 Filter (Authentication) @

// staging/src/k8s.io/apiserver/pkg/authentication/authenticator/interfaces.go
type Request interface {
    AuthenticateRequest(req *http.Request) (*Response, bool, error)
}

type Response struct {
    User user.Info  // 认证后的用户信息
    Audiences Audiences
}

认证方式链（按顺序尝试，见 pkg/kubeapiserver/authenticator/config.go）：

方式	说明
RequestHeader	前置代理（front-proxy）通过请求头传递的客户端证书身份
Client Certificate	X.509 客户端证书
Bearer Token	Token 链依次尝试：Token File → ServiceAccount → Bootstrap Token → OIDC → Webhook
Anonymous	匿名用户（以上都未认证时兜底）

// 认证链实现
// staging/src/k8s.io/apiserver/pkg/authentication/request/union/union.go
type unionAuthRequestHandler struct {
    Handlers []authenticator.Request
}

// 链式调用，任一成功即返回
func (authHandler *unionAuthRequestHandler) AuthenticateRequest(req *http.Request) (*authenticator.Response, bool, error) {
    for _, authenticator := range authHandler.Handlers {
        if resp, ok, err := authenticator.AuthenticateRequest(req); ok {
            return resp, ok, err
        }
    }
    return nil, false, aggregateErr
}

4.2 授权 Filter (Authorization) @

// staging/src/k8s.io/apiserver/pkg/authorization/authorizer/interfaces.go
type Authorizer interface {
    Authorize(ctx context.Context, a Attributes) (Decision, string, error)
}

type Decision int
const (
    DecisionDeny Decision = iota
    DecisionAllow
    DecisionNoOpinion
)

授权方式：

方式	说明
RBAC	基于角色的访问控制（最常用）
Node	节点专用授权（kubelet）
Webhook	外部授权服务
ABAC	基于属性的访问控制

// 授权链实现
// staging/src/k8s.io/apiserver/pkg/authorization/union/union.go
func (authzHandler unionAuthzHandler) Authorize(ctx context.Context, a Attributes) (Decision, string, error) {
    for _, authz := range authzHandler.chain {
        decision, reason, err := authz.Authorizer.Authorize(ctx, a)
        if decision == DecisionAllow || decision == DecisionDeny {
            return decision, reason, err
        }
        // DecisionNoOpinion：记录 reason，继续尝试下一个
    }
    // 没有任何插件给出 Allow/Deny：返回 NoOpinion，并拼接各插件的 reason
    return DecisionNoOpinion, strings.Join(reasonList, "\n"), aggregateErr
}

4.3 准入控制 (Admission) @

准入不在 HTTP Filter Chain 中，而是在 REST Handler（Create/Update/Delete 等）内部执行。

// staging/src/k8s.io/apiserver/pkg/admission/interfaces.go
// Interface 只有 Handles；Admit 在 MutationInterface，Validate 在 ValidationInterface
type Interface interface {
    Handles(operation Operation) bool
}

type MutationInterface interface {
    Interface
    Admit(ctx context.Context, a Attributes, o ObjectInterfaces) error
}

type ValidationInterface interface {
    Interface
    Validate(ctx context.Context, a Attributes, o ObjectInterfaces) error
}

准入插件执行顺序：

MutatingAdmission（变更型）：修改请求对象
- 添加默认值
- 注入 Sidecar 容器
- 设置资源限制
ValidatingAdmission（验证型）：验证请求对象
- 检查约束条件
- 验证业务规则

// 准入链执行
func (h *AdmissionHandler) Admit(ctx context.Context, a Attributes, o ObjectInterfaces) error {
    // 1. 执行所有 Mutating 插件
    for _, plugin := range h.mutatingPlugins {
        if plugin.Handles(a.GetOperation()) {
            if err := plugin.Admit(ctx, a, o); err != nil {
                return err
            }
        }
    }
    
    // 2. 执行所有 Validating 插件
    for _, plugin := range h.validatingPlugins {
        if plugin.Handles(a.GetOperation()) {
            if err := plugin.Validate(ctx, a, o); err != nil {
                return err
            }
        }
    }
    return nil
}

五、REST Handler @

请求通过 Filter Chain 后，Director 根据 URL 分发到对应的 REST Handler。

5.1 路由注册 @

// staging/src/k8s.io/apiserver/pkg/endpoints/installer.go
func (a *APIInstaller) Install() ([]metav1.APIResource, []*storageversion.ResourceInfo, *restful.WebService, []error) {
    // 注册 /api/v1/pods
    ws.Path(a.prefix).Doc("...")
    
    // 注册 Create
    routes := []restful.RouteBuilder{}
    routes = append(routes, restful.POST(a.prefix).
        To(handler).
        Operation("create").
        ...
    )
    
    // 注册 Get
    routes = append(routes, restful.GET(a.prefix).
        To(handler).
        Operation("get").
        ...
    )
    
    // 注册 List
    routes = append(routes, restful.GET(a.prefix).
        To(handler).
        Operation("list").
        ...
    )
    
    // 注册 Update
    routes = append(routes, restful.PUT(a.prefix).
        To(handler).
        Operation("update").
        ...
    )
    
    // 注册 Delete
    routes = append(routes, restful.DELETE(a.prefix).
        To(handler).
        Operation("delete").
        ...
    )
    
    // 注册 Watch
    routes = append(routes, restful.GET(a.prefix).
        To(handler).
        Operation("watch").
        ...
    )
}

5.2 REST Storage 接口 @

// staging/src/k8s.io/apiserver/pkg/registry/rest/rest.go
type Storage interface {
    New() runtime.Object
    Destroy()
}

type Creater interface {
    Storage
    Create(ctx context.Context, obj runtime.Object, createValidation ValidateObjectFunc, opts *metav1.CreateOptions) (runtime.Object, error)
}

type Getter interface {
    Storage
    Get(ctx context.Context, name string, opts *metav1.GetOptions) (runtime.Object, error)
}

type Updater interface {
    Storage
    Update(ctx context.Context, name string, objInfo UpdatedObjectInfo, createValidation ValidateObjectFunc, updateValidation ValidateObjectUpdateFunc, forceAllowCreate bool, opts *metav1.UpdateOptions) (runtime.Object, bool, error)
}

// 删除单个对象
type GracefulDeleter interface {
    Storage
    Delete(ctx context.Context, name string, deleteValidation ValidateObjectFunc, opts *metav1.DeleteOptions) (runtime.Object, bool, error)
}

// 删除整个集合
type CollectionDeleter interface {
    Storage
    DeleteCollection(ctx context.Context, deleteValidation ValidateObjectFunc, opts *metav1.DeleteOptions, listOptions *metainternalversion.ListOptions) (runtime.Object, error)
}

type Watcher interface {
    Storage
    Watch(ctx context.Context, opts *metainternalversion.ListOptions) (watch.Interface, error)
}

5.3 Create Handler 示例 @

// staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go
func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Interface, includeName bool) http.HandlerFunc {
    return func(w http.ResponseWriter, req *http.Request) {
        // 1. 解析请求
        namespace, name, _ := scope.Namer.Name(req)
        
        // 2. 内容协商（JSON/Protobuf）
        outputMediaType, _, _ := negotiation.NegotiateOutputMediaType(req, scope.Serializer, scope)
        
        // 3. 反序列化请求体（解码到内部版本）
        obj := r.New()
        decoder := scope.Serializer.DecoderToVersion(decodeSerializer, scope.HubGroupVersion)
        obj, gvk, err := decoder.Decode(body, &defaultGVK, obj)
        
        // 4. 执行 Mutating 准入
        if mutatingAdmission, ok := admit.(admission.MutationInterface); ok {
            err = mutatingAdmission.Admit(ctx, admissionAttributes, scope)
        }
        
        // 5. 执行创建（Validating 准入经 AdmissionToValidateObjectFunc 传入存储层）
        obj, err = r.Create(ctx, name, obj,
            rest.AdmissionToValidateObjectFunc(admit, admissionAttributes, scope),
            options)
        
        // 6. 返回结果
        responsewriters.WriteObject(ctx, outputMediaType, scope.Serializer, obj, w, req, statusCode)
    }
}

六、Storage Layer（存储层） @

6.1 Cacher（缓存层） @

API Server 使用 Cacher 作为存储抽象层，减少 etcd 压力：

// staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
type Cacher struct {
    storage    storage.Interface // 底层 etcd3 存储
    watchCache *watchCache       // 近期事件的滑动窗口 + 对象当前状态
    reflector  *cache.Reflector  // 持续 watch etcd，异步填充 watchCache
}

func (c *Cacher) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error {
    // 等待 watchCache 同步到足够新的版本后，直接从缓存读取
    obj, exists, readRV, err := c.watchCache.WaitUntilFreshAndGet(ctx, getRV, key)
    if err != nil {
        return err
    }
    if !exists {
        return storage.NewKeyNotFoundError(key, int64(readRV))
    }
    // 将缓存对象写入 out
    ...
}

func (c *Cacher) Watch(ctx context.Context, opts storage.ListOptions) (watch.Interface, error) {
    // 通过 watchCache 实现高效 Watch
    ...
}

6.2 etcd3 存储 @

// staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
type store struct {
    client    *clientv3.Client
    codec     runtime.Codec
    versioner storage.Versioner
}

func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
    // 1. 校验 RV 必须为 0（Create 不允许对象自带 RV）
    if version, err := s.versioner.ObjectResourceVersion(obj); err == nil && version != 0 {
        return storage.ErrResourceVersionSetOnCreate
    }
    // 写入前清空 resourceVersion / selfLink
    s.versioner.PrepareObjectForStorage(obj)
    
    // 2. 序列化
    data, err := runtime.Encode(s.codec, obj)
    
    // 3. 乐观写入 etcd（仅当 key 不存在时成功）
    txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, key, data, 0, ...)
    if !txnResp.Succeeded {
        return storage.NewKeyExistsError(key, 0)
    }
    
    return err
}

func (s *store) Get(ctx context.Context, key string, opts storage.GetOptions, out runtime.Object) error {
    // 1. 从 etcd 读取
    resp, err := s.client.Get(ctx, key)
    
    // 2. 反序列化
    return runtime.Decode(s.codec, resp.Kvs[0].Value, out)
}

七、Resource Version（资源版本） @

Resource Version (RV) 是 API Server 实现乐观并发控制的关键机制：

7.1 作用 @

并发控制：防止多个客户端同时修改同一对象
Watch 机制：客户端可以通过 RV 获取增量更新
一致性保证：保证读取到的数据是最新的

7.2 工作原理 @

# 客户端读取 Pod
GET /api/v1/namespaces/default/pods/my-pod
# 返回: metadata.resourceVersion = "12345"

# 客户端更新 Pod
PUT /api/v1/namespaces/default/pods/my-pod
{
  "metadata": {
    "resourceVersion": "12345"  # 必须匹配当前 RV
  },
  "spec": { ... }
}

如果 RV 不匹配，API Server 返回 409 Conflict：

{
  "kind": "Status",
  "apiVersion": "v1",
  "status": "Failure",
  "message": "Operation cannot be fulfilled on pods \"my-pod\": the object has been modified; please apply your changes to the latest version and try again",
  "reason": "Conflict",
  "code": 409
}

7.3 实现 @

// staging/src/k8s.io/apiserver/pkg/storage/api_object_versioner.go
type APIObjectVersioner struct{}

func (v APIObjectVersioner) PrepareObjectForStorage(obj runtime.Object) error {
    // 写入 etcd 前清空 resourceVersion 和 selfLink；
    // 新对象的 RV 来自 etcd 写入成功后的 revision，并非固定为 1
    accessor, _ := meta.Accessor(obj)
    accessor.SetResourceVersion("")
    accessor.SetSelfLink("")
    return nil
}

func (v APIObjectVersioner) ObjectResourceVersion(obj runtime.Object) (uint64, error) {
    accessor, _ := meta.Accessor(obj)
    version := accessor.GetResourceVersion()
    if version == "" {
        return 0, nil
    }
    return strconv.ParseUint(version, 10, 64)
}

八、Watch 机制 @

Watch 是 Kubernetes 实现事件驱动的核心机制：

8.1 客户端 Watch @

// client-go 中的 Watch（由 gentype 生成的类型化客户端）
// staging/src/k8s.io/client-go/gentype/type.go
func (c *Client[T]) Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
    opts.Watch = true
    // 发送 Watch 请求
    // GET /api/v1/namespaces/default/pods?watch=true&resourceVersion=12345
    return c.client.Get().
        NamespaceIfScoped(c.namespace, c.namespace != "").
        Resource(c.resource).
        VersionedParams(&opts, c.parameterCodec).
        Watch(ctx)
}

8.2 服务端 WatchCache @

// staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go
type watchCache struct {
    sync.RWMutex
    cond            *sync.Cond               // 等待缓存推进到足够新的 RV
    resourceVersion uint64                   // 缓存已推进到的 RV
    history         *watchCacheHistory       // 近期事件历史（滑动窗口）
    storage         *store.WatchCacheStorage // 对象当前状态
}

func (w *watchCache) Add(obj interface{}) error {
    // 事件 RV 取自事件对象自身的 resourceVersion
    object, resourceVersion, err := w.objectToVersionedRuntimeObject(obj)
    event := watch.Event{Type: watch.Added, Object: object}
    return w.processEvent(event, resourceVersion)
}

// 从指定 RV 开始返回事件（未导出方法，需在锁内调用）
func (w *watchCache) getAllEventsSinceLocked(resourceVersion uint64, key string, opts storage.ListOptions) (*watchCacheInterval, error) {
    ...
    return w.history.GetIntervalLocked(resourceVersion, ...)
}

8.3 Watch 事件流 @

客户端 Watch → API Server → etcd Watch
                    ↓
            WatchCache (本地缓存)
                    ↓
            事件推送给客户端

九、请求处理完整流程 @

以 kubectl create -f pod.yaml 为例：

1. kubectl 发送 POST 请求
   POST /api/v1/namespaces/default/pods
   Content-Type: application/json
   Authorization: Bearer <token>
   
   {
     "apiVersion": "v1",
     "kind": "Pod",
     "metadata": { "name": "my-pod" },
     "spec": { ... }
   }
       ↓
2. API Server 接收请求
       ↓
3. Filter Chain 处理：
   - WithRequestInfo: 解析请求信息
   - WithTimeout: 设置超时
   - WithCORS: 检查跨域头
   - WithAuthentication: 认证（验证 Token）
   - WithAudit: 记录审计日志
   - WithMaxInFlight/APF: 限流检查
   - WithAuthorization: 授权（检查 RBAC）
       ↓
4. Director 路由分发：
   URL 匹配 /api/v1/namespaces/{namespace}/pods → Create Handler
       ↓
5. Create Handler 处理：
   - 内容协商（JSON）
   - 反序列化请求体
   - 执行准入控制（Mutating 直接调用；Validating 传入存储层）
     - Mutating: 添加默认值
     - Validating: 验证约束
   - 调用 Storage.Create()
       ↓
6. Storage Layer：
   - Cacher: 写操作直接委托底层 etcd3 store 写入 etcd
   - 本地缓存由 reflector 对 etcd 的 watch 异步填充
       ↓
7. 返回响应：
   201 Created
   {
     "metadata": {
       "name": "my-pod",
       "namespace": "default",
       "resourceVersion": "12345",
       "creationTimestamp": "..."
     },
     "spec": { ... },
     "status": {}
   }

十、设计亮点总结 @

责任链模式：Filter Chain 设计使得每个职责解耦，易于扩展。
读写分离：Cacher 层实现读写分离——读从本地缓存读取（微秒级），写写入 etcd（毫秒级）。
乐观并发控制：Resource Version 实现无锁并发，避免数据竞争。
Watch 机制：客户端通过 Watch 实时感知变化，实现事件驱动架构。
插件化架构：认证、授权、准入均可插拔，支持自定义扩展。

十一、总结 @

API Server 的设计体现了几个核心思想：

单一入口：所有组件通过 API Server 交互，便于审计和控制
分层架构：Filter Chain → Handler → Storage，职责清晰
高性能：Cacher 缓存 + Watch 机制，减少 etcd 压力
可扩展：插件化架构，认证/授权/准入均可自定义
一致性：Resource Version 保证并发安全

理解 API Server，就掌握了 Kubernetes 集群的"咽喉"。后续的控制器、调度器、kubelet，都是 API Server 的客户端。