kubectl 工具是 kubernetes API Server 的客戶端。它的主要工作是向 kubernetes API Server 發起http請求。k8s是完全以資源為中心的系統，而kubectl會發起HTTP請求來操縱資源（對資源進行CURD操作），來對集群系統進行維護。
下面主要分析kubectl create nginx_pod.yaml這條命令的創建過程:
apiVersion: v1
kind: Pod
metadata:
name: nginx-pod
spec:
containers:
- name: nginx
image: nginx:1.8
1、cobra命令行參數解析
k8s的客戶端都是使用的cobra來構建的。使用方法見我另一篇文章
2、kubectl - create的調用邏輯
main
func main() {
// 如果不調用rand.Seed，每次重新運行這個main函數，rand下的函數返回值始終一致
// Seed即隨機的種子，每次用時間戳作為種子，就能保證隨機性
rand.Seed(time.Now().UnixNano())
// 創建了kubectl命令的默認參數
command := cmd.NewDefaultKubectlCommand()
// TODO: once we switch everything over to Cobra commands, we can go back to calling
// cliflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
// normalize func and add the go flag set by hand.
pflag.CommandLine.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
pflag.CommandLine.AddGoFlagSet(goflag.CommandLine)
// cliflag.InitFlags()
// 日志的初始化與退出
logs.InitLogs()
defer logs.FlushLogs()
// 運行command
if err := command.Execute(); err != nil {
os.Exit(1)
}
}
match
func NewDefaultKubectlCommand() *cobra.Command {
return NewDefaultKubectlCommandWithArgs(NewDefaultPluginHandler(plugin.ValidPluginFilenamePrefixes), os.Args, os.Stdin, os.Stdout, os.Stderr)
}
func NewDefaultKubectlCommandWithArgs(pluginHandler PluginHandler, args []string, in io.Reader, out, errout io.Writer) *cobra.Command {
// 初始化NewKubectlCommand，采用標准輸入、輸出、錯誤輸出
cmd := NewKubectlCommand(in, out, errout)
if pluginHandler == nil {
return cmd
}
if len(args) > 1 {
// 這里為傳入的參數，即 create -f nginx_pod.yaml 部分
cmdPathPieces := args[1:]
// 調用cobra的Find去匹配args
if _, _, err := cmd.Find(cmdPathPieces); err != nil {
if err := HandlePluginCommand(pluginHandler, cmdPathPieces); err != nil {
fmt.Fprintf(errout, "%v\n", err)
os.Exit(1)
}
}
}
return cmd
}
command
代碼比較長，只摘錄關鍵的代碼
func NewKubectlCommand(in io.Reader, out, err io.Writer) *cobra.Command {
warningHandler := rest.NewWarningWriter(err, rest.WarningWriterOptions{Deduplicate: true, Color: term.AllowsColorOutput(err)})
warningsAsErrors := false
// 創建主命令
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/overview/`),
Run: runHelp,
// 初始化后，在運行指令前的鈎子
PersistentPreRunE: func(*cobra.Command, []string) error {
// 這里是做 pprof 性能分析，跳轉到對應代碼可以看到，我們可以用參數 --profile xxx 來采集性能指標，默認保存在當前目錄下的profile.pprof中
return initProfiling()
},
// 運行指令后的鈎子
PersistentPostRunE: func(*cobra.Command, []string) error {
// 保存pprof性能分析指標
return flushProfiling()
},
// bash自動補齊功能，可通過 kubectl completion bash 命令查看
// 具體安裝可參考 https://kubernetes.io/docs/tasks/tools/install-kubectl/#enabling-shell-autocompletion
BashCompletionFunction: bashCompletionFunc,
}
// 實例化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, ioStreams),
expose.NewCmdExposeService(f, ioStreams),
run.NewCmdRun(f, ioStreams),
set.NewCmdSet(f, ioStreams),
},
},
{
// 2. 中級命令，包括explain/get/edit/delete
Message: "Basic Commands (Intermediate):",
Commands: []*cobra.Command{
explain.NewCmdExplain("kubectl", f, ioStreams),
get.NewCmdGet("kubectl", f, ioStreams),
edit.NewCmdEdit(f, ioStreams),
delete.NewCmdDelete(f, ioStreams),
},
},
{
// 3. 部署命令，包括 rollout/scale/autoscale
Message: "Deploy Commands:",
Commands: []*cobra.Command{
rollout.NewCmdRollout(f, ioStreams),
scale.NewCmdScale(f, ioStreams),
autoscale.NewCmdAutoscale(f, ioStreams),
},
},
{
// 4. 集群管理命令，包括 cerfificate/cluster-info/top/cordon/drain/taint
Message: "Cluster Management Commands:",
Commands: []*cobra.Command{
certificates.NewCmdCertificate(f, ioStreams),
clusterinfo.NewCmdClusterInfo(f, ioStreams),
top.NewCmdTop(f, ioStreams),
drain.NewCmdCordon(f, ioStreams),
drain.NewCmdUncordon(f, ioStreams),
drain.NewCmdDrain(f, ioStreams),
taint.NewCmdTaint(f, ioStreams),
},
},
{
// 5. 故障排查和調試，包括 describe/logs/attach/exec/port-forward/proxy/cp/auth
Message: "Troubleshooting and Debugging Commands:",
Commands: []*cobra.Command{
describe.NewCmdDescribe("kubectl", f, ioStreams),
logs.NewCmdLogs(f, ioStreams),
attach.NewCmdAttach(f, ioStreams),
cmdexec.NewCmdExec(f, ioStreams),
portforward.NewCmdPortForward(f, ioStreams),
proxy.NewCmdProxy(f, ioStreams),
cp.NewCmdCp(f, ioStreams),
auth.NewCmdAuth(f, ioStreams),
},
},
{
// 6. 高級命令，包括diff/apply/patch/replace/wait/convert/kustomize
Message: "Advanced Commands:",
Commands: []*cobra.Command{
diff.NewCmdDiff(f, ioStreams),
apply.NewCmdApply("kubectl", f, ioStreams),
patch.NewCmdPatch(f, ioStreams),
replace.NewCmdReplace(f, ioStreams),
wait.NewCmdWait(f, ioStreams),
convert.NewCmdConvert(f, ioStreams),
kustomize.NewCmdKustomize(ioStreams),
},
},
{
// 7. 設置命令，包括label，annotate，completion
Message: "Settings Commands:",
Commands: []*cobra.Command{
label.NewCmdLabel(f, ioStreams),
annotate.NewCmdAnnotate("kubectl", f, ioStreams),
completion.NewCmdCompletion(ioStreams.Out, ""),
},
},
}
groups.Add(cmds)
filters := []string{"options"}
// alpha相關的子命令
alpha := cmdpkg.NewCmdAlpha(f, ioStreams)
if !alpha.HasSubCommands() {
filters = append(filters, alpha.Name())
}
templates.ActsAsRootCommand(cmds, filters, groups...)
// 代碼補全相關
for name, completion := range bashCompletionFlags {
if cmds.Flag(name) != nil {
if cmds.Flag(name).Annotations == nil {
cmds.Flag(name).Annotations = map[string][]string{}
}
cmds.Flag(name).Annotations[cobra.BashCompCustom] = append(
cmds.Flag(name).Annotations[cobra.BashCompCustom],
completion,
)
}
}
// 添加其余子命令，包括 alpha/config/plugin/version/api-versions/api-resources/options
cmds.AddCommand(alpha)
cmds.AddCommand(cmdconfig.NewCmdConfig(f, clientcmd.NewDefaultPathOptions(), ioStreams))
cmds.AddCommand(plugin.NewCmdPlugin(f, ioStreams))
cmds.AddCommand(version.NewCmdVersion(f, ioStreams))
cmds.AddCommand(apiresources.NewCmdAPIVersions(f, ioStreams))
cmds.AddCommand(apiresources.NewCmdAPIResources(f, ioStreams))
cmds.AddCommand(options.NewCmdOptions(ioStreams.Out))
return cmds
}
create
func NewCmdCreate(f cmdutil.Factory, ioStreams genericclioptions.IOStreams) *cobra.Command {
// create子命令的相關選項
o := NewCreateOptions(ioStreams)
// create子命令的相關說明
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) {
if cmdutil.IsFilenameSliceEmpty(o.FilenameOptions.Filenames, o.FilenameOptions.Kustomize) {
ioStreams.ErrOut.Write([]byte("Error: must specify one of -f and -k\n\n"))
defaultRunFunc := cmdutil.DefaultSubCommandRun(ioStreams.ErrOut)
defaultRunFunc(cmd, args)
return
}
cmdutil.CheckErr(o.Complete(f, cmd))
cmdutil.CheckErr(o.ValidateArgs(cmd, args))
// 核心的運行代碼邏輯是在這里的RunCreate
cmdutil.CheckErr(o.RunCreate(f, cmd))
},
}
o.RecordFlags.AddFlags(cmd)
usage := "to use to create the resource"
// 加入文件名選項的flag -f，保存到o.FilenameOptions.Filenames中，對應上面
cmdutil.AddFilenameOptionFlags(cmd, &o.FilenameOptions, usage)
cmdutil.AddValidateFlags(cmd)
cmd.Flags().BoolVar(&o.EditBeforeCreate, "edit", o.EditBeforeCreate, "Edit the API resource before creating")
cmd.Flags().Bool("windows-line-endings", runtime.GOOS == "windows",
"Only relevant if --edit=true. Defaults to the line ending native to your platform.")
cmdutil.AddApplyAnnotationFlags(cmd)
cmdutil.AddDryRunFlag(cmd)
cmd.Flags().StringVarP(&o.Selector, "selector", "l", o.Selector, "Selector (label query) to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2)")
cmd.Flags().StringVar(&o.Raw, "raw", o.Raw, "Raw URI to POST to the server. Uses the transport specified by the kubeconfig file.")
cmdutil.AddFieldManagerFlagVar(cmd, &o.fieldManager, "kubectl-create")
o.PrintFlags.AddFlags(cmd)
// create的子命令，指定create對象
cmd.AddCommand(NewCmdCreateNamespace(f, ioStreams))
cmd.AddCommand(NewCmdCreateQuota(f, ioStreams))
cmd.AddCommand(NewCmdCreateSecret(f, ioStreams))
cmd.AddCommand(NewCmdCreateConfigMap(f, ioStreams))
cmd.AddCommand(NewCmdCreateServiceAccount(f, ioStreams))
cmd.AddCommand(NewCmdCreateService(f, ioStreams))
cmd.AddCommand(NewCmdCreateDeployment(f, ioStreams))
cmd.AddCommand(NewCmdCreateClusterRole(f, ioStreams))
cmd.AddCommand(NewCmdCreateClusterRoleBinding(f, ioStreams))
cmd.AddCommand(NewCmdCreateRole(f, ioStreams))
cmd.AddCommand(NewCmdCreateRoleBinding(f, ioStreams))
cmd.AddCommand(NewCmdCreatePodDisruptionBudget(f, ioStreams))
cmd.AddCommand(NewCmdCreatePriorityClass(f, ioStreams))
cmd.AddCommand(NewCmdCreateJob(f, ioStreams))
cmd.AddCommand(NewCmdCreateCronJob(f, ioStreams))
return cmd
}
runCreate
func (o *CreateOptions) RunCreate(f cmdutil.Factory, cmd *cobra.Command) error {
// f為傳入的Factory，主要是封裝了與kube-apiserver交互客戶端
schema, err := f.Validator(cmdutil.GetFlagBool(cmd, "validate"))
if err != nil {
return err
}
cmdNamespace, enforceNamespace, err := f.ToRawKubeConfigLoader().Namespace()
if err != nil {
return err
}
// 實例化Builder，這塊的邏輯比較復雜，我們先關注文件部分
r := f.NewBuilder().
Unstructured().
Schema(schema).
ContinueOnError().
NamespaceParam(cmdNamespace).DefaultNamespace().
// 讀取文件信息，發現除了支持簡單的本地文件，也支持標准輸入和http/https協議訪問的文件，保存為Visitor
FilenameParam(enforceNamespace, &o.FilenameOptions).
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 {
// 打印結果 xxxx created
return o.PrintObj(info.Object)
})
return nil
}
3、創建資源對象過程
原理：客戶端與服務端進行一次http請求的交互。主要流程可分為：實例化Factory接口、通過Builder和Visitor將資源對象描述文件（nginx_pod.yaml）文本格式轉換成資源對象。將資源對象以http請求的方式發送給api server，並得到響應結果。最終根據Visitor匿名函數集的errors判斷是否成功創建了資源對象。
4、實例化Factory接口
在執行每一個kubectl命令之前，都需要執行實例化 cmdutil Factory 接口對象的操作。Factory 是一個通用對象，它提供了與 kube-apiserver 的交互方式，以及驗證資源對象等方法。cmdutil Factory接口代碼如下：
// 實例化Factory接口，工廠模式
f := cmdutil.NewFactory(matchVersionKubeConfigFlags)
Factory接口如下:
type Factory interface {
// DynamicClient returns a dynamic client ready for use
DynamicClient() (dynamic.Interface, error)
// KubernetesClientSet gives you back an external clientset
KubernetesClientSet() (*kubernetes.Clientset, error)
// Returns a RESTClient for accessing Kubernetes resources or an error.
RESTClient() (*restclient.RESTClient, error)
// NewBuilder returns an object that assists in loading objects from both disk and the server
// and which implements the common patterns for CLI interactions with generic resources.
NewBuilder() *resource.Builder
// Returns a schema that can validate objects stored on disk.
Validator(validate bool) (validation.Schema, error)
}
接口說明如下：
- DynamicClient：動態客戶端。
- KubernetesClientSet：ClientSet客戶端。
- RESTClient：RESTClient客戶端。
- NewBuilder：實例化builder，builder用於將命令獲取的參數轉換成資源對象。
- Validator：驗證資源對象。
cmd Factory 接口封裝了3種 client-go 客戶端與 kube-apiserver 交互的方式，分別是DynamicClient、KubernetesClientSet（簡稱ClientSet）及RESTClient。3種交互方式各有不同的應用場景。
5、kubectl 設計模式中Visitor的實現
訪問者模式：允許一個或者多個操作應用到對象上，解耦操作和對象本身
對於一個程序來說主要表現在：
1、表面：某個對象執行了一個方法
2、內部：對象內部調用了多個方法，最后統一返回結果
舉個例子：
1、表面：調用一個查詢接口
2、內部：先從緩存中查詢，沒查到再去熱點數據庫查詢，還沒查到則去歸檔數據庫查詢
Visitor
訪問者模式的定義
// Visitor 即為訪問者這個對象
type Visitor interface {
Visit(VisitorFunc) error
}
// VisitorFunc對應這個對象的方法，也就是定義中的“操作”
type VisitorFunc func(*Info, error) error
基本的數據結構很簡單，從當前的數據結構看有2個問題：
1、單個操作可以直接調用Visit方法，那多個操作如何實現呢？
2、在應用多個操作時，如果出現了error，該退出還是繼續下一個操作呢？
VisitorList
封裝多個Visitor為一個，出現錯誤就立即終止並返回
// VisitorList定義為[]Visitor，又實現了Visit方法，也就是將多個[]Visitor封裝為一個Visitor
type VisitorList []Visitor
// Visit implements Visitor
// 發生error就立刻返回，不繼續遍歷
func (l VisitorList) Visit(fn VisitorFunc) error {
for i := range l {
if err := l[i].Visit(fn); err != nil {
return err
}
}
return nil
}
EagerVisitorList
封裝多個Visitor為一個，出現錯誤暫存下來，全部遍歷完再聚合所有的錯誤並返回
// EagerVisitorList 也是將多個[]Visitor封裝為一個Visitor
type EagerVisitorList []Visitor
// 返回的錯誤暫存到[]error中，統一聚合
// Visit implements Visitor, and gathers errors that occur during processing until
// all sub visitors have been visited.
func (l EagerVisitorList) Visit(fn VisitorFunc) error {
errs := []error(nil)
for i := range l {
if err := l[i].Visit(func(info *Info, err error) error {
if err != nil {
errs = append(errs, err)
return nil
}
if err := fn(info, nil); err != nil {
errs = append(errs, err)
}
return nil
}); err != nil {
errs = append(errs, err)
}
}
return utilerrors.NewAggregate(errs)
}
DecoratedVisitor
這里借鑒了裝飾器的設計模式，將一個Visitor調用多個VisitorFunc方法，封裝為調用一個VisitorFunc
// 裝飾器Visitor
type DecoratedVisitor struct {
visitor Visitor
decorators []VisitorFunc
}
// visitor遍歷調用decorators中所有函數，有失敗立即返回
// Visit implements Visitor
func (v DecoratedVisitor) Visit(fn VisitorFunc) error {
return v.visitor.Visit(func(info *Info, err error) error {
if err != nil {
return err
}
for i := range v.decorators {
if err := v.decorators[i](info, nil); err != nil {
return err
}
}
return fn(info, nil)
})
}
ContinueOnErrorVisitor
// 報錯依舊繼續
type ContinueOnErrorVisitor struct {
Visitor
}
// 報錯不立即返回，聚合所有錯誤后返回
func (v ContinueOnErrorVisitor) Visit(fn VisitorFunc) error {
errs := []error{}
err := v.Visitor.Visit(func(info *Info, err error) error {
if err != nil {
errs = append(errs, err)
return nil
}
if err := fn(info, nil); err != nil {
errs = append(errs, err)
}
return nil
})
if err != nil {
errs = append(errs, err)
}
if len(errs) == 1 {
return errs[0]
}
return utilerrors.NewAggregate(errs)
}
Visitor的實現
StreamVisitor
最基礎的Visitor
type StreamVisitor struct {
// 讀取信息的來源，實現了Read這個接口，這個"流式"的概念，
//包括了常見的HTTP、文件、標准輸入等各類輸入
io.Reader
*mapper
Source string
Schema ContentValidator
}
FileVisitor
文件的訪問，包括標准輸入，底層調用StreamVisitor來訪問
// 表示文件路徑或者STDIN
type FileVisitor struct {
Path string
*StreamVisitor
}
URLVisitor
HTTP用GET方法獲取數據，底層也是復用StreamVisitor
type URLVisitor struct {
URL *url.URL
*StreamVisitor
// 提供錯誤重試次數
HttpAttemptCount int
}
KustomizeVisitor
自定義的Visitor，針對自定義的文件系統
type KustomizeVisitor struct {
Path string
*StreamVisitor
}
6、資源創建http請求發送
// 在RunCreate函數中，關鍵的發送函數
obj, err := resource.
NewHelper(info.Client, info.Mapping).
DryRun(o.DryRunStrategy == cmdutil.DryRunServer).
WithFieldManager(o.fieldManager).
Create(info.Namespace, true, info.Object)
// 進入create函數，查看到
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()
}
/*
到這里，我們發現了2個關鍵性的定義:
1. RESTClient 與kube-apiserver交互的RESTful風格的客戶端
2. runtime.Object 資源對象的抽象，包括Pod/Deployment/Service等各類資源
*/
RESTful Client
// 從傳入參數來看，數據來源於Info這個結構
r.Visit(func(info *resource.Info, err error) error{})
// 而info來源於前面的Builder，前面部分都是將Builder參數化，核心的生成為Do函數
r := f.NewBuilder().
Unstructured().
Schema(schema).
ContinueOnError().
NamespaceParam(cmdNamespace).DefaultNamespace().
FilenameParam(enforceNamespace, &o.FilenameOptions).
LabelSelectorParam(o.Selector).
Flatten().
Do()
// 大致看一下這些函數，我們可以在Unstructured()中看到getClient函數，其實這就是我們要找的函數
func (b *Builder) getClient(gv schema.GroupVersion) (RESTClient, error)
// 從返回值來看，client包括默認的REST client和配置選項
NewClientWithOptions(client, b.requestTransforms...)
// 這個Client會在kubernetes項目中大量出現，它是與kube-apiserver交互的核心組件，以后再深入。
Object
Object這個對象是怎么獲取到的呢？因為我們的數據源是來自文件的，那么我們最直觀的想法就是FileVisitor
func (v *FileVisitor) Visit(fn VisitorFunc) error {
// 省略讀取這塊的代碼，底層調用的是StreamVisitor的邏輯
return v.StreamVisitor.Visit(fn)
}
func (v *StreamVisitor) Visit(fn VisitorFunc) error {
d := yaml.NewYAMLOrJSONDecoder(v.Reader, 4096)
for {
// 這里就是返回info的地方
info, err := v.infoForData(ext.Raw, v.Source)
}
}
// 再往下一層看，來到mapper層，也就是kubernetes的資源對象映射關系
func (m *mapper) infoForData(data []byte, source string) (*Info, error){
// 這里就是我們返回Object的地方，其中GVK是Group/Version/Kind的縮寫，后續我們會涉及
obj, gvk, err := m.decoder.Decode(data, nil, nil)
}
這時，我們想回頭去看，這個mapper是在什么時候被定義的？
// 在Builder初始化中，我們就找到了
func (b *Builder) Unstructured() *Builder {
b.mapper = &mapper{
localFn: b.isLocal,
restMapperFn: b.restMapperFn,
clientFn: b.getClient,
// 我們查找資源用到的是這個decoder
decoder: &metadataValidatingDecoder{unstructured.UnstructuredJSONScheme},
}
return b
}
// 逐層往下找，對應的Decode方法的實現，就是對應的數據解析成data：
func (s unstructuredJSONScheme) decode(data []byte) (runtime.Object, error) {
// 細節暫時忽略
}
給api server發送Post請求
// RESTful接口風格中，POST請求對應的就是CREATE方法
c.Post().
NamespaceIfScoped(namespace, m.NamespaceScoped).
Resource(resource).
VersionedParams(options, metav1.ParameterCodec).
Body(obj).
Do(context.TODO()).
Get()
// Do方法，發送請求
err := r.request(ctx, func(req *http.Request, resp *http.Response) {
result = r.transformResponse(resp, req)
})
// Get方法，獲取請求的返回結果，用來打印狀態
switch t := out.(type) {
case *metav1.Status:
if t.Status != metav1.StatusSuccess {
return nil, errors.FromObject(t)
}
}