kubectl 源码分析之exec

 欢迎关注我的公众号：

 目前刚开始写一个月，一共写了18篇原创文章，文章目录如下：

​​istio多集群探秘，部署了50次多集群后我得出的结论​​

​​istio多集群链路追踪，附实操视频​​

​​istio防故障利器，你知道几个,istio新手不要读，太难！​​

​​istio业务权限控制，原来可以这么玩​​

​​istio实现非侵入压缩，微服务之间如何实现压缩​​

​​不懂envoyfilter也敢说精通istio系列-http-rbac-不要只会用AuthorizationPolicy配置权限​​

​​不懂envoyfilter也敢说精通istio系列-02-http-corsFilter-不要只会vs​​

​​不懂envoyfilter也敢说精通istio系列-03-http-csrf filter-再也不用再代码里写csrf逻辑了​​

​​不懂envoyfilter也敢说精通istio系列http-jwt_authn-不要只会RequestAuthorization​​

​​不懂envoyfilter也敢说精通istio系列-05-fault-filter-故障注入不止是vs​​

​​不懂envoyfilter也敢说精通istio系列-06-http-match-配置路由不只是vs​​

​​不懂envoyfilter也敢说精通istio系列-07-负载均衡配置不止是dr​​

​​不懂envoyfilter也敢说精通istio系列-08-连接池和断路器​​

​​不懂envoyfilter也敢说精通istio系列-09-http-route filter​​

​​不懂envoyfilter也敢说精通istio系列-network filter-redis proxy​​

​​不懂envoyfilter也敢说精通istio系列-network filter-HttpConnectionManager​​

​​不懂envoyfilter也敢说精通istio系列-ratelimit-istio ratelimit完全手册​​

————————————————

type ExecOptions struct {//exec结构体
  StreamOptions

  ResourceName string
  Command      []string

  ParentCommandName       string
  EnableSuggestedCmdUsage bool

  Builder          func() *resource.Builder
  ExecutablePodFn  polymorphichelpers.AttachablePodForObjectFunc
  restClientGetter genericclioptions.RESTClientGetter

  Pod           *corev1.Pod
  Executor      RemoteExecutor
  PodClient     coreclient.PodsGetter
  GetPodTimeout time.Duration
  Config        *restclient.Config
}

type StreamOptions struct {//stream结构体
  Namespace     string
  PodName       string
  ContainerName string
  Stdin         bool
  TTY           bool
  // minimize unnecessary output
  Quiet bool
  // InterruptParent, if set, is used to handle interrupts while attached
  InterruptParent *interrupt.Handler

  genericclioptions.IOStreams

  // for testing
  overrideStreams func() (io.ReadCloser, io.Writer, io.Writer)
  isTerminalIn    func(t term.TTY) bool
}

//创建exec命令
func NewCmdExec(f cmdutil.Factory, streams genericclioptions.IOStreams) *cobra.Command {
  options := &ExecOptions{//初始化结构体
    StreamOptions: StreamOptions{
      IOStreams: streams,
    },

    Executor: &DefaultRemoteExecutor{},
  }
  cmd := &cobra.Command{//创建cobra命令
    Use:                   "exec (POD | TYPE/NAME) [-c CONTAINER] [flags] -- COMMAND [args...]",
    DisableFlagsInUseLine: true,
    Short:                 i18n.T("Execute a command in a container"),
    Long:                  "Execute a command in a container.",
    Example:               execExample,
    Run: func(cmd *cobra.Command, args []string) {
      argsLenAtDash := cmd.ArgsLenAtDash()
      cmdutil.CheckErr(options.Complete(f, cmd, args, argsLenAtDash))//准备
      cmdutil.CheckErr(options.Validate())//校验
      cmdutil.CheckErr(options.Run())//运行
    },
  }
  cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodExecTimeout)//pod-running-timeout信息
  // TODO support UID
  cmd.Flags().StringVarP(&options.ContainerName, "container", "c", options.ContainerName, "Container name. If omitted, the first container in the pod will be chosen")// container选项
  cmd.Flags().BoolVarP(&options.Stdin, "stdin", "i", options.Stdin, "Pass stdin to the container")//-i选项
  cmd.Flags().BoolVarP(&options.TTY, "tty", "t", options.TTY, "Stdin is a TTY")//-t选项
  return cmd
}

//准备函数
func (p *ExecOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, argsIn []string, argsLenAtDash int) error {
  // Let kubectl exec follow rules for `--`, see #13004 issue
  if len(argsIn) == 0 || argsLenAtDash == 0 {//没有参数报错
    return cmdutil.UsageErrorf(cmd, execUsageStr)
  }

  p.ResourceName = argsIn[0]//设置资源名称
  p.Command = argsIn[1:]//设置运行命令

  var err error

  p.Namespace, _, err = f.ToRawKubeConfigLoader().Namespace()//设置namespace
  if err != nil {
    return err
  }

  p.ExecutablePodFn = polymorphichelpers.AttachablePodForObjectFn//设置查找pod函数

  p.GetPodTimeout, err = cmdutil.GetPodRunningTimeoutFlag(cmd)//设置pod-running-timeout选项
  if err != nil {
    return cmdutil.UsageErrorf(cmd, err.Error())
  }

  p.Builder = f.NewBuilder//设置builder
  p.restClientGetter = f//设置restClientGetter

  cmdParent := cmd.Parent()//获取父命令
  if cmdParent != nil {
    p.ParentCommandName = cmdParent.CommandPath()//设置父命令名称
  }
  if len(p.ParentCommandName) > 0 && cmdutil.IsSiblingCommandExists(cmd, "describe") {
    p.EnableSuggestedCmdUsage = true//设置是否提示用法
  }

  p.Config, err = f.ToRESTConfig()//设置restConfig
  if err != nil {
    return err
  }

  clientset, err := f.KubernetesClientSet()//设置clientset
  if err != nil {
    return err
  }
  p.PodClient = clientset.CoreV1()//设置PodClient

  return nil
}

//校验
func (p *ExecOptions) Validate() error {
  if len(p.PodName) == 0 && len(p.ResourceName) == 0 {//resourcename和podName不能同时为空
    return fmt.Errorf("pod or type/name must be specified")
  }
  if len(p.Command) == 0 {//命令不能为空
    return fmt.Errorf("you must specify at least one command for the container")
  }
  if p.Out == nil || p.ErrOut == nil {//Out和ErrOut必须都不为空
    return fmt.Errorf("both output and error output must be provided")
  }
  return nil
}

//运行
func (p *ExecOptions) Run() error {
  var err error
  // we still need legacy pod getter when PodName in ExecOptions struct is provided,
  // since there are any other command run this function by providing Podname with PodsGetter
  // and without resource builder, eg: `kubectl cp`.
  if len(p.PodName) != 0 {//pod名称不为空
    p.Pod, err = p.PodClient.Pods(p.Namespace).Get(p.PodName, metav1.GetOptions{})//通过podClient获取pod
    if err != nil {
      return err
    }
  } else {
    builder := p.Builder().
      WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
      NamespaceParam(p.Namespace).DefaultNamespace().ResourceNames("pods", p.ResourceName)//用构造器获取result对象

    obj, err := builder.Do().Object()//从result获取Object
    if err != nil {
      return err
    }

    p.Pod, err = p.ExecutablePodFn(p.restClientGetter, obj, p.GetPodTimeout)//从obj获取可执行的pod
    if err != nil {
      return err
    }
  }

  pod := p.Pod

  if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed {//如果pod为complete状态返回错误
    return fmt.Errorf("cannot exec into a container in a completed pod; current phase is %s", pod.Status.Phase)
  }

  containerName := p.ContainerName
  if len(containerName) == 0 {//指定了container
    if len(pod.Spec.Containers) > 1 {// 如果pod container大于1个，提示信息
      fmt.Fprintf(p.ErrOut, "Defaulting container name to %s.\n", pod.Spec.Containers[0].Name)
      if p.EnableSuggestedCmdUsage {
        fmt.Fprintf(p.ErrOut, "Use '%s describe pod/%s -n %s' to see all of the containers in this pod.\n", p.ParentCommandName, pod.Name, p.Namespace)
      }
    }
    containerName = pod.Spec.Containers[0].Name//容器名称为第0个容器名称
  }

  // ensure we can recover the terminal while attached
  t := p.SetupTTY()//设置tty

  var sizeQueue remotecommand.TerminalSizeQueue
  if t.Raw {
    // this call spawns a goroutine to monitor/update the terminal size
    sizeQueue = t.MonitorSize(t.GetSize())

    // unset p.Err if it was previously set because both stdout and stderr go over p.Out when tty is
    // true
    p.ErrOut = nil
  }

  fn := func() error {//exec函数
    restClient, err := restclient.RESTClientFor(p.Config)//获取client
    if err != nil {
      return err
    }

    // TODO: consider abstracting into a client invocation or client helper
    req := restClient.Post().
      Resource("pods").
      Name(pod.Name).
      Namespace(pod.Namespace).
      SubResource("exec")// 构造request对象
    req.VersionedParams(&corev1.PodExecOptions{
      Container: containerName,
      Command:   p.Command,
      Stdin:     p.Stdin,
      Stdout:    p.Out != nil,
      Stderr:    p.ErrOut != nil,
      TTY:       t.Raw,
    }, scheme.ParameterCodec)

    return p.Executor.Execute("POST", req.URL(), p.Config, p.In, p.Out, p.ErrOut, t.Raw, sizeQueue)//执行exec
  }

  if err := t.Safe(fn); err != nil {//执行t safe
    return err
  }

  return nil
}

//执行exec
func (*DefaultRemoteExecutor) Execute(method string, url *url.URL, config *restclient.Config, stdin io.Reader, stdout, stderr io.Writer, tty bool, terminalSizeQueue remotecommand.TerminalSizeQueue) error {
  exec, err := remotecommand.NewSPDYExecutor(config, method, url)//创建executer
  if err != nil {
    return err
  }
  return exec.Stream(remotecommand.StreamOptions{//执行stream
    Stdin:             stdin,
    Stdout:            stdout,
    Stderr:            stderr,
    Tty:               tty,
    TerminalSizeQueue: terminalSizeQueue,
  })
}