kubectl源码分析之port-forward

 欢迎关注我的公众号：

 目前刚开始写一个月，一共写了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 PortForwardOptions struct {//port-forward结构体
  Namespace     string
  PodName       string
  RESTClient    *restclient.RESTClient
  Config        *restclient.Config
  PodClient     corev1client.PodsGetter
  Address       []string
  Ports         []string
  PortForwarder portForwarder
  StopChannel   chan struct{}
  ReadyChannel  chan struct{}
}

//创建port-forward命令
func NewCmdPortForward(f cmdutil.Factory, streams genericclioptions.IOStreams) *cobra.Command {
  opts := &PortForwardOptions{//初始化结构体
    PortForwarder: &defaultPortForwarder{
      IOStreams: streams,
    },
  }
  cmd := &cobra.Command{//创建cobra命令
    Use:                   "port-forward TYPE/NAME [options] [LOCAL_PORT:]REMOTE_PORT [...[LOCAL_PORT_N:]REMOTE_PORT_N]",
    DisableFlagsInUseLine: true,
    Short:                 i18n.T("Forward one or more local ports to a pod"),
    Long:                  portforwardLong,
    Example:               portforwardExample,
    Run: func(cmd *cobra.Command, args []string) {
      if err := opts.Complete(f, cmd, args); err != nil {//先准备
        cmdutil.CheckErr(err)
      }
      if err := opts.Validate(); err != nil {//校验
        cmdutil.CheckErr(cmdutil.UsageErrorf(cmd, "%v", err.Error()))
      }
      if err := opts.RunPortForward(); err != nil {//运行
        cmdutil.CheckErr(err)
      }
    },
  }
  cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodPortForwardWaitTimeout)//pod-running-timeout选项
  cmd.Flags().StringSliceVar(&opts.Address, "address", []string{"localhost"}, "Addresses to listen on (comma separated). Only accepts IP addresses or localhost as a value. When localhost is supplied, kubectl will try to bind on both 127.0.0.1 and ::1 and will fail if neither of these addresses are available to bind.")//address选项
  // TODO support UID
  return cmd
}

//准备函数
func (o *PortForwardOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []string) error {
  var err error
  if len(args) < 2 {//参数小于2个报错
    return cmdutil.UsageErrorf(cmd, "TYPE/NAME and list of ports are required for port-forward")
  }

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

  builder := f.NewBuilder().
    WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
    ContinueOnError().
    NamespaceParam(o.Namespace).DefaultNamespace()//构造result对象

  getPodTimeout, err := cmdutil.GetPodRunningTimeoutFlag(cmd)//获取pod-running-timeout选项
  if err != nil {
    return cmdutil.UsageErrorf(cmd, err.Error())
  }

  resourceName := args[0]//获取资源
  builder.ResourceNames("pods", resourceName)

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

  forwardablePod, err := polymorphichelpers.AttachablePodForObjectFn(f, obj, getPodTimeout)//从obj获取forward的pod
  if err != nil {
    return err
  }

  o.PodName = forwardablePod.Name//设置pod名称

  // handle service port mapping to target port if needed
  switch t := obj.(type) {//断言obj类型
  case *corev1.Service://如果是service
    o.Ports, err = translateServicePortToTargetPort(args[1:], *t, *forwardablePod)//转换service的port
    if err != nil {
      return err
    }
  default://默认
    o.Ports, err = convertPodNamedPortToNumber(args[1:], *forwardablePod)//转换pod的port
    if err != nil {
      return err
    }
  }

  clientset, err := f.KubernetesClientSet()//获取clientset
  if err != nil {
    return err
  }

  o.PodClient = clientset.CoreV1()//从clientset获取podclient

  o.Config, err = f.ToRESTConfig()//设置restconfig
  if err != nil {
    return err
  }
  o.RESTClient, err = f.RESTClient()//设置restClient
  if err != nil {
    return err
  }

  o.StopChannel = make(chan struct{}, 1)//定义通道
  o.ReadyChannel = make(chan struct{})
  return nil
}

//校验
func (o PortForwardOptions) Validate() error {
  if len(o.PodName) == 0 {//pod名称不能为空
    return fmt.Errorf("pod name or resource type/name must be specified")
  }

  if len(o.Ports) < 1 {//端口不能小于个
    return fmt.Errorf("at least 1 PORT is required for port-forward")
  }

  if o.PortForwarder == nil || o.PodClient == nil || o.RESTClient == nil || o.Config == nil {//这四个对象不能为空
    return fmt.Errorf("client, client config, restClient, and portforwarder must be provided")
  }
  return nil
}

//运行
func (o PortForwardOptions) RunPortForward() error {
  pod, err := o.PodClient.Pods(o.Namespace).Get(o.PodName, metav1.GetOptions{})//获取pod
  if err != nil {
    return err
  }

  if pod.Status.Phase != corev1.PodRunning {//pod状态不为running返回错误
    return fmt.Errorf("unable to forward port because pod is not running. Current status=%v", pod.Status.Phase)
  }

  signals := make(chan os.Signal, 1)
  signal.Notify(signals, os.Interrupt)
  defer signal.Stop(signals)

  go func() {
    <-signals
    if o.StopChannel != nil {
      close(o.StopChannel)
    }
  }()

  req := o.RESTClient.Post().
    Resource("pods").
    Namespace(o.Namespace).
    Name(pod.Name).
    SubResource("portforward")//通过restclient的post方法后去request

  return o.PortForwarder.ForwardPorts("POST", req.URL(), o)//执行forward
}

//执行forward
func (f *defaultPortForwarder) ForwardPorts(method string, url *url.URL, opts PortForwardOptions) error {
  transport, upgrader, err := spdy.RoundTripperFor(opts.Config)//获取transport
  if err != nil {
    return err
  }
  dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, method, url)//获取dialer
  fw, err := portforward.NewOnAddresses(dialer, opts.Address, opts.Ports, opts.StopChannel, opts.ReadyChannel, f.Out, f.ErrOut)//构造forwarder
  if err != nil {
    return err
  }
  return fw.ForwardPorts()//执行forward
}

//拆分port
func splitPort(port string) (local, remote string) {
  parts := strings.Split(port, ":")//用冒号分割
  if len(parts) == 2 {//有两个
    return parts[0], parts[1]
  }

  return parts[0], parts[0]
}

//翻译svc port到target port
func translateServicePortToTargetPort(ports []string, svc corev1.Service, pod corev1.Pod) ([]string, error) {
  var translated []string
  for _, port := range ports {//遍历ports
    localPort, remotePort := splitPort(port)//分割port

    portnum, err := strconv.Atoi(remotePort)//把远程port转为整数
    if err != nil {//如果转换失败
      svcPort, err := util.LookupServicePortNumberByName(svc, remotePort)//通过名称查找service的port
      if err != nil {
        return nil, err
      }
      portnum = int(svcPort)//把svcPort转整数

      if localPort == remotePort {//如果本地port和远程port相等
        localPort = strconv.Itoa(portnum)
      }
    }
    containerPort, err := util.LookupContainerPortNumberByServicePort(svc, pod, int32(portnum))//后去svc的target-port
    if err != nil {
      // can't resolve a named port, or Service did not declare this port, return an error
      return nil, err
    }

    if int32(portnum) != containerPort {//如果容器的port的贝蒂svc的port不同则，
      translated = append(translated, fmt.Sprintf("%s:%d", localPort, containerPort))
    } else {
      translated = append(translated, port)
    }
  }
  return translated, nil
}

func convertPodNamedPortToNumber(ports []string, pod corev1.Pod) ([]string, error) {
  var converted []string
  for _, port := range ports {//遍历ports
    localPort, remotePort := splitPort(port)//分割port

    containerPortStr := remotePort
    _, err := strconv.Atoi(remotePort)//转换remoteport为整形
    if err != nil {
      containerPort, err := util.LookupContainerPortNumberByName(pod, remotePort)//从pod里查找port
      if err != nil {
        return nil, err
      }

      containerPortStr = strconv.Itoa(int(containerPort))
    }

    if localPort != remotePort {//如果本地port不等于远程port
      converted = append(converted, fmt.Sprintf("%s:%s", localPort, containerPortStr))
    } else {
      converted = append(converted, containerPortStr)
    }
  }

  return converted, nil
}