devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Kubernetes源码之kubectl剖析_error: must specify one of -f and -k

作者：羊村懒王 | 2024-06-04 12:23:48

赞

踩

error: must specify one of -f and -k

目录

1.Kubernetes架构简介

2.kuberctl入口源码剖析

2.1 源码入口

2.2 kubectl命令行客户端代码

2.3 剖析kubectl create子命令

2.4 cobra.Command命令的执行模板

3.ApiServer服务端代码剖析

3.1 ApiServer启动入口

3.2 ApiServer命令启动执行方法

3.3 APiServer启动http server监听http请求

3.4 各Rest请求处理的Handler构建

4.总结

5.参考文章

1.Kubernetes架构简介

Kubernetes架构分为三层，即：控制平面、数据平面、计算平面。控制平面主要是指kube-apiserver、kube-controller-manager、kube-scheduler三个组件；数据平面是指etcd集群；计算平面是指kubelet、kube-proxy组件。我们通常会将控制平面和数据平面部署在同一组节点上，通常称其为Master节点，而计算平面称为Node节点。控制平面是整个集群的大脑，负责控制、调度集群资源；计算平面负责运行工作负载，是控制平面调度的对象；而数据平面则用来存储整个集群持久化数据，比如我们提交的配置文件以及集群状态信息等。我们通常通过kubectl这个命令行客户端来控制Kubernetes集群。

kubernetes集群中各个模块之间的依赖关系如下图所示：

Kubernetes 组件

 

图1 kubernetes架构图（源自：https://kubernetes.io/zh/docs/concepts/architecture/cloud-controller/）

kubectl通过命令行向apiserver发起rest请求，进行相应的资源调度管理相关操作，由apiserver收到命令之后落地到etcd数据库，然后其他各个模块与apiserver交互，实现相关资源调度工作的执行。

图2 服务调用依赖关系图（源自：https://blog.csdn.net/u014458692/article/details/108068088）

各个组件说明：

kube-apiserver 对外暴露Kubernetes API，所有对集群的操作都是通过这组API完成，包括客户端下达应用编排命令给Kubernetes集群；kubelet上报集群资源使用情况；以及各个组件之间的交互都是通过这套API完成的。

kube-controller-manager 负责整个Kubernetes的管理工作，保证集群中各种资源处于期望状态，当监控到集群中某个资源状态与期望状态不符时，controller-manager会触发调度操作。

kube-scheduler 调度器负责Kubernetes集群的具体调度工作，接收来自于controller-manager触发的调度操作请求，然后根据请求规格、调度约束、整体资源情况进行调度计算，最后将任务发送到目标节点由kubelet组件执行。

etcd 是一个高效KV存储系统。在Kubernetes环境中主要用于存储所有需要持久化的数据。

kubelet 是Node节点上的核心组件，负责与docker daemon进行交互运行docker容器；配置网络和数据卷；监控并上报节点资源使用情况以供调度器使用。

kube-proxy 主要负责Service Endpoint到POD实例的请求转发及负载均衡的规则管理。

 

2.kuberctl入口源码剖析

本文剖析源码的版本：kubernetes的源码版本为1.20， golang的源码版本为：1.14

2.1 源码入口

kubectl的源码入口：cmd/kubectl/kubectl.go:35

cmd包里面包含kubernetes常用的各种命令行工具，其中kubectl工具的入口方法为kubectl.go文件中的main方法，通过command := cmd.NewDefaultKubectlCommand()方法实现新建一个kubectl的命令行对象，然后调用command.Execute()方法执行命令，实现与控制面板的服务端服务对接。具体入口源码如下：

  1. func main() {
  2. 	rand.Seed(time.Now().UnixNano())
  3.  
  4. 	command := cmd.NewDefaultKubectlCommand()
  5.  
  6. 	// TODO: once we switch everything over to Cobra commands, we can go back to calling
  7. 	// cliflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
  8. 	// normalize func and add the go flag set by hand.
  9. 	pflag.CommandLine.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
  10. 	pflag.CommandLine.AddGoFlagSet(goflag.CommandLine)
  11. 	// cliflag.InitFlags()
  12. 	logs.InitLogs()
  13. 	defer logs.FlushLogs()
  14.  
  15. 	if err := command.Execute(); err != nil {
  16. 		os.Exit(1)
  17. 	}
  18. }

2.2 kubectl命令行客户端代码

命令行工具采用的是Cobra的工具包对参数进行解析，解析成功之后，采用一整套的模板方法实现整个执行过程的调度。

首先分析一下kubectl命令行对象的创建，具体的创建该对象的函数为：pkg/kubectl/cmd/cmd.go:432 NewKubectlCommand 函数，源代码如下：

  1. // NewKubectlCommand creates the `kubectl` command and its nested children.
  2. func NewKubectlCommand(in io.Reader, out, err io.Writer) *cobra.Command {
  3. 	warningHandler := rest.NewWarningWriter(err, rest.WarningWriterOptions{Deduplicate: true, Color: term.AllowsColorOutput(err)})
  4. 	warningsAsErrors := false
  5.  
  6. 	// Parent command to which all subcommands are added.
  7. 	cmds := &cobra.Command{
  8. 		Use:   "kubectl",
  9. 		Short: i18n.T("kubectl controls the Kubernetes cluster manager"),
  10. 		Long: templates.LongDesc(`
  11.       kubectl controls the Kubernetes cluster manager.
  13.       Find more information at:
  14.             https://kubernetes.io/docs/reference/kubectl/overview/`),
  15. 		Run: runHelp,
  16. 		// Hook before and after Run initialize and write profiles to disk,
  17. 		// respectively.
  18. 		PersistentPreRunE: func(*cobra.Command, []string) error {
  19. 			rest.SetDefaultWarningHandler(warningHandler)
  20. 			return initProfiling()
  21. 		},
  22. 		PersistentPostRunE: func(*cobra.Command, []string) error {
  23. 			if err := flushProfiling(); err != nil {
  24. 				return err
  25. 			}
  26. 			if warningsAsErrors {
  27. 				count := warningHandler.WarningCount()
  28. 				switch count {
  29. 				case 0:
  30. 					// no warnings
  31. 				case 1:
  32. 					return fmt.Errorf("%d warning received", count)
  33. 				default:
  34. 					return fmt.Errorf("%d warnings received", count)
  35. 				}
  36. 			}
  37. 			return nil
  38. 		},
  39. 		BashCompletionFunction: bashCompletionFunc,
  40. 	}
  41.  
  42. 	flags := cmds.PersistentFlags()
  43. 	flags.SetNormalizeFunc(cliflag.WarnWordSepNormalizeFunc) // Warn for "_" flags
  44.  
  45. 	// Normalize all flags that are coming from other packages or pre-configurations
  46. 	// a.k.a. change all "_" to "-". e.g. glog package
  47. 	flags.SetNormalizeFunc(cliflag.WordSepNormalizeFunc)
  48.  
  49. 	addProfilingFlags(flags)
  50.  
  51. 	flags.BoolVar(&warningsAsErrors, "warnings-as-errors", warningsAsErrors, "Treat warnings received from the server as errors and exit with a non-zero exit code")
  52.  
  53. 	kubeConfigFlags := genericclioptions.NewConfigFlags(true).WithDeprecatedPasswordFlag()
  54. 	kubeConfigFlags.AddFlags(flags)
  55. 	matchVersionKubeConfigFlags := cmdutil.NewMatchVersionFlags(kubeConfigFlags)
  56. 	matchVersionKubeConfigFlags.AddFlags(cmds.PersistentFlags())
  57.  
  58. 	cmds.PersistentFlags().AddGoFlagSet(flag.CommandLine)
  59.  
  60. 	f := cmdutil.NewFactory(matchVersionKubeConfigFlags)
  61.  
  62. 	// Sending in 'nil' for the getLanguageFn() results in using
  63. 	// the LANG environment variable.
  64. 	//
  65. 	// TODO: Consider adding a flag or file preference for setting
  66. 	// the language, instead of just loading from the LANG env. variable.
  67. 	i18n.LoadTranslations("kubectl", nil)
  68.  
  69. 	// From this point and forward we get warnings on flags that contain "_" separators
  70. 	cmds.SetGlobalNormalizationFunc(cliflag.WarnWordSepNormalizeFunc)
  71.  
  72. 	ioStreams := genericclioptions.IOStreams{In: in, Out: out, ErrOut: err}
  73.  
  74. 	groups := templates.CommandGroups{
  75. 		{
  76. 			Message: "Basic Commands (Beginner):",
  77. 			Commands: []*cobra.Command{
  78. 				create.NewCmdCreate(f, ioStreams),
  79. 				expose.NewCmdExposeService(f, ioStreams),
  80. 				run.NewCmdRun(f, ioStreams),
  81. 				set.NewCmdSet(f, ioStreams),
  82. 			},
  83. 		},
  84. 		{
  85. 			Message: "Basic Commands (Intermediate):",
  86. 			Commands: []*cobra.Command{
  87. 				explain.NewCmdExplain("kubectl", f, ioStreams),
  88. 				get.NewCmdGet("kubectl", f, ioStreams),
  89. 				edit.NewCmdEdit(f, ioStreams),
  90. 				delete.NewCmdDelete(f, ioStreams),
  91. 			},
  92. 		},
  93. 		{
  94. 			Message: "Deploy Commands:",
  95. 			Commands: []*cobra.Command{
  96. 				rollout.NewCmdRollout(f, ioStreams),
  97. 				scale.NewCmdScale(f, ioStreams),
  98. 				autoscale.NewCmdAutoscale(f, ioStreams),
  99. 			},
  100. 		},
  101. 		{
  102. 			Message: "Cluster Management Commands:",
  103. 			Commands: []*cobra.Command{
  104. 				certificates.NewCmdCertificate(f, ioStreams),
  105. 				clusterinfo.NewCmdClusterInfo(f, ioStreams),
  106. 				top.NewCmdTop(f, ioStreams),
  107. 				drain.NewCmdCordon(f, ioStreams),
  108. 				drain.NewCmdUncordon(f, ioStreams),
  109. 				drain.NewCmdDrain(f, ioStreams),
  110. 				taint.NewCmdTaint(f, ioStreams),
  111. 			},
  112. 		},
  113. 		{
  114. 			Message: "Troubleshooting and Debugging Commands:",
  115. 			Commands: []*cobra.Command{
  116. 				describe.NewCmdDescribe("kubectl", f, ioStreams),
  117. 				logs.NewCmdLogs(f, ioStreams),
  118. 				attach.NewCmdAttach(f, ioStreams),
  119. 				cmdexec.NewCmdExec(f, ioStreams),
  120. 				portforward.NewCmdPortForward(f, ioStreams),
  121. 				proxy.NewCmdProxy(f, ioStreams),
  122. 				cp.NewCmdCp(f, ioStreams),
  123. 				auth.NewCmdAuth(f, ioStreams),
  124. 			},
  125. 		},
  126. 		{
  127. 			Message: "Advanced Commands:",
  128. 			Commands: []*cobra.Command{
  129. 				diff.NewCmdDiff(f, ioStreams),
  130. 				apply.NewCmdApply("kubectl", f, ioStreams),
  131. 				patch.NewCmdPatch(f, ioStreams),
  132. 				replace.NewCmdReplace(f, ioStreams),
  133. 				wait.NewCmdWait(f, ioStreams),
  134. 				convert.NewCmdConvert(f, ioStreams),
  135. 				kustomize.NewCmdKustomize(ioStreams),
  136. 			},
  137. 		},
  138. 		{
  139. 			Message: "Settings Commands:",
  140. 			Commands: []*cobra.Command{
  141. 				label.NewCmdLabel(f, ioStreams),
  142. 				annotate.NewCmdAnnotate("kubectl", f, ioStreams),
  143. 				completion.NewCmdCompletion(ioStreams.Out, ""),
  144. 			},
  145. 		},
  146. 	}
  147. 	groups.Add(cmds)
  148.  
  149. 	filters := []string{"options"}
  150.  
  151. 	// Hide the "alpha" subcommand if there are no alpha commands in this build.
  152. 	alpha := cmdpkg.NewCmdAlpha(f, ioStreams)
  153. 	if !alpha.HasSubCommands() {
  154. 		filters = append(filters, alpha.Name())
  155. 	}
  156.  
  157. 	templates.ActsAsRootCommand(cmds, filters, groups...)
  158.  
  159. 	for name, completion := range bashCompletionFlags {
  160. 		if cmds.Flag(name) != nil {
  161. 			if cmds.Flag(name).Annotations == nil {
  162. 				cmds.Flag(name).Annotations = map[string][]string{}
  163. 			}
  164. 			cmds.Flag(name).Annotations[cobra.BashCompCustom] = append(
  165. 				cmds.Flag(name).Annotations[cobra.BashCompCustom],
  166. 				completion,
  167. 			)
  168. 		}
  169. 	}
  170.  
  171. 	cmds.AddCommand(alpha)
  172. 	cmds.AddCommand(cmdconfig.NewCmdConfig(f, clientcmd.NewDefaultPathOptions(), ioStreams))
  173. 	cmds.AddCommand(plugin.NewCmdPlugin(f, ioStreams))
  174. 	cmds.AddCommand(version.NewCmdVersion(f, ioStreams))
  175. 	cmds.AddCommand(apiresources.NewCmdAPIVersions(f, ioStreams))
  176. 	cmds.AddCommand(apiresources.NewCmdAPIResources(f, ioStreams))
  177. 	cmds.AddCommand(options.NewCmdOptions(ioStreams.Out))
  178.  
  179. 	return cmds
  180. }

kubectl的命令行cobra.Command是一个父子树，其中kubectl命令作为树的根节点，然后添加了子节点（groups变量），各个子节点以kubectl Command节点为父节点，如根据文件创建，编辑文件，创建命名空间，创建部署服务，创建证书等等。同样，kubectl的子节点还可以拥有自己的子节点，创建都类似。

2.3 剖析kubectl create子命令

参考一个kubectl create 命令解析的入口： vendor/k8s.io/kubectl/pkg/cmd/create/create.go:100 NewCmdCreate方法，其源代码如下：

  1. // NewCmdCreate returns new initialized instance of create sub command
  2. func NewCmdCreate(f cmdutil.Factory, ioStreams genericclioptions.IOStreams) *cobra.Command {
  3. 	o := NewCreateOptions(ioStreams)
  4.  
  5. 	cmd := &cobra.Command{
  6. 		Use:                   "create -f FILENAME",
  7. 		DisableFlagsInUseLine: true,
  8. 		Short:                 i18n.T("Create a resource from a file or from stdin."),
  9. 		Long:                  createLong,
  10. 		Example:               createExample,
  11. 		Run: func(cmd *cobra.Command, args []string) {
  12. 			if cmdutil.IsFilenameSliceEmpty(o.FilenameOptions.Filenames, o.FilenameOptions.Kustomize) {
  13. 				ioStreams.ErrOut.Write([]byte("Error: must specify one of -f and -k\n\n"))
  14. 				defaultRunFunc := cmdutil.DefaultSubCommandRun(ioStreams.ErrOut)
  15. 				defaultRunFunc(cmd, args)
  16. 				return
  17. 			}
  18. 			cmdutil.CheckErr(o.Complete(f, cmd))
  19. 			cmdutil.CheckErr(o.ValidateArgs(cmd, args))
  20. 			cmdutil.CheckErr(o.RunCreate(f, cmd))
  21. 		},
  22. 	}
  23.  
  24. 	// bind flag structs
  25. 	o.RecordFlags.AddFlags(cmd)
  26.  
  27. 	usage := "to use to create the resource"
  28. 	cmdutil.AddFilenameOptionFlags(cmd, &o.FilenameOptions, usage)
  29. 	cmdutil.AddValidateFlags(cmd)
  30. 	cmd.Flags().BoolVar(&o.EditBeforeCreate, "edit", o.EditBeforeCreate, "Edit the API resource before creating")
  31. 	cmd.Flags().Bool("windows-line-endings", runtime.GOOS == "windows",
  32. 		"Only relevant if --edit=true. Defaults to the line ending native to your platform.")
  33. 	cmdutil.AddApplyAnnotationFlags(cmd)
  34. 	cmdutil.AddDryRunFlag(cmd)
  35. 	cmd.Flags().StringVarP(&o.Selector, "selector", "l", o.Selector, "Selector (label query) to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2)")
  36. 	cmd.Flags().StringVar(&o.Raw, "raw", o.Raw, "Raw URI to POST to the server.  Uses the transport specified by the kubeconfig file.")
  37. 	cmdutil.AddFieldManagerFlagVar(cmd, &o.fieldManager, "kubectl-create")
  38.  
  39. 	o.PrintFlags.AddFlags(cmd)
  40.  
  41. 	// create subcommands
  42. 	cmd.AddCommand(NewCmdCreateNamespace(f, ioStreams))
  43. 	cmd.AddCommand(NewCmdCreateQuota(f, ioStreams))
  44. 	cmd.AddCommand(NewCmdCreateSecret(f, ioStreams))
  45. 	cmd.AddCommand(NewCmdCreateConfigMap(f, ioStreams))
  46. 	cmd.AddCommand(NewCmdCreateServiceAccount(f, ioStreams))
  47. 	cmd.AddCommand(NewCmdCreateService(f, ioStreams))
  48. 	cmd.AddCommand(NewCmdCreateDeployment(f, ioStreams))
  49. 	cmd.AddCommand(NewCmdCreateClusterRole(f, ioStreams))
  50. 	cmd.AddCommand(NewCmdCreateClusterRoleBinding(f, ioStreams))
  51. 	cmd.AddCommand(NewCmdCreateRole(f, ioStreams))
  52. 	cmd.AddCommand(NewCmdCreateRoleBinding(f, ioStreams))
  53. 	cmd.AddCommand(NewCmdCreatePodDisruptionBudget(f, ioStreams))
  54. 	cmd.AddCommand(NewCmdCreatePriorityClass(f, ioStreams))
  55. 	cmd.AddCommand(NewCmdCreateJob(f, ioStreams))
  56. 	cmd.AddCommand(NewCmdCreateCronJob(f, ioStreams))
  57. 	return cmd
  58. }

代码中，通过cmdutil.AddFilenameOptionFlags，cmdutil.AddValidateFlags，cmd.Flags().BoolVar，cmd.Flags().Bool等方法为kubectl命令添加对应的命令行参数； 通过cmd.AddCommand() 方法为create命令创建子命令，如namespace，secret， configmap，role， job等等，以此构建一整套命令行的树结构。

在NewCmdCreate方法中，采用cobra.Command命令的Run方法中，首先对参数调用cmdutil.CheckErr(o.Complete(f, cmd))方法实现参数的补齐，其次调用cmdutil.CheckErr(o.ValidateArgs(cmd, args))方法实现对参数的合理性进行校验，最后通过cmdutil.CheckErr(o.RunCreate(f, cmd))方法实现对命令的执行。

查看RunCreate方法源码，vendor/k8s.io/kubectl/pkg/cmd/create/create.go:227 RunCreate方法，通过该方法创建，方法中通过链路形式创建request请求，通过 obj, err := resource.NewHelper(info.Client, info.Mapping).DryRun(o.DryRunStrategy == cmdutil.DryRunServer).WithFieldManager(o.fieldManager).Create(info.Namespace, true, info.Object)调用，实现Rest调用触发服务端API服务，完成相应的创建动作。

  1. // RunCreate performs the creation
  2. func (o *CreateOptions) RunCreate(f cmdutil.Factory, cmd *cobra.Command) error {
  3. 	// raw only makes sense for a single file resource multiple objects aren't likely to do what you want.
  4. 	// the validator enforces this, so
  5. 	if len(o.Raw) > 0 {
  6. 		restClient, err := f.RESTClient()
  7. 		if err != nil {
  8. 			return err
  9. 		}
  10. 		return rawhttp.RawPost(restClient, o.IOStreams, o.Raw, o.FilenameOptions.Filenames[0])
  11. 	}
  12.  
  13. 	if o.EditBeforeCreate {
  14. 		return RunEditOnCreate(f, o.PrintFlags, o.RecordFlags, o.IOStreams, cmd, &o.FilenameOptions, o.fieldManager)
  15. 	}
  16. 	schema, err := f.Validator(cmdutil.GetFlagBool(cmd, "validate"))
  17. 	if err != nil {
  18. 		return err
  19. 	}
  20.  
  21. 	cmdNamespace, enforceNamespace, err := f.ToRawKubeConfigLoader().Namespace()
  22. 	if err != nil {
  23. 		return err
  24. 	}
  25.  
  26. 	r := f.NewBuilder().
  27. 		Unstructured().
  28. 		Schema(schema).
  29. 		ContinueOnError().
  30. 		NamespaceParam(cmdNamespace).DefaultNamespace().
  31. 		FilenameParam(enforceNamespace, &o.FilenameOptions).
  32. 		LabelSelectorParam(o.Selector).
  33. 		Flatten().
  34. 		Do()
  35. 	err = r.Err()
  36. 	if err != nil {
  37. 		return err
  38. 	}
  39.  
  40. 	count := 0
  41. 	err = r.Visit(func(info *resource.Info, err error) error {
  42. 		if err != nil {
  43. 			return err
  44. 		}
  45. 		if err := util.CreateOrUpdateAnnotation(cmdutil.GetFlagBool(cmd, cmdutil.ApplyAnnotationsFlag), info.Object, scheme.DefaultJSONEncoder()); err != nil {
  46. 			return cmdutil.AddSourceToErr("creating", info.Source, err)
  47. 		}
  48.  
  49. 		if err := o.Recorder.Record(info.Object); err != nil {
  50. 			klog.V(4).Infof("error recording current command: %v", err)
  51. 		}
  52.  
  53. 		if o.DryRunStrategy != cmdutil.DryRunClient {
  54. 			if o.DryRunStrategy == cmdutil.DryRunServer {
  55. 				if err := o.DryRunVerifier.HasSupport(info.Mapping.GroupVersionKind); err != nil {
  56. 					return cmdutil.AddSourceToErr("creating", info.Source, err)
  57. 				}
  58. 			}
  59. 			obj, err := resource.
  60. 				NewHelper(info.Client, info.Mapping).
  61. 				DryRun(o.DryRunStrategy == cmdutil.DryRunServer).
  62. 				WithFieldManager(o.fieldManager).
  63. 				Create(info.Namespace, true, info.Object)
  64. 			if err != nil {
  65. 				return cmdutil.AddSourceToErr("creating", info.Source, err)
  66. 			}
  67. 			info.Refresh(obj, true)
  68. 		}
  69.  
  70. 		count++
  71.  
  72. 		return o.PrintObj(info.Object)
  73. 	})
  74. 	if err != nil {
  75. 		return err
  76. 	}
  77. 	if count == 0 {
  78. 		return fmt.Errorf("no objects passed to create")
  79. 	}
  80. 	return nil
  81. }

rest调用通过cliet-go发起远程调用 源码可以追溯到staging/src/k8s.io/client-go/rest/request.go:826 request 方法，如下源码：

  1. // request connects to the server and invokes the provided function when a server response is
  2. // received. It handles retry behavior and up front validation of requests. It will invoke
  3. // fn at most once. It will return an error if a problem occurred prior to connecting to the
  4. // server - the provided function is responsible for handling server errors.
  5. func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Response)) error {
  6. 	//Metrics for total request latency
  7. 	start := time.Now()
  8. 	defer func() {
  9. 		metrics.RequestLatency.Observe(r.verb, r.finalURLTemplate(), time.Since(start))
  10. 	}()
  11.  
  12. 	if r.err != nil {
  13. 		klog.V(4).Infof("Error in request: %v", r.err)
  14. 		return r.err
  15. 	}
  16.  
  17. 	if err := r.requestPreflightCheck(); err != nil {
  18. 		return err
  19. 	}
  20.  
  21. 	client := r.c.Client
  22. 	if client == nil {
  23. 		client = http.DefaultClient
  24. 	}
  25.  
  26. 	// Throttle the first try before setting up the timeout configured on the
  27. 	// client. We don't want a throttled client to return timeouts to callers
  28. 	// before it makes a single request.
  29. 	if err := r.tryThrottle(ctx); err != nil {
  30. 		return err
  31. 	}
  32.  
  33. 	if r.timeout > 0 {
  34. 		var cancel context.CancelFunc
  35. 		ctx, cancel = context.WithTimeout(ctx, r.timeout)
  36. 		defer cancel()
  37. 	}
  38.  
  39. 	// Right now we make about ten retry attempts if we get a Retry-After response.
  40. 	retries := 0
  41. 	for {
  42.  
  43. 		url := r.URL().String()
  44. 		req, err := http.NewRequest(r.verb, url, r.body)
  45. 		if err != nil {
  46. 			return err
  47. 		}
  48. 		req = req.WithContext(ctx)
  49. 		req.Header = r.headers
  50.  
  51. 		r.backoff.Sleep(r.backoff.CalculateBackoff(r.URL()))
  52. 		if retries > 0 {
  53. 			// We are retrying the request that we already send to apiserver
  54. 			// at least once before.
  55. 			// This request should also be throttled with the client-internal rate limiter.
  56. 			if err := r.tryThrottle(ctx); err != nil {
  57. 				return err
  58. 			}
  59. 		}
  60. 		resp, err := client.Do(req)
  61. 		updateURLMetrics(r, resp, err)
  62. 		if err != nil {
  63. 			r.backoff.UpdateBackoff(r.URL(), err, 0)
  64. 		} else {
  65. 			r.backoff.UpdateBackoff(r.URL(), err, resp.StatusCode)
  66. 		}
  67. 		if err != nil {
  68. 			// "Connection reset by peer" or "apiserver is shutting down" are usually a transient errors.
  69. 			// Thus in case of "GET" operations, we simply retry it.
  70. 			// We are not automatically retrying "write" operations, as
  71. 			// they are not idempotent.
  72. 			if r.verb != "GET" {
  73. 				return err
  74. 			}
  75. 			// For connection errors and apiserver shutdown errors retry.
  76. 			if net.IsConnectionReset(err) || net.IsProbableEOF(err) {
  77. 				// For the purpose of retry, we set the artificial "retry-after" response.
  78. 				// TODO: Should we clean the original response if it exists?
  79. 				resp = &http.Response{
  80. 					StatusCode: http.StatusInternalServerError,
  81. 					Header:     http.Header{"Retry-After": []string{"1"}},
  82. 					Body:       ioutil.NopCloser(bytes.NewReader([]byte{})),
  83. 				}
  84. 			} else {
  85. 				return err
  86. 			}
  87. 		}
  88.  
  89. 		done := func() bool {
  90. 			// Ensure the response body is fully read and closed
  91. 			// before we reconnect, so that we reuse the same TCP
  92. 			// connection.
  93. 			defer func() {
  94. 				const maxBodySlurpSize = 2 << 10
  95. 				if resp.ContentLength <= maxBodySlurpSize {
  96. 					io.Copy(ioutil.Discard, &io.LimitedReader{R: resp.Body, N: maxBodySlurpSize})
  97. 				}
  98. 				resp.Body.Close()
  99. 			}()
  100.  
  101. 			retries++
  102. 			if seconds, wait := checkWait(resp); wait && retries <= r.maxRetries {
  103. 				if seeker, ok := r.body.(io.Seeker); ok && r.body != nil {
  104. 					_, err := seeker.Seek(0, 0)
  105. 					if err != nil {
  106. 						klog.V(4).Infof("Could not retry request, can't Seek() back to beginning of body for %T", r.body)
  107. 						fn(req, resp)
  108. 						return true
  109. 					}
  110. 				}
  111.  
  112. 				klog.V(4).Infof("Got a Retry-After %ds response for attempt %d to %v", seconds, retries, url)
  113. 				r.backoff.Sleep(time.Duration(seconds) * time.Second)
  114. 				return false
  115. 			}
  116. 			fn(req, resp)
  117. 			return true
  118. 		}()
  119. 		if done {
  120. 			return nil
  121. 		}
  122. 	}
  123. }

其中构造URL参数依赖方法 staging/src/k8s.io/client-go/rest/request.go:468 URL() 方法，在该方法中完成命令行参数转变为url path的拼接封装，如下源码：

  1. staging/src/k8s.io/client-go/rest/request.go:468
  2.  
  3. // URL returns the current working URL.
  4. func (r *Request) URL() *url.URL {
  5. 	p := r.pathPrefix
  6. 	if r.namespaceSet && len(r.namespace) > 0 {
  7. 		p = path.Join(p, "namespaces", r.namespace)
  8. 	}
  9. 	if len(r.resource) != 0 {
  10. 		p = path.Join(p, strings.ToLower(r.resource))
  11. 	}
  12. 	// Join trims trailing slashes, so preserve r.pathPrefix's trailing slash for backwards compatibility if nothing was changed
  13. 	if len(r.resourceName) != 0 || len(r.subpath) != 0 || len(r.subresource) != 0 {
  14. 		p = path.Join(p, r.resourceName, r.subresource, r.subpath)
  15. 	}
  16.  
  17. 	finalURL := &url.URL{}
  18. 	if r.c.base != nil {
  19. 		*finalURL = *r.c.base
  20. 	}
  21. 	finalURL.Path = p
  22.  
  23. 	query := url.Values{}
  24. 	for key, values := range r.params {
  25. 		for _, value := range values {
  26. 			query.Add(key, value)
  27. 		}
  28. 	}
  29.  
  30. 	// timeout is handled specially here.
  31. 	if r.timeout != 0 {
  32. 		query.Set("timeout", r.timeout.String())
  33. 	}
  34. 	finalURL.RawQuery = query.Encode()
  35. 	return finalURL
  36. }

2.4 cobra.Command命令的执行模板

kubectl命令行，采用cobra.Command创建，其中命令的执行是有一整套的命令执行模板方法，该执行方法的入口为：vendor/github.com/spf13/cobra/command.go:892 ExecuteC方法，具体源码参考如下代码段。代码中执行

  1. // ExecuteC executes the command.
  2. func (c *Command) ExecuteC() (cmd *Command, err error) {
  3. 	if c.ctx == nil {
  4. 		c.ctx = context.Background()
  5. 	}
  6.  
  7. 	// Regardless of what command execute is called on, run on Root only
  8. 	if c.HasParent() {
  9. 		return c.Root().ExecuteC()
  10. 	}
  11.  
  12. 	// windows hook
  13. 	if preExecHookFn != nil {
  14. 		preExecHookFn(c)
  15. 	}
  16.  
  17. 	// initialize help as the last point possible to allow for user
  18. 	// overriding
  19. 	c.InitDefaultHelpCmd()
  20.  
  21. 	args := c.args
  22.  
  23. 	// Workaround FAIL with "go test -v" or "cobra.test -test.v", see #155
  24. 	if c.args == nil && filepath.Base(os.Args[0]) != "cobra.test" {
  25. 		args = os.Args[1:]
  26. 	}
  27.  
  28. 	// initialize the hidden command to be used for bash completion
  29. 	c.initCompleteCmd(args)
  30.  
  31. 	var flags []string
  32. 	if c.TraverseChildren {
  33. 		cmd, flags, err = c.Traverse(args)
  34. 	} else {
  35. 		cmd, flags, err = c.Find(args)
  36. 	}
  37. 	if err != nil {
  38. 		// If found parse to a subcommand and then failed, talk about the subcommand
  39. 		if cmd != nil {
  40. 			c = cmd
  41. 		}
  42. 		if !c.SilenceErrors {
  43. 			c.Println("Error:", err.Error())
  44. 			c.Printf("Run '%v --help' for usage.\n", c.CommandPath())
  45. 		}
  46. 		return c, err
  47. 	}
  48.  
  49. 	cmd.commandCalledAs.called = true
  50. 	if cmd.commandCalledAs.name == "" {
  51. 		cmd.commandCalledAs.name = cmd.Name()
  52. 	}
  53.  
  54. 	// We have to pass global context to children command
  55. 	// if context is present on the parent command.
  56. 	if cmd.ctx == nil {
  57. 		cmd.ctx = c.ctx
  58. 	}
  59.  
  60. 	err = cmd.execute(flags)
  61. 	if err != nil {
  62. 		// Always show help if requested, even if SilenceErrors is in
  63. 		// effect
  64. 		if err == flag.ErrHelp {
  65. 			cmd.HelpFunc()(cmd, args)
  66. 			return cmd, nil
  67. 		}
  68.  
  69. 		// If root command has SilentErrors flagged,
  70. 		// all subcommands should respect it
  71. 		if !cmd.SilenceErrors && !c.SilenceErrors {
  72. 			c.Println("Error:", err.Error())
  73. 		}
  74.  
  75. 		// If root command has SilentUsage flagged,
  76. 		// all subcommands should respect it
  77. 		if !cmd.SilenceUsage && !c.SilenceUsage {
  78. 			c.Println(cmd.UsageString())
  79. 		}
  80. 	}
  81. 	return cmd, err
  82. }

执行模板中依赖的执行execute方法vendor/github.com/spf13/cobra/command.go:755 execute方法，该方法定义了各个方法的前后依赖关系，执行顺序如下：

  1. // The *Run functions are executed in the following order:
  2. //   * PersistentPreRun()
  3. //   * PreRun()
  4. //   * Run()
  5. //   * PostRun()
  6. //   * PersistentPostRun()
  7. // All functions get the same args, the arguments after the command name.

调用服务端通过rest方法调用，依赖go语言底层的http的包实现，具体参考：vendor/k8s.io/client-go/rest/request.go:826 request 方法。

  1. func (c *Command) execute(a []string) (err error) {
  2. 	if c == nil {
  3. 		return fmt.Errorf("Called Execute() on a nil Command")
  4. 	}
  5.  
  6. 	if len(c.Deprecated) > 0 {
  7. 		c.Printf("Command %q is deprecated, %s\n", c.Name(), c.Deprecated)
  8. 	}
  9.  
  10. 	// initialize help and version flag at the last point possible to allow for user
  11. 	// overriding
  12. 	c.InitDefaultHelpFlag()
  13. 	c.InitDefaultVersionFlag()
  14.  
  15. 	err = c.ParseFlags(a)
  16. 	if err != nil {
  17. 		return c.FlagErrorFunc()(c, err)
  18. 	}
  19.  
  20. 	// If help is called, regardless of other flags, return we want help.
  21. 	// Also say we need help if the command isn't runnable.
  22. 	helpVal, err := c.Flags().GetBool("help")
  23. 	if err != nil {
  24. 		// should be impossible to get here as we always declare a help
  25. 		// flag in InitDefaultHelpFlag()
  26. 		c.Println("\"help\" flag declared as non-bool. Please correct your code")
  27. 		return err
  28. 	}
  29.  
  30. 	if helpVal {
  31. 		return flag.ErrHelp
  32. 	}
  33.  
  34. 	// for back-compat, only add version flag behavior if version is defined
  35. 	if c.Version != "" {
  36. 		versionVal, err := c.Flags().GetBool("version")
  37. 		if err != nil {
  38. 			c.Println("\"version\" flag declared as non-bool. Please correct your code")
  39. 			return err
  40. 		}
  41. 		if versionVal {
  42. 			err := tmpl(c.OutOrStdout(), c.VersionTemplate(), c)
  43. 			if err != nil {
  44. 				c.Println(err)
  45. 			}
  46. 			return err
  47. 		}
  48. 	}
  49.  
  50. 	if !c.Runnable() {
  51. 		return flag.ErrHelp
  52. 	}
  53.  
  54. 	c.preRun()
  55.  
  56. 	argWoFlags := c.Flags().Args()
  57. 	if c.DisableFlagParsing {
  58. 		argWoFlags = a
  59. 	}
  60.  
  61. 	if err := c.ValidateArgs(argWoFlags); err != nil {
  62. 		return err
  63. 	}
  64.  
  65. 	for p := c; p != nil; p = p.Parent() {
  66. 		if p.PersistentPreRunE != nil {
  67. 			if err := p.PersistentPreRunE(c, argWoFlags); err != nil {
  68. 				return err
  69. 			}
  70. 			break
  71. 		} else if p.PersistentPreRun != nil {
  72. 			p.PersistentPreRun(c, argWoFlags)
  73. 			break
  74. 		}
  75. 	}
  76. 	if c.PreRunE != nil {
  77. 		if err := c.PreRunE(c, argWoFlags); err != nil {
  78. 			return err
  79. 		}
  80. 	} else if c.PreRun != nil {
  81. 		c.PreRun(c, argWoFlags)
  82. 	}
  83.  
  84. 	if err := c.validateRequiredFlags(); err != nil {
  85. 		return err
  86. 	}
  87. 	if c.RunE != nil {
  88. 		if err := c.RunE(c, argWoFlags); err != nil {
  89. 			return err
  90. 		}
  91. 	} else {
  92. 		c.Run(c, argWoFlags)
  93. 	}
  94. 	if c.PostRunE != nil {
  95. 		if err := c.PostRunE(c, argWoFlags); err != nil {
  96. 			return err
  97. 		}
  98. 	} else if c.PostRun != nil {
  99. 		c.PostRun(c, argWoFlags)
  100. 	}
  101. 	for p := c; p != nil; p = p.Parent() {
  102. 		if p.PersistentPostRunE != nil {
  103. 			if err := p.PersistentPostRunE(c, argWoFlags); err != nil {
  104. 				return err
  105. 			}
  106. 			break
  107. 		} else if p.PersistentPostRun != nil {
  108. 			p.PersistentPostRun(c, argWoFlags)
  109. 			break
  110. 		}
  111. 	}
  112.  
  113. 	return nil
  114. }


3.ApiServer服务端代码剖析

上一节讲了客户端kubectl客户端发起命令，最终转换为http rest请求，触发服务调用。那么服务端是如何接收并处理相应的请求呢？服务端由apiserver接收来自kubectl客户端的调用，并进行相应的业务逻辑处理。服务端的整体框架的构建参考如下章节描述。

3.1 ApiServer启动入口

服务端apiserver启动入口：cmd/kube-apiserver/apiserver.go:32 main()方法，通过该方法启动apiserver。源码入口如下：

  1. func main() {
  2. 	rand.Seed(time.Now().UnixNano())
  3.  
  4. 	command := app.NewAPIServerCommand()
  5.  
  6. 	// TODO: once we switch everything over to Cobra commands, we can go back to calling
  7. 	// utilflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
  8. 	// normalize func and add the go flag set by hand.
  9. 	// utilflag.InitFlags()
  10. 	logs.InitLogs()
  11. 	defer logs.FlushLogs()
  12.  
  13. 	if err := command.Execute(); err != nil {
  14. 		os.Exit(1)
  15. 	}
  16. }

3.2 ApiServer命令启动执行方法

通过层层代码跟进，发现服务端启动APIServer，执行命令分创建服务调用链路，然后准备执行，最后执行启动Server方法，入口为：cmd/kube-apiserver/app/server.go:161 Run()方法，源码如下：

  1. cmd/kube-apiserver/app/server.go:161
  2. // Run runs the specified APIServer.  This should never exit.
  3. func Run(completeOptions completedServerRunOptions, stopCh <-chan struct{}) error {
  4. 	// To help debugging, immediately log version
  5. 	klog.Infof("Version: %+v", version.Get())
  6.  
  7. 	server, err := CreateServerChain(completeOptions, stopCh)
  8. 	if err != nil {
  9. 		return err
  10. 	}
  11.  
  12. 	prepared, err := server.PrepareRun()
  13. 	if err != nil {
  14. 		return err
  15. 	}
  16.  
  17. 	return prepared.Run(stopCh)
  18. }
  19.  
  20.  
  21. // 层层分析深入到如下代码：
  22. staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go:316
  23.  
  24. // Run spawns the secure http server. It only returns if stopCh is closed
  25. // or the secure port cannot be listened on initially.
  26. func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
  27. 	delayedStopCh := make(chan struct{})
  28.  
  29. 	go func() {
  30. 		defer close(delayedStopCh)
  31.  
  32. 		<-stopCh
  33.  
  34. 		// As soon as shutdown is initiated, /readyz should start returning failure.
  35. 		// This gives the load balancer a window defined by ShutdownDelayDuration to detect that /readyz is red
  36. 		// and stop sending traffic to this server.
  37. 		close(s.readinessStopCh)
  38.  
  39. 		time.Sleep(s.ShutdownDelayDuration)
  40. 	}()
  41.  
  42. 	// close socket after delayed stopCh
  43. 	stoppedCh, err := s.NonBlockingRun(delayedStopCh)
  44. 	if err != nil {
  45. 		return err
  46. 	}
  47.  
  48. 	<-stopCh
  49.  
  50. 	// run shutdown hooks directly. This includes deregistering from the kubernetes endpoint in case of kube-apiserver.
  51. 	err = s.RunPreShutdownHooks()
  52. 	if err != nil {
  53. 		return err
  54. 	}
  55.  
  56. 	// wait for the delayed stopCh before closing the handler chain (it rejects everything after Wait has been called).
  57. 	<-delayedStopCh
  58. 	// wait for stoppedCh that is closed when the graceful termination (server.Shutdown) is finished.
  59. 	<-stoppedCh
  60.  
  61. 	// Wait for all requests to finish, which are bounded by the RequestTimeout variable.
  62. 	s.HandlerChainWaitGroup.Wait()
  63.  
  64. 	return nil
  65. }

3.3 APiServer启动http server监听http请求

通过深入分析源码，进入到启动http Server的入口： staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go:360 NonBlockingRun()方法，如下：

  1. // NonBlockingRun spawns the secure http server. An error is
  2. // returned if the secure port cannot be listened on.
  3. // The returned channel is closed when the (asynchronous) termination is finished.
  4. func (s preparedGenericAPIServer) NonBlockingRun(stopCh <-chan struct{}) (<-chan struct{}, error) {
  5. 	// Use an stop channel to allow graceful shutdown without dropping audit events
  6. 	// after http server shutdown.
  7. 	auditStopCh := make(chan struct{})
  8.  
  9. 	// Start the audit backend before any request comes in. This means we must call Backend.Run
  10. 	// before http server start serving. Otherwise the Backend.ProcessEvents call might block.
  11. 	if s.AuditBackend != nil {
  12. 		if err := s.AuditBackend.Run(auditStopCh); err != nil {
  13. 			return nil, fmt.Errorf("failed to run the audit backend: %v", err)
  14. 		}
  15. 	}
  16.  
  17. 	// Use an internal stop channel to allow cleanup of the listeners on error.
  18. 	internalStopCh := make(chan struct{})
  19. 	var stoppedCh <-chan struct{}
  20. 	if s.SecureServingInfo != nil && s.Handler != nil {
  21. 		var err error
  22. 		stoppedCh, err = s.SecureServingInfo.Serve(s.Handler, s.ShutdownTimeout, internalStopCh)
  23. 		if err != nil {
  24. 			close(internalStopCh)
  25. 			close(auditStopCh)
  26. 			return nil, err
  27. 		}
  28. 	}
  29.  
  30. 	// Now that listener have bound successfully, it is the
  31. 	// responsibility of the caller to close the provided channel to
  32. 	// ensure cleanup.
  33. 	go func() {
  34. 		<-stopCh
  35. 		close(internalStopCh)
  36. 		if stoppedCh != nil {
  37. 			<-stoppedCh
  38. 		}
  39. 		s.HandlerChainWaitGroup.Wait()
  40. 		close(auditStopCh)
  41. 	}()
  42.  
  43. 	s.RunPostStartHooks(stopCh)
  44.  
  45. 	if _, err := systemd.SdNotify(true, "READY=1\n"); err != nil {
  46. 		klog.Errorf("Unable to send systemd daemon successful start message: %v\n", err)
  47. 	}
  48.  
  49. 	return stoppedCh, nil
  50. }

启动http Server 方法入口： staging/src/k8s.io/apiserver/pkg/server/secure_serving.go:147 Serve()方法，源码如下所示。该方法中最后执行RunServer()方法启动http server监听请求服务，并进行相应的资源处理。

  1. // Serve runs the secure http server. It fails only if certificates cannot be loaded or the initial listen call fails.
  2. // The actual server loop (stoppable by closing stopCh) runs in a go routine, i.e. Serve does not block.
  3. // It returns a stoppedCh that is closed when all non-hijacked active requests have been processed.
  4. func (s *SecureServingInfo) Serve(handler http.Handler, shutdownTimeout time.Duration, stopCh <-chan struct{}) (<-chan struct{}, error) {
  5. 	if s.Listener == nil {
  6. 		return nil, fmt.Errorf("listener must not be nil")
  7. 	}
  8.  
  9. 	tlsConfig, err := s.tlsConfig(stopCh)
  10. 	if err != nil {
  11. 		return nil, err
  12. 	}
  13.  
  14. 	secureServer := &http.Server{
  15. 		Addr:           s.Listener.Addr().String(),
  16. 		Handler:        handler,
  17. 		MaxHeaderBytes: 1 << 20,
  18. 		TLSConfig:      tlsConfig,
  19. 	}
  20.  
  21. 	// At least 99% of serialized resources in surveyed clusters were smaller than 256kb.
  22. 	// This should be big enough to accommodate most API POST requests in a single frame,
  23. 	// and small enough to allow a per connection buffer of this size multiplied by `MaxConcurrentStreams`.
  24. 	const resourceBody99Percentile = 256 * 1024
  25.  
  26. 	http2Options := &http2.Server{}
  27.  
  28. 	// shrink the per-stream buffer and max framesize from the 1MB default while still accommodating most API POST requests in a single frame
  29. 	http2Options.MaxUploadBufferPerStream = resourceBody99Percentile
  30. 	http2Options.MaxReadFrameSize = resourceBody99Percentile
  31.  
  32. 	// use the overridden concurrent streams setting or make the default of 250 explicit so we can size MaxUploadBufferPerConnection appropriately
  33. 	if s.HTTP2MaxStreamsPerConnection > 0 {
  34. 		http2Options.MaxConcurrentStreams = uint32(s.HTTP2MaxStreamsPerConnection)
  35. 	} else {
  36. 		http2Options.MaxConcurrentStreams = 250
  37. 	}
  38.  
  39. 	// increase the connection buffer size from the 1MB default to handle the specified number of concurrent streams
  40. 	http2Options.MaxUploadBufferPerConnection = http2Options.MaxUploadBufferPerStream * int32(http2Options.MaxConcurrentStreams)
  41.  
  42. 	if !s.DisableHTTP2 {
  43. 		// apply settings to the server
  44. 		if err := http2.ConfigureServer(secureServer, http2Options); err != nil {
  45. 			return nil, fmt.Errorf("error configuring http2: %v", err)
  46. 		}
  47. 	}
  48.  
  49. 	// use tlsHandshakeErrorWriter to handle messages of tls handshake error
  50. 	tlsErrorWriter := &tlsHandshakeErrorWriter{os.Stderr}
  51. 	tlsErrorLogger := log.New(tlsErrorWriter, "", 0)
  52. 	secureServer.ErrorLog = tlsErrorLogger
  53.  
  54. 	klog.Infof("Serving securely on %s", secureServer.Addr)
  55. 	return RunServer(secureServer, s.Listener, shutdownTimeout, stopCh)
  56. }

3.4 各Rest请求处理的Handler构建

在上面3.2节中的Run方法中，通过CreateServerChain方法构建ApiServer服务请求的整套体系，该方法中调用CreateKubeAPIServer()方法，该方法中kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer)对New方法的调用，然后对s, err := c.GenericConfig.New("kube-apiserver", delegationTarget)方法调用，在NewAPIServerHandler方法中创建了路由关系的处理，即：gorestfulContainer.Router(restful.CurlyRouter{}) // e.g. for proxy/{kind}/{name}/{*}

具体源码见：vendor/k8s.io/apiserver/pkg/server/handler.go:73 NewAPIServerHandler() 如下：

  1. cmd/kube-apiserver/app/server.go:228
  2.  
  3. // CreateKubeAPIServer creates and wires a workable kube-apiserver
  4. func CreateKubeAPIServer(kubeAPIServerConfig *controlplane.Config, delegateAPIServer genericapiserver.DelegationTarget) (*controlplane.Instance, error) {
  5. 	kubeAPIServer, err := kubeAPIServerConfig.Complete().New(delegateAPIServer)
  6. 	if err != nil {
  7. 		return nil, err
  8. 	}
  9.  
  10. 	return kubeAPIServer, nil
  11. }
  12.  
  13. vendor/k8s.io/apiserver/pkg/server/config.go:522
  14. / New creates a new server which logically combines the handling chain with the passed server.
  15. // name is used to differentiate for logging. The handler chain in particular can be difficult as it starts delgating.
  16. // delegationTarget may not be nil.
  17. func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*GenericAPIServer, error) {
  18. ..........
  19. apiServerHandler := NewAPIServerHandler(name, c.Serializer, handlerChainBuilder, delegationTarget.UnprotectedHandler())
  20. ..........
  21. }
  22.  
  23. vendor/k8s.io/apiserver/pkg/server/handler.go:73
  24. func NewAPIServerHandler(name string, s runtime.NegotiatedSerializer, handlerChainBuilder HandlerChainBuilderFn, notFoundHandler http.Handler) *APIServerHandler {
  25. 	nonGoRestfulMux := mux.NewPathRecorderMux(name)
  26. 	if notFoundHandler != nil {
  27. 		nonGoRestfulMux.NotFoundHandler(notFoundHandler)
  28. 	}
  29.  
  30. 	gorestfulContainer := restful.NewContainer()
  31. 	gorestfulContainer.ServeMux = http.NewServeMux()
  32. 	gorestfulContainer.Router(restful.CurlyRouter{}) // e.g. for proxy/{kind}/{name}/{*}
  33. 	gorestfulContainer.RecoverHandler(func(panicReason interface{}, httpWriter http.ResponseWriter) {
  34. 		logStackOnRecover(s, panicReason, httpWriter)
  35. 	})
  36. 	gorestfulContainer.ServiceErrorHandler(func(serviceErr restful.ServiceError, request *restful.Request, response *restful.Response) {
  37. 		serviceErrorHandler(s, serviceErr, request, response)
  38. 	})
  39.  
  40. 	director := director{
  41. 		name:               name,
  42. 		goRestfulContainer: gorestfulContainer,
  43. 		nonGoRestfulMux:    nonGoRestfulMux,
  44. 	}
  45.  
  46. 	return &APIServerHandler{
  47. 		FullHandlerChain:   handlerChainBuilder(director),
  48. 		GoRestfulContainer: gorestfulContainer,
  49. 		NonGoRestfulMux:    nonGoRestfulMux,
  50. 		Director:           director,
  51. 	}
  52. }

关于各个Rest的处理，依赖各个WebService服务进行处理，通过循环匹配path和webservice对象，实现服务之间的映射匹配，选择匹配最成功的一个进行相应的request的处理，具体的参考源码vendor/github.com/emicklei/go-restful/curly.go:19 CurlyRouter.SelectRoute()方法，具体源码如下：

  1. vendor/github.com/emicklei/go-restful/curly.go:19
  2.  
  3. // SelectRoute is part of the Router interface and returns the best match
  4. // for the WebService and its Route for the given Request.
  5. func (c CurlyRouter) SelectRoute(
  6. 	webServices []*WebService,
  7. 	httpRequest *http.Request) (selectedService *WebService, selected *Route, err error) {
  8.  
  9. 	requestTokens := tokenizePath(httpRequest.URL.Path)
  10.  
  11. 	detectedService := c.detectWebService(requestTokens, webServices)
  12. 	if detectedService == nil {
  13. 		if trace {
  14. 			traceLogger.Printf("no WebService was found to match URL path:%s\n", httpRequest.URL.Path)
  15. 		}
  16. 		return nil, nil, NewError(http.StatusNotFound, "404: Page Not Found")
  17. 	}
  18. 	candidateRoutes := c.selectRoutes(detectedService, requestTokens)
  19. 	if len(candidateRoutes) == 0 {
  20. 		if trace {
  21. 			traceLogger.Printf("no Route in WebService with path %s was found to match URL path:%s\n", detectedService.rootPath, httpRequest.URL.Path)
  22. 		}
  23. 		return detectedService, nil, NewError(http.StatusNotFound, "404: Page Not Found")
  24. 	}
  25. 	selectedRoute, err := c.detectRoute(candidateRoutes, httpRequest)
  26. 	if selectedRoute == nil {
  27. 		return detectedService, nil, err
  28. 	}
  29. 	return detectedService, selectedRoute, nil
  30. }
  31.  
  32.  
  33. vendor/github.com/emicklei/go-restful/curly.go:122
  34. // detectWebService returns the best matching webService given the list of path tokens.
  35. // see also computeWebserviceScore
  36. func (c CurlyRouter) detectWebService(requestTokens []string, webServices []*WebService) *WebService {
  37. 	var best *WebService
  38. 	score := -1
  39. 	for _, each := range webServices {
  40. 		matches, eachScore := c.computeWebserviceScore(requestTokens, each.pathExpr.tokens)
  41. 		if matches && (eachScore > score) {
  42. 			best = each
  43. 			score = eachScore
  44. 		}
  45. 	}
  46. 	return best
  47. }

关于WebService与Container的关系，参考源码vendor/github.com/emicklei/go-restful/container.go:88 Add()方法实现将WebService添加到Container对象中。具体的关于服务启动的时候，何时添加服务到Container中，有待进一步分析相关源码。关于WebService struct源码中描述如下： WebService holds a collection of Route values that bind a Http Method + URL Path to a function.

  1. vendor/github.com/emicklei/go-restful/container.go:88
  2. // Add a WebService to the Container. It will detect duplicate root paths and exit in that case.
  3. func (c *Container) Add(service *WebService) *Container {
  4. 	c.webServicesLock.Lock()
  5. 	defer c.webServicesLock.Unlock()
  6.  
  7. 	// if rootPath was not set then lazy initialize it
  8. 	if len(service.rootPath) == 0 {
  9. 		service.Path("/")
  10. 	}
  11.  
  12. 	// cannot have duplicate root paths
  13. 	for _, each := range c.webServices {
  14. 		if each.RootPath() == service.RootPath() {
  15. 			log.Printf("WebService with duplicate root path detected:['%v']", each)
  16. 			os.Exit(1)
  17. 		}
  18. 	}
  19.  
  20. 	// If not registered on root then add specific mapping
  21. 	if !c.isRegisteredOnRoot {
  22. 		c.isRegisteredOnRoot = c.addHandler(service, c.ServeMux)
  23. 	}
  24. 	c.webServices = append(c.webServices, service)
  25. 	return c
  26. }

pkg/kubelet/cri/streaming/server.go:108 NewServer

4.总结

上述章节，以kubectl create为例，阐述了通过kubectl命令行，创建Command命令，然后通过client-go的rest方法发送请求到服务端的关键流程节点；然后描述了服务端apiserver的启动流程，以及WebService服务注册以及针对http request的路由处理链路，从而实现了端到端的整个流程的流转。关于WebService的构建以及后续的服务逻辑的处理，后续在逐步研究分享。

 

5.参考文章

https://blog.csdn.net/u014458692/article/details/108068088

https://kubernetes.io/zh/docs/concepts/architecture/cloud-controller/

https://www.cnblogs.com/waken-captain/p/10509705.html

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/671830
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号