浅析Consul集群的管理

// consul:agent/consul/server.go:398
func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server) (*Server, error) {
    ...
    if config.SerfWANConfig != nil {
		s.serfWAN, s.serfWANConfig, err = s.setupSerf(setupSerfOptions{
			Config:       config.SerfWANConfig,
			EventCh:      s.eventChWAN,
			SnapshotPath: serfWANSnapshot,
			WAN:          true,
			Listener:     s.Listener,
		})
		...
	}
    ...
	// Initialize the LAN Serf for the default network segment.
	if err := s.setupSerfLAN(config); err != nil {
		s.Shutdown()
		return nil, fmt.Errorf("Failed to start LAN Serf: %v", err)
	}
    ...
    go s.lanEventHandler()
    ...
}

// consul:agent/consul/server_serf.go:52
func (s *Server) setupSerf(opts setupSerfOptions) (*serf.Serf, *serf.Config, error) {
	conf, err := s.setupSerfConfig(opts)
	if err != nil {
		return nil, nil, err
	}

	cluster, err := serf.Create(conf)
	if err != nil {
		return nil, nil, err
	}

	return cluster, conf, nil
}

// consul:agent/consul/server_serf.go:37
type setupSerfOptions struct {
	Config       *serf.Config
	EventCh      chan serf.Event
	SnapshotPath string
	Listener     net.Listener

	// WAN only
	WAN bool

	// LAN only
	Segment   string
	Partition string
}

// consul:agent/consul/server.go:398
func NewServer(config *Config, flat Deps, externalGRPCServer *grpc.Server) (*Server, error) {
    ...
    go s.lanEventHandler()
    ...
    // Add a "static route" to the WAN Serf and hook it up to Serf events.
	if s.serfWAN != nil {
		if err := s.router.AddArea(types.AreaWAN, s.serfWAN, s.connPool); err != nil {
			s.Shutdown()
			return nil, fmt.Errorf("Failed to add WAN serf route: %v", err)
		}
		go router.HandleSerfEvents(s.logger, s.router, types.AreaWAN, s.serfWAN.ShutdownCh(), s.eventChWAN, s.wanMembershipNotifyCh)

		// Fire up the LAN <-> WAN join flooder.
		addrFn := func(s *metadata.Server) (string, error) {
			if s.WanJoinPort == 0 {
				return "", fmt.Errorf("no wan join  port for server: %s", s.Addr.String())
			}
			addr, _, err := net.SplitHostPort(s.Addr.String())
			if err != nil {
				return "", err
			}
			return fmt.Sprintf("%s:%d", addr, s.WanJoinPort), nil
		}
		go s.Flood(addrFn, s.serfWAN)
	}
    ...
}

// consul:agent/consul/server_serf.go:265
func (s *Server) lanEventHandler() {
	for {
		select {
		case e := <-s.eventChLAN:
			switch e.EventType() {
			case serf.EventMemberJoin:
				s.lanNodeJoin(e.(serf.MemberEvent))
				s.localMemberEvent(e.(serf.MemberEvent))

			case serf.EventMemberLeave, serf.EventMemberFailed, serf.EventMemberReap:
				s.lanNodeFailed(e.(serf.MemberEvent))
				s.localMemberEvent(e.(serf.MemberEvent))

			case serf.EventUser:
				s.localEvent(e.(serf.UserEvent))
			case serf.EventMemberUpdate:
				s.lanNodeUpdate(e.(serf.MemberEvent))
				s.localMemberEvent(e.(serf.MemberEvent))
			case serf.EventQuery: // Ignore
			default:
				s.logger.Warn("Unhandled LAN Serf Event", "event", e)
			}

		case <-s.shutdownCh:
			return
		}
	}
}

// consul:agent/router/serf_adapter.go:58
func HandleSerfEvents(
	logger hclog.Logger,
	router *Router,
	areaID types.AreaID,
	shutdownCh <-chan struct{},
	eventCh <-chan serf.Event,
	membershipNotifyCh chan<- struct{},
) {
	for {
		select {
		case <-shutdownCh:
			return

		case e := <-eventCh:
			switch e.EventType() {
			case serf.EventMemberJoin:
				handleMemberEvent(logger, router.AddServer, areaID, e)
				notifyMembershipPossibleChange(membershipNotifyCh)

			case serf.EventMemberLeave, serf.EventMemberReap:
				handleMemberEvent(logger, router.RemoveServer, areaID, e)
				notifyMembershipPossibleChange(membershipNotifyCh)

			case serf.EventMemberFailed:
				handleMemberEvent(logger, router.FailServer, areaID, e)
				notifyMembershipPossibleChange(membershipNotifyCh)

			case serf.EventMemberUpdate:
				handleMemberEvent(logger, router.AddServer, areaID, e)
				notifyMembershipPossibleChange(membershipNotifyCh)

			// All of these event types are ignored.
			case serf.EventUser:
			case serf.EventQuery:

			default:
				logger.Warn("Unhandled Serf Event", "event", e)
			}
		}
	}
}

// consul:agent/consul/server.go:313
func (s *Server) lanNodeJoin(me serf.MemberEvent) {
	for _, m := range me.Members {
		ok, serverMeta := metadata.IsConsulServer(m)
		if !ok || serverMeta.Segment != "" {
			continue
		}
		s.logger.Info("Adding LAN server", "server", serverMeta.String())

		// Update server lookup
		s.serverLookup.AddServer(serverMeta)

		// If we're still expecting to bootstrap, may need to handle this.
		if s.config.BootstrapExpect != 0 {
			s.maybeBootstrap()
		}

		// Kick the join flooders.
		s.FloodNotify()
	}
}

// consul:agent/consul/server_lookup.go:25
func (sl *ServerLookup) AddServer(server *metadata.Server) {
	sl.lock.Lock()
	defer sl.lock.Unlock()
	sl.addressToServer[raft.ServerAddress(server.Addr.String())] = server
	sl.idToServer[raft.ServerID(server.ID)] = server
}

// consul:agent/consul/client.go:94
func NewClient(config *Config, deps Deps) (*Client, error) {
	if err := config.CheckProtocolVersion(); err != nil {
		return nil, err
	}
	if config.DataDir == "" {
		return nil, fmt.Errorf("Config must provide a DataDir")
	}
	if err := config.CheckACL(); err != nil {
		return nil, err
	}

	c := &Client{
		config:          config,
		connPool:        deps.ConnPool,
		eventCh:         make(chan serf.Event, serfEventBacklog),
		logger:          deps.Logger.NamedIntercept(logging.ConsulClient),
		shutdownCh:      make(chan struct{}),
		tlsConfigurator: deps.TLSConfigurator,
	}

	c.rpcLimiter.Store(rate.NewLimiter(config.RPCRateLimit, config.RPCMaxBurst))

	if err := c.initEnterprise(deps); err != nil {
		c.Shutdown()
		return nil, err
	}

	aclConfig := ACLResolverConfig{
		Config:          config.ACLResolverSettings,
		Backend:         &clientACLResolverBackend{Client: c},
		Logger:          c.logger,
		DisableDuration: aclClientDisabledTTL,
		CacheConfig:     clientACLCacheConfig,
		ACLConfig:       newACLConfig(&partitionInfoNoop{}, c.logger),
		Tokens:          deps.Tokens,
	}
	var err error
	if c.ACLResolver, err = NewACLResolver(&aclConfig); err != nil {
		c.Shutdown()
		return nil, fmt.Errorf("Failed to create ACL resolver: %v", err)
	}

	// Initialize the LAN Serf
	c.serf, err = c.setupSerf(config.SerfLANConfig, c.eventCh, serfLANSnapshot)
	if err != nil {
		c.Shutdown()
		return nil, fmt.Errorf("Failed to start lan serf: %v", err)
	}

	if err := deps.Router.AddArea(types.AreaLAN, c.serf, c.connPool); err != nil {
		c.Shutdown()
		return nil, fmt.Errorf("Failed to add LAN area to the RPC router: %w", err)
	}
	c.router = deps.Router

	// Start LAN event handlers after the router is complete since the event
	// handlers depend on the router and the router depends on Serf.
	go c.lanEventHandler()

	return c, nil
}

// consul:agent/consul/client_serf.go:19
func (c *Client) setupSerf(conf *serf.Config, ch chan serf.Event, path string) (*serf.Serf, error) {
	conf.Init()
	...
	conf.EventCh = ch
	...
	return serf.Create(conf)
}

func main() {
	ch := make(chan string, 1024)
	ch = A(ch)
	ch = B(aCh)
	ff := []string{"AFunc", "BFunc"}
	go func() {
		for {
			n := ff[rand.Int()%2]
			fmt.Println("main: send ", n)
			ch <- n
			time.Sleep(time.Second * 5)
		}
	}()
	time.Sleep(time.Second * 60)
}

func A(ch chan string) chan string {
	newCh := make(chan string, 1024)
	go func() {
		for {
			select {
			case e := <-newCh:
				if strings.Contains(e, "AFunc") {
					fmt.Println("A func do something")
				} else {
					ch <- e
				}
			}
		}
	}()
	return newCh
}

func B(ch chan string) chan string {
	newCh := make(chan string, 1024)
	go func() {
		for {
			select {
			case e := <-newCh:
				if strings.Contains(e, "BFunc") {
					fmt.Println("B func do something")
				} else {
					ch <- e
				}
			}
		}
	}()
	return newCh
}

// serf:serf/serf.go:249
func Create(conf *Config) (*Serf, error) {
	...
		// Check if serf member event coalescing is enabled
	if conf.CoalescePeriod > 0 && conf.QuiescentPeriod > 0 && conf.EventCh != nil {
		c := &memberEventCoalescer{
			lastEvents:   make(map[string]EventType),
			latestEvents: make(map[string]coalesceEvent),
		}

		conf.EventCh = coalescedEventCh(conf.EventCh, serf.shutdownCh,
			conf.CoalescePeriod, conf.QuiescentPeriod, c)
	}

	// Check if user event coalescing is enabled
	if conf.UserCoalescePeriod > 0 && conf.UserQuiescentPeriod > 0 && conf.EventCh != nil {
		c := &userEventCoalescer{
			events: make(map[string]*latestUserEvents),
		}

		conf.EventCh = coalescedEventCh(conf.EventCh, serf.shutdownCh,
			conf.UserCoalescePeriod, conf.UserQuiescentPeriod, c)
	}

	outCh, err := newSerfQueries(serf, serf.logger, conf.EventCh, serf.shutdownCh)
	if err != nil {
		return nil, fmt.Errorf("Failed to setup serf query handler: %v", err)
	}
	conf.EventCh = outCh
	...
}

// serf:serf/serf.go:641
func (s *Serf) Join(existing []string, ignoreOld bool) (int, error) {
	...
	num, err := s.memberlist.Join(existing)

	// If we joined any nodes, broadcast the join message
	if num > 0 {
		// Start broadcasting the update
		if err := s.broadcastJoin(s.clock.Time()); err != nil {
			return num, err
		}
	}
	return num, err
}

// serf:serf/serf.go:249
func Create(conf *Config) (*Serf, error) {
	...
	serf.broadcasts = &memberlist.TransmitLimitedQueue{
		NumNodes:       serf.NumNodes,
		RetransmitMult: conf.MemberlistConfig.RetransmitMult,
	}
	serf.eventBroadcasts = &memberlist.TransmitLimitedQueue{
		NumNodes:       serf.NumNodes,
		RetransmitMult: conf.MemberlistConfig.RetransmitMult,
	}
	serf.queryBroadcasts = &memberlist.TransmitLimitedQueue{
		NumNodes:       serf.NumNodes,
		RetransmitMult: conf.MemberlistConfig.RetransmitMult,
	}
	...
	memberlist, err := memberlist.Create(conf.MemberlistConfig)
	if err != nil {
		return nil, fmt.Errorf("Failed to create memberlist: %v", err)
	}
	serf.memberlist = memberlist
	...
}

// memberlist:memberlist.go:97
func newMemberlist(conf *Config) (*Memberlist, error) {
	...
	go m.streamListen()
	go m.packetListen()
	go m.packetHandler()
	go m.checkBroadcastQueueDepth()
	...
}

// memberlist:memberlist.go:245
func Create(conf *Config) (*Memberlist, error) {
	m, err := newMemberlist(conf)
	if err != nil {
		return nil, err
	}
	if err := m.setAlive(); err != nil {
		m.Shutdown()
		return nil, err
	}
	m.schedule()
	return m, nil
}

// memberlist: state.go:122
func (m *Memberlist) schedule() {
	m.tickerLock.Lock()
	defer m.tickerLock.Unlock()

	// If we already have tickers, then don't do anything, since we're
	// scheduled
	if len(m.tickers) > 0 {
		return
	}

	// Create the stop tick channel, a blocking channel. We close this
	// when we should stop the tickers.
	stopCh := make(chan struct{})

	// Create a new probeTicker
	if m.config.ProbeInterval > 0 {
		t := time.NewTicker(m.config.ProbeInterval)
		go m.triggerFunc(m.config.ProbeInterval, t.C, stopCh, m.probe)
		m.tickers = append(m.tickers, t)
	}

	// Create a push pull ticker if needed
	if m.config.PushPullInterval > 0 {
		go m.pushPullTrigger(stopCh)
	}

	// Create a gossip ticker if needed
	if m.config.GossipInterval > 0 && m.config.GossipNodes > 0 {
		t := time.NewTicker(m.config.GossipInterval)
		go m.triggerFunc(m.config.GossipInterval, t.C, stopCh, m.gossip)
		m.tickers = append(m.tickers, t)
	}

	// If we made any tickers, then record the stopTick channel for
	// later.
	if len(m.tickers) > 0 {
		m.stopTick = stopCh
	}
}

// memberlist:memberlist.go:267
func (m *Memberlist) Join(existing []string) (int, error) {
	numSuccess := 0
	var errs error
	for _, exist := range existing {
		addrs, err := m.resolveAddr(exist)
		if err != nil {
			err = fmt.Errorf("Failed to resolve %s: %v", exist, err)
			errs = multierror.Append(errs, err)
			m.logger.Printf("[WARN] memberlist: %v", err)
			continue
		}

		for _, addr := range addrs {
			hp := joinHostPort(addr.ip.String(), addr.port)
			a := Address{Addr: hp, Name: addr.nodeName}
			if err := m.pushPullNode(a, true); err != nil {
				err = fmt.Errorf("Failed to join %s: %v", a.Addr, err)
				errs = multierror.Append(errs, err)
				m.logger.Printf("[DEBUG] memberlist: %v", err)
				continue
			}
			numSuccess++
		}

	}
	if numSuccess > 0 {
		errs = nil
	}
	return numSuccess, errs
}

// memberlist:state.go:670
func (m *Memberlist) pushPullNode(a Address, join bool) error {
	defer metrics.MeasureSinceWithLabels([]string{"memberlist", "pushPullNode"}, time.Now(), m.metricLabels)

	// Attempt to send and receive with the node
	remote, userState, err := m.sendAndReceiveState(a, join)
	if err != nil {
		return err
	}

	if err := m.mergeRemoteState(join, remote, userState); err != nil {
		return err
	}
	return nil
}