新增etcd认证配置

cluster/etcddiscovery.go

@@ -13,10 +13,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"go.uber.org/zap"
 	"path"
 	"strings"
 	"sync/atomic"
+	"io/ioutil"
+	"crypto/x509"
+	"crypto/tls"
 )
 
 const originDir = "/origin"
@@ -40,11 +42,16 @@ type EtcdDiscoveryService struct {
 	mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
 }
 
+var  etcdDiscovery *EtcdDiscoveryService
 func getEtcdDiscovery() IServiceDiscovery {
-	etcdDiscovery := &EtcdDiscoveryService{}
+	if etcdDiscovery == nil {
+		etcdDiscovery = &EtcdDiscoveryService{}
+	}
+
 	return etcdDiscovery
 }
 
+
 func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
 	ed.localNodeId = localNodeId
 
@@ -87,15 +94,44 @@ func (ed *EtcdDiscoveryService) OnInit() error {
 	}
 
 	for i := 0; i < len(etcdDiscoveryCfg.EtcdList); i++ {
-		client, cerr := clientv3.New(clientv3.Config{
+		var client *clientv3.Client
+		var tlsConfig *tls.Config
+
+		if etcdDiscoveryCfg.EtcdList[i].Cert != "" {
+			// load cert
+			cert, cerr := tls.LoadX509KeyPair(etcdDiscoveryCfg.EtcdList[i].Cert, etcdDiscoveryCfg.EtcdList[i].CertKey)
+			if cerr != nil {
+				log.Error("load cert error", log.ErrorField("err", cerr))
+				return cerr
+			}
+
+			// load root ca
+			caData, cerr := ioutil.ReadFile(etcdDiscoveryCfg.EtcdList[i].Ca)
+			if cerr != nil {
+				log.Error("load root ca error", log.ErrorField("err", cerr))
+				return cerr
+			}
+			pool := x509.NewCertPool()
+			pool.AppendCertsFromPEM(caData)
+			tlsConfig = &tls.Config{
+				Certificates: []tls.Certificate{cert},
+				RootCAs:      pool,
+			}
+		}
+
+		client, err = clientv3.New(clientv3.Config{
 			Endpoints:   etcdDiscoveryCfg.EtcdList[i].Endpoints,
 			DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
-			Logger:      zap.NewNop(),
+			Username: etcdDiscoveryCfg.EtcdList[i].UserName,
+			Password: etcdDiscoveryCfg.EtcdList[i].Password,
+			Logger:      log.GetLogger().Logger,
+			TLS: tlsConfig,
 		})
 
-		if cerr != nil {
-			log.Error("etcd discovery init fail", log.ErrorField("err", cerr))
-			return cerr
+
+		if err != nil {
+			log.Error("etcd discovery init fail", log.ErrorField("err", err))
+			return err
 		}
 
 		ctx, _ := context.WithTimeout(context.Background(), time.Second*3)

cluster/parsecfg.go

@@ -15,9 +15,15 @@ import (
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
 
+
 type EtcdList struct {
 	NetworkName []string
 	Endpoints   []string
+	UserName string
+	Password string
+	Cert string
+	CertKey string
+	Ca string
 }
 
 type EtcdDiscovery struct {

sysmodule/netmodule/wsmodule/WSModule.go

@@ -14,7 +14,7 @@ import (
 type WSModule struct {
 	service.Module
 
-	wsServer network.WSServer
+	WSServer network.WSServer
 
 	mapClientLocker sync.RWMutex
 	mapClient       map[string]*WSClient
@@ -57,16 +57,16 @@ func (ws *WSModule) OnInit() error {
 		return fmt.Errorf("please call the Init function correctly")
 	}
 
-	ws.wsServer.MaxConnNum = ws.wsCfg.MaxConnNum
-	ws.wsServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
-	ws.wsServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
-	ws.wsServer.Addr = ws.wsCfg.ListenAddr
+	ws.WSServer.MaxConnNum = ws.wsCfg.MaxConnNum
+	ws.WSServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
+	ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
+	ws.WSServer.Addr = ws.wsCfg.ListenAddr
 
 	//3.设置解析处理器
 	ws.process.SetByteOrder(ws.wsCfg.LittleEndian)
 
-	ws.mapClient = make(map[string]*WSClient, ws.wsServer.MaxConnNum)
-	ws.wsServer.NewAgent = ws.NewWSClient
+	ws.mapClient = make(map[string]*WSClient, ws.WSServer.MaxConnNum)
+	ws.WSServer.NewAgent = ws.NewWSClient
 
 	//4.设置网络事件处理
 	ws.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_WebSocket, ws.GetEventHandler(), ws.wsEventHandler)
@@ -80,7 +80,7 @@ func (ws *WSModule) Init(wsCfg *WSCfg, process processor.IRawProcessor) {
 }
 
 func (ws *WSModule) Start() error {
-	return ws.wsServer.Start()
+	return ws.WSServer.Start()
 }
 
 func (ws *WSModule) wsEventHandler(ev event.IEvent) {
@@ -197,3 +197,7 @@ func (ws *WSModule) SendRawMsg(clientId string, msg []byte) error {
 	ws.mapClientLocker.Unlock()
 	return client.wsConn.WriteMsg(msg)
 }
+
+func (ws *WSModule) SetMessageType(messageType int) {
+	ws.WSServer.SetMessageType(messageType)
+}

util/smath/smath.go

@@ -32,10 +32,10 @@ func Abs[NumType typ.Signed | typ.Float](Num NumType) NumType {
 func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
 	ret := number1 + number2
 	if number2 > 0 && ret < number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	} else if number2 < 0 && ret > number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	}
 
@@ -45,10 +45,10 @@ func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
 func SubSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
 	ret := number1 - number2
 	if number2 > 0 && ret > number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	} else if number2 < 0 && ret < number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	}
 
@@ -65,7 +65,7 @@ func MulSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
 		return ret, true
 	}
 
-	log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+	log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 	return ret, true
 }