【技术】从POD创建看Kubernetes源码实现（二）- kube-apiserver
·
✍️ 作者：茶水间Tech
🏷️ 标签：#云计算#云原生#kubernetes#容器
📖 前言
kubernetes的模块比较多，架构复杂，代码量更是庞大，看代码比较麻烦，我们从现实场景出发，从创建POD分析在Kubernetes内部的代码流程，本系列文章从POD创建，整体梳理Kubernetes源码实现，其中本节主要分析kube-apiserver侧的流程实现。
本文基于 Client Version: v1.34.3 ， Server Version: v1.34.2
📌 POD创建的整体架构图：
💻 正文
📑 一、关于kube-apiserver
整个Kubernetes技术体系由声明式API以及Controller构成，而kube-apiserver是Kubernetes的声明式api server，作为整个Kubernetes集群操作etcd的唯一入口，负责Kubernetes各资源的认证&鉴权，校验以及CRUD等操作，提供RESTful APIs，供其它组件调用
官方地址：https://github.com/kubernetes/apiserver
k8s API Server提供了k8s各类资源对象（pod,RC,Service等）的增删改查及watch等HTTP Rest接口，是整个系统的数据总线和数据中心。
kubernetes API Server的功能：
- 提供了集群管理的REST API接口(包括认证授权、数据校验以及集群状态变更)；
- 提供其他模块之间的数据交互和通信的枢纽（其他模块通过API Server查询或修改数据，只有API Server才直接操作etcd）;
- 是资源配额控制的入口；
- 拥有完备的集群安全机制.
整个kubeAPIServer提供了三类API Resource接口：
- core group：主要在
/api/v1下； - named groups：其 path 为
/apis/$GROUP/$VERSION； - 系统状态的一些 API：如
/metrics、/version等；
而API的URL大致以 /apis/{group}/{version}/namespaces/{namespace}/resource/{name} 组成，结构如下图所示：
📑 二、代码分析
上文中，kubectl 最终请求 https://:/apis/apps/v1/namespaces//deployments 创建了nginx的deployment
代码逻辑图
详细流程如下：
2.1 程序入口：CreateResource (create.go)
apiserver 启动的时候添加了为handler 添加了WithAuthentication/WithAuthorization等中间件filter，经filter过滤后创建POD的代码在 CreateResource中（installer.go 中的 registerResourceHandlers 中将POST请求转过来），实际createHandler() 中 实现了完整的资源创建流程，包括参数验证、准入控制WithAdmission、对象转换、存储和响应处理等所有必要步骤。
代码路径：kubernetes/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go
// 参数: 资源的创建接口, 请求作用域以及准入控制接口
func CreateResource(r rest.Creater, scope *RequestScope, admission admission.Interface) http.HandlerFunc {
return createHandler(&namedCreaterAdapter{r}, scope, admission, false)
}
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, err := scope.Namer.Name(req)
//...(略)
// 设置 34 秒的上限超时
ctx, cancel := context.WithTimeout(ctx, requestTimeoutUpperBound)
//...(略)
//读取POST请求体body（带大小限制）
body, err := limitedReadBodyWithRecordMetric(ctx, req, scope.MaxRequestBodyBytes, scope.Resource.GroupResource(), requestmetrics.Create)
if err != nil {
span.AddEvent("limitedReadBody failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req)
return
}
span.AddEvent("limitedReadBody succeeded", attribute.Int("len", len(body)))
// 解析创建选项
options := &metav1.CreateOptions{}
values := req.URL.Query()
if err := metainternalversionscheme.ParameterCodec.DecodeParameters(values, scope.MetaGroupVersion, options); err != nil {
err = errors.NewBadRequest(err.Error())
scope.err(err, w, req)
return
}
if errs := validation.ValidateCreateOptions(options); len(errs) > 0 {
err := errors.NewInvalid(schema.GroupKind{Group: metav1.GroupName, Kind: "CreateOptions"}, "", errs)
scope.err(err, w, req)
return
}
options.TypeMeta.SetGroupVersionKind(metav1.SchemeGroupVersion.WithKind("CreateOptions"))
defaultGVK := scope.Kind
original := r.New()
// 字段验证
validationDirective := fieldValidation(options.FieldValidation)
decodeSerializer := s.Serializer
if validationDirective == metav1.FieldValidationWarn || validationDirective == metav1.FieldValidationStrict {
decodeSerializer = s.StrictSerializer
}
// 将body按照编码正常期望版本的obj
decoder := scope.Serializer.DecoderToVersion(decodeSerializer, scope.HubGroupVersion)
span.AddEvent("About to convert to expected version")
obj, gvk, err := decoder.Decode(body, &defaultGVK, original)
// ...(略)
// 验证API版本
if !scope.AcceptsGroupVersion(objGV) {
err = errors.NewBadRequest(fmt.Sprintf("the API version in the data (%s) does not match the expected API version (%v)", objGV.String(), gv.String()))
scope.err(err, w, req)
return
}
span.AddEvent("Conversion done")
// 处理创建的name 以及namespace
if len(name) == 0 {
_, name, _ = scope.Namer.ObjectName(obj)
}
if len(namespace) == 0 && scope.Resource == namespaceGVR {
namespace = name
}
ctx = request.WithNamespace(ctx, namespace)
admit = admission.WithAudit(admit)
audit.LogRequestObject(req.Context(), obj, objGV, scope.Resource, scope.Subresource, scope.Serializer)
userInfo, _ := request.UserFrom(ctx)
if objectMeta, err := meta.Accessor(obj); err == nil {
preserveObjectMetaSystemFields := false
if c, ok := r.(rest.SubresourceObjectMetaPreserver); ok && len(scope.Subresource) > 0 {
preserveObjectMetaSystemFields = c.PreserveRequestObjectMetaSystemFieldsOnSubresourceCreate()
}
if !preserveObjectMetaSystemFields {
rest.WipeObjectMetaSystemFields(objectMeta)
}
// ensure namespace on the object is correct, or error if a conflicting namespace was set in the object
if err := rest.EnsureObjectNamespaceMatchesRequestNamespace(rest.ExpectedNamespaceForResource(namespace, scope.Resource), objectMeta); err != nil {
scope.err(err, w, req)
return
}
}
span.AddEvent("About to store object in database")
//准备准入控制所需的属性插件内容
admissionAttributes := admission.NewAttributesRecord(obj, nil, scope.Kind, namespace, name, scope.Resource, scope.Subresource, admission.Create, options, dryrun.IsDryRun(options.DryRun), userInfo)
// 创建请求函数
requestFunc := func() (runtime.Object, error) {
return r.Create(
ctx,
name,
obj,
rest.AdmissionToValidateObjectFunc(admit, admissionAttributes, scope),
options,
)
}
// Dedup owner references before updating managed fields
dedupOwnerReferencesAndAddWarning(obj, req.Context(), false)
//执行创建
result, err := finisher.FinishRequest(ctx, func() (runtime.Object, error) {
//创建新对象
liveObj, err := scope.Creater.New(scope.Kind)
if err != nil {
return nil, fmt.Errorf("failed to create new object (Create for %v): %v", scope.Kind, err)
}
//更新字段管理器
obj = scope.FieldManager.UpdateNoErrors(liveObj, obj, managerOrUserAgent(options.FieldManager, req.UserAgent()))
admit = fieldmanager.NewManagedFieldsValidatingAdmissionController(admit)
//执行 Validating Admission (验证型准入)
if mutatingAdmission, ok := admit.(admission.MutationInterface); ok && mutatingAdmission.Handles(admission.Create) {
if err := mutatingAdmission.Admit(ctx, admissionAttributes, scope); err != nil {
return nil, err
}
}
// Dedup owner references again after mutating admission happens
dedupOwnerReferencesAndAddWarning(obj, req.Context(), true)
//执行之际创建
result, err := requestFunc()
// If the object wasn't committed to storage because it's serialized size was too large,
// it is safe to remove managedFields (which can be large) and try again.
if isTooLargeError(err) {
if accessor, accessorErr := meta.Accessor(obj); accessorErr == nil {
accessor.SetManagedFields(nil)
result, err = requestFunc()
}
}
return result, err
})
if err != nil {
span.AddEvent("Write to database call failed", attribute.Int("len", len(body)), attribute.String("err", err.Error()))
scope.err(err, w, req)
return
}
span.AddEvent("Write to database call succeeded", attribute.Int("len", len(body)))
// 设置状态码
code := http.StatusCreated
status, ok := result.(*metav1.Status)
if ok && status.Code == 0 {
status.Code = int32(code)
}
span.AddEvent("About to write a response")
defer span.AddEvent("Writing http response done")
// 转换响应对象并返回
transformResponseObject(ctx, scope, req, w, code, outputMediaType, result)
}
}
handler 过滤器
func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
handler := apiHandler
handler = filterlatency.TrackCompleted(handler)
handler = genericapifilters.WithAuthorization(handler, c.Authorization.Authorizer, c.Serializer)
//...(略)
//认证过滤器
handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences, c.Authentication.RequestHeaderConfig)
//...(略)
handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver)
handler = genericapifilters.WithRequestReceivedTimestamp(handler)
handler = genericapifilters.WithMuxAndDiscoveryComplete(handler, c.lifecycleSignals.MuxAndDiscoveryComplete.Signaled())
handler = genericfilters.WithPanicRecovery(handler, c.RequestInfoResolver)
handler = genericapifilters.WithAuditInit(handler)
return handler
}
2.2 实际Create流程：create （store.go）
requestFunc 调用Store.create() 准备deployment对象，再调用e.Storage.Create() 往etcd中写入deployment序列化后的数据
代码路径：kubernetes/staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go
func (e *Store) create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
var finishCreate FinishFunc = finishNothing
// Init metadata as early as possible.
if objectMeta, err := meta.Accessor(obj); err != nil {
return nil, err
} else {
rest.FillObjectMetaSystemFields(objectMeta)
if len(objectMeta.GetGenerateName()) > 0 && len(objectMeta.GetName()) == 0 {
objectMeta.SetName(e.CreateStrategy.GenerateName(objectMeta.GetGenerateName()))
}
}
if e.BeginCreate != nil {
fn, err := e.BeginCreate(ctx, obj, options)
if err != nil {
return nil, err
}
finishCreate = fn
defer func() {
finishCreate(ctx, false)
}()
}
if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
return nil, err
}
// at this point we have a fully formed object. It is time to call the validators that the apiserver
// handling chain wants to enforce.
if createValidation != nil {
if err := createValidation(ctx, obj.DeepCopyObject()); err != nil {
return nil, err
}
}
name, err := e.ObjectNameFunc(obj)
if err != nil {
return nil, err
}
// 生成存储 key
key, err := e.KeyFunc(ctx, name)
if err != nil {
return nil, err
}
qualifiedResource := e.qualifiedResourceFromContext(ctx)
// 准备存储对象 , 计算ttl
ttl, err := e.calculateTTL(obj, 0, false)
if err != nil {
return nil, err
}
// 关键：调用底层存储接口写入 etcd
out := e.NewFunc()
if err := e.Storage.Create(ctx, key, obj, out, ttl, dryrun.IsDryRun(options.DryRun)); err != nil {
err = storeerr.InterpretCreateError(err, qualifiedResource, name)
err = rest.CheckGeneratedNameError(ctx, e.CreateStrategy, err, obj)
if !apierrors.IsAlreadyExists(err) {
return nil, err
}
if errGet := e.Storage.Get(ctx, key, storage.GetOptions{}, out); errGet != nil {
return nil, err
}
accessor, errGetAcc := meta.Accessor(out)
if errGetAcc != nil {
return nil, err
}
if accessor.GetDeletionTimestamp() != nil {
msg := &err.(*apierrors.StatusError).ErrStatus.Message
*msg = fmt.Sprintf("object is being deleted: %s", *msg)
}
return nil, err
}
// The operation has succeeded. Call the finish function if there is one,
// and then make sure the defer doesn't call it again.
fn := finishCreate
finishCreate = finishNothing
fn(ctx, true)
if e.AfterCreate != nil {
e.AfterCreate(out, options)
}
if e.Decorator != nil {
e.Decorator(out)
}
return out, nil
}
📝 总结与展望
kube-apiserver 接收http请求后，先后对请求进行认证（Authentication）、鉴权（Authorization）以及准入控制（Admission Control）后，调用CreateResource() 创建deployment资源对象，然后调用通用Store处理Create（）将body反序列化成资源创建对象，最终序列化后存储到etcd，整个流程完成。
📚 参考资料
- https://juejin.cn/post/7359138355180847131
- https://kubernetes.io/zh-cn/docs/reference/command-line-tools-reference/kube-apiserver/
更多推荐
12
0- 0
已为社区贡献15条内容
所有评论(0)