func (s *KingbusServer) configure(ctx context.Context, cc raftpb.ConfChange) ([]*membership.Member, error) {
	cc.ID = s.reqIDGen.Next()
	ch := s.wait.Register(cc.ID)
	if err := s.raftNode.ProposeConfChange(ctx, cc); err != nil {
		s.wait.Trigger(cc.ID, nil)
		return nil, err
	}
	select {
	case x := <-ch:
		if x == nil {
			log.Log.Panicf("configure trigger value should never be nil")
		}
		resp := x.(*confChangeResponse)
		return resp.membs, resp.err
	case <-ctx.Done():
		s.wait.Trigger(cc.ID, nil)  
		return nil, ctx.Err()
	case <-s.stopping:
		return nil, errors.New("kingbus server stoping")
	}
}
configure sends a configuration change through consensus and
then waits for it to be applied to the server. It
will block until the change is performed or there is an error.
package server
import (
	"context"
	"fmt"
	"net/url"
	"time"
	"errors"
	"encoding/json"
	"sync"
	"strings"
	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/idutil"
	"github.com/coreos/etcd/pkg/transport"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/pkg/wait"
	etcdraft "github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/rafthttp"
	"github.com/flike/kingbus/api"
	"github.com/flike/kingbus/config"
	"github.com/flike/kingbus/log"
	"github.com/flike/kingbus/mysql"
	"github.com/flike/kingbus/raft"
	"github.com/flike/kingbus/raft/membership"
	"github.com/flike/kingbus/storage"
	"github.com/flike/kingbus/utils"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/rcrowley/go-metrics"
	"go.uber.org/atomic"
)
const (
	maxInflightMsgs = 512
	HealthInterval = 5 * time.Second
	DialTimeout = 5 * time.Second
	ProposeMaxRetryCount = 1000
)
type KingbusServer struct {
	Cfg            *config.KingbusServerConfig
	adminSvr       *api.AdminServer
	syncer         *Syncer
	master         *BinlogServer
	binlogProgress *BinlogProgress
	prometheusSvr  *PrometheusServer
	id       types.ID
	raftNode *raft.Node
	cluster  *membership.RaftCluster
	store    storage.Storage
	lead            *atomic.Uint64
	leadElectedTime time.Time
	Mu     sync.RWMutex
	errorc chan error
	wait     wait.Wait
	reqIDGen *idutil.Generator
	applyBroadcast *utils.Broadcast
	stats  *stats.ServerStats
	lstats *stats.LeaderStats
	stopping chan struct{}
	started  *atomic.Bool
}
func (s *KingbusServer) starRaft(cfg config.RaftNodeConfig) error {
	var (
		etcdRaftNode etcdraft.Node
		id           types.ID
		cl           *membership.RaftCluster
		remotes      []*membership.Member
		appliedIndex uint64
	)
	prt, err := rafthttp.NewRoundTripper(transport.TLSInfo{}, DialTimeout)
	if err != nil {
		return err
	}
	store, err := storage.NewDiskStorage(cfg.DataDir, cfg.ReserveDataSize)
	if err != nil {
		log.Log.Fatalf("NewKingbusServer:NewDiskStorage error,err:%s,dir:%s", err.Error(), cfg.DataDir)
	}
	defer func() {
		if err != nil {
			store.Close()
		}
	}()
	logExist := utils.ExistLog(cfg.DataDir)
	switch {
	case !logExist && !cfg.NewCluster:
		if err = cfg.VerifyJoinExisting(); err != nil {
			return err
		}
		cl, err = membership.NewClusterFromURLsMap(cfg.InitialPeerURLsMap)
		if err != nil {
			return err
		}
		remotePeerURLs := membership.GetRemotePeerURLs(cl, cfg.Name)
		existingCluster, gerr := membership.GetClusterFromRemotePeers(remotePeerURLs, prt)
		if gerr != nil {
			return fmt.Errorf("cannot fetch cluster info from peer urls: %v", gerr)
		}
		if err = membership.ValidateClusterAndAssignIDs(cl, existingCluster); err != nil {
			return fmt.Errorf("error validating peerURLs %s: %v", existingCluster, err)
		}
		remotes = existingCluster.Members()
		cl.SetID(existingCluster.GetID())
		cl.SetStore(store)
		id, etcdRaftNode = startEtcdRaftNode(cfg, store, cl, nil)
	case !logExist && cfg.NewCluster:
		if err = cfg.VerifyBootstrap(); err != nil {
			return err
		}
		cl, err = membership.NewClusterFromURLsMap(cfg.InitialPeerURLsMap)
		if err != nil {
			return err
		}
		m := cl.MemberByName(cfg.Name)
		if membership.IsMemberBootstrapped(cl, cfg.Name, prt, DialTimeout) {
			return fmt.Errorf("member %s has already been bootstrapped", m.ID)
		}
		cl.SetStore(store)
		id, etcdRaftNode = startEtcdRaftNode(cfg, store, cl, cl.MemberIDs())
	case logExist:
		if err = utils.IsDirWriteable(cfg.DataDir); err != nil {
			return fmt.Errorf("cannot write to member directory: %v", err)
		}
		appliedIndex = raft.MustGetAppliedIndex(store)
		cfg.AppliedIndex = appliedIndex
		id, etcdRaftNode, cl = restartEtcdNode(cfg, store)
		cl.SetStore(store)
	default:
		return fmt.Errorf("unsupported bootstrap config")
	}
	s.raftNode = raft.NewNode(
		raft.NodeConfig{
			IsIDRemoved: func(id uint64) bool { return cl.IsIDRemoved(types.ID(id)) },
			Node:        etcdRaftNode,
			Heartbeat:   cfg.HeartbeatMs,
			Storage:     store,
		},
	)
	s.raftNode.SetAppliedIndex(appliedIndex)
	s.id = id
	s.wait = wait.New()
	s.reqIDGen = idutil.NewGenerator(uint16(id), time.Now())
	s.stopping = make(chan struct{})
	s.errorc = make(chan error)
	s.applyBroadcast = utils.NewBroadcast()
	s.stats = stats.NewServerStats(cfg.Name, id.String())
	s.lstats = stats.NewLeaderStats(id.String())
	s.store = store
	tr := &rafthttp.Transport{
		TLSInfo:     transport.TLSInfo{},
		DialTimeout: DialTimeout,
		ID:          id,
		URLs:        cfg.PeerURLs,
		ClusterID:   cl.GetID(),
		Raft:        s,
		ServerStats: s.stats,
		LeaderStats: s.lstats,
		ErrorC:      s.errorc,
	}
	if err = tr.Start(); err != nil {
		return err
	}
	for _, m := range remotes {
		if m.ID != id {
			tr.AddRemote(m.ID, m.PeerURLs)
		}
	}
	for _, m := range cl.Members() {
		if m.ID != id {
			tr.AddPeer(m.ID, m.PeerURLs)
		}
	}
	s.raftNode.Transport = tr
	s.cluster = cl
	return nil
}
func (s *KingbusServer) startRaftPeer(peerURLs types.URLs) error {
	err := s.raftNode.NewPeerListener(peerURLs)
	if err != nil {
		return err
	}
	s.raftNode.SetPeerHandler()
	s.raftNode.PeerHandlerServe()
	log.Log.Infof("startRaftPeer success")
	return nil
}
func (s *KingbusServer) startAdminServer(urls types.URLs) error {
	if len(urls) != 1 {
		return ErrArgs
	}
	addr := urls[0].Host
	s.adminSvr = api.NewAdminServer(addr, s, s.cluster)
	return nil
}
func (s *KingbusServer) startPrometheus(addr string) error {
	if len(addr) == 0 {
		log.Log.Errorf("metrics addr is nil")
		return ErrArgs
	}
	s.prometheusSvr = NewPrometheusServer(addr, metrics.DefaultRegistry, prometheus.DefaultRegisterer, time.Second)
	return nil
}
func NewKingbusServer(cfg *config.KingbusServerConfig) (*KingbusServer, error) {
	s := new(KingbusServer)
	s.Cfg = cfg
	s.lead = atomic.NewUint64(0)
	err := s.starRaft(s.Cfg.RaftNodeCfg)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startRaft error,err:%s", err)
		return nil, err
	}
	err = s.startRaftPeer(s.Cfg.RaftNodeCfg.PeerURLs)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startRaftPeer error,err:%s", err)
		return nil, err
	}
	err = s.startAdminServer(s.Cfg.AdminURLs)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startAdminServer error,err:%s", err)
		return nil, err
	}
	err = s.newBinlogProgress()
	if err != nil {
		log.Log.Errorf("NewKingbusServer:newBinlogProgress error,err:%s", err)
		return nil, err
	}
	err = s.startPrometheus(s.Cfg.MetricsAddr)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startPrometheus error,err:%s", err)
		return nil, err
	}
	s.started = atomic.NewBool(false)
	return s, nil
}
func (s *KingbusServer) updateLeadership(leadChange bool) {
	log.Log.Debugf("updateLeadership:leadChange:%v,s.IsLeader:%v,s.IsSyncerStarted:%v,",
		leadChange, s.IsLeader(), s.IsSyncerStarted())
	log.Log.Debugf("updateLeadership:appliedIndex:%d,committedIndex:%d", s.getAppliedIndex(), s.getCommittedIndex())
	if s.IsLeader() == false && s.IsSyncerStarted() {
		log.Log.Infof("server[%s] is not leader,stop syncer", s.Cfg.RaftNodeCfg.PeerURLs.String())
		s.StopServer(config.SyncerServerType)
		return
	}
	if leadChange && s.IsLeader() {
		go func() {
			t := time.Now()
			s.Mu.Lock()
			s.leadElectedTime = t
			s.Mu.Unlock()
			<-time.After(time.Second * 10)
			value, err := s.store.Get(utils.StringToBytes(storage.SyncerArgsKey))
			if err != nil {
				log.Log.Errorf("store Get value error,err:%s,key:%s,not start syncer",
					err, storage.SyncerArgsKey)
				return
			}
			if value == nil {
				log.Log.Infof("syncer didn't start, ignore start syncer")
				return
			}
			var syncerArgs config.SyncerArgs
			err = json.Unmarshal(value, &syncerArgs)
			if err != nil {
				log.Log.Fatalf("json unmarshal error,err:%s,value:%v", err, value)
			}
			err = s.StartServer(config.SyncerServerType, &syncerArgs)
			if err != nil {
				log.Log.Errorf("Start syncer error,err:%s,syncerAgs:%v", err, syncerArgs)
				return
			}
			log.Log.Infof("server[%s] become leader,start syncer automatically", s.Cfg.RaftNodeCfg.PeerURLs[0].Host)
		}()
	}
	if s.stats != nil {
		s.stats.BecomeLeader()
	}
}
func (s *KingbusServer) runRaft() {
	rh := &raft.ReadyHandler{
		GetLead:          func() (lead uint64) { return s.getLead() },
		UpdateLead:       func(lead uint64) { s.setLead(lead) },
		UpdateLeadership: s.updateLeadership,
		UpdateCommittedIndex: func(ci uint64) {
			cci := s.getCommittedIndex()
			if ci > cci {
				s.setCommittedIndex(ci)
			}
		},
	}
	s.raftNode.Run(rh)
	for {
		select {
		case ap := <-s.raftNode.Apply():
			s.applyAll(&ap)
		case <-s.raftNode.DoneC():
			log.Log.Infof("runRaft:quit success")
			return
		}
	}
}
func (s *KingbusServer) Run() {
	if s.started.Load() {
		return
	}
	go s.runRaft()
	go s.adminSvr.Run()
	go s.SyncAdminURL()
	go s.prometheusSvr.Run()
	s.started.Store(true)
	for {
		select {
		case err := <-s.errorc:
			log.Log.Errorf("KingbusServer.Start: run error,need quit. error:%s", err)
			s.Stop()
			return
		case <-s.stopping:
			log.Log.Infof("KingbusServer.Start quit")
			return
		}
	}
}
func (s *KingbusServer) Stop() {
	if s.started.Load() == false {
		return
	}
	s.adminSvr.Stop()
	s.prometheusSvr.Stop()
	s.StopServer(config.SyncerServerType)
	s.StopServer(config.BinlogServerType)
	s.raftNode.Stop()
	close(s.stopping)
	s.started.Store(false)
	return
}
func startEtcdRaftNode(cfg config.RaftNodeConfig, store storage.Storage, cl *membership.RaftCluster, ids []types.ID) (
	id types.ID, n etcdraft.Node) {
	member := cl.MemberByName(cfg.Name)
	peers := make([]etcdraft.Peer, len(ids))
	for i, id := range ids {
		ctx, err := json.Marshal((*cl).Member(id))
		if err != nil {
			log.Log.Panicf("marshal member should never fail: %v", err)
		}
		peers[i] = etcdraft.Peer{ID: uint64(id), Context: ctx}
	}
	id = member.ID
	log.Log.Infof("starting member %s in cluster %s", id, cl.GetID())
	c := &etcdraft.Config{
		ID:                        uint64(id),
		ElectionTick:              int(cfg.ElectionTimeoutMs / cfg.HeartbeatMs),
		HeartbeatTick:             1,
		Storage:                   store,
		MaxSizePerMsg:             cfg.MaxRequestBytes,
		MaxInflightMsgs:           maxInflightMsgs,
		CheckQuorum:               true,
		PreVote:                   cfg.PreVote,
		DisableProposalForwarding: true,
		Logger:                    log.Log,
	}
	n = etcdraft.StartNode(c, peers)
	raft.AdvanceTicks(n, c.ElectionTick)
	return id, n
}
func restartEtcdNode(cfg config.RaftNodeConfig, store storage.Storage) (
	types.ID, etcdraft.Node, *membership.RaftCluster) {
	cl, err := membership.GetRaftClusterFromStorage(store)
	if err != nil {
		if err != nil {
			log.Log.Panic("GetRaftClusterFromStorage error:%s", err.Error())
		}
	}
	log.Log.Debugf("restartEtcdNode:get raft cluster from storage,cluster:%v", cl.String())
	member := cl.MemberByName(cfg.Name)
	if member == nil {
		log.Log.Fatalf("restartEtcdNode:member not in raft cluster,cluster:%v,memberName:%s",
			cl.String(), cfg.Name)
	}
	c := &etcdraft.Config{
		ID:                        uint64(member.ID),
		ElectionTick:              int(cfg.ElectionTimeoutMs / cfg.HeartbeatMs),
		HeartbeatTick:             1,
		Applied:                   cfg.AppliedIndex,  
		Storage:                   store,
		MaxSizePerMsg:             cfg.MaxRequestBytes,
		MaxInflightMsgs:           maxInflightMsgs,
		CheckQuorum:               true,
		PreVote:                   cfg.PreVote,
		DisableProposalForwarding: true,
		Logger:                    log.Log,
	}
	n := etcdraft.RestartNode(c)
	return member.ID, n, cl
}
func (s *KingbusServer) SyncAdminURL() {
	var id types.ID
	var leaderAdminURL string
	id = s.id
	urls := s.Cfg.AdminURLs.StringSlice()
	attributes := config.Attributes{
		ID:   id,
		Name: s.cluster.MembersMap[id].Name,
	}
	attributes.AdminURLs = make([]string, len(urls))
	copy(attributes.AdminURLs, urls)
	for uint64(s.Leader()) == etcdraft.None {
		time.Sleep(time.Millisecond * 500)
		log.Log.Debugf("SyncAdminURL:cluster has no leader,wait for 500ms and then check again")
	}
	if s.IsLeader() {
		data, err := attributes.EncodeWithType()
		if err != nil {
			log.Log.Errorf("SyncAdminURL:attributes EncodeWithType error,err:%v,attributes:%v", err, attributes)
			return
		}
		err = s.Propose(data)
		if err != nil {
			log.Log.Errorf("SyncAdminURL:Propose error,err:%v,attributes:%v", err, attributes)
			return
		}
		return
	}
	for len(leaderAdminURL) == 0 {
		leaderID := s.Leader()
		leader := s.cluster.Member(leaderID)
		if len(leader.AdminURLs) == 1 {
			leaderAdminURL = leader.AdminURLs[0]
			log.Log.Debugf("leader admin url is 1,leader:%v", *leader)
			break
		}
		log.Log.Debugf("SyncAdminURL:leader admin url is not 1,leader:%v", *leader)
	}
	leaderURL, err := url.Parse(leaderAdminURL)
	if err != nil {
		log.Log.Errorf("SyncAdminURL:Parse error,url:%s", leaderAdminURL)
		return
	}
	updateAdminURL := leaderURL.Scheme + "://" + leaderURL.Host + "/admin/url"
	req, err := json.Marshal(attributes)
	if err != nil {
		log.Log.Errorf("SyncAdminURL:json marshal error,err:%s,attributes:%v", err, attributes)
		return
	}
	resp, err := utils.SendRequest("PUT", updateAdminURL, req)
	if err != nil {
		log.Log.Errorf("SyncAdminURL:SendRequest error,err:%s,url:%s", err, updateAdminURL)
		return
	}
	if resp.Message != "success" {
		log.Log.Errorf("SyncAdminURL:resp.Message is not success,err:%s", resp.Message)
		return
	}
	log.Log.Infof("SyncAdminURL:sync admin url to leader success,leader admin url is:%s",
		attributes.AdminURLs[0])
}
func (s *KingbusServer) StartProposeBinlog(ctx context.Context) {
	go func() {
		for s.IsLeader() && s.IsSyncerStarted() {
			select {
			case <-ctx.Done():
				log.Log.Infof("StartProposeBinlog: receive cancel, stop propose binlog event")
				return
			case <-s.stopping:
				log.Log.Infof("kingbus server stopping")
				return
			case e := <-s.syncer.BinlogEventC():
				dataWithType, err := utils.EncodeBinlogEvent(e)
				if err != nil {
					select {
					case s.errorc <- err:
						log.Log.Errorf("EncodeBinlogEvent error,err:%s,event:%v", err, *e)
					default:
					}
					return
				}
				s.ProposeWithRetry(ctx, dataWithType)
			case d := <-s.syncer.DataC():
				s.ProposeWithRetry(ctx, d)
			}
		}
	}()
}