Merge branch 'pipiaha-dev' into dev

Author: rogeralsing

cluster/clusterproviders/etcd/config.go

@@ -0,0 +1,39 @@
+package etcd
+
+import (
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+type RoleChangedListener interface {
+	OnRoleChanged(RoleType)
+}
+
+type Option func(*config)
+
+func WithBaseKey(baseKey string) Option {
+	return func(o *config) {
+		o.BaseKey = baseKey
+	}
+}
+
+func WithEtcdConfig(cfg clientv3.Config) Option {
+	return func(o *config) {
+		o.cfg = cfg
+	}
+}
+
+func WithRoleChangedListener(l RoleChangedListener) Option {
+	return func(o *config) {
+		o.RoleChanged = l
+	}
+}
+
+type config struct {
+	BaseKey     string
+	cfg         clientv3.Config
+	RoleChanged RoleChangedListener
+}
+
+func defaultConfig() *config {
+	return &config{}
+}

cluster/clusterproviders/etcd/etcd_provider.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -31,6 +32,10 @@ type Provider struct {
 	retryInterval time.Duration
 	revision      uint64
 	// deregisterCritical time.Duration
+	schedulers          []*SingletonScheduler
+	role                RoleType
+	roleChangedChan     chan RoleType
+	roleChangedListener RoleChangedListener
 }
 
 func New() (*Provider, error) {
@@ -40,18 +45,27 @@ func New() (*Provider, error) {
 	})
 }
 
-func NewWithConfig(baseKey string, cfg clientv3.Config) (*Provider, error) {
-	client, err := clientv3.New(cfg)
+func NewWithConfig(baseKey string, cfg clientv3.Config, opts ...Option) (*Provider, error) {
+	c := defaultConfig()
+	WithBaseKey(baseKey)(c)
+	WithEtcdConfig(cfg)(c)
+	for _, opt := range opts {
+		opt(c)
+	}
+	client, err := clientv3.New(c.cfg)
 	if err != nil {
 		return nil, err
 	}
 	p := &Provider{
-		client:        client,
-		keepAliveTTL:  3 * time.Second,
-		retryInterval: 1 * time.Second,
-		baseKey:       baseKey,
-		members:       map[string]*Node{},
-		cancelWatchCh: make(chan bool),
+		client:              client,
+		keepAliveTTL:        3 * time.Second,
+		retryInterval:       1 * time.Second,
+		baseKey:             c.BaseKey,
+		members:             map[string]*Node{},
+		cancelWatchCh:       make(chan bool),
+		role:                Follower,
+		roleChangedChan:     make(chan RoleType, 1),
+		roleChangedListener: c.RoleChanged,
 	}
 	return p, nil
 }
@@ -78,7 +92,12 @@ func (p *Provider) StartMember(c *cluster.Cluster) error {
 	if err := p.init(c); err != nil {
 		return err
 	}
+	// register self
+	if err := p.registerService(); err != nil {
+		return err
+	}
 
+	p.startRoleChangedNotifyLoop()
 	// fetch memberlist
 	nodes, err := p.fetchNodes()
 	if err != nil {
@@ -89,12 +108,9 @@ func (p *Provider) StartMember(c *cluster.Cluster) error {
 	p.publishClusterTopologyEvent()
 	p.startWatching()
 
-	// register self
-	if err := p.registerService(); err != nil {
-		return err
-	}
 	ctx := context.TODO()
 	p.startKeepAlive(ctx)
+	p.updateLeadership(nodes)
 	return nil
 }
 
@@ -116,6 +132,7 @@ func (p *Provider) StartClient(c *cluster.Cluster) error {
 func (p *Provider) Shutdown(graceful bool) error {
 	p.shutdown = true
 	if !p.deregistered {
+		p.updateLeadership(nil)
 		err := p.deregisterService()
 		if err != nil {
 			p.cluster.Logger().Error("deregisterMember", slog.Any("error", err))
@@ -254,6 +271,8 @@ func (p *Provider) handleWatchResponse(resp clientv3.WatchResponse) map[string]*
 				p.cluster.Logger().Debug("New member.", slog.String("key", key))
 			}
 			changes[nodeId] = node
+			node.SetMeta(metaKeySeq, nodeId)
+			node.SetMeta(metaKeySeq, fmt.Sprintf("%d", ev.Kv.Lease))
 		case clientv3.EventTypeDelete:
 			node, ok := p.members[nodeId]
 			if !ok {
@@ -267,6 +286,8 @@ func (p *Provider) handleWatchResponse(resp clientv3.WatchResponse) map[string]*
 			cloned := *node
 			cloned.SetAlive(false)
 			changes[nodeId] = &cloned
+			node.SetMeta(metaKeySeq, nodeId)
+			node.SetMeta(metaKeySeq, fmt.Sprintf("%d", ev.Kv.Lease))
 		default:
 			p.cluster.Logger().Error("Invalid etcd event.type.", slog.String("key", key),
 				slog.String("type", ev.Type.String()))
@@ -294,6 +315,11 @@ func (p *Provider) _keepWatching(stream clientv3.WatchChan) error {
 		}
 		nodesChanges := p.handleWatchResponse(resp)
 		p.updateNodesWithChanges(nodesChanges)
+		l := make([]*Node, 0)
+		for _, node := range nodesChanges {
+			l = append(l, node)
+		}
+		p.updateLeadership(l)
 		p.publishClusterTopologyEvent()
 	}
 	return nil
@@ -304,6 +330,17 @@ func (p *Provider) startWatching() {
 	ctx, cancel := context.WithCancel(ctx)
 	p.cancelWatch = cancel
 	go func() {
+		//recover
+		defer func() {
+			if r := recover(); r != nil {
+				p.cluster.Logger().Error("Recovered from panic in keepWatching.", slog.Any("error", r))
+				p.clusterError = fmt.Errorf("keepWatching panic: %v", r)
+			}
+			if p.cancelWatchCh != nil {
+				close(p.cancelWatchCh)
+			}
+			p.cancelWatch = nil
+		}()
 		for !p.shutdown {
 			if err := p.keepWatching(ctx); err != nil {
 				p.cluster.Logger().Error("Failed to keepWatching.", slog.Any("error", err))
@@ -339,6 +376,10 @@ func (p *Provider) fetchNodes() ([]*Node, error) {
 			return nil, err
 		}
 		nodes = append(nodes, &n)
+		if v.Lease > 0 {
+			n.SetMeta(metaKeySeq, fmt.Sprintf("%d", v.Lease))
+			n.SetMeta(metaKeyID, n.ID)
+		}
 	}
 	p.revision = uint64(resp.Header.GetRevision())
 	// plog.Debug("fetch nodes",
@@ -393,6 +434,7 @@ func (p *Provider) getLeaseID() clientv3.LeaseID {
 
 func (p *Provider) setLeaseID(leaseID clientv3.LeaseID) {
 	atomic.StoreInt64((*int64)(&p.leaseID), (int64)(leaseID))
+	p.self.SetMeta(metaKeySeq, fmt.Sprintf("%d", leaseID))
 }
 
 func (p *Provider) newLeaseID() (clientv3.LeaseID, error) {
@@ -417,3 +459,88 @@ func splitHostPort(addr string) (host string, port int, err error) {
 	}
 	return
 }
+
+func (p *Provider) RegisterSingletonScheduler(scheduler *SingletonScheduler) {
+	p.schedulers = append(p.schedulers, scheduler)
+}
+
+// 修改现有的角色变化处理逻辑
+func (p *Provider) updateLeadership(ns []*Node) {
+	role := Follower
+	ns, err := p.fetchNodes()
+	if err != nil {
+		p.cluster.Logger().Error("Failed to fetch nodes in updateLeadership.", slog.Any("error", err))
+	}
+
+	if p.isLeaderOf(ns) {
+		role = Leader
+	}
+	if role != p.role {
+		p.cluster.Logger().Info("Role changed.", slog.String("from", p.role.String()), slog.String("to", role.String()))
+		p.role = role
+		p.roleChangedChan <- role
+
+		// 通知所有注册的 SingletonScheduler
+		for _, scheduler := range p.schedulers {
+			safeRun(p.cluster.Logger(), func() {
+				scheduler.OnRoleChanged(role)
+			})
+		}
+	}
+}
+
+func safeRun(logger *slog.Logger, fn func()) {
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Warn("OnRoleChanged.", slog.Any("error", fmt.Errorf("%v\n%s", r, string(getRunTimeStack()))))
+		}
+	}()
+	fn()
+}
+
+func getRunTimeStack() []byte {
+	const size = 64 << 10
+	buf := make([]byte, size)
+	return buf[:runtime.Stack(buf, false)]
+}
+
+func (p *Provider) isLeaderOf(ns []*Node) bool {
+	if ns == nil {
+		return false
+	}
+	if len(ns) == 1 && p.self != nil && ns[0].ID == p.self.ID {
+		return true
+	}
+	var minSeq int
+	for _, node := range ns {
+		if !node.IsAlive() {
+			continue
+		}
+		if seq := node.GetSeq(); (seq > 0 && seq < minSeq) || minSeq == 0 {
+			minSeq = seq
+		}
+	}
+	//for _, node := range ns {
+	//	if p.self != nil && node.ID == p.self.ID {
+	//		return minSeq > 0 && minSeq == p.self.GetSeq()
+	//	}
+	//}
+	if minSeq <= 0 { // 没有在线actor
+		return true
+	}
+	if p.self != nil && p.self.GetSeq() <= minSeq {
+		return true
+	}
+	return false
+}
+
+func (p *Provider) startRoleChangedNotifyLoop() {
+	go func() {
+		for !p.shutdown {
+			role := <-p.roleChangedChan
+			if lis := p.roleChangedListener; lis != nil {
+				safeRun(p.cluster.Logger(), func() { lis.OnRoleChanged(role) })
+			}
+		}
+	}()
+}

cluster/clusterproviders/etcd/node.go

@@ -2,10 +2,16 @@ package etcd
 
 import (
 	"encoding/json"
+	"strconv"
 
 	"github.com/asynkron/protoactor-go/cluster"
 )
 
+const (
+	metaKeyID  = "id"
+	metaKeySeq = "seq"
+)
+
 type Node struct {
 	ID      string            `json:"id"`
 	Name    string            `json:"name"`
@@ -64,6 +70,17 @@ func (n *Node) GetMeta(name string) (string, bool) {
 	val, ok := n.Meta[name]
 	return val, ok
 }
+func (n *Node) GetSeq() int {
+	if seqStr, ok := n.GetMeta(metaKeySeq); ok {
+		return strToInt(seqStr)
+	}
+	return 0
+}
+
+func strToInt(s string) int {
+	i, _ := strconv.ParseInt(s, 10, 64)
+	return int(i)
+}
 
 func (n *Node) MemberStatus() *cluster.Member {
 	host, port := n.GetAddress()

cluster/clusterproviders/etcd/singleton.go

@@ -0,0 +1,71 @@
+package etcd
+
+import (
+	"github.com/asynkron/protoactor-go/actor"
+	"sync"
+)
+
+// RoleType 表示节点在集群中的角色
+type RoleType int
+
+const (
+	Follower RoleType = iota
+	Leader
+)
+
+func (r RoleType) String() string {
+	switch r {
+	case Leader:
+		return "Leader"
+	default:
+		return "Follower"
+	}
+}
+
+// SingletonScheduler 管理在领导节点上运行的单例 actor
+type SingletonScheduler struct {
+	sync.Mutex
+	root  *actor.RootContext
+	props []*actor.Props
+	pids  []*actor.PID
+}
+
+func NewSingletonScheduler(rc *actor.RootContext) *SingletonScheduler {
+	return &SingletonScheduler{root: rc}
+}
+
+func (s *SingletonScheduler) FromFunc(f actor.ReceiveFunc) *SingletonScheduler {
+	s.Lock()
+	defer s.Unlock()
+	s.props = append(s.props, actor.PropsFromFunc(f))
+	return s
+}
+
+func (s *SingletonScheduler) FromProducer(f actor.Producer) *SingletonScheduler {
+	s.Lock()
+	defer s.Unlock()
+	s.props = append(s.props, actor.PropsFromProducer(f))
+	return s
+}
+
+func (s *SingletonScheduler) OnRoleChanged(rt RoleType) {
+	s.Lock()
+	defer s.Unlock()
+	if rt == Follower {
+		if len(s.pids) > 0 {
+			s.root.Logger().Info("I am follower, poison singleton actors")
+			for _, pid := range s.pids {
+				s.root.Poison(pid)
+			}
+			s.pids = nil
+		}
+	} else if rt == Leader {
+		if len(s.props) > 0 {
+			s.root.Logger().Info("I am leader now, start singleton actors")
+			s.pids = make([]*actor.PID, len(s.props))
+			for i, p := range s.props {
+				s.pids[i] = s.root.Spawn(p)
+			}
+		}
+	}
+}