mirror of
https://github.com/duanhf2012/origin.git
synced 2026-03-04 04:47:35 +08:00
新增etcd服务发现
This commit is contained in:
boyce
443
cluster/etcddiscovery.go
Normal file
443
cluster/etcddiscovery.go
Normal file
@@ -0,0 +1,443 @@
|
||||
package cluster
|
||||
|
||||
|
||||
import (
|
||||
"github.com/duanhf2012/origin/v2/event"
|
||||
"github.com/duanhf2012/origin/v2/log"
|
||||
"github.com/duanhf2012/origin/v2/rpc"
|
||||
"github.com/duanhf2012/origin/v2/service"
|
||||
"github.com/duanhf2012/origin/v2/util/timer"
|
||||
"go.etcd.io/etcd/api/v3/mvccpb"
|
||||
"go.etcd.io/etcd/client/v3"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"time"
|
||||
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
const originDir = "/origin"
|
||||
const testKey = originDir+"/_inner/_test_7501f3ed-b716-44c2-0090-fc1ed0166d7a"
|
||||
|
||||
type etcdClientInfo struct {
|
||||
watchKeys []string
|
||||
leaseID clientv3.LeaseID
|
||||
keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse
|
||||
}
|
||||
|
||||
type EtcdDiscoveryService struct {
|
||||
service.Service
|
||||
funDelNode FunDelNode
|
||||
funSetNode FunSetNode
|
||||
localNodeId string
|
||||
|
||||
byteLocalNodeInfo string
|
||||
mapClient map[*clientv3.Client]*etcdClientInfo
|
||||
isClose int32
|
||||
|
||||
mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
|
||||
}
|
||||
|
||||
func getEtcdDiscovery() (IServiceDiscovery) {
|
||||
etcdDiscovery := &EtcdDiscoveryService{}
|
||||
return etcdDiscovery
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string,funDelNode FunDelNode,funSetNode FunSetNode) error {
|
||||
ed.localNodeId = localNodeId
|
||||
|
||||
ed.funDelNode = funDelNode
|
||||
ed.funSetNode = funSetNode
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
const(
|
||||
eeGets = 0
|
||||
eePut = 1
|
||||
eeDelete = 2
|
||||
)
|
||||
|
||||
type etcdDiscoveryEvent struct {
|
||||
typ int
|
||||
watchKey string
|
||||
Kvs []*mvccpb.KeyValue
|
||||
}
|
||||
|
||||
func (ee *etcdDiscoveryEvent) GetEventType() event.EventType{
|
||||
return event.Sys_Event_EtcdDiscovery
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnInit() error {
|
||||
ed.mapClient = make(map[*clientv3.Client]*etcdClientInfo,1)
|
||||
ed.mapDiscoveryNodeId = make(map[string]map[string]struct{})
|
||||
|
||||
ed.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_EtcdDiscovery, ed.GetEventHandler(), ed.OnEtcdDiscovery)
|
||||
|
||||
err := ed.marshalNodeInfo()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
etcdDiscoveryCfg := cluster.GetEtcdDiscovery()
|
||||
if etcdDiscoveryCfg == nil {
|
||||
return errors.New("etcd discovery config is nil.")
|
||||
}
|
||||
|
||||
for i:=0;i<len(etcdDiscoveryCfg.EtcdList);i++{
|
||||
client, cerr := clientv3.New(clientv3.Config{
|
||||
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
|
||||
DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
|
||||
})
|
||||
|
||||
if cerr != nil {
|
||||
log.Error("etcd discovery init fail",log.ErrorAttr("err",cerr))
|
||||
return cerr
|
||||
}
|
||||
|
||||
ctx,_:=context.WithTimeout(context.Background(),time.Second*3)
|
||||
_,err = client.Put(ctx,testKey,"")
|
||||
if err != nil {
|
||||
log.Error("etcd discovery init fail",log.Any("endpoint",etcdDiscoveryCfg.EtcdList[i].Endpoints),log.ErrorAttr("err",err))
|
||||
return err
|
||||
}
|
||||
|
||||
ec := &etcdClientInfo{}
|
||||
for _, networkName := range etcdDiscoveryCfg.EtcdList[i].NetworkName {
|
||||
ec.watchKeys = append(ec.watchKeys,fmt.Sprintf("%s/%s",originDir,networkName))
|
||||
//ec.etcdKey = append(ec.etcdKey,fmt.Sprintf("%s/%s/%s",originDir,networkName,nd.localNodeId))
|
||||
}
|
||||
|
||||
ed.mapClient[client] = ec
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) getRegisterKey(watchkey string) string{
|
||||
return watchkey+"/"+ed.localNodeId
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) registerServiceByClient(client *clientv3.Client,etcdClient *etcdClientInfo) {
|
||||
// 创建租约
|
||||
var err error
|
||||
var resp *clientv3.LeaseGrantResponse
|
||||
resp, err = client.Grant(context.Background(), cluster.GetEtcdDiscovery().TTLSecond)
|
||||
if err != nil {
|
||||
log.Error("etcd registerService fail",log.ErrorAttr("err",err))
|
||||
ed.tryRegisterService(client,etcdClient)
|
||||
return
|
||||
}
|
||||
|
||||
etcdClient.leaseID = resp.ID
|
||||
for _,watchKey:= range etcdClient.watchKeys {
|
||||
// 注册服务节点到 etcd
|
||||
_, err = client.Put(context.Background(), ed.getRegisterKey(watchKey), ed.byteLocalNodeInfo, clientv3.WithLease(resp.ID))
|
||||
if err != nil {
|
||||
log.Error("etcd Put fail",log.ErrorAttr("err",err))
|
||||
ed.tryRegisterService(client,etcdClient)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
etcdClient.keepAliveChan,err = client.KeepAlive(context.Background(), etcdClient.leaseID)
|
||||
if err != nil {
|
||||
log.Error("etcd KeepAlive fail",log.ErrorAttr("err",err))
|
||||
ed.tryRegisterService(client,etcdClient)
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case _, ok := <-etcdClient.keepAliveChan:
|
||||
//log.Debug("ok",log.Any("addr",client.Endpoints()))
|
||||
if !ok {
|
||||
log.Error("etcd keepAliveChan fail",log.Any("watchKeys",etcdClient.watchKeys))
|
||||
ed.tryRegisterService(client,etcdClient)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
|
||||
func (ed *EtcdDiscoveryService) tryRegisterService(client *clientv3.Client,etcdClient *etcdClientInfo){
|
||||
if ed.isStop() {
|
||||
return
|
||||
}
|
||||
|
||||
ed.AfterFunc(time.Second*3, func(t *timer.Timer) {
|
||||
ed.registerServiceByClient(client,etcdClient)
|
||||
})
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) tryWatch(client *clientv3.Client,etcdClient *etcdClientInfo) {
|
||||
if ed.isStop() {
|
||||
return
|
||||
}
|
||||
ed.AfterFunc(time.Second*3, func(t *timer.Timer) {
|
||||
ed.watchByClient(client,etcdClient)
|
||||
})
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnRetire(){
|
||||
atomic.StoreInt32(&ed.isClose,1)
|
||||
ed.close()
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnRelease(){
|
||||
atomic.StoreInt32(&ed.isClose,1)
|
||||
ed.close()
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) isStop() bool{
|
||||
return atomic.LoadInt32(&ed.isClose) == 1
|
||||
}
|
||||
|
||||
func (nd *EtcdDiscoveryService) OnStart() {
|
||||
for c, ec := range nd.mapClient {
|
||||
nd.tryRegisterService(c,ec)
|
||||
nd.tryWatch(c,ec)
|
||||
}
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) marshalNodeInfo() error{
|
||||
nInfo := cluster.GetLocalNodeInfo()
|
||||
var nodeInfo rpc.NodeInfo
|
||||
nodeInfo.NodeId = nInfo.NodeId
|
||||
nodeInfo.ListenAddr = nInfo.ListenAddr
|
||||
nodeInfo.Retire = nInfo.Retire
|
||||
nodeInfo.PublicServiceList = nInfo.PublicServiceList
|
||||
nodeInfo.MaxRpcParamLen = nInfo.MaxRpcParamLen
|
||||
|
||||
byteLocalNodeInfo,err := proto.Marshal(&nodeInfo)
|
||||
if err ==nil{
|
||||
ed.byteLocalNodeInfo = string(byteLocalNodeInfo)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) setNodeInfo(networkName string,nodeInfo *rpc.NodeInfo) bool{
|
||||
if nodeInfo == nil || nodeInfo.Private == true || nodeInfo.NodeId == ed.localNodeId {
|
||||
return false
|
||||
}
|
||||
|
||||
//筛选关注的服务
|
||||
var discoverServiceSlice = make([]string, 0, 24)
|
||||
for _, pubService := range nodeInfo.PublicServiceList {
|
||||
if cluster.CanDiscoveryService(networkName,pubService) == true {
|
||||
discoverServiceSlice = append(discoverServiceSlice,pubService)
|
||||
}
|
||||
}
|
||||
|
||||
if len(discoverServiceSlice) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
var nInfo NodeInfo
|
||||
nInfo.ServiceList = discoverServiceSlice
|
||||
nInfo.PublicServiceList = discoverServiceSlice
|
||||
nInfo.NodeId = nodeInfo.NodeId
|
||||
nInfo.ListenAddr = nodeInfo.ListenAddr
|
||||
nInfo.MaxRpcParamLen = nodeInfo.MaxRpcParamLen
|
||||
nInfo.Retire = nodeInfo.Retire
|
||||
nInfo.Private = nodeInfo.Private
|
||||
|
||||
ed.funSetNode(&nInfo)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) close(){
|
||||
for c, ec := range ed.mapClient {
|
||||
if _, err := c.Revoke(context.Background(), ec.leaseID); err != nil {
|
||||
log.Error("etcd Revoke fail",log.ErrorAttr("err",err))
|
||||
}
|
||||
c.Watcher.Close()
|
||||
err := c.Close()
|
||||
if err != nil {
|
||||
log.Error("etcd Close fail",log.ErrorAttr("err",err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) getServices(client *clientv3.Client,etcdClient *etcdClientInfo,watchKey string) bool {
|
||||
// 根据前缀获取现有的key
|
||||
resp, err := client.Get(context.Background(), watchKey, clientv3.WithPrefix())
|
||||
if err != nil {
|
||||
log.Error("etcd Get fail", log.ErrorAttr("err", err))
|
||||
ed.tryWatch(client, etcdClient)
|
||||
return false
|
||||
}
|
||||
|
||||
// 遍历获取得到的k和v
|
||||
ed.notifyGets(watchKey,resp.Kvs)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) watchByClient(client *clientv3.Client,etcdClient *etcdClientInfo){
|
||||
//先关闭所有的watcher
|
||||
for _, watchKey := range etcdClient.watchKeys {
|
||||
// 监视前缀，修改变更server
|
||||
go ed.watcher(client,etcdClient, watchKey)
|
||||
}
|
||||
}
|
||||
|
||||
// watcher 监听Key的前缀
|
||||
func (ed *EtcdDiscoveryService) watcher(client *clientv3.Client,etcdClient *etcdClientInfo,watchKey string) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
buf := make([]byte, 4096)
|
||||
l := runtime.Stack(buf, false)
|
||||
errString := fmt.Sprint(r)
|
||||
log.Dump(string(buf[:l]),log.String("error",errString))
|
||||
|
||||
ed.tryWatch(client,etcdClient)
|
||||
}
|
||||
}()
|
||||
|
||||
log.Debug(">>try watcher")
|
||||
rch := client.Watch(context.Background(), watchKey, clientv3.WithPrefix())
|
||||
|
||||
if ed.getServices(client,etcdClient,watchKey) == false {
|
||||
return
|
||||
}
|
||||
|
||||
for wresp := range rch {
|
||||
for _, ev := range wresp.Events {
|
||||
switch ev.Type {
|
||||
case clientv3.EventTypePut: // 修改或者新增
|
||||
ed.notifyPut(watchKey,ev.Kv)
|
||||
case clientv3.EventTypeDelete: // 删除
|
||||
ed.notifyDelete(watchKey,ev.Kv)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ed.tryWatch(client,etcdClient)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) setNode(netWorkName string,byteNode []byte) string{
|
||||
var nodeInfo rpc.NodeInfo
|
||||
err := proto.Unmarshal(byteNode,&nodeInfo)
|
||||
if err != nil {
|
||||
log.Error("Unmarshal fail",log.String("netWorkName",netWorkName),log.ErrorAttr("err",err))
|
||||
return ""
|
||||
}
|
||||
|
||||
ed.setNodeInfo(netWorkName,&nodeInfo)
|
||||
|
||||
return nodeInfo.NodeId
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) delNode(fullKey string) string{
|
||||
nodeId := ed.getNodeId(fullKey)
|
||||
if nodeId == ed.localNodeId {
|
||||
return ""
|
||||
}
|
||||
|
||||
ed.funDelNode(nodeId,true)
|
||||
return nodeId
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) getNetworkNameByWatchKey(watchKey string) string{
|
||||
return watchKey[strings.LastIndex(watchKey,"/")+1:]
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) getNetworkNameByFullKey(fullKey string) string{
|
||||
return fullKey[len(originDir)+1:strings.LastIndex(fullKey,"/")]
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) getNodeId(fullKey string) string{
|
||||
return fullKey[strings.LastIndex(fullKey,"/")+1:]
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnEtcdDiscovery(ev event.IEvent){
|
||||
disEvent := ev.(*etcdDiscoveryEvent)
|
||||
switch disEvent.typ {
|
||||
case eeGets:
|
||||
ed.OnEventGets(disEvent.watchKey,disEvent.Kvs)
|
||||
case eePut:
|
||||
if len(disEvent.Kvs) == 1 {
|
||||
ed.OnEventPut(disEvent.watchKey,disEvent.Kvs[0])
|
||||
}
|
||||
case eeDelete:
|
||||
if len(disEvent.Kvs) == 1 {
|
||||
ed.OnEventDelete(disEvent.watchKey,disEvent.Kvs[0])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) notifyGets(watchKey string,Kvs []*mvccpb.KeyValue) {
|
||||
var ev etcdDiscoveryEvent
|
||||
ev.typ = eeGets
|
||||
ev.watchKey = watchKey
|
||||
ev.Kvs = Kvs
|
||||
ed.NotifyEvent(&ev)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) notifyPut(watchKey string,Kvs *mvccpb.KeyValue) {
|
||||
var ev etcdDiscoveryEvent
|
||||
ev.typ = eePut
|
||||
ev.watchKey = watchKey
|
||||
ev.Kvs = append(ev.Kvs,Kvs)
|
||||
ed.NotifyEvent(&ev)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) notifyDelete(watchKey string,Kvs *mvccpb.KeyValue) {
|
||||
var ev etcdDiscoveryEvent
|
||||
ev.typ = eeDelete
|
||||
ev.watchKey = watchKey
|
||||
ev.Kvs = append(ev.Kvs,Kvs)
|
||||
ed.NotifyEvent(&ev)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnEventGets(watchKey string,Kvs []*mvccpb.KeyValue) {
|
||||
mapNode := make(map[string]struct{},32)
|
||||
for _, kv := range Kvs {
|
||||
nodeId := ed.setNode(ed.getNetworkNameByFullKey(string(kv.Key)), kv.Value)
|
||||
mapNode[nodeId] = struct{}{}
|
||||
ed.addNodeId(watchKey,nodeId)
|
||||
}
|
||||
|
||||
log.Debug(">>etcd OnEventGets",log.String("watchKey",watchKey),log.Any("mapNode",mapNode))
|
||||
// 此段代码为遍历并删除过期节点的逻辑。
|
||||
// 对于mapDiscoveryNodeId中与watchKey关联的所有节点ID，遍历该集合。
|
||||
// 如果某个节点ID不在mapNode中且不是本地节点ID，则调用funDelNode函数删除该节点。
|
||||
mapLastNodeId := ed.mapDiscoveryNodeId[watchKey] // 根据watchKey获取对应的节点ID集合
|
||||
for nodeId := range mapLastNodeId { // 遍历所有节点ID
|
||||
if _,ok := mapNode[nodeId];ok == false && nodeId != ed.localNodeId { // 检查节点是否不存在于mapNode且不是本地节点
|
||||
ed.funDelNode(nodeId,true) // 调用函数删除该节点
|
||||
delete(ed.mapDiscoveryNodeId[watchKey],nodeId)
|
||||
log.Debug(">>etcd OnEventGets Delete",log.String("watchKey",watchKey),log.String("nodeId",nodeId))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnEventPut(watchKey string,Kv *mvccpb.KeyValue) {
|
||||
log.Debug(">>etcd OnEventPut",log.String("watchKey",watchKey),log.String("nodeId",ed.getNodeId(string(Kv.Key))))
|
||||
nodeId := ed.setNode(ed.getNetworkNameByFullKey(string(Kv.Key)), Kv.Value)
|
||||
ed.addNodeId(watchKey,nodeId)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) OnEventDelete(watchKey string,Kv *mvccpb.KeyValue) {
|
||||
log.Debug(">>etcd OnEventDelete",log.String("watchKey",watchKey),log.String("nodeId",ed.getNodeId(string(Kv.Key))))
|
||||
nodeId := ed.delNode(string(Kv.Key))
|
||||
delete(ed.mapDiscoveryNodeId[watchKey],nodeId)
|
||||
}
|
||||
|
||||
func (ed *EtcdDiscoveryService) addNodeId(watchKey string,nodeId string) {
|
||||
if _,ok := ed.mapDiscoveryNodeId[watchKey];ok == false {
|
||||
ed.mapDiscoveryNodeId[watchKey] = make(map[string]struct{})
|
||||
}
|
||||
|
||||
ed.mapDiscoveryNodeId[watchKey][nodeId] = struct{}{}
|
||||
}
|
||||
Reference in New Issue
Block a user