Compare commits

...

4 Commits

Author SHA1 Message Date
boyce
50dd80b082 整理代码 2025-03-10 11:35:19 +08:00
boyce
a6487dd41e 将默认日志改为rotatelogs 2025-01-25 00:14:18 +08:00
boyce
d5299294d8 优化日志,新增rotatelogs库支持 2025-01-25 00:04:31 +08:00
boyce
4d36e525a5 优化配置读取,去消默认cluster目录 2025-01-16 13:45:06 +08:00
8 changed files with 196 additions and 117 deletions

View File

@@ -40,8 +40,6 @@ type NodeInfo struct {
DiscoveryService []DiscoveryService //筛选发现的服务,如果不配置,不进行筛选
status NodeStatus
Retire bool
NetworkName string
}
type NodeRpcInfo struct {

View File

@@ -3,24 +3,23 @@ package cluster
import "github.com/duanhf2012/origin/v2/rpc"
type ConfigDiscovery struct {
funDelNode FunDelNode
funSetNode FunSetNode
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
}
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode FunDelNode,funSetNode FunSetNode) error{
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
discovery.localNodeId = localNodeId
discovery.funDelNode = funDelNode
discovery.funSetNode = funSetNode
//解析本地其他服务配置
_,nodeInfoList,_,err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull)
_, nodeInfoList, _, err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull)
if err != nil {
return err
}
for _,nodeInfo := range nodeInfoList {
for _, nodeInfo := range nodeInfoList {
if nodeInfo.NodeId == localNodeId {
continue
}
@@ -30,5 +29,3 @@ func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode Fu
return nil
}

View File

@@ -5,17 +5,17 @@ import (
"github.com/duanhf2012/origin/v2/service"
)
func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.discoveryInfo.getDiscoveryType() == OriginType { //origin类型服务发现
return cls.setupOriginDiscovery(localNodeId,setupServiceFun)
}else if cls.discoveryInfo.getDiscoveryType() == EtcdType{//etcd类型服务发现
return cls.setupEtcdDiscovery(localNodeId,setupServiceFun)
return cls.setupOriginDiscovery(localNodeId, setupServiceFun)
} else if cls.discoveryInfo.getDiscoveryType() == EtcdType { //etcd类型服务发现
return cls.setupEtcdDiscovery(localNodeId, setupServiceFun)
}
return cls.setupConfigDiscovery(localNodeId,setupServiceFun)
return cls.setupConfigDiscovery(localNodeId, setupServiceFun)
}
func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}
@@ -27,6 +27,7 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
}
cls.serviceDiscovery = getOriginDiscovery()
//2.如果为动态服务发现安装本地发现服务
if localMaster == true {
setupServiceFun(&masterService)
@@ -36,11 +37,10 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
setupServiceFun(&clientService)
cls.AddDiscoveryService(OriginDiscoveryClientName, true)
return nil
}
func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}
@@ -48,12 +48,12 @@ func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun Setup
//setup etcd service
cls.serviceDiscovery = getEtcdDiscovery()
setupServiceFun(cls.serviceDiscovery.(service.IService))
cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(),false)
cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(), false)
return nil
}
func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{
func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup")
}

View File

@@ -1,6 +1,11 @@
package cluster
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"github.com/duanhf2012/origin/v2/event"
"github.com/duanhf2012/origin/v2/log"
"github.com/duanhf2012/origin/v2/rpc"
@@ -9,16 +14,11 @@ import (
"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/client/v3"
"google.golang.org/protobuf/proto"
"time"
"context"
"errors"
"fmt"
"os"
"path"
"strings"
"sync/atomic"
"io/ioutil"
"crypto/x509"
"crypto/tls"
"time"
)
const originDir = "/origin"
@@ -42,7 +42,8 @@ type EtcdDiscoveryService struct {
mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
}
var etcdDiscovery *EtcdDiscoveryService
var etcdDiscovery *EtcdDiscoveryService
func getEtcdDiscovery() IServiceDiscovery {
if etcdDiscovery == nil {
etcdDiscovery = &EtcdDiscoveryService{}
@@ -51,7 +52,6 @@ func getEtcdDiscovery() IServiceDiscovery {
return etcdDiscovery
}
func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
ed.localNodeId = localNodeId
@@ -99,17 +99,17 @@ func (ed *EtcdDiscoveryService) OnInit() error {
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
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
caData, cErr := os.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)
@@ -122,13 +122,12 @@ func (ed *EtcdDiscoveryService) OnInit() error {
client, err = clientv3.New(clientv3.Config{
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
Username: etcdDiscoveryCfg.EtcdList[i].UserName,
Password: etcdDiscoveryCfg.EtcdList[i].Password,
Username: etcdDiscoveryCfg.EtcdList[i].UserName,
Password: etcdDiscoveryCfg.EtcdList[i].Password,
Logger: log.GetLogger().Logger,
TLS: tlsConfig,
TLS: tlsConfig,
})
if err != nil {
log.Error("etcd discovery init fail", log.ErrorField("err", err))
return err

View File

@@ -7,6 +7,7 @@ import (
"github.com/duanhf2012/origin/v2/rpc"
jsoniter "github.com/json-iterator/go"
"gopkg.in/yaml.v3"
"io/fs"
"os"
"path/filepath"
"strings"
@@ -70,12 +71,8 @@ type NodeInfoList struct {
NodeList []NodeInfo
}
func validConfigFile(f os.DirEntry) bool {
if f.IsDir() == true || (filepath.Ext(f.Name()) != ".json" && filepath.Ext(f.Name()) != ".yml" && filepath.Ext(f.Name()) != ".yaml") {
return false
}
return true
func validConfigFile(f string) bool {
return strings.HasSuffix(f, ".json")|| strings.HasSuffix(f, ".yml") || strings.HasSuffix(f, ".yaml")
}
func yamlToJson(data []byte, v interface{}) ([]byte, error) {
@@ -277,32 +274,33 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
var discoveryInfo DiscoveryInfo
var rpcMode RpcMode
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return discoveryInfo, nil, rpcMode, fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _, f := range fileInfoList {
if !validConfigFile(f) {
continue
err := filepath.Walk(configDir, func(path string, info fs.FileInfo, err error)error {
if info.IsDir() {
return nil
}
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
fileNodeInfoList, rErr := cls.ReadClusterConfig(filePath)
if err != nil {
return err
}
if !validConfigFile(info.Name()) {
return nil
}
fileNodeInfoList, rErr := cls.ReadClusterConfig(path)
if rErr != nil {
return discoveryInfo, nil, rpcMode, fmt.Errorf("read file path %s is error:%+v", filePath, rErr)
return fmt.Errorf("read file path %s is error:%+v", path, rErr)
}
err = cls.SetRpcMode(&fileNodeInfoList.RpcMode, &rpcMode)
if err != nil {
return discoveryInfo, nil, rpcMode, err
return err
}
err = discoveryInfo.setDiscovery(&fileNodeInfoList.Discovery)
if err != nil {
return discoveryInfo, nil, rpcMode, err
return err
}
for _, nodeInfo := range fileNodeInfoList.NodeList {
@@ -310,6 +308,12 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
nodeInfoList = append(nodeInfoList, nodeInfo)
}
}
return nil
})
if err != nil {
return discoveryInfo, nil, rpcMode, err
}
if nodeId != rpc.NodeIdNull && (len(nodeInfoList) != 1) {
@@ -331,32 +335,32 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
}
func (cls *Cluster) readLocalService(localNodeId string) error {
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
var globalCfg interface{}
publicService := map[string]interface{}{}
nodeService := map[string]interface{}{}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _, f := range fileInfoList {
if !validConfigFile(f) {
continue
err := filepath.Walk(configDir, func(path string, info fs.FileInfo, err error)error{
if info.IsDir() {
return nil
}
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
if err != nil {
continue
return err
}
if !validConfigFile(info.Name()) {
return nil
}
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(path)
if err != nil {
return err
}
if currGlobalCfg != nil {
//不允许重复的配置global配置
if globalCfg != nil {
return fmt.Errorf("[Global] does not allow repeated configuration in %s", f.Name())
return fmt.Errorf("[Global] does not allow repeated configuration in %s", info.Name())
}
globalCfg = currGlobalCfg
}
@@ -372,7 +376,7 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
pubCfg, ok := serviceConfig[s]
if ok == true {
if _, publicOk := publicService[s]; publicOk == true {
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s", s, f.Name())
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s", s, info.Name())
}
publicService[s] = pubCfg
}
@@ -388,12 +392,17 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
}
if _, nodeOK := nodeService[s]; nodeOK == true {
return fmt.Errorf("NodeService NodeId[%s] Service[%s] does not allow repeated configuration in %s", cls.localNodeInfo.NodeId, s, f.Name())
return fmt.Errorf("NodeService NodeId[%s] Service[%s] does not allow repeated configuration in %s", cls.localNodeInfo.NodeId, s, info.Name())
}
nodeService[s] = nodeCfg
break
}
}
return nil
})
if err != nil {
return err
}
//组合所有的配置

View File

@@ -1,31 +1,37 @@
package log
import (
"github.com/duanhf2012/rotatelogs"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"gopkg.in/natefinch/lumberjack.v2"
"os"
"path/filepath"
"strings"
"time"
)
var isSetLogger bool
var gLogger = NewDefaultLogger()
var LogLevel zapcore.Level
var MaxSize int
var LogPath string
var OpenConsole *bool
var LogChanLen int
type Logger struct {
*zap.Logger
stack bool
OpenConsole *bool
LogPath string
FileName string
Skip int
LogLevel zapcore.Level
Encoder zapcore.Encoder
LogConfig *lumberjack.Logger
SugaredLogger *zap.SugaredLogger
CoreList []zapcore.Core
FileName string
Skip int
Encoder zapcore.Encoder
SugaredLogger *zap.SugaredLogger
CoreList []zapcore.Core
WriteSyncerFun []func() zapcore.WriteSyncer
}
// 设置Logger
func SetLogger(logger *Logger) {
if logger != nil {
gLogger = logger
@@ -33,6 +39,15 @@ func SetLogger(logger *Logger) {
}
}
// 设置ZapLogger
func SetZapLogger(zapLogger *zap.Logger) {
if zapLogger != nil {
gLogger = &Logger{}
gLogger.Logger = zapLogger
isSetLogger = true
}
}
func GetLogger() *Logger {
return gLogger
}
@@ -47,8 +62,8 @@ func (logger *Logger) SetSkip(skip int) {
func GetJsonEncoder() zapcore.Encoder {
encoderConfig := zap.NewProductionEncoderConfig()
encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
encoderConfig.EncodeCaller = zapcore.ShortCallerEncoder
encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
enc.AppendString(t.Format("2006-01-02 15:04:05.000"))
}
@@ -67,51 +82,100 @@ func GetTxtEncoder() zapcore.Encoder {
return zapcore.NewConsoleEncoder(encoderConfig)
}
func getLogConfig() *lumberjack.Logger {
func (logger *Logger) getLogConfig() *lumberjack.Logger {
return &lumberjack.Logger{
Filename: "",
MaxSize: 2048,
Filename: filepath.Join(LogPath, logger.FileName),
MaxSize: MaxSize,
MaxBackups: 0,
MaxAge: 0,
Compress: false,
LocalTime: true,
}
}
func NewDefaultLogger() *Logger {
logger := Logger{}
logger.Encoder = GetJsonEncoder()
logger.LogConfig = getLogConfig()
logger.LogConfig.LocalTime = true
core := zapcore.NewCore(logger.Encoder, zapcore.AddSync(os.Stdout), zap.InfoLevel)
logger.Logger = zap.New(core, zap.AddCaller(), zap.AddCallerSkip(1))
logger.Init()
return &logger
}
func (logger *Logger) SetLogLevel(level zapcore.Level) {
logger.LogLevel = level
func (logger *Logger) SetSyncers(syncers ...func() zapcore.WriteSyncer) {
logger.WriteSyncerFun = syncers
}
func (logger *Logger) AppendSyncerFun(syncerFun func() zapcore.WriteSyncer) {
logger.WriteSyncerFun = append(logger.WriteSyncerFun, syncerFun)
}
func SetLogLevel(level zapcore.Level) {
LogLevel = level
}
func (logger *Logger) Enabled(zapcore.Level) bool {
return logger.stack
}
func (logger *Logger) NewLumberjackWriter() zapcore.WriteSyncer {
return zapcore.AddSync(
&lumberjack.Logger{
Filename: filepath.Join(LogPath, logger.FileName),
MaxSize: MaxSize,
MaxBackups: 0,
MaxAge: 0,
Compress: false,
LocalTime: true,
})
}
func (logger *Logger) NewRotatelogsWriter() zapcore.WriteSyncer {
var options []rotatelogs.Option
if MaxSize > 0 {
options = append(options, rotatelogs.WithRotateMaxSize(int64(MaxSize)))
}
if LogChanLen > 0 {
options = append(options, rotatelogs.WithChannelLen(LogChanLen))
}
options = append(options, rotatelogs.WithRotationTime(time.Hour*24))
fileName := strings.TrimRight(logger.FileName, filepath.Ext(logger.FileName))
rotateLogs, err := rotatelogs.NewRotateLogs(LogPath, "20060102/"+fileName+"_20060102_150405", options...)
if err != nil {
panic(err)
}
return zapcore.AddSync(rotateLogs)
}
func (logger *Logger) Init() {
if isSetLogger {
return
}
var syncerList []zapcore.WriteSyncer
if logger.WriteSyncerFun == nil {
syncerList = append(syncerList, logger.NewRotatelogsWriter())
} else {
for _, syncer := range logger.WriteSyncerFun {
syncerList = append(syncerList, syncer())
}
}
var coreList []zapcore.Core
if logger.OpenConsole == nil || *logger.OpenConsole {
core := zapcore.NewCore(logger.Encoder, zapcore.AddSync(os.Stdout), logger.LogLevel)
if OpenConsole == nil || *OpenConsole {
syncerList = append(syncerList, zapcore.AddSync(os.Stdout))
}
for _, writer := range syncerList {
core := zapcore.NewCore(logger.Encoder, writer, LogLevel)
coreList = append(coreList, core)
}
if logger.CoreList != nil {
coreList = append(coreList, logger.CoreList...)
}else if logger.LogPath != "" {
WriteSyncer := zapcore.AddSync(logger.LogConfig)
core := zapcore.NewCore(logger.Encoder, WriteSyncer, logger.LogLevel)
coreList = append(coreList, core)
}
core := zapcore.NewTee(coreList...)

View File

@@ -17,7 +17,6 @@ import (
_ "net/http/pprof"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
@@ -59,6 +58,7 @@ func init() {
console.RegisterCommandString("loglevel", "debug", "<-loglevel debug|info|warn|error|stackerror|fatal> Set loglevel.", setLevel)
console.RegisterCommandString("logpath", "", "<-logpath path> Set log file path.", setLogPath)
console.RegisterCommandInt("logsize", 0, "<-logsize size> Set log size(MB).", setLogSize)
console.RegisterCommandInt("logchanlen", 0, "<-logchanlen len> Set log channel len.", setLogChanLen)
console.RegisterCommandString("pprof", "", "<-pprof ip:port> Open performance analysis.", setPprof)
}
@@ -220,7 +220,7 @@ func initNode(id string) {
func initLog() error {
logger := log.GetLogger()
if logger.LogPath == "" {
if log.LogPath == "" {
err := setLogPath("./log")
if err != nil {
return err
@@ -230,7 +230,6 @@ func initLog() error {
localNodeInfo := cluster.GetCluster().GetLocalNodeInfo()
fileName := fmt.Sprintf("%s.log", localNodeInfo.NodeId)
logger.FileName = fileName
logger.LogConfig.Filename = filepath.Join(logger.LogPath, logger.FileName)
logger.Init()
return nil
@@ -440,10 +439,10 @@ func openConsole(args interface{}) error {
strOpen := strings.ToLower(strings.TrimSpace(args.(string)))
if strOpen == "false" {
bOpenConsole := false
log.GetLogger().OpenConsole = &bOpenConsole
log.OpenConsole = &bOpenConsole
} else if strOpen == "true" {
bOpenConsole := true
log.GetLogger().OpenConsole = &bOpenConsole
log.OpenConsole = &bOpenConsole
} else {
return errors.New("parameter console error")
}
@@ -458,17 +457,17 @@ func setLevel(args interface{}) error {
strlogLevel := strings.TrimSpace(args.(string))
switch strlogLevel {
case "debug":
log.GetLogger().LogLevel = zapcore.DebugLevel
log.LogLevel = zapcore.DebugLevel
case "info":
log.GetLogger().LogLevel = zapcore.InfoLevel
log.LogLevel = zapcore.InfoLevel
case "warn":
log.GetLogger().LogLevel = zapcore.WarnLevel
log.LogLevel = zapcore.WarnLevel
case "error":
log.GetLogger().LogLevel = zapcore.ErrorLevel
log.LogLevel = zapcore.ErrorLevel
case "stackerror":
log.GetLogger().LogLevel = zapcore.ErrorLevel
log.LogLevel = zapcore.ErrorLevel
case "fatal":
log.GetLogger().LogLevel = zapcore.FatalLevel
log.LogLevel = zapcore.FatalLevel
default:
return errors.New("unknown level: " + strlogLevel)
}
@@ -481,7 +480,7 @@ func setLogPath(args interface{}) error {
}
logPath := strings.TrimSpace(args.(string))
dir, err := os.Stat(logPath)
if err == nil && dir.IsDir() == false {
if err != nil || dir.IsDir() == false {
return errors.New("Not found dir " + logPath)
}
@@ -492,7 +491,7 @@ func setLogPath(args interface{}) error {
}
}
log.GetLogger().LogPath = logPath
log.LogPath = logPath
return nil
}
@@ -505,7 +504,20 @@ func setLogSize(args interface{}) error {
return nil
}
log.GetLogger().LogConfig.MaxSize = logSize
log.MaxSize = logSize
return nil
}
func setLogChanLen(args interface{}) error {
logChanLen, ok := args.(int)
if ok == false {
return errors.New("param logsize is error")
}
if logChanLen == 0 {
return nil
}
log.LogChanLen = logChanLen
return nil
}

View File

@@ -192,7 +192,7 @@ func (s *Service) run() {
break
}
if s.profiler != nil {
analyzer = s.profiler.Push("[Req]" + rpcRequest.RpcRequestData.GetServiceMethod())
analyzer = s.profiler.Push("[RpcReq]" + rpcRequest.RpcRequestData.GetServiceMethod()+"."+strconv.Itoa(int(rpcRequest.RpcRequestData.GetRpcMethodId())))
}
s.GetRpcHandler().HandlerRpcRequest(rpcRequest)