基于 Golang 实现的负载均衡网关：gobetween 分析（一）

type Config struct {
	Logging  LoggingConfig     `toml:"logging" json:"logging"`
	Api      ApiConfig         `toml:"api" json:"api"`
	Metrics  MetricsConfig     `toml:"metrics" json:"metrics"`
	Defaults ConnectionOptions `toml:"defaults" json:"defaults"`
	Acme     *AcmeConfig       `toml:"acme" json:"acme"`
	Profiler *ProfilerConfig   `toml:"profiler" json:"profiler"`
	Servers  map[string]Server `toml:"servers" json:"servers"`
}

type Server struct {
	ConnectionOptions

	// hostname:port
	Bind string `toml:"bind" json:"bind"`

	// tcp | udp | tls
	Protocol string `toml:"protocol" json:"protocol"`

	// weight | leastconn | roundrobin
	Balance string `toml:"balance" json:"balance"`

	// Optional configuration for server name indication
	Sni *Sni `toml:"sni" json:"sni"`

	// Optional configuration for protocol = tls
	Tls *Tls `toml:"tls" json:"tls"`

	// Optional configuration for backend_tls_enabled = true
	BackendsTls *BackendsTls `toml:"backends_tls" json:"backends_tls"`

	// Optional configuration for protocol = udp
	Udp *Udp `toml:"udp" json:"udp"`

	// Access configuration
	Access *AccessConfig `toml:"access" json:"access"`

	// ProxyProtocol configuration
	ProxyProtocol *ProxyProtocol `toml:"proxy_protocol" json:"proxy_protocol"`

	// Discovery configuration
	Discovery *DiscoveryConfig `toml:"discovery" json:"discovery"`

	// Healthcheck configuration
	Healthcheck *HealthcheckConfig `toml:"healthcheck" json:"healthcheck"`
}

/**
 * Balancer interface
 */
type Balancer interface {
	/**
	 * Elect backend based on Balancer implementation
	 */
	Elect(Context, []*Backend) (*Backend, error)
}

type Server interface {
	/**
	 * Start server
	 */
	Start() error

	/**
	 * Stop server and wait until it stop
	 */
	Stop()

	/**
	 * Get server configuration
	 */
	Cfg() config.Server
}

type ReadWriteCount struct {
	/* Read bytes count */
	CountRead uint

	/* Write bytes count */
	CountWrite uint

	Target Target
}

type Context interface {
	String() string
	Ip() net.IP
	Port() int
	Sni() string
}

/**
 * Proxy tcp context
 */
type TcpContext struct {
	Hostname string
	/**
	 * Current client connection
	 */
	Conn net.Conn
}

/**
 * Service is a global facility that could be Enabled or Disabled for a number
 * of core.Server instances, depending on their configration. See services/registry
 * for exact examples.
 */
type Service interface {
	/**
	 * Enable service for Server
	 */
	Enable(Server) error

	/**
	 * Disable service for Server
	 */
	Disable(Server) error
}

/**
 * Backend means upstream server
 * with all needed associate information
 */
type Backend struct {
	Target
	Priority int          `json:"priority"`
	Weight   int          `json:"weight"`
	Sni      string       `json:"sni,omitempty"`
	Stats    BackendStats `json:"stats"`
}

/**
 * Backend status
 */
type BackendStats struct {
	Live               bool   `json:"live"`
	Discovered         bool   `json:"discovered"`
	TotalConnections   int64  `json:"total_connections"`
	ActiveConnections  uint   `json:"active_connections"`
	RefusedConnections uint64 `json:"refused_connections"`
	RxBytes            uint64 `json:"rx"`
	TxBytes            uint64 `json:"tx"`
	RxSecond           uint   `json:"rx_second"`
	TxSecond           uint   `json:"tx_second"`
}

func main(){
    ...
    // Process flags and start
	cmd.Execute(func(cfg *config.Config) {
		// Configure logging
		logging.Configure(cfg.Logging.Output, cfg.Logging.Level, cfg.Logging.Format)
		// Start manager
		manager.Initialize(*cfg)
		/* setup metrics */
		metrics.Start((*cfg).Metrics)
		// Start API
		api.Start((*cfg).Api)
		// block forever
		<-(chan string)(nil)
    })
}

func Initialize(cfg config.Config) {
	......
	// Go through config and start servers for each server
	for name, serverCfg := range cfg.Servers {
		err := Create(name, serverCfg)
		if err != nil {
			log.Fatal(err)
		}
	}
	......
}

// 启动单个代理
func Create(name string, cfg config.Server) error {

	servers.Lock()
	defer servers.Unlock()

	if _, ok := servers.m[name]; ok {
		return errors.New("Server with this name already exists:" + name)
	}

	// 根据配置初始化 Server 结构
	c, err := prepareConfig(name, cfg, defaults)
	if err != nil {
		return err
	}

	// 初始化
	server, err := server.New(name, c)

	if err != nil {
		return err
	}

	for _, srv := range services {
		err = srv.Enable(server)
		if err != nil {
			return err
		}
	}

	// 启动代理
	if err = server.Start(); err != nil {
		return err
	}
	servers.m[name] = server
	return nil
}

type Server struct {
	/* Server friendly name */
	name string

	/* Listener */
	listener net.Listener

	/* Configuration */
	cfg config.Server

	/* Scheduler deals with discovery, balancing and healthchecks */
	scheduler scheduler.Scheduler		// 每一个代理结构都包含一个调度器 scheduler

	/* Current clients connection */
	clients map[string]net.Conn			//map 结构保持了 TCP 代理的所有连接（前置）

	/* Stats handler */
	statsHandler *stats.Handler			// 代理的统计回调

	/* ----- channels ----- */

	/* Channel for new connections */
	connect chan (*core.TcpContext)

	/* Channel for dropping connections or connectons to drop */
	disconnect chan (net.Conn)

	/* Stop channel */
	stop chan bool

	/* Tls config used to connect to backends */
	backendsTlsConfg *tls.Config

	/* Tls config used for incoming connections */
	tlsConfig *tls.Config

	/* Get certificate filled by external service */
	GetCertificate func(*tls.ClientHelloInfo) (*tls.Certificate, error)

	/* ----- modules ----- */
	/* Access module checks if client is allowed to connect */
	access *access.Access
}

// 初始化 Server 的方法：
func New(name string, cfg config.Server) (*Server, error) {
	......
	log := logging.For("server")

	var err error = nil
	statsHandler := stats.NewHandler(name)

	// Create server
	server := &Server{
		name:         name,
		cfg:          cfg,
		stop:         make(chan bool),
		disconnect:   make(chan net.Conn),
		connect:      make(chan *core.TcpContext),
		clients:      make(map[string]net.Conn),
		statsHandler: statsHandler,
		scheduler: scheduler.Scheduler{		// 初始化 scheduler
			Balancer:     balance.New(cfg.Sni, cfg.Balance),		// 负载均衡器
			Discovery:    discovery.New(cfg.Discovery.Kind, *cfg.Discovery),	// 服务发现
			Healthcheck:  healthcheck.New(cfg.Healthcheck.Kind, *cfg.Healthcheck),	// 健康检查
			StatsHandler: statsHandler,		// 状态上报 & 监控
		},
	}
	......
}

func (this *Server) Start() error {
	var err error
	this.tlsConfig, err = tlsutil.MakeTlsConfig(this.cfg.Tls, this.GetCertificate)
	if err != nil {
		return err
	}

	// 代理工作的核心循环
	go func() {
		for {
			select {
				// 处理连接断开的事件
			case client := <-this.disconnect:
				this.HandleClientDisconnect(client)
				// 处理新连接事件
			case ctx := <-this.connect:
				this.HandleClientConnect(ctx)
				// 处理代理退出事件
			case <-this.stop:
				this.scheduler.Stop()
				this.statsHandler.Stop()
				if this.listener != nil {
					this.listener.Close()
					for _, conn := range this.clients {
						conn.Close()
					}
				}
				this.clients = make(map[string]net.Conn)
				return
			}
		}
	}()

	// Start stats handler
	this.statsHandler.Start()

	// Start scheduler
	this.scheduler.Start()

	// Start listening
	if err := this.Listen(); err != nil {
		this.Stop()
		return err
	}

	return nil
}

func (this *Server) Listen() (err error) {

	log := logging.For("server.Listen")

	// create tcp listener
	this.listener, err = net.Listen("tcp", this.cfg.Bind)

	if err != nil {
		log.Error("Error starting", this.cfg.Protocol+"server:", err)
		return err
	}

	sniEnabled := this.cfg.Sni != nil

	// 在子 goroutine 中接受连接，不优雅（参考 fasthttp）
	go func() {
		for {
			conn, err := this.listener.Accept()

			if err != nil {
				log.Error(err)
				return
			}
			// 处理新连接
			go this.wrap(conn, sniEnabled)
		}
	}()

	return nil
}

// 处理新连接的方法
func (this *Server) wrap(conn net.Conn, sniEnabled bool) {
	log := logging.For("server.Listen.wrap")

	var hostname string
	var err error

	if sniEnabled {
		var sniConn net.Conn
		sniConn, hostname, err = sni.Sniff(conn, utils.ParseDurationOrDefault(this.cfg.Sni.ReadTimeout, time.Second*2))

		if err != nil {
			log.Error("Failed to get / parse ClientHello for sni:", err)
			conn.Close()
			return
		}

		conn = sniConn
	}

	if this.tlsConfig != nil {
		conn = tls.Server(conn, this.tlsConfig)
	}

	// 将新连接封装为 core.TcpContext 结构，放入 channel
	this.connect <- &core.TcpContext{
		hostname,
		conn,
	}
}

type Scheduler struct {
	/* Balancer impl */
	Balancer core.Balancer
	/* Discovery impl */
	Discovery *discovery.Discovery
	/* Healthcheck impl */
	Healthcheck *healthcheck.Healthcheck
	/* ----- backends ------*/
	/* Current cached backends map */
	backends map[core.Target]*core.Backend
	/* Stats */
	StatsHandler *stats.Handler
	/* ----- channels ----- */
	/* Backend operation channel */
	ops chan Op
	/* Stop channel */
	stop chan bool
	/* Elect backend channel */
	elect chan ElectRequest
}

func (this *Scheduler) Start() {

	log := logging.For("scheduler")

	log.Info("Starting scheduler", this.StatsHandler.Name)

	this.ops = make(chan Op)
	this.elect = make(chan ElectRequest)
	this.stop = make(chan bool)
	this.backends = make(map[core.Target]*core.Backend)

	this.Discovery.Start()
	this.Healthcheck.Start()

	// backends stats pusher ticker
	backendsPushTicker := time.NewTicker(2 * time.Second)

	/**
	 * Goroutine updates and manages backends
	 */
	go func() {
		for {
			select {

			/* ----- discovery ----- */

			// handle newly discovered backends
			case backends := <-this.Discovery.Discover():
				this.HandleBackendsUpdate(backends)
				this.Healthcheck.In <- this.Targets()
				this.StatsHandler.BackendsCounter.In <- this.Targets()

			/* ------ healthcheck ----- */

			// handle backend healthcheck result
			case checkResult := <-this.Healthcheck.Out:
				this.HandleBackendLiveChange(checkResult.Target, checkResult.Status == healthcheck.Healthy)

			/* ----- stats ----- */

			// push current backends to stats handler
			case <-backendsPushTicker.C:
				this.StatsHandler.Backends <- this.Backends()

			// handle new bandwidth stats of a backend
			case bs := <-this.StatsHandler.BackendsCounter.Out:
				this.HandleBackendStatsChange(bs.Target, &bs)

			/* ----- operations ----- */

			// handle backend operation
			case op := <-this.ops:
				this.HandleOp(op)

			// elect backend
			case electReq := <-this.elect:
				this.HandleBackendElect(electReq)

			/* ----- stop ----- */

			// handle scheduler stop
			case <-this.stop:
				log.Info("Stopping scheduler", this.StatsHandler.Name)
				backendsPushTicker.Stop()
				this.Discovery.Stop()
				this.Healthcheck.Stop()
				metrics.RemoveServer(fmt.Sprintf("%s", this.StatsHandler.Name), this.backends)
				return
			}
		}
	}()
}

func init() {
	registry["consul"] = NewConsulDiscovery
}

type Discovery struct {
	/**
	 * Cached backends
	 */
	backends *[]core.Backend	// 缓存从 consul-API 获取到的后端节点
	/**
	 * Function to fetch / discovery backends
	 */
	fetch FetchFunc				//consul-API 拉取后端节点的方法
	/**
	 * Options for fetch
	 */
	opts DiscoveryOpts			// 服务发现的配置选项
	/**
	 * Discovery configuration
	 */
	cfg config.DiscoveryConfig
	/**
	 * Channel where to push newly discovered backends
	 */
	out chan ([]core.Backend)	// 用于和 scheduler 模块通信的 channel，用于传送最新的 backends（后端节点）
	/**
	 * Channel for stopping discovery
	 */
	stop chan bool
}

func consulFetch(cfg config.DiscoveryConfig) (*[]core.Backend, error) {
	......
	scheme := "http"
	transport := &http.Transport{
		DisableKeepAlives: true,
	}

	// Enable tls if needed
	if cfg.ConsulTlsEnabled {
		tlsConfig := &consul.TLSConfig{
			Address:  cfg.ConsulHost,
			CertFile: cfg.ConsulTlsCertPath,
			KeyFile:  cfg.ConsulTlsKeyPath,
			CAFile:   cfg.ConsulTlsCacertPath,
		}
		tlsClientConfig, err := consul.SetupTLSConfig(tlsConfig)
		if err != nil {
			return nil, err
		}
		transport.TLSClientConfig = tlsClientConfig
		scheme = "https"
	}

	// Parse http timeout
	timeout := utils.ParseDurationOrDefault(cfg.Timeout, consulTimeout)

	// Create consul client
	client, _ := consul.NewClient(&consul.Config{
		Token:      cfg.ConsulAclToken,
		Scheme:     scheme,
		Address:    cfg.ConsulHost,
		Datacenter: cfg.ConsulDatacenter,
		HttpAuth: &consul.HttpBasicAuth{
			Username: cfg.ConsulAuthUsername,
			Password: cfg.ConsulAuthPassword,
		},
		HttpClient: &http.Client{Timeout: timeout, Transport: transport},
	})

	// Query service
	service, _, err := client.Health().Service(cfg.ConsulServiceName, cfg.ConsulServiceTag, cfg.ConsulServicePassingOnly, nil)
	if err != nil {
		return nil, err
	}

	// Gather backends
	backends := []core.Backend{}
	for _, entry := range service {
		s := entry.Service
		sni := ""

		for _, tag := range s.Tags {
			split := strings.SplitN(tag, "=", 2)

			if len(split) != 2 {
				continue
			}

			if split[0] != "sni" {
				continue
			}
			sni = split[1]
		}

		var host string
		if s.Address != "" {
			host = s.Address
		} else {
			host = entry.Node.Address
		}

		backends = append(backends, core.Backend{
			Target: core.Target{
				Host: host,
				Port: fmt.Sprintf("%v", s.Port),
			},
			Priority: 1,
			Weight:   1,
			Stats: core.BackendStats{
				Live: true,
			},
			Sni: sni,
		})
	}

	return &backends, nil
}

var typeRegistry = make(map[string]reflect.Type)

func init() {
	typeRegistry["roundrobin"] = reflect.TypeOf(RoundrobinBalancer{})
}

func New(sniConf *config.Sni, balance string) core.Balancer {
	balancer := reflect.New(typeRegistry[balance]).Elem().Addr().Interface().(core.Balancer)

	if sniConf == nil {
		return balancer
	}

	return &middleware.SniBalancer{
		SniConf:  sniConf,
		Delegate: balancer,
	}
}

/**
 * Roundrobin balancer
 */
type RoundrobinBalancer struct {

	/* Current backend position */
	current int
}

/**
 * Elect backend using roundrobin strategy
 */
func (b *RoundrobinBalancer) Elect(context core.Context, backends []*core.Backend) (*core.Backend, error) {

	if len(backends) == 0 {
		return nil, errors.New("Can't elect backend, Backends empty")
	}

	sort.SliceStable(backends, func(i, j int) bool {
		return backends[i].Target.String() < backends[j].Target.String()
	})

	if b.current >= len(backends) {
		b.current = 0
	}

	backend := backends[b.current]
	b.current += 1

	return backend, nil
}

type Healthcheck struct {
	/* Healthcheck function */
	check CheckFunc
	/* Healthcheck configuration */
	cfg config.HealthcheckConfig
	/* Input channel to accept targets */
	In chan []core.Target		// 与 scheduler 模块通信 channel，接收待探测的后端节点
	/* Output channel to send check results for individual target */
	Out chan CheckResult		// 与 scheduler 模块通信 channel，将拨测结果告知 scheduler
	/* Current check workers */
	workers []*Worker
	/* Channel to handle stop */
	stop chan bool
}

func init() {
	registry["ping"] = ping
	registry["probe"] = probe
	registry["exec"] = exec
	registry["none"] = nil
}

func New(strategy string, cfg config.HealthcheckConfig) *Healthcheck {
	// 根据 strategy 的名字获取对应的探测方法的指针
	check := registry[strategy]
	h := Healthcheck{
		check:   check,	// 初始化
		cfg:     cfg,
		In:      make(chan []core.Target),
		Out:     make(chan CheckResult),
		workers: []*Worker{},		// 初始化为空
		stop:    make(chan bool),
	}

	return &h
}

func (this *Healthcheck) Start() {

	go func() {
		for {
			select {

			/* got new targets */
			case targets := <-this.In:
				this.UpdateWorkers(targets)

			/* got stop requst */
			case <-this.stop:
				// Stop all workers
				for i := range this.workers {
					this.workers[i].Stop()
				}
				// And free it's memory
				this.workers = []*Worker{}

				return
			}
		}
	}()
}

func (this *Healthcheck) UpdateWorkers(targets []core.Target) {
	//result 为本次更新的 workerlist
	result := []*Worker{}
	// Keep or add needed workers
	for _, t := range targets {
		var keep *Worker
		// 遍历旧的 workers 列表
		for i := range this.workers {
			c := this.workers[i]
			if t.EqualTo(c.target) {
				// 还在使用的 worker
				keep = c
				break
			}
		}

		// 如果没找到针对 targets 的 worker，那么就新建一个 worker
		if keep == nil {
			keep = &Worker{
				target: t,
				stop:   make(chan bool),
				out:    this.Out,		//this.Out 为 healthycheck 的 channel
				cfg:    this.cfg,
				check:  this.check,
				LastResult: CheckResult{
					Status: Initial,
				},
			}
			// 启动 worker
			keep.Start()
		}
		// 将目前在使用的 worker 加入列表
		result = append(result, keep)
	}

	// Stop needed workers
	// 移除不再使用的 worker
	for i := range this.workers {
		c := this.workers[i]
		remove := true
		for _, t := range targets {
			if c.target.EqualTo(t) {
				remove = false
				break
			}
		}

		if remove {
			// 向 worker-goroutine 发送退出 channel
			c.Stop()
		}
	}

	// 更新 worker 列表
	this.workers = result
}

func (this *Healthcheck) Stop() {
	this.stop <- true
}

type Worker struct {
	/* Target to monitor and check */
	target core.Target
	/* Function that does actual check */
	check CheckFunc
	/* Channel to write changed check results */
	out chan<- CheckResult
	/* Healthcheck configuration */
	cfg config.HealthcheckConfig
	/* Stop channel to worker to stop */
	stop chan bool
	/* Last confirmed check result */
	LastResult CheckResult
	/* Current passes count, if LastResult.Live = true */
	passes int
	/* Current fails count, if LastResult.Live = false */
	fails int
}

/**
 * Start worker
 */
func (this *Worker) Start() {
	// Special case for no healthcheck, don't actually start worker
	if this.cfg.Kind == "none" {
		return
	}

	interval, _ := time.ParseDuration(this.cfg.Interval)

	ticker := time.NewTicker(interval)
	c := make(chan CheckResult, 1)

	go func() {
		/* Check health before any delay*/
		log.Debug("Initial check", this.cfg.Kind, "for", this.target)
		go this.check(this.target, this.cfg, c)
		for {
			select {

			/* new check interval has reached */
			case <-ticker.C:
				log.Debug("Next check", this.cfg.Kind, "for", this.target)
				go this.check(this.target, this.cfg, c)

			/* new check result is ready */
			case checkResult := <-c:
				log.Debug("Got check result", this.cfg.Kind, ":", checkResult)
				this.process(checkResult)

			/* request to stop worker */
			case <-this.stop:
				ticker.Stop()
				//close(c) // TODO: Check!
				return
			}
		}
	}()
}

func (this *Worker) process(checkResult CheckResult) {
	//......
	if checkResult.Status == this.LastResult.Status {
		// check status not changed
		return
	}

	if checkResult.Status == Unhealthy {
		this.passes = 0
		this.fails++
	} else if checkResult.Status == Healthy {
		this.fails = 0
		this.passes++
	}

	if this.passes == 0 && this.fails >= this.cfg.Fails ||
		this.fails == 0 && this.passes >= this.cfg.Passes {
		this.LastResult = checkResult

		log.Info("Sending to scheduler:", this.LastResult)
		// 通过 out 这个 channel（本质是 healthy 的 Out）将结果通知给 Scheduler
		this.out <- checkResult
	}
}

serverCount             *prometheus.GaugeVec
serverActiveConnections *prometheus.GaugeVec
serverRxTotal           *prometheus.GaugeVec
serverTxTotal           *prometheus.GaugeVec
serverRxSecond          *prometheus.GaugeVec
serverTxSecond          *prometheus.GaugeVec

backendActiveConnections  *prometheus.GaugeVec
backendRefusedConnections *prometheus.GaugeVec
backendTotalConnections   *prometheus.GaugeVec
backendRxBytes            *prometheus.GaugeVec
backendTxBytes            *prometheus.GaugeVec
backendRxSecond           *prometheus.GaugeVec
backendTxSecond           *prometheus.GaugeVec
backendLive               *prometheus.GaugeVec