【技术】从POD创建看Kubernetes源码实现（一）- kubectl
·
【技术】从POD创建看Kubernetes源码实现（一）- kubectl
✍️ 作者：茶水间Tech
🏷️ 标签：#云计算#云原生#kubernetes#容器
📖 前言
kubernetes的模块比较多，架构复杂，代码量更是庞大，看代码比较麻烦，我们从现实场景出发，从创建POD分析在Kubernetes内部的代码流程，本系列文章从POD创建，整体梳理Kubernetes源码实现，其中本节主要分析kubectl 侧的流程实现。
本文基于 Client Version: v1.34.3 ， Server Version: v1.34.2
📌 POD创建的整体架构图：
💻 正文
📑 一、关于kubectl
kubelet 是Kubernetes CLI 版本的瑞士军刀，是用于管理Kubernetes集群资源的主要工具，
官方地址：https://github.com/kubernetes/kubectl
1.1 kubectl 常规用法
在了解创建POD的流程前，先得了解一下kubectl 的常规用法，便于下文代码理解。
a. 普通内置命令
#kubectl get pods
b. 插件命令
#kubectl foo bar
c. 子命令插件命令
#kubectl create customresource
d. 带标志的内置命令
#kubectl create -f file.yaml
1.2 通过kubectl 创建POD
#kubectl create -f nginx_pod.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: my-nginx
namespace: default
spec:
replicas: 3
selector:
matchLabels:
app: nginxs
template:
metadata:
labels:
app: nginx
spec:
containers:
- name: nginx
image: nginx:latest
ports:
- containerPort: 80
📑 二、代码分析
代码逻辑图
kubectl
|
|-- 1. CLI 初始化 & 命令解析
|-- 2. 读取并解析 YAML
|-- 3. 构建 Kubernetes 对象（runtime.Object）
|-- 4. REST 请求（POST）
详细流程如下：
2.1 程序入口：main (kubectl.go)
kubectl 是kubernetes的一部分，被整合到kubernetes大项目中，作为staging中的一个库存在
代码路径：kubernetes/cmd/kubectl/kubectl.go
func main() {
// We need to manually parse the arguments looking for verbosity flag and
// set appropriate level here, because in the normal flow the flag parsing,
// including the logging verbosity, happens inside cli.RunNoErrOutput.
// Doing it here ensures we can continue using klog during kubectl command
// construction, which includes handling plugins and parsing .kuberc file,
// for example.
logs.GlogSetter(cmd.GetLogVerbosity(os.Args)) // nolint:errcheck
command := cmd.NewDefaultKubectlCommand()
if err := cli.RunNoErrOutput(command); err != nil {
// Pretty-print the error and exit with an error.
util.CheckErr(err)
}
}
2.2 命令注册：NewDefaultKubectlCommand （cmd.go）
kubectl main() 入口函数调用了 NewDefaultKubectlCommand() => NewDefaultKubectlCommandWithArgs 做cobra的命令初始化。
- 预先注册：所有内置命令在
NewKubectlCommand()中注册到 cobra 命令树 - 查找判断：通过
cmd.Find()在命令树中查找，能找到就是内置命令 - 错误处理：找不到的命令才会尝试作为插件处理
代码路径：kubernetes/staging/src/k8s.io/kubectl/pkg/cmd/cmd.go
func NewDefaultKubectlCommand() *cobra.Command {
ioStreams := genericiooptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr}
return NewDefaultKubectlCommandWithArgs(KubectlOptions{
PluginHandler: NewDefaultPluginHandler(plugin.ValidPluginFilenamePrefixes),
Arguments: os.Args,
ConfigFlags: defaultConfigFlags().WithWarningPrinter(ioStreams),
IOStreams: ioStreams,
})
}
// NewDefaultKubectlCommandWithArgs creates the `kubectl` command with arguments
func NewDefaultKubectlCommandWithArgs(o KubectlOptions) *cobra.Command {
//初始化KubectlCommand
cmd := NewKubectlCommand(o)
if o.PluginHandler == nil {
return cmd
}
if len(o.Arguments) > 1 {
// 这里为传入的参数，即 create -f nginx_pod.yaml 部分
cmdPathPieces := o.Arguments[1:]
// only look for suitable extension executables if
// the specified command does not already exist
// 调用cobra的Find去匹配args
if foundCmd, foundArgs, err := cmd.Find(cmdPathPieces); err != nil {
// Also check the commands that will be added by Cobra.
// These commands are only added once rootCmd.Execute() is called, so we
// need to check them explicitly here.
var cmdName string // first "non-flag" arguments
for _, arg := range cmdPathPieces {
if !strings.HasPrefix(arg, "-") {
cmdName = arg
break
}
}
switch cmdName {
case "help", cobra.ShellCompRequestCmd, cobra.ShellCompNoDescRequestCmd:
// Don't search for a plugin
default:
if err := HandlePluginCommand(o.PluginHandler, cmdPathPieces, 1); err != nil {
fmt.Fprintf(o.IOStreams.ErrOut, "Error: %v\n", err)
os.Exit(1)
}
}
} else if err == nil {
// Command exists(e.g. kubectl create), but it is not certain that
// subcommand also exists (e.g. kubectl create networkpolicy)
// we also have to eliminate kubectl create -f
if IsSubcommandPluginAllowed(foundCmd.Name()) && len(foundArgs) >= 1 && !strings.HasPrefix(foundArgs[0], "-") {
subcommand := foundArgs[0]
builtinSubcmdExist := false
for _, subcmd := range foundCmd.Commands() {
if subcmd.Name() == subcommand {
builtinSubcmdExist = true
break
}
}
if !builtinSubcmdExist {
if err := HandlePluginCommand(o.PluginHandler, cmdPathPieces, len(cmdPathPieces)-len(foundArgs)+1); err != nil {
fmt.Fprintf(o.IOStreams.ErrOut, "Error: %v\n", err)
os.Exit(1)
}
}
}
}
}
return cmd
}
**NewDefaultKubectlCommandWithArgs()**中将所有内置命令 在 NewKubectlCommand() 中注册到cobra的命令树中，
并实例化 Factory 对象f ，用于创建与 Kubernetes 交互所需的各种对象和客户端，并通过上下文参数一直传下去。
// kubernetes/staging/src/k8s.io/kubectl/pkg/cmd/cmd.go
func NewKubectlCommand(o KubectlOptions) *cobra.Command {
// ...(略)
// 创建主命令
cmds := &cobra.Command{
Use: "kubectl",
Short: i18n.T("kubectl controls the Kubernetes cluster manager"),
Long: templates.LongDesc(`
kubectl controls the Kubernetes cluster manager.
Find more information at:
https://kubernetes.io/docs/reference/kubectl/`),
Run: runHelp,
// Hook before and after Run initialize and write profiles to disk,
// respectively.
// 初始化后，在运行指令前的钩子
PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
rest.SetDefaultWarningHandler(warningHandler)
if cmd.Name() == cobra.ShellCompRequestCmd {
// This is the __complete or __completeNoDesc command which
// indicates shell completion has been requested.
plugin.SetupPluginCompletion(cmd, args)
}
return initProfiling()
},
// 运行指令后的钩子
PersistentPostRunE: func(*cobra.Command, []string) error {
if err := flushProfiling(); err != nil {
return err
}
if warningsAsErrors {
count := warningHandler.WarningCount()
switch count {
case 0:
// no warnings
case 1:
return fmt.Errorf("%d warning received", count)
default:
return fmt.Errorf("%d warnings received", count)
}
}
return nil
},
}
// ...(略)
// 实例化Factory接口，工厂模式
f := cmdutil.NewFactory(matchVersionKubeConfigFlags)
// kubectl定义了7类命令，结合Message和各个子命令的package名来看
groups := templates.CommandGroups{
{
// 1. 初级命令，包括 create/expose/run/set
Message: "Basic Commands (Beginner):",
Commands: []*cobra.Command{
create.NewCmdCreate(f, o.IOStreams),
expose.NewCmdExposeService(f, o.IOStreams),
run.NewCmdRun(f, o.IOStreams),
set.NewCmdSet(f, o.IOStreams),
},
},
{
// 2. 中级命令，包括explain/get/edit/delete
Message: "Basic Commands (Intermediate):",
Commands: []*cobra.Command{
explain.NewCmdExplain("kubectl", f, o.IOStreams),
getCmd,
edit.NewCmdEdit(f, o.IOStreams),
delete.NewCmdDelete(f, o.IOStreams),
},
},
{
// 3. 部署命令，包括 rollout/scale/autoscale
Message: "Deploy Commands:",
Commands: []*cobra.Command{
rollout.NewCmdRollout(f, o.IOStreams),
scale.NewCmdScale(f, o.IOStreams),
autoscale.NewCmdAutoscale(f, o.IOStreams),
},
},
{
// 4. 集群管理命令，包括 cerfificate/cluster-info/top/cordon/drain/taint
Message: "Cluster Management Commands:",
Commands: []*cobra.Command{
certificates.NewCmdCertificate(f, o.IOStreams),
clusterinfo.NewCmdClusterInfo(f, o.IOStreams),
top.NewCmdTop(f, o.IOStreams),
drain.NewCmdCordon(f, o.IOStreams),
drain.NewCmdUncordon(f, o.IOStreams),
drain.NewCmdDrain(f, o.IOStreams),
taint.NewCmdTaint(f, o.IOStreams),
},
},
{
// 5. 故障排查和调试，包括 describe/logs/attach/exec/port-forward/proxy/cp/auth
Message: "Troubleshooting and Debugging Commands:",
Commands: []*cobra.Command{
describe.NewCmdDescribe("kubectl", f, o.IOStreams),
logs.NewCmdLogs(f, o.IOStreams),
attach.NewCmdAttach(f, o.IOStreams),
cmdexec.NewCmdExec(f, o.IOStreams),
portforward.NewCmdPortForward(f, o.IOStreams),
proxyCmd,
cp.NewCmdCp(f, o.IOStreams),
auth.NewCmdAuth(f, o.IOStreams),
debugCmd,
events.NewCmdEvents(f, o.IOStreams),
},
},
{
// 6. 高级命令，包括diff/apply/patch/replace/wait/convert/kustomize
Message: "Advanced Commands:",
Commands: []*cobra.Command{
diff.NewCmdDiff(f, o.IOStreams),
apply.NewCmdApply("kubectl", f, o.IOStreams),
patch.NewCmdPatch(f, o.IOStreams),
replace.NewCmdReplace(f, o.IOStreams),
wait.NewCmdWait(f, o.IOStreams),
kustomize.NewCmdKustomize(o.IOStreams),
},
},
{
// 7. 设置命令，包括label，annotate，completion
Message: "Settings Commands:",
Commands: []*cobra.Command{
label.NewCmdLabel(f, o.IOStreams),
annotate.NewCmdAnnotate("kubectl", f, o.IOStreams),
completion.NewCmdCompletion(o.IOStreams.Out, ""),
},
},
}
groups.Add(cmds)
// ...(略)
// 添加其余子命令，包括 alpha/config/plugin/version/api-versions/api-resources/options
cmds.AddCommand(alpha)
cmds.AddCommand(cmdconfig.NewCmdConfig(f, clientcmd.NewDefaultPathOptions(), o.IOStreams))
cmds.AddCommand(plugin.NewCmdPlugin(o.IOStreams))
cmds.AddCommand(version.NewCmdVersion(f, o.IOStreams))
cmds.AddCommand(apiresources.NewCmdAPIVersions(f, o.IOStreams))
cmds.AddCommand(apiresources.NewCmdAPIResources(f, o.IOStreams))
cmds.AddCommand(options.NewCmdOptions(o.IOStreams.Out))
// ...(略)
return cmds
}
关于Factory
type factoryImpl struct {
clientGetter genericclioptions.RESTClientGetter
// Caches OpenAPI document and parsed resources
openAPIParser *openapi.CachedOpenAPIParser
oapi *openapi.CachedOpenAPIGetter
parser sync.Once
getter sync.Once
}
func NewFactory(clientGetter genericclioptions.RESTClientGetter) Factory {
if clientGetter == nil {
panic("attempt to instantiate client_access_factory with nil clientGetter")
}
f := &factoryImpl{
clientGetter: clientGetter,
}
return f
}
func (f *factoryImpl) ToRESTConfig() (*restclient.Config, error) {
return f.clientGetter.ToRESTConfig()
}
func (f *factoryImpl) ToRESTMapper() (meta.RESTMapper, error) {
return f.clientGetter.ToRESTMapper()
}
func (f *factoryImpl) ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error) {
return f.clientGetter.ToDiscoveryClient()
}
2.3 命令执行 : NewCmdCreate (create.go)
因为是内置标志命令，转到了 NewCmdCreate()，通过填充默认值，并校验参数合法性，最终执行创建操作。
代码路径：kubernetes/staging/src/k8s.io/kubectl/pkg/cmd/create/create.go
NewCmdCreate调用链
NewCmdCreate()
└── RunCreate()
└── o.RunCreate()
└── resource.Builder.Do()
└── Result.Visit()
└── FileVisitor.Visit()
└── StreamVisitor.Visit()
└── VisitorFunc
└── resource.Create()
func NewCmdCreate(f cmdutil.Factory, ioStreams genericiooptions.IOStreams) *cobra.Command {
// create子命令的相关选项
o := NewCreateOptions(ioStreams)
cmd := &cobra.Command{
Use: "create -f FILENAME",
DisableFlagsInUseLine: true,
Short: i18n.T("Create a resource from a file or from stdin"),
Long: createLong,
Example: createExample,
Run: func(cmd *cobra.Command, args []string) {
// 验证参数并运行
cmdutil.CheckErr(o.Complete(f, cmd, args))
cmdutil.CheckErr(o.Validate())
cmdutil.CheckErr(o.RunCreate(f, cmd))
},
}
// ...(略)
return cmd
}
2.4 核心逻辑 : NewCmdCreate (create.go)
o.RunCreate(f,cmd) 是kubectl create命令的核心执行逻辑，函数中先绑定kubeconfig，通过 f.NewBuilder() 构造了builder流水线链，其中链中FilenameParam 中遍历生成了File Visitor，builder的Do()函数中，生成了带有visitor的result object, 最终调用builder的Visit函数中 创建resource的Helper助手，进行Create 请求apiserver
代码路径：kubernetes/staging/src/k8s.io/kubectl/pkg/cmd/create/create.go
Visit 函数的处理流程：
- 添加注解：记录 kubectl 的最后应用配置
- 记录命令：用于审计和回滚
- 创建资源：调用 API Server 创建
- 输出结果：按指定格式输出
func (o *CreateOptions) RunCreate(f cmdutil.Factory, cmd *cobra.Command) error {
// ...(略)
if len(o.Raw) > 0 { //kubectl create --raw /api/v1/xxxx 场景
restClient, err := f.RESTClient()
if err != nil {
return err
}
return rawhttp.RawPost(restClient, o.IOStreams, o.Raw, o.FilenameOptions.Filenames[0])
}
// ...(略)
schema, err := f.Validator(o.ValidationDirective)
if err != nil {
return err
}
cmdNamespace, enforceNamespace, err := f.ToRawKubeConfigLoader().Namespace() //绑定kubeconfig
if err != nil {
return err
}
//构造builder，生成带有具体 visitor的result
r := f.NewBuilder().
Unstructured(). // 使用非结构化数据（支持任何资源类型）
Schema(schema). //设置验证模式
ContinueOnError(). // 遇到错误继续处理其他资源
NamespaceParam(cmdNamespace).DefaultNamespace(). //设置命名空间
FilenameParam(enforceNamespace, &o.FilenameOptions). //处理文件输入生成visitor
LabelSelectorParam(o.Selector). //应用标签选择器
Flatten(). //展平嵌套的资源列表
Do()
err = r.Err()
if err != nil {
return err
}
count := 0
// 调用visit函数，创建资源
err = r.Visit(func(info *resource.Info, err error) error {
if err != nil {
return err
}
// 1. 添加或更新注解
if err := util.CreateOrUpdateAnnotation(cmdutil.GetFlagBool(cmd, cmdutil.ApplyAnnotationsFlag), info.Object, scheme.DefaultJSONEncoder()); err != nil {
return cmdutil.AddSourceToErr("creating", info.Source, err)
}
// 2. 记录命令历史
if err := o.Recorder.Record(info.Object); err != nil {
klog.V(4).Infof("error recording current command: %v", err)
}
// 3. 执行创建操作
if o.DryRunStrategy != cmdutil.DryRunClient { //不是客户端模拟，那就得发送到apiserver
obj, err := resource.
NewHelper(info.Client, info.Mapping).
DryRun(o.DryRunStrategy == cmdutil.DryRunServer).
WithFieldManager(o.fieldManager).
WithFieldValidation(o.ValidationDirective).
Create(info.Namespace, true, info.Object) //调用apiserver
if err != nil {
return cmdutil.AddSourceToErr("creating", info.Source, err)
}
info.Refresh(obj, true)
}
count++
// 4. 输出结果
return o.PrintObj(info.Object)
})
if err != nil {
return err
}
if count == 0 {
return fmt.Errorf("no objects passed to create")
}
return nil
}
其中，builder 生成的visitor 在FileVisitor的Visit中读取文件并做编码转换，最终转给了StreamVisitor的Visit
// kubernetes/staging/src/k8s.io/cli-runtime/pkg/resource/visitor.go
func (v *FileVisitor) Visit(fn VisitorFunc) error {
var f *os.File
if v.Path == constSTDINstr {
f = os.Stdin
} else {
var err error
f, err = os.Open(v.Path)
if err != nil {
return err
}
defer f.Close()
}
// TODO: Consider adding a flag to force to UTF16, apparently some
// Windows tools don't write the BOM
utf16bom := unicode.BOMOverride(unicode.UTF8.NewDecoder())
v.StreamVisitor.Reader = transform.NewReader(f, utf16bom)
return v.StreamVisitor.Visit(fn)
}
StreamVisitor的Visit中解析yaml/json 最终调用VisitorFunc 处理
// kubernetes/staging/src/k8s.io/cli-runtime/pkg/resource/visitor.go
func (v *StreamVisitor) Visit(fn VisitorFunc) error {
d := yaml.NewYAMLOrJSONDecoder(v.Reader, 4096)
for {
ext := runtime.RawExtension{}
if err := d.Decode(&ext); err != nil {
if err == io.EOF {
return nil
}
return fmt.Errorf("error parsing %s: %v", v.Source, err)
}
// TODO: This needs to be able to handle object in other encodings and schemas.
ext.Raw = bytes.TrimSpace(ext.Raw)
if len(ext.Raw) == 0 || bytes.Equal(ext.Raw, []byte("null")) {
continue
}
if err := ValidateSchema(ext.Raw, v.Schema); err != nil {
return fmt.Errorf("error validating %q: %v", v.Source, err)
}
info, err := v.infoForData(ext.Raw, v.Source) //反序列化为Info object
if err != nil {
if fnErr := fn(info, err); fnErr != nil {
return fnErr
}
continue
}
if err := fn(info, nil); err != nil { //调用VisitorFunc 处理资源
return err
}
}
}
2.5 请求Api Server：Create (helper.go)
Helper.Create() 通过调用restclient 封装HTTP 进行 POST请求 apiserver
请求地址：https://:/apis/apps/v1/namespaces//deployments
代码路径：kubernetes/staging/src/k8s.io/cli-runtime/pkg/resource/helper.go
func (m *Helper) Create(namespace string, modify bool, obj runtime.Object) (runtime.Object, error) {
return m.CreateWithOptions(namespace, modify, obj, nil)
}
func (m *Helper) CreateWithOptions(namespace string, modify bool, obj runtime.Object, options *metav1.CreateOptions) (runtime.Object, error) {
// ...(略)
return m.createResource(m.RESTClient, m.Resource, namespace, obj, options)
}
func (m *Helper) createResource(c RESTClient, resource, namespace string, obj runtime.Object, options *metav1.CreateOptions) (runtime.Object, error) {
return c.Post().
NamespaceIfScoped(namespace, m.NamespaceScoped).
Resource(resource).
VersionedParams(options, metav1.ParameterCodec).
Body(obj).
Do(context.TODO()).
Get()
}
📝 总结与展望
概括kubectl create -f *.yaml 的过程中，逻辑比较简单，通过cobra解析kubctl 传参，构造kubectl 完整命令树，进而解析YAML/JSON文档，最终执行VisitorFunc 封装了HTTP POST请求，最终请求转给了Kube-apiserver。
📚 参考资料
https://blog.huweihuang.com/kubernetes-notes/principle/flow/pod-flow/
https://www.cnblogs.com/liweiboy/p/16100586.html
更多推荐
17
0- 0
已为社区贡献15条内容
所有评论(0)