pitaya框架中etcd实现服务发现源码注释

package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/clientv3/namespace"
	"github.com/topfreegames/pitaya/config"
	"github.com/topfreegames/pitaya/constants"
	"github.com/topfreegames/pitaya/logger"
	"github.com/topfreegames/pitaya/util"
)

//ETCD实现服务发现的核心思想： key存储、 租约、 key监听
// 本地使用一个结构Server保存服务的信息 id type
// 将id和type组装成key server进行Json序列化
// 连接ETCD生成etcd client
// 生成一个租约设置租约时间并且对租约keepalive，同时对租约续租响应通道进行监听(如果keepalive失败则重新一遍本部及下一步的流程)
// 同时注册本地server到ETCD授权租约之后同步ETCD上的server到本地
// 开启定时器定期同步ETCD上服务器信息（get key）
// 使用key监听的方式 监听ETCD上传来的Server的添加或者删除用来更新本地的信息

//server想etcd注册一个key(type/id)写入自身的信息，同时开启定时器获取ETCD上的server同步到本地，
//监听ETCD上指定前缀的key用来监听ETCD上的服务端的变化同时同步到本地

type etcdServiceDiscovery struct {
	cli                  *clientv3.Client              //etcd v3客户端
	config               *config.Config                //viper配置
	syncServersInterval  time.Duration                 //servers 同步时间间隔
	heartbeatTTL         time.Duration                 //心跳间隔
	logHeartbeat         bool                          //是否日志心跳
	lastHeartbeatTime    time.Time                     //上次心跳
	leaseID              clientv3.LeaseID              //租约id
	mapByTypeLock        sync.RWMutex                  //读写所
	serverMapByType      map[string]map[string]*Server //类型 server映射
	serverMapByID        sync.Map                      //同步的map
	etcdEndpoints        []string                      //etcd 节点
	etcdPrefix           string
	etcdDialTimeout      time.Duration
	running              bool
	server               *Server //本机服务
	stopChan             chan bool
	stopLeaseChan        chan bool
	lastSyncTime         time.Time     //上次同步 事件
	listeners            []SDListener  //服务发现监听
	revokeTimeout        time.Duration //撤销超时
	grantLeaseTimeout    time.Duration //设置租约时间超时
	grantLeaseMaxRetries int
	grantLeaseInterval   time.Duration
	shutdownDelay        time.Duration
	appDieChan           chan bool
}

// NewEtcdServiceDiscovery ctor
func NewEtcdServiceDiscovery(
	config *config.Config,
	server *Server,
	appDieChan chan bool,
	cli ...*clientv3.Client,
) (ServiceDiscovery, error) {
	var client *clientv3.Client
	if len(cli) > 0 {
		client = cli[0]
	}
	sd := &etcdServiceDiscovery{
		config:          config,
		running:         false,
		server:          server,
		serverMapByType: make(map[string]map[string]*Server),
		listeners:       make([]SDListener, 0),
		stopChan:        make(chan bool),
		stopLeaseChan:   make(chan bool),
		appDieChan:      appDieChan,
		cli:             client,
	}

	sd.configure()

	return sd, nil
}

// etcdServiceDiscovery 设置服务发现的配置
func (sd *etcdServiceDiscovery) configure() {
	sd.etcdEndpoints = sd.config.GetStringSlice("pitaya.cluster.sd.etcd.endpoints")                  //逗号分隔的etcd端点列表
	sd.etcdDialTimeout = sd.config.GetDuration("pitaya.cluster.sd.etcd.dialtimeout")                 //拨号超时值传递给服务发现etcd客户端
	sd.etcdPrefix = sd.config.GetString("pitaya.cluster.sd.etcd.prefix")                             //避免不同的pitaya冲突 服务器必须具有相同的前缀才能相互看到
	sd.heartbeatTTL = sd.config.GetDuration("pitaya.cluster.sd.etcd.heartbeat.ttl")                  //etcd租约的心跳间隔
	sd.logHeartbeat = sd.config.GetBool("pitaya.cluster.sd.etcd.heartbeat.log")                      //是否应在调试模式下记录etcd心跳
	sd.syncServersInterval = sd.config.GetDuration("pitaya.cluster.sd.etcd.syncservers.interval")    //服务发现模块执行的服务器同步之间的间隔
	sd.revokeTimeout = sd.config.GetDuration("pitaya.cluster.sd.etcd.revoke.timeout")                //etcd的撤销功能超时
	sd.grantLeaseTimeout = sd.config.GetDuration("pitaya.cluster.sd.etcd.grantlease.timeout")        //etcd租期超时
	sd.grantLeaseMaxRetries = sd.config.GetInt("pitaya.cluster.sd.etcd.grantlease.maxretries")       //etcd授予租约的最大尝试次数
	sd.grantLeaseInterval = sd.config.GetDuration("pitaya.cluster.sd.etcd.grantlease.retryinterval") //每次授予租约尝试之间的间隔
	sd.shutdownDelay = sd.config.GetDuration("pitaya.cluster.sd.etcd.shutdown.delay")                //从服务发现注销后等待关闭的时间
}

// watchLeaseChan 监听续租的相应管道LeaseKeepAliveResponse 会收到续租相应的事件
func (sd *etcdServiceDiscovery) watchLeaseChan(c <-chan *clientv3.LeaseKeepAliveResponse) {
	failedGrantLeaseAttempts := 0
	for {
		select {
		case <-sd.stopChan: //服务发现停止
			return
		case <-sd.stopLeaseChan: //是否停止续住监听
			return
		case leaseKeepAliveResponse := <-c: //从续租响应管道获取到通知事件
			if leaseKeepAliveResponse != nil {
				if sd.logHeartbeat {
					logger.Log.Debugf("sd: etcd lease %x renewed", leaseKeepAliveResponse.ID)
				}
				failedGrantLeaseAttempts = 0
				continue
			}
			logger.Log.Warn("sd: error renewing etcd lease, reconfiguring")
			for {
				err := sd.renewLease()
				if err != nil {
					failedGrantLeaseAttempts = failedGrantLeaseAttempts + 1
					if err == constants.ErrEtcdGrantLeaseTimeout {
						logger.Log.Warn("sd: timed out trying to grant etcd lease")
						if sd.appDieChan != nil {
							sd.appDieChan <- true
						}
						return
					}
					if failedGrantLeaseAttempts >= sd.grantLeaseMaxRetries {
						logger.Log.Warn("sd: exceeded max attempts to renew etcd lease")
						if sd.appDieChan != nil {
							sd.appDieChan <- true
						}
						return
					}
					logger.Log.Warnf("sd: error granting etcd lease, will retry in %d seconds", uint64(sd.grantLeaseInterval.Seconds()))
					time.Sleep(sd.grantLeaseInterval)
					continue
				}
				return
			}
		}
	}
}

// renewLease reestablishes connection with etcd 创新生成租约重新启动后服务
func (sd *etcdServiceDiscovery) renewLease() error {
	c := make(chan error)
	go func() {
		defer close(c)
		logger.Log.Infof("waiting for etcd lease")
		err := sd.grantLease()
		if err != nil {
			c <- err
			return
		}
		err = sd.bootstrapServer(sd.server)
		c <- err
	}()
	select {
	case err := <-c:
		return err
	case <-time.After(sd.grantLeaseTimeout):
		return constants.ErrEtcdGrantLeaseTimeout
	}
}

// grantLease创建租约并且续租同时监听续租的相应管道
func (sd *etcdServiceDiscovery) grantLease() error {
	// 创建一个租约到期时间即心跳时间
	l, err := sd.cli.Grant(context.TODO(), int64(sd.heartbeatTTL.Seconds()))
	if err != nil {
		return err
	}
	//记录租约id
	sd.leaseID = l.ID
	logger.Log.Debugf("sd: got leaseID: %x", l.ID)
	// this will keep alive forever, when channel c is closed
	// it means we probably have to rebootstrap the lease
	//租约定期续租保持alive 返回租约续租相应管道
	c, err := sd.cli.KeepAlive(context.TODO(), sd.leaseID)
	if err != nil {
		return err
	}
	// need to receive here as per etcd docs
	//读取管道的值
	<-c
	go sd.watchLeaseChan(c)
	return nil
}

//服务注册 将Server的type和id组装成key 序列化信息做为value将 kv写入etcd
func (sd *etcdServiceDiscovery) addServerIntoEtcd(server *Server) error {
	_, err := sd.cli.Put(
		context.TODO(),
		getKey(server.ID, server.Type), //servers/serverType/serverID 组装key
		server.AsJSONString(),          //server的信息序列化为json数据
		clientv3.WithLease(sd.leaseID), //给key授权租约
	)
	return err
}

// bootstrapServer 启动服务
func (sd *etcdServiceDiscovery) bootstrapServer(server *Server) error {
	//注册服务 将服务信息作为kv写入etcd同时授权租约
	if err := sd.addServerIntoEtcd(server); err != nil {
		return err
	}

	//湖区etcd上的注册的服务器信息对本地保存的服务器信息进行同步
	sd.SyncServers()
	return nil
}

// AddListener adds a listener to etcd service discovery
func (sd *etcdServiceDiscovery) AddListener(listener SDListener) {
	sd.listeners = append(sd.listeners, listener)
}

// AfterInit executes after Init
func (sd *etcdServiceDiscovery) AfterInit() {
}

// notifyListeners 添加删除服务器时进行监听通知
func (sd *etcdServiceDiscovery) notifyListeners(act Action, sv *Server) {
	for _, l := range sd.listeners {
		if act == DEL {
			l.RemoveServer(sv)
		} else if act == ADD {
			l.AddServer(sv)
		}
	}
}

// 加锁执行f
func (sd *etcdServiceDiscovery) writeLockScope(f func()) {
	sd.mapByTypeLock.Lock()
	defer sd.mapByTypeLock.Unlock()
	f()
}

//根据id删除本地的server信息记录
func (sd *etcdServiceDiscovery) deleteServer(serverID string) {
	if actual, ok := sd.serverMapByID.Load(serverID); ok {
		sv := actual.(*Server)
		sd.serverMapByID.Delete(sv.ID)
		sd.writeLockScope(func() {
			if svMap, ok := sd.serverMapByType[sv.Type]; ok {
				delete(svMap, sv.ID)
			}
		})
		sd.notifyListeners(DEL, sv)
	}
}

func (sd *etcdServiceDiscovery) deleteLocalInvalidServers(actualServers []string) {
	sd.serverMapByID.Range(func(key interface{}, value interface{}) bool {
		k := key.(string)
		if !util.SliceContainsString(actualServers, k) {
			logger.Log.Warnf("deleting invalid local server %s", k)
			sd.deleteServer(k)
		}
		return true
	})
}

// getKey 类型和id组装成key servers/serverType/serverID
func getKey(serverID, serverType string) string {
	return fmt.Sprintf("servers/%s/%s", serverType, serverID)
}

// getServerFromEtcd 查询etcd上的server序列化信息
func (sd *etcdServiceDiscovery) getServerFromEtcd(serverType, serverID string) (*Server, error) {
	svKey := getKey(serverID, serverType)
	svEInfo, err := sd.cli.Get(context.TODO(), svKey)
	if err != nil {
		return nil, fmt.Errorf("error getting server: %s from etcd, error: %s", svKey, err.Error())
	}
	if len(svEInfo.Kvs) == 0 {
		return nil, fmt.Errorf("didn't found server: %s in etcd", svKey)
	}
	return parseServer(svEInfo.Kvs[0].Value)
}

// GetServersByType returns a slice with all the servers of a certain type
func (sd *etcdServiceDiscovery) GetServersByType(serverType string) (map[string]*Server, error) {
	sd.mapByTypeLock.RLock()
	defer sd.mapByTypeLock.RUnlock()
	if m, ok := sd.serverMapByType[serverType]; ok && len(m) > 0 {
		// Create a new map to avoid concurrent read and write access to the
		// map, this also prevents accidental changes to the list of servers
		// kept by the service discovery.
		ret := make(map[string]*Server, len(sd.serverMapByType))
		for k, v := range sd.serverMapByType[serverType] {
			ret[k] = v
		}
		return ret, nil
	}
	return nil, constants.ErrNoServersAvailableOfType
}

// GetServers returns a slice with all the servers
func (sd *etcdServiceDiscovery) GetServers() []*Server {
	ret := make([]*Server, 0)
	sd.serverMapByID.Range(func(k, v interface{}) bool {
		ret = append(ret, v.(*Server))
		return true
	})
	return ret
}

func (sd *etcdServiceDiscovery) bootstrap() error {
	//生成租约 监听续租响应管道
	if err := sd.grantLease(); err != nil {
		return err
	}

	//将sd.server信息写入ETCD同时授权租约，之后同步ETCD 上的Server到本地
	if err := sd.bootstrapServer(sd.server); err != nil {
		return err
	}

	return nil
}

// GetServer returns a server given it's id
func (sd *etcdServiceDiscovery) GetServer(id string) (*Server, error) {
	if sv, ok := sd.serverMapByID.Load(id); ok {
		return sv.(*Server), nil
	}
	return nil, constants.ErrNoServerWithID
}

// Init starts the service discovery client 注册模块儿的时候回进行调用
func (sd *etcdServiceDiscovery) Init() error {
	sd.running = true
	var cli *clientv3.Client
	var err error
	if sd.cli == nil { //连接ETCD
		cli, err = clientv3.New(clientv3.Config{
			Endpoints:   sd.etcdEndpoints,
			DialTimeout: sd.etcdDialTimeout,
		})
		if err != nil {
			return err
		}
		sd.cli = cli
	}

	// namespaced etcd :) Package namespace is a clientv3 wrapper that translates all keys to begin with a given prefix.
	sd.cli.KV = namespace.NewKV(sd.cli.KV, sd.etcdPrefix)                // 重写ectd client interfaces 给所有的key 监听 租约加上前缀
	sd.cli.Watcher = namespace.NewWatcher(sd.cli.Watcher, sd.etcdPrefix) //监听
	sd.cli.Lease = namespace.NewLease(sd.cli.Lease, sd.etcdPrefix)       //租约

	//生成租约且设置租约时间并且租约定期续租 同时注册本地server到ETCD授权租约之后同步ETCD上的server到本地
	if err = sd.bootstrap(); err != nil {
		return err
	}

	// update servers 开启定时器定期同步服务器信息
	syncServersTicker := time.NewTicker(sd.syncServersInterval)
	go func() {
		for sd.running {
			select {
			case <-syncServersTicker.C:
				err := sd.SyncServers()
				if err != nil {
					logger.Log.Errorf("error resyncing servers: %s", err.Error())
				}
			case <-sd.stopChan:
				return
			}
		}
	}()

	//使用key监听的方式 监听ETCD上传来的Server的添加或者删除用来更新本地的信息
	go sd.watchEtcdChanges()
	return nil
}

func parseEtcdKey(key string) (string, string, error) {
	splittedServer := strings.Split(key, "/")
	if len(splittedServer) != 3 {
		return "", "", fmt.Errorf("error parsing etcd key %s (server name can't contain /)", key)
	}
	svType := splittedServer[1]
	svID := splittedServer[2]
	return svType, svID, nil
}

func parseServer(value []byte) (*Server, error) {
	var sv *Server
	err := json.Unmarshal(value, &sv)
	if err != nil {
		logger.Log.Warnf("failed to load server %s, error: %s", sv, err.Error())
	}
	return sv, nil
}

func (sd *etcdServiceDiscovery) printServers() {
	sd.mapByTypeLock.RLock()
	defer sd.mapByTypeLock.RUnlock()
	for k, v := range sd.serverMapByType {
		logger.Log.Debugf("type: %s, servers: %+v", k, v)
	}
}

// SyncServers gets all servers from etcd 从etcd获取左右的server对本地保存的server信息进行同步（增删）
func (sd *etcdServiceDiscovery) SyncServers() error {
	//读取servers目录下的所有key
	keys, err := sd.cli.Get(
		context.TODO(),
		"servers/",
		clientv3.WithPrefix(),
		clientv3.WithKeysOnly(),
	)
	if err != nil {
		return err
	}

	// delete invalid servers (local ones that are not in etcd)
	allIds := make([]string, 0)

	// filter servers I need to grab info 获取所有etcd上的server根据key获取server然后添加到本地
	for _, kv := range keys.Kvs {
		svType, svID, err := parseEtcdKey(string(kv.Key)) //根据etcd的key解析出 svType 和svID
		if err != nil {
			logger.Log.Warnf("failed to parse etcd key %s, error: %s", kv.Key, err.Error())
		}
		allIds = append(allIds, svID)
		// TODO is this slow? if so we can paralellize
		if _, ok := sd.serverMapByID.Load(svID); !ok { //查看id是否存在 如果本地存在记录
			logger.Log.Debugf("loading info from missing server: %s/%s", svType, svID)
			//根据svType 和svId组装成key获取Etcd上的Server序列化信息
			sv, err := sd.getServerFromEtcd(svType, svID)
			if err != nil {
				logger.Log.Errorf("error getting server from etcd: %s, error: %s", svID, err.Error())
				continue
			}
			//将sever添加到本地存储server的两种map
			sd.addServer(sv)
		}
	}
	//删除ETCD上没有本地存在的server
	sd.deleteLocalInvalidServers(allIds)

	sd.printServers()
	sd.lastSyncTime = time.Now() //记录下同步的时间
	return nil
}

// BeforeShutdown executes before shutting down and will remove the server from the list
func (sd *etcdServiceDiscovery) BeforeShutdown() {
	sd.revoke()
	time.Sleep(sd.shutdownDelay) // Sleep for a short while to ensure shutdown has propagated
}

// Shutdown executes on shutdown and will clean etcd
func (sd *etcdServiceDiscovery) Shutdown() error {
	sd.running = false
	close(sd.stopChan)
	return nil
}

// revoke prevents Pitaya from crashing when etcd is not available
func (sd *etcdServiceDiscovery) revoke() error {
	close(sd.stopLeaseChan)
	c := make(chan error)
	defer close(c)
	go func() {
		logger.Log.Debug("waiting for etcd revoke")
		_, err := sd.cli.Revoke(context.TODO(), sd.leaseID)
		c <- err
		logger.Log.Debug("finished waiting for etcd revoke")
	}()
	select {
	case err := <-c:
		return err // completed normally
	case <-time.After(sd.revokeTimeout):
		logger.Log.Warn("timed out waiting for etcd revoke")
		return nil // timed out
	}
}

// addServer 将Server存入serverMapByID（map[id]*Server） 和 serverMapByType(map[type]map[id]*Server)
func (sd *etcdServiceDiscovery) addServer(sv *Server) {
	if _, loaded := sd.serverMapByID.LoadOrStore(sv.ID, sv); !loaded {
		//在对serverMapByType进行修改时使用加锁
		sd.writeLockScope(func() {
			mapSvByType, ok := sd.serverMapByType[sv.Type]
			if !ok {
				mapSvByType = make(map[string]*Server)
				sd.serverMapByType[sv.Type] = mapSvByType
			}
			mapSvByType[sv.ID] = sv
		})
		if sv.ID != sd.server.ID {
			//通知监听添加服务器
			sd.notifyListeners(ADD, sv)
		}
	}
}

// 监听所有ETCD上的Server的key
func (sd *etcdServiceDiscovery) watchEtcdChanges() {
	//监视etcd上key前缀为servers的key
	w := sd.cli.Watch(context.Background(), "servers/", clientv3.WithPrefix())

	go func(chn clientv3.WatchChan) {
		for sd.running { //sd在运行
			select {
			case wResp := <-chn: //监听到key有变化
				for _, ev := range wResp.Events {
					switch ev.Type {
					case clientv3.EventTypePut: //添加或者更新事件
						var sv *Server
						var err error
						if sv, err = parseServer(ev.Kv.Value); err != nil {
							logger.Log.Errorf("Failed to parse server from etcd: %v", err)
							continue
						}
						sd.addServer(sv) //添加服务到本地
						logger.Log.Debugf("server %s added", ev.Kv.Key)
						sd.printServers()
					case clientv3.EventTypeDelete: //删除服务事件
						_, svID, err := parseEtcdKey(string(ev.Kv.Key))
						if err != nil {
							logger.Log.Warnf("failed to parse key from etcd: %s", ev.Kv.Key)
							continue
						}
						sd.deleteServer(svID) //删除本地服务
						logger.Log.Debugf("server %s deleted", svID)
						sd.printServers()
					}
				}
			case <-sd.stopChan: //停止
				return
			}
		}
	}(w)
}