从源码上看k8s创建pod全流程(上) 简介 kubectl

这是我参与11月更文挑战的第6天，活动详情查看：2021最后一次更文挑战

简介

在K8s中最重要的基础资源类型是pod，其他高级的使用模式都是基于pod来运作的，如果想对K8s有更深入清晰的认识，我们有必要来看看pod到底是怎么被创建出来的。创建pod最简单的方式是通过kubectl run命令，根据官方给出的说明，”Create and run a particular image in a pod”，意思是在一个pod中创建并运行一个指定的镜像。

Create and run a particular image in a pod.

Examples:

Start a nginx pod.

kubectl run nginx –image=nginx

接下来我们将从开始敲回车之后，到我们可以查看到pod创建并启动成功，这背后到底发生了什么。从宏观层面看，整个故事发生在三个场地，一个是本地，主角当然是kubectl，一个则是远端的K8s集群主节点，主角是kube-apiserver、etcd以及调度器，另一个是K8s集群的工作节点，主角是kubelet。

首先我们先看一下整体大的流程，在每个环节都做了些什么，然后我们再逐个查看各个环节的代码实现。由于K8s核心代码已经多到两百多万行，而且还使用了很多高级的设计模式，本文将尽量对这份优秀的代码进行解读，一家之言可能有很多谬误，不妥之处，请不吝指出，我们可以一起来讨论学习。

以下代码均是基于K8s正式发布版V1.20.2进行分析。

整体流程图

从整体具体的上看，kubectl负责接收用户的输入，做初步的处理后，按照kube-apiserver的处理要求生成具体的请求；kube-apiserver是所有请求的入口，它负责所有请求的通用检查和分发以及对外提供资源状态的查询等等；kubelet负责具体pod的生命周期管理和节点整体状态数据的上报。

Kubernetes是基于事件+控制器模式实现的，因此在代码中并没有一个贯穿pod创建控制流程始终的代码存在。涉及到的组件更像是团队接力赛选手，大家在完成自己的工作后，就把棒交出去了(更新信息/触发某种事件)，下一个选手根据自己感兴趣的事件去选择要做的事情(例如watch监听)，并努力把自己分内的事做好，然后再交出去(更新信息/触发某种事件)。

kubectl run命令的历史简介

在早期的版本中，run子命令不仅可以创建pod，还可以创建job(含cronjob)、deployment以及replication controller

在v1.20.2版本中kubectl run命令已经大大被简化，generator选项被移除，只能创建最基础的pod。在一些细节上也做了改进，例如dry-run选项增加了区分client还是server，控制更加细腻。

kubectl

客户端参数检查验证和对象生成

代码入口：

vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L246

参数检查与验证

主要包含镜像名称校验，镜像拉取策略校验等

go复制代码// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L276 
imageName := o.Image 
if imageName == "" { 
  return fmt.Errorf("--image is required") 
} 
validImageRef := reference.ReferenceRegexp.MatchString(imageName)
if !validImageRef { 
  return fmt.Errorf("Invalid image name %q: %v", imageName, reference.ErrReferenceInvalidFormat)   
}

// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L310 
if err := verifyImagePullPolicy(cmd); err != nil {
  return err
}

对象生成

获取pod默认生成器

go复制代码// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L314 
generators := generateversioned.GeneratorFn("run") 
// 加载run下说有的生成器，目前只剩下一个pod的生成器，历史版本上还有job、deployment等等，参考kubectl run命令历史小节 
generator, found := generators[generateversioned.RunPodV1GeneratorName] // "run-pod/v1" 
if !found { 
  return cmdutil.UsageErrorf(cmd, "generator %q not found", o.Generator)   
} 
// vendor/k8s.io/kubectl/pkg/generate/versioned/generator.go#L94 
case "run": // run子命令下注册的默认生成器 
	generator = map[string]generate.Generator{ RunPodV1GeneratorName: BasicPod{},

生成运行时对象

go复制代码// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L330 
var createdObjects = []*RunObject{} 
runObject, err := o.createGeneratedObject(f, cmd, generator, names, params, cmdutil.GetFlagString(cmd, "overrides")) // 这里开始发起创建运行时对象 
if err != nil { 
  return err 
} 
createdObjects = append(createdObjects, runObject) 
// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L616 
func (o *RunOptions) createGeneratedObject(f cmdutil.Factory, cmd *cobra.Command, generator generate.Generator, names []generate.GeneratorParam, params map[string]interface{}, overrides string) (*RunObject, error) { 
  // 验证生成器参数  
  err := generate.ValidateParams(names, params)   
  // 生成器生成对象  obj, err := generator.Generate(params)  
  // API分组和版本协调  
  mapper, err := f.ToRESTMapper() 
  // run has compiled knowledge of the thing is creating 
  gvks, _, err := scheme.Scheme.ObjectKinds(obj) 
  mapping, err := mapper.RESTMapping(gvks[0].GroupKind(), gvks[0].Version)  
  if o.DryRunStrategy != cmdutil.DryRunClient {  
    // 客户端实例构建 
    client, err := f.ClientForMapping(mapping) 
    // 具体实例取决于f是怎么实例化的 // 发送HTTP请求   
    actualObj, err = resource. NewHelper(client, mapping). DryRun(o.DryRunStrategy == cmdutil.DryRunServer). // 动态配置server side dry run 
    WithFieldManager(o.fieldManager). // 更新管理者   
    Create(o.Namespace, false, obj)  
  } 
}

关于API groups和version发现与协商

Kubernetes使用的API是带版本号并且被分成了API groups。一个API group是指一组操作资源类似的API集合。Kubernetes一般支持多版本的API groups，kubectl为了找到最合适的API，需要只通过发现机制来获取kube-api暴露的schema文档(通过OpenAPI格式）。为了提高性能，一般kubectl会在本地~/.kube/cache/discovery目录缓存这些schema文件。

处理返回结果

得到api-server返回值后，进行后续处理。按照正确的格式输出创建的对象。

go复制代码// vendor/k8s.io/kubectl/pkg/cmd/run/run.go#L430 
if runObject != nil { 
	if err := o.PrintObj(runObject.Object); err != nil { 
		return err 
	}
}

客户端认证支持

为了确保请求发送成功，kubectl需要具备认证的能力。用户凭证几乎总是存储在本地磁盘的kubeconfig文件中。为了定位该文件，kubectl会按照以下步骤加载该文件

1. 如果–kubeconfig指定了文件，则使用这个文件
2. 如果$KUBECONFIG环境变量定义了，则使用该环境变量指向的文件
3. 在本地home目录下，例如~/.kube，搜索并使用第一个找到的文件

在文件解析完成后，kubectl就可以确定当前使用的上下文，指向的集群以及当前用户关联的认证信息。

kube-apiserver

认证

kube-apiserver是客户端与系统组件之间主要的持久化和查询集群状态界面。首要的kube-apiserver需要知道请求的发起者是谁。

apiserver如何对请求做认证？当服务第一次启动时，它会查看用户提供的所有命令行参数，然后组装成一个合适的认证器列表。每个请求到来后都要逐个通过认证器的检查，直到有一个认证通过。

go复制代码// vendor/k8s.io/apiserver/pkg/authentication/request/union/union.go#L53 
// AuthenticateRequest authenticates the request using a chain of authenticator.Request objects. 
func (authHandler *unionAuthRequestHandler) AuthenticateRequest(req *http.Request) (*authenticator.Response, bool, error) { 
  var errlist []error 
  for _, currAuthRequestHandler := range authHandler.Handlers { 
    resp, ok, err := currAuthRequestHandler.AuthenticateRequest(req) 
    if err != nil { 
      if authHandler.FailOnError {
        return resp, ok, err
      }
      errlist = append(errlist, err) continue
    }
    if ok {
      return resp, ok, err
    }
  }
  return nil, false, utilerrors.NewAggregate(errlist)
}

认证器的初始化流程

go复制代码// pkg/kubeapiserver/authenticator/config.go#L95
// New returns an authenticator.Request or an error that supports the standard
// Kubernetes authentication mechanisms.
func (config Config) New() (authenticator.Request, *spec.SecurityDefinitions, error) {
  var authenticators []authenticator.Request
  authenticator := union.New(authenticators...)
  // 各种初始化
  authenticator = group.NewAuthenticatedGroupAdder(authenticator)
  if config.Anonymous {
    // If the authenticator chain returns an error, return an error (don't consider a bad bearer token
    // or invalid username/password combination anonymous).
    authenticator = union.NewFailOnError(authenticator, anonymous.NewAuthenticator())
  }
  return authenticator, &securityDefinitions, nil
}

如下图所示，假设所有的认证器都被启用，当客户端发送请求到kube-apiserver服务，该请求会进入Authentication Handler函数（处理认证相关的Handler函数），在Authentication Handler函数中，会遍历已启用的认证器列表，尝试执行每个认证器，当有一个认证器返回true时，则认证成功，否则继续尝试下一个认证器。

当所有认证器都认证失败后，请求将会被拒绝，合并后的错误会返回给客户端。如果认证成功，Authorization头信息将会从请求中移除，用户信息会添加到请求的上下文信息中。这样后续步骤就可以访问到认证阶段确定的请求用户的信息了。

鉴权

虽然现在kube-apiserver已经成功地验证了请求者的身份信息，但是在进行下一步之前还得确保请求者是否有权限去操作。身份认证和鉴权不是同一个事情，要想进一步使用，kube-apiserver需要对我们进行鉴权。

类似认证器的处理方法，kube-apiserver需要基于用户提供的命令行参数，来组装一个合适的鉴权器列表来处理每一个请求。当所有的鉴权器都拒绝该请求时，请求会终止，并且请求方会得到Forbidden的答复。如果任何一个鉴权器批准了请求，那么请求鉴权成功，将会进入下一阶段处理。

鉴权器初始化

kube-apiserver目前提供了6种授权机制，分别是AlwaysAllow、AlwaysDeny、ABAC、Webhook、RBAC、Node，可通过指定–authorization-mode参数设置授权机制，至少需要指定一个。

go复制代码// pkg/kubeapiserver/authorizer/config.go#L71
// New returns the right sort of union of multiple authorizer.Authorizer objects // based on the authorizationMode or an error.
func (config Config) New() (authorizer.Authorizer, authorizer.RuleResolver, error) {
  if len(config.AuthorizationModes) == 0 {
    return nil, nil, fmt.Errorf("at least one authorization mode must be passed")
  }
}

鉴权器决策状态

go复制代码// vendor/k8s.io/apiserver/pkg/authorization/authorizer/interfaces.go#L149
type Decision int const
( 
  // DecisionDeny means that an authorizer decided to deny the action.
  DecisionDeny
  Decision = iota
  // DecisionAllow means that an authorizer decided to allow the action.
  DecisionAllow
  // DecisionNoOpionion means that an authorizer has no opinion on whether
  // to allow or deny an action.
  DecisionNoOpinion
)

当决策状态是DecisionDeny或DecisionNoOpinion时会交由下一个鉴权器继续处理，如果没有下一个鉴权器则鉴权失败。当决策状态是DecisionAllow时鉴权成功，请求被接受。

Admission control

在认证和授权之后，对象被持久化之前，拦截kube-apiserver的请求，对请求的资源对象进行自定义操作（校验、修改或者拒绝请求）。为什么需要有这一个环节？为了集群的稳定性，在资源对象被正式接纳前，需要由系统内其他组件对待创建的资源先进行一系列的检查，确保符合整个集群的预期和规则，从而防患于未然。这是在etcd创建资源前的最后一道保障。

插件实现接口

go复制代码// vendor/k8s.io/apiserver/pkg/admission/interfaces.go#L123
// Interface is an abstract, pluggable interface for Admission Control decisions.
type Interface interface {
	// Handles returns true if this admission controller can handle the given operation
	// where operation can be one of CREATE, UPDATE, DELETE, or CONNECT
	Handles(operation Operation) bool
}

type MutationInterface interface {
	Interface

	// Admit makes an admission decision based on the request attributes.
	// Context is used only for timeout/deadline/cancellation and tracing information.
	Admit(ctx context.Context, a Attributes, o ObjectInterfaces) (err error)
}

// ValidationInterface is an abstract, pluggable interface for Admission Control decisions.
type ValidationInterface interface {
	Interface

	// Validate makes an admission decision based on the request attributes.  It is NOT allowed to mutate
	// Context is used only for timeout/deadline/cancellation and tracing information.
	Validate(ctx context.Context, a Attributes, o ObjectInterfaces) (err error)
}

两种准入控制器

1. 变更准入控制器（Mutating Admission Controller）用于变更信息，能够修改用户提交的资源对象信息。
2. 验证准入控制器（Validating Admission Controller）用于身份验证，能够验证用户提交的资源对象信息。

变更准入控制器运行在验证准入控制器之前。

准入控制器的运行方式与认证和鉴权方式类似，不同的地方是任何一个准入控制器失败后，整个准入控制流程就会结束，请求失败。

准入控制器以插件的形式运行在kube-apiserver进程中，插件化的好处在于可扩展插件并单独启用/禁用指定插件，也可以将每个准入控制器称为准入控制器插件。

客户端发起一个请求，在请求经过准入控制器列表时，只要有一个准入控制器拒绝了该请求，则整个请求被拒绝（HTTP 403Forbidden）并返回一个错误给客户端。

ETCD存储

经过身份认证、鉴权以及准入控制检查后，kube-apiserver将反序列化HTTP请求（解码），构造运行时对象（runtime object），并将它持久化到etcd。

横向扩展

kube-apiserver如何知道某一个资源的操作该如何处理呢？这在服务刚启动的时候会有非常复杂的配置步骤，让我们粗略看一下：

1. 当kube-apiserver启动时，会创建一个服务链(server chain)，允许apiserver进行聚合，这是提供多个apiserver的基础方式

go复制代码// cmd/kube-apiserver/app/server.go#L184
server, err := CreateServerChain(completeOptions, stopCh)

1. 作为默认实现的通用apiserver会被创建

go复制代码// cmd/kube-apiserver/app/server.go#L215
apiExtensionsServer, err := createAPIExtensionsServer(apiExtensionsConfig, genericapiserver.NewEmptyDelegate())

1. 生成的OpenAPI信息(schema)会填充到apiserver的配置中

go复制代码// cmd/kube-apiserver/app/server.go#L477
	genericConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(generatedopenapi.GetOpenAPIDefinitions, openapinamer.NewDefinitionNamer(legacyscheme.Scheme, extensionsapiserver.Scheme, aggregatorscheme.Scheme))

1. kube-apiserver为每个API组配置一个存储服务提供器，它就是kube-apiserver访问和修改资源状态时的代理。

go复制代码// pkg/controlplane/instance.go#L591
for _, restStorageBuilder := range restStorageProviders {
		groupName := restStorageBuilder.GroupName()
		if !apiResourceConfigSource.AnyVersionForGroupEnabled(groupName) {
			klog.V(1).Infof("Skipping disabled API group %q.", groupName)
			continue
		}
		apiGroupInfo, enabled, err := restStorageBuilder.NewRESTStorage(apiResourceConfigSource, restOptionsGetter)
		if err != nil {
			return fmt.Errorf("problem initializing API group %q : %v", groupName, err)
		}
		if !enabled {
			klog.Warningf("API group %q is not enabled, skipping.", groupName)
			continue
        }
}

1. 为每一个不同版本的API组添加REST路由映射信息。这会运行kube-apiserver将请求映射到所匹配到的正确代理。

go复制代码// vendor/k8s.io/apiserver/pkg/server/genericapiserver.go#L439
r, err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer)

1. 在我们这个特定场景下，POST处理器会被注册，它将代理资源的创建操作

go复制代码// vendor/k8s.io/apiserver/pkg/endpoints/installer.go#816
case "POST": // Create a resource.
			var handler restful.RouteFunction
			if isNamedCreater {
				handler = restfulCreateNamedResource(namedCreater, reqScope, admit)
			} else {
				handler = restfulCreateResource(creater, reqScope, admit)
      }

小结一下，至此kube-apiserver完成了路由到内部资源操作代理的映射配置，当请求匹配后，就可以触发指定的操作代理了。

pod存储流程

我们继续看pod创建的流程：

1. 基于注册的路由信息，当请求匹配到处理器链条中的某一个时，就会交由该处理器去处理。如果没有匹配的处理器，就返回给基于路径的处理器进行处理。但是如果没有注册路径处理器，则由notfound处理器返回404错误信息。
2. 幸运的是我们已经注册过createHandler了。它会做些什么呢？首先，它会解码HTTP请求体，并进行基础的验证，如提供的json是否符合相关版本API资源的要求
3. 进行审计和最终的准入检查
4. 通过存储代理将资源存储到etcd中。通常etcd的key是如下格式：/，它可以通过配置继续修改

go复制代码Create(ctx context.Context, name string, obj runtime.Object, createValidation ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error)

1. 检查是否有任何错误，没有错误时存储代理会通过get调用来确保资源对象确实被创建了。然后它会触发post-create handler和其他额外要求的装饰器。
2. 构造HTTP请求返回内容并发送

go复制代码// vendor/k8s.io/apiserver/pkg/endpoints/handlers/create.go#L49
func createHandler(r rest.NamedCreater, scope *RequestScope, admit admission.Interface, includeName bool) http.HandlerFunc {
		gv := scope.Kind.GroupVersion()
		s, err := negotiation.NegotiateInputSerializer(req, false, scope.Serializer)
		if err != nil {
			scope.err(err, w, req)
			return
		}

		// 这个decoder构建比较关键，后边对body的解析就是通过它来完成的
		decoder := scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion)
		

		// 读取包体：序列化后的runtime
		body, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
		if err != nil {
			scope.err(err, w, req)
			return
        }
		// 检查创建参数

		// 开始解码转换
		defaultGVK := scope.Kind
		original := r.New()
		trace.Step("About to convert to expected version")
		obj, gvk, err := decoder.Decode(body, &defaultGVK, original)
		// 省略
		trace.Step("Conversion done")

		// admission
		
		// 开始存储etcd
		trace.Step("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,
			)
		}
		// 省略
		trace.Step("Object stored in database")

		// 构造HTTP返回结果
		code := http.StatusCreated
		status, ok := result.(*metav1.Status)
		if ok && err == nil && status.Code == 0 {
			status.Code = int32(code)
		}

		transformResponseObject(ctx, scope, trace, req, w, code, outputMediaType, result)
  // 至此，整个创建pod的HTTP请求就会返回，同时会返回创建后的对象

总结

到现在为止，kube-apiserver完成了大量的工作，pod资源已经存储到etcd中了，但是它对外还不可见，还需要最后一步。篇幅原因，后续文章我们再继续讲解pod的调度和实际创建。

参考文档

书籍

《Kubernetes源码剖析》郑东旭

网文

Kubernetes 弃用 Docker 来龙去脉

kubelet创建pod工作流程

v1.14版的kubectl run创建pod流程

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有