Command

Etcd 使用命令方式创建可执行功能，在 Gateway 部分中，首先创建一个主命令，然后再向主命令下挂接子命令（第 6 行），如下所示。

  1. func newGatewayCommand() *cobra.Command {
  2.     lpc := &cobra.Command{
  3.         Use:   "gateway <subcommand>",
  4.         Short: "gateway related command",
  5.     }
  6.     lpc.AddCommand(newGatewayStartCommand())
  8.     return lpc
  9. }

子命令也是一个 Command 实例，其中的 Run 方法是该命令执行时的关键方法，第 8 ～ 16 行用于处理与子命令相关的参数。

  1. func newGatewayStartCommand() *cobra.Command {
  2.     cmd := cobra.Command{
  3.         Use:   "start",
  4.         Short: "start the gateway",
  5.         Run:   startGateway,
  6.     }
  8.     cmd.Flags().StringVar(&gatewayListenAddr, "listen-addr", "127.0.0.1:23790", "listen address")
  9.     cmd.Flags().StringVar(&gatewayDNSCluster, "discovery-srv", "", "DNS domain used to bootstrap initial cluster")
  10.     cmd.Flags().StringVar(&gatewayDNSClusterServiceName, "discovery-srv-name", "", "service name to query when using DNS discovery")
  11.     cmd.Flags().BoolVar(&gatewayInsecureDiscovery, "insecure-discovery", false, "accept insecure SRV records")
  12.     cmd.Flags().StringVar(&gatewayCA, "trusted-ca-file", "", "path to the client server TLS CA file for verifying the discovered endpoints when discovery-srv is provided.")
  14.     cmd.Flags().StringSliceVar(&gatewayEndpoints, "endpoints", []string{"127.0.0.1:2379"}, "comma separated etcd cluster endpoints")
  16.     cmd.Flags().DurationVar(&gatewayRetryDelay, "retry-delay", time.Minute, "duration of delay before retrying failed endpoints")
  18.     return &cmd
  19. }

Execution

Gateway 命令的执行部分为 startGateway。大致过程如下，首先获取目标节点（第 11 行）；然后对获得的目标地址进行处理（第 17 ～ 29 行）；接下来获取 Gateway 监听地址信息并进行处理（第 31 ～ 64 行）；最后启动 TCPProxy。

  1. func startGateway(cmd *cobra.Command, args []string) {
  2.     lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
  3.     if err != nil {
  4.         fmt.Fprintln(os.Stderr, err)
  5.         os.Exit(1)
  6.     }
  8.     // We use os.Args to show all the arguments (not only passed-through Cobra).
  9.     lg.Info("Running: ", zap.Strings("args", os.Args))
  11.     srvs := discoverEndpoints(lg, gatewayDNSCluster, gatewayCA, gatewayInsecureDiscovery, gatewayDNSClusterServiceName)
  12.     if len(srvs.Endpoints) == 0 {
  13.         // no endpoints discovered, fall back to provided endpoints
  14.         srvs.Endpoints = gatewayEndpoints
  15.     }
  16.     // Strip the schema from the endpoints because we start just a TCP proxy
  17.     srvs.Endpoints = stripSchema(srvs.Endpoints)
  18.     if len(srvs.SRVs) == 0 {
  19.         for _, ep := range srvs.Endpoints {
  20.             h, p, serr := net.SplitHostPort(ep)
  21.             if serr != nil {
  22.                 fmt.Printf("error parsing endpoint %q", ep)
  23.                 os.Exit(1)
  24.             }
  25.             var port uint16
  26.             fmt.Sscanf(p, "%d", &port)
  27.             srvs.SRVs = append(srvs.SRVs, &net.SRV{Target: h, Port: port})
  28.         }
  29.     }
  31.     lhost, lport, err := net.SplitHostPort(gatewayListenAddr)
  32.     if err != nil {
  33.         fmt.Println("failed to validate listen address:", gatewayListenAddr)
  34.         os.Exit(1)
  35.     }
  37.     laddrs, err := net.LookupHost(lhost)
  38.     if err != nil {
  39.         fmt.Println("failed to resolve listen host:", lhost)
  40.         os.Exit(1)
  41.     }
  42.     laddrsMap := make(map[string]bool)
  43.     for _, addr := range laddrs {
  44.         laddrsMap[addr] = true
  45.     }
  47.     for _, srv := range srvs.SRVs {
  48.         var eaddrs []string
  49.         eaddrs, err = net.LookupHost(srv.Target)
  50.         if err != nil {
  51.             fmt.Println("failed to resolve endpoint host:", srv.Target)
  52.             os.Exit(1)
  53.         }
  54.         if fmt.Sprintf("%d", srv.Port) != lport {
  55.             continue
  56.         }
  58.         for _, ea := range eaddrs {
  59.             if laddrsMap[ea] {
  60.                 fmt.Printf("SRV or endpoint (%s:%d->%s:%d) should not resolve to the gateway listen addr (%s)\n", srv.Target, srv.Port, ea, srv.Port, gatewayListenAddr)
  61.                 os.Exit(1)
  62.             }
  63.         }
  64.     }
  66.     if len(srvs.Endpoints) == 0 {
  67.         fmt.Println("no endpoints found")
  68.         os.Exit(1)
  69.     }
  71.     var l net.Listener
  72.     l, err = net.Listen("tcp", gatewayListenAddr)
  73.     if err != nil {
  74.         fmt.Fprintln(os.Stderr, err)
  75.         os.Exit(1)
  76.     }
  78.     tp := tcpproxy.TCPProxy{
  79.         Logger:          lg,
  80.         Listener:        l,
  81.         Endpoints:       srvs.SRVs,
  82.         MonitorInterval: gatewayRetryDelay,
  83.     }
  85.     // At this point, etcd gateway listener is initialized
  86.     notifySystemd(lg)
  88.     tp.Run()
  89. }

Proxy

TCPProxy 通过 Run 方法启动，主要工作是创建一个 Monitor 协程，及监听 Listener 实例的新连接请求，监听到新连接后通过 serve 协程独立处理。

  1. func (tp *TCPProxy) Run() error {
  2.     tp.donec = make(chan struct{})
  3.     if tp.MonitorInterval == 0 {
  4.         tp.MonitorInterval = 5 * time.Minute
  5.     }
  6.     for _, srv := range tp.Endpoints {
  7.         addr := formatAddr(srv.Target, srv.Port)
  8.         tp.remotes = append(tp.remotes, &remote{srv: srv, addr: addr})
  9.     }
  11.     eps := []string{}
  12.     for _, ep := range tp.Endpoints {
  13.         eps = append(eps, fmt.Sprintf("%s:%d", ep.Target, ep.Port))
  14.     }
  15.     if tp.Logger != nil {
  16.         tp.Logger.Info("ready to proxy client requests", zap.Strings("endpoints", eps))
  17.     }
  19.     go tp.runMonitor()
  20.     for {
  21.         in, err := tp.Listener.Accept()
  22.         if err != nil {
  23.             return err
  24.         }
  26.         go tp.serve(in)
  27.     }
  28. }

Serve Requests

代理请求处理过程非常简单，首先通过 pick 方法获取一个目标端，并创建到目标端的 TCP 连接，如果目标端不可达，那么标记该目标端为未激活状态（第 19 行）。如果目标端可以到达，那么通过一个从请求到目标的协程（第 30 ～ 34 行）与自己所在协程的目标到请求端（第 36 ～ 38 行）双协程构建一个双全工通道。

  1. func (tp *TCPProxy) serve(in net.Conn) {
  2.     var (
  3.         err error
  4.         out net.Conn
  5.     )
  7.     for {
  8.         tp.mu.Lock()
  9.         remote := tp.pick()
  10.         tp.mu.Unlock()
  11.         if remote == nil {
  12.             break
  13.         }
  14.         // TODO: add timeout
  15.         out, err = net.Dial("tcp", remote.addr)
  16.         if err == nil {
  17.             break
  18.         }
  19.         remote.inactivate()
  20.         if tp.Logger != nil {
  21.             tp.Logger.Warn("deactivated endpoint", zap.String("address", remote.addr), zap.Duration("interval", tp.MonitorInterval), zap.Error(err))
  22.         }
  23.     }
  25.     if out == nil {
  26.         in.Close()
  27.         return
  28.     }
  30.     go func() {
  31.         io.Copy(in, out)
  32.         in.Close()
  33.         out.Close()
  34.     }()
  36.     io.Copy(out, in)
  37.     out.Close()
  38.     in.Close()
  39. }

Monitor

监控协程会在固定时间间隔后触发，遍历全部的远端节点，并尝试检查未激活的远端节点是否已经恢复可用。

  1. func (tp *TCPProxy) runMonitor() {
  2.     for {
  3.         select {
  4.         case <-time.After(tp.MonitorInterval):
  5.             tp.mu.Lock()
  6.             for _, rem := range tp.remotes {
  7.                 if rem.isActive() {
  8.                     continue
  9.                 }
  10.                 go func(r *remote) {
  11.                     if err := r.tryReactivate(); err != nil {
  12.                         if tp.Logger != nil {
  13.                             tp.Logger.Warn("failed to activate endpoint (stay inactive for another interval)", zap.String("address", r.addr), zap.Duration("interval", tp.MonitorInterval), zap.Error(err))
  14.                         }
  15.                     } else {
  16.                         if tp.Logger != nil {
  17.                             tp.Logger.Info("activated", zap.String("address", r.addr))
  18.                         }
  19.                     }
  20.                 }(rem)
  21.             }
  22.             tp.mu.Unlock()
  23.         case <-tp.donec:
  24.             return
  25.         }
  26.     }
  27. }

检查过程非常简单，只是简单进行连接。

  1. func (r *remote) tryReactivate() error {
  2.     conn, err := net.Dial("tcp", r.addr)
  3.     if err != nil {
  4.         return err
  5.     }
  6.     conn.Close()
  7.     r.mu.Lock()
  8.     defer r.mu.Unlock()
  9.     r.inactive = false
  10.     return nil
  11. }

Pick a Remote

根据远端是否可达，优先级及权重综合选择一个合适的远端节点。如果远端目标不可达，不选择（第 10 行）；第 11 ～ 24 行对优先级进行选择，注意，通过这个方法获得的 weighted、unweighted 两个切片中的远端节点优先级绝对值是递减的。

  1. func (tp *TCPProxy) pick() *remote {
  2.     var weighted []*remote
  3.     var unweighted []*remote
  5.     bestPr := uint16(65535)
  6.     w := 0
  7.     // find best priority class
  8.     for _, r := range tp.remotes {
  9.         switch {
  10.         case !r.isActive():
  11.         case r.srv.Priority < bestPr:
  12.             bestPr = r.srv.Priority
  13.             w = 0
  14.             weighted = nil
  15.             unweighted = nil
  16.             fallthrough
  17.         case r.srv.Priority == bestPr:
  18.             if r.srv.Weight > 0 {
  19.                 weighted = append(weighted, r)
  20.                 w += int(r.srv.Weight)
  21.             } else {
  22.                 unweighted = append(unweighted, r)
  23.             }
  24.         }
  25.     }
  26.     if weighted != nil {
  27.         if len(unweighted) > 0 && rand.Intn(100) == 1 {
  28.             // In the presence of records containing weights greater
  29.             // than 0, records with weight 0 should have a very small
  30.             // chance of being selected.
  31.             r := unweighted[tp.pickCount%len(unweighted)]
  32.             tp.pickCount++
  33.             return r
  34.         }
  35.         // choose a uniform random number between 0 and the sum computed
  36.         // (inclusive), and select the RR whose running sum value is the
  37.         // first in the selected order
  38.         choose := rand.Intn(w)
  39.         for i := 0; i < len(weighted); i++ {
  40.             choose -= int(weighted[i].srv.Weight)
  41.             if choose <= 0 {
  42.                 return weighted[i]
  43.             }
  44.         }
  45.     }
  46.     if unweighted != nil {
  47.         for i := 0; i < len(tp.remotes); i++ {
  48.             picked := tp.remotes[tp.pickCount%len(tp.remotes)]
  49.             tp.pickCount++
  50.             if picked.isActive() {
  51.                 return picked
  52.             }
  53.         }
  54.     }
  55.     return nil
  56. }