zishuo/NetHive
1
0
Fork 0
mirror of https://github.com/wlynxg/NetHive.git synced 2025-09-26 19:11:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: refactor the log module

Browse Source
This commit is contained in:
lynx
2024-04-10 15:35:16 +08:00
parent ab4ad8480d
commit df4633046d
7 changed files with 171 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
core/engine/conn.go
View File

@@ -13,7 +13,7 @@ import (
)
func (e *Engine) addConnByDst(dst netip.Addr) (PacketChan, error) {
e.log.Debugf(e.ctx, "Try to connect to the corresponding node of %s", dst)
e.log.Debugf("Try to connect to the corresponding node of %s", dst)
var conn PacketChan
e.routeTable.set.Range(func(id peer.ID, prefix *netipx.IPSet) bool {
@@ -45,7 +45,7 @@ func (e *Engine) addConnByDst(dst netip.Addr) (PacketChan, error) {
}
func (e *Engine) addConnByID(id peer.ID) (PacketChan, error) {
e.log.Debugf(e.ctx, "Try to connect to the corresponding node of %s", string(id))
e.log.Debugf("Try to connect to the corresponding node of %s", string(id))
if conn, ok := e.routeTable.id.Load(id); ok {
return conn, nil
@@ -64,7 +64,7 @@ func (e *Engine) addConnByID(id peer.ID) (PacketChan, error) {
func (e *Engine) addConn(peerChan PacketChan, id peer.ID) {
dev := &devWrapper{w: e.devWriter, r: peerChan}
e.log.Infof(e.ctx, "start find peer %s", string(id))
e.log.Infof("start find peer %s", string(id))
var (
stream network.Stream
@@ -73,7 +73,7 @@ func (e *Engine) addConn(peerChan PacketChan, id peer.ID) {
idr, err := base58.Decode(string(id))
if err != nil {
e.log.Infof(e.ctx, "base58 decode failed: %s", err)
e.log.Infof("base58 decode failed: %s", err)
}
info := e.host.Peerstore().PeerInfo(peer.ID(idr))
@@ -82,7 +82,7 @@ func (e *Engine) addConn(peerChan PacketChan, id peer.ID) {
if err != nil {
peerc, err := e.discovery.FindPeers(e.ctx, string(id))
if err != nil {
e.log.Warningf(e.ctx, "Finding node by dht %s failed because %s", string(id), err)
e.log.Warnf("Finding node by dht %s failed because %s", string(id), err)
return
}
@@ -94,7 +94,7 @@ func (e *Engine) addConn(peerChan PacketChan, id peer.ID) {
}
}
}
e.log.Warningf(e.ctx, "Connection establishment with node %s failed", string(id))
e.log.Warnf("Connection establishment with node %s failed", string(id))
return
}
}
@@ -103,19 +103,19 @@ func (e *Engine) addConn(peerChan PacketChan, id peer.ID) {
return
}
e.log.Infof(e.ctx, "Peer [%s] connect success", string(id))
e.log.Infof("Peer [%s] connect success", string(id))
defer stream.Close()
go func() {
defer stream.Close()
_, err := io.Copy(stream, dev)
if err != nil && err != io.EOF {
e.log.Errorf(e.ctx, "Peer [%s] stream write error: %s", string(id), err)
e.log.Errorf("Peer [%s] stream write error: %s", string(id), err)
}
}()
_, err = io.Copy(dev, stream)
if err != nil && err != io.EOF {
e.log.Errorf(e.ctx, "Peer [%s] stream read error: %s", string(id), err)
e.log.Errorf("Peer [%s] stream read error: %s", string(id), err)
}
}

30
core/engine/engine.go
View File

@@ -12,9 +12,9 @@ import (
"github.com/wlynxg/NetHive/core/device"
"github.com/wlynxg/NetHive/core/protocol"
"github.com/wlynxg/NetHive/core/route"
mlog "github.com/wlynxg/NetHive/pkgs/log"
"github.com/wlynxg/NetHive/pkgs/xsync"
"github.com/gogf/gf/v2/os/glog"
"github.com/libp2p/go-libp2p"
dht "github.com/libp2p/go-libp2p-kad-dht"
"github.com/libp2p/go-libp2p/core/host"
@@ -35,7 +35,7 @@ const (
type PacketChan chan Payload
type Engine struct {
log *glog.Logger
log *mlog.Logger
ctx context.Context
cancel context.CancelFunc
cfg *config.Config
@@ -65,7 +65,7 @@ func New(cfg *config.Config) (*Engine, error) {
var (
e = new(Engine)
)
e.log = glog.New()
e.log = mlog.New("engine")
e.cfg = cfg
e.ctx, e.cancel = context.WithCancel(context.Background())
e.devWriter = make(PacketChan, ChanSize)
@@ -119,7 +119,7 @@ func New(cfg *config.Config) (*Engine, error) {
}
e.host = node
e.log.Infof(e.ctx, "host ID: %s", node.ID().String())
e.log.Infof("host ID: %s", node.ID().String())
e.dht, err = dht.New(e.ctx, e.host)
if err != nil {
return nil, err
@@ -149,7 +149,7 @@ func (e *Engine) Start() error {
go func() {
defer wg.Done()
if err := e.host.Connect(e.ctx, info); err != nil {
e.log.Warningf(e.ctx, "connection %s fails, because of error :%s", info.String(), err)
e.log.Warnf("connection %s fails, because of error :%s", info.String(), err)
}
}()
}
@@ -171,7 +171,7 @@ func (e *Engine) Start() error {
defer stream.Close()
_, err := io.Copy(stream, dev)
if err != nil && err != io.EOF {
e.log.Errorf(e.ctx, "Peer [%s] stream write error: %s", string(id), err)
e.log.Errorf("Peer [%s] stream write error: %s", string(id), err)
}
}()
@@ -196,7 +196,7 @@ func (e *Engine) Start() error {
}()
_, err := io.Copy(dev, stream)
if err != nil && err != io.EOF {
e.log.Errorf(e.ctx, "Peer [%s] stream read error: %s", string(id), err)
e.log.Errorf("Peer [%s] stream read error: %s", string(id), err)
}
})
@@ -208,7 +208,7 @@ func (e *Engine) Start() error {
case <-ticker.C:
peers, err := e.dht.GetClosestPeers(e.ctx, string(e.host.ID()))
if err != nil {
e.log.Warningf(e.ctx, "Failed to get nearest node: %s", err)
e.log.Warnf("Failed to get nearest node: %s", err)
continue
}
@@ -250,7 +250,7 @@ func (e *Engine) RoutineTUNReader() {
}
ip, err := protocol.ParseIP(buff[:n])
if err != nil {
e.log.Warningf(e.ctx, "[RoutineTUNReader] drop packet, because %s", err)
e.log.Warnf("[RoutineTUNReader] drop packet, because %s", err)
continue
}
payload := Payload{
@@ -262,7 +262,7 @@ func (e *Engine) RoutineTUNReader() {
select {
case e.devReader <- payload:
default:
e.log.Warningf(e.ctx, "[RoutineTUNReader] drop packet: %s, because the sending queue is already full", payload.Dst)
e.log.Warnf("[RoutineTUNReader] drop packet: %s, because the sending queue is already full", payload.Dst)
}
}
}
@@ -300,21 +300,21 @@ func (e *Engine) RoutineRouteTableWriter() {
select {
case conn <- payload:
default:
e.log.Warningf(e.ctx, "[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
e.log.Warnf("[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
}
return true
}
conn, err := e.addConnByID(key)
if err != nil {
e.log.Warningf(e.ctx, "[RoutineRouteTableWriter] drop packet: %s, because %s", payload.Dst, err)
e.log.Warnf("[RoutineRouteTableWriter] drop packet: %s, because %s", payload.Dst, err)
return true
}
select {
case conn <- payload:
default:
e.log.Warningf(e.ctx, "[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
e.log.Warnf("[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
}
return true
@@ -326,7 +326,7 @@ func (e *Engine) RoutineRouteTableWriter() {
} else {
c, err := e.addConnByDst(payload.Dst)
if err != nil {
e.log.Warningf(e.ctx, "[RoutineRouteTableWriter] drop packet: %s, because %s", payload.Dst, err)
e.log.Warnf("[RoutineRouteTableWriter] drop packet: %s, because %s", payload.Dst, err)
continue
}
conn = c
@@ -335,7 +335,7 @@ func (e *Engine) RoutineRouteTableWriter() {
select {
case conn <- payload:
default:
e.log.Warningf(e.ctx, "[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
e.log.Warnf("[RoutineRouteTableWriter] drop packet: %s, because the sending queue is already full", payload.Dst)
}
}
}

2
core/engine/mdns.go
View File

@@ -7,6 +7,6 @@ import (
)
func (e *Engine) HandlePeerFound(pi peer.AddrInfo) {
e.log.Infof(e.ctx, "find %s by mDNS", pi)
e.log.Infof("find %s by mDNS", pi)
e.host.Peerstore().AddAddrs(pi.ID, pi.Addrs, 5*time.Minute)
}

5
go.mod
View File

@@ -8,6 +8,7 @@ require (
github.com/libp2p/go-libp2p-kad-dht v0.25.2
github.com/mr-tron/base58 v1.2.0
github.com/pkg/errors v0.9.1
go.uber.org/zap v1.27.0
go4.org/netipx v0.0.0-20231129151722-fdeea329fbba
golang.org/x/sys v0.19.0
)
@@ -24,7 +25,6 @@ require (
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
github.com/docker/go-units v0.5.0 // indirect
github.com/elastic/gosigar v0.14.2 // indirect
github.com/fatih/color v1.15.0 // indirect
github.com/flynn/noise v1.1.0 // indirect
github.com/francoispqt/gojay v1.2.13 // indirect
github.com/fsnotify/fsnotify v1.7.0 // indirect
@@ -68,7 +68,6 @@ require (
github.com/libp2p/zeroconf/v2 v2.2.0 // indirect
github.com/magiconair/properties v1.8.6 // indirect
github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/miekg/dns v1.1.58 // indirect
github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b // indirect
@@ -102,13 +101,11 @@ require (
go.opencensus.io v0.24.0 // indirect
go.opentelemetry.io/otel v1.16.0 // indirect
go.opentelemetry.io/otel/metric v1.16.0 // indirect
go.opentelemetry.io/otel/sdk v1.14.0 // indirect
go.opentelemetry.io/otel/trace v1.16.0 // indirect
go.uber.org/dig v1.17.1 // indirect
go.uber.org/fx v1.20.1 // indirect
go.uber.org/mock v0.4.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.27.0 // indirect
golang.org/x/crypto v0.19.0 // indirect
golang.org/x/exp v0.0.0-20240213143201-ec583247a57a // indirect
golang.org/x/mod v0.15.0 // indirect

2
go.sum
View File

@@ -239,7 +239,6 @@ github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd/go.mod h1:QuCEs
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
@@ -513,7 +512,6 @@ golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20210426080607-c94f62235c83/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=

131
pkgs/log/core.go Normal file
View File

@@ -0,0 +1,131 @@
package mlog
import (
"os"
"path/filepath"
"strings"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
type CoreConfig struct {
OutputType string
OutputPath string
Level string
EncodeType string
EncodeColor bool
}
var (
coreConfigs []CoreConfig
)
func SetOutputTypes(configs ...CoreConfig) {
coreConfigs = append(coreConfigs, configs...)
}
func NewCore() zapcore.Core {
cores := make([]zapcore.Core, 0, len(coreConfigs))
for _, cfg := range coreConfigs {
var core zapcore.Core
switch cfg.OutputType {
case "file":
core = FileCore(cfg)
case "console":
core = ConsoleCore(cfg)
}
if core != nil {
cores = append(cores, core)
}
}
if len(cores) == 0 {
cores = append(cores, ConsoleCore(CoreConfig{EncodeColor: true}))
}
return zapcore.NewTee(cores...)
}
func ConsoleCore(cfg CoreConfig) zapcore.Core {
encoderConfig := zapcore.EncoderConfig{
// Keys can be anything except the empty string.
TimeKey: "T",
LevelKey: "L",
NameKey: "N",
CallerKey: "C",
FunctionKey: zapcore.OmitKey,
MessageKey: "M",
StacktraceKey: "S",
LineEnding: zapcore.DefaultLineEnding,
EncodeLevel: zapcore.CapitalLevelEncoder,
EncodeDuration: zapcore.StringDurationEncoder,
EncodeCaller: zapcore.ShortCallerEncoder,
ConsoleSeparator: "\t",
}
if cfg.EncodeColor {
encoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder
}
level, err := zap.ParseAtomicLevel(cfg.Level)
if err != nil {
level = zap.NewAtomicLevelAt(zap.InfoLevel)
}
encoder := zapcore.NewConsoleEncoder(encoderConfig)
switch cfg.EncodeType {
case "json":
encoder = zapcore.NewJSONEncoder(encoderConfig)
}
out := "stdout"
if strings.ToLower(cfg.OutputPath) == "stderr" {
out = "stderr"
}
writer, closeFn, err := zap.Open(out)
if err != nil {
closeFn()
return nil
}
return zapcore.NewCore(encoder, writer, level)
}
func FileCore(cfg CoreConfig) zapcore.Core {
encoderConfig := zapcore.EncoderConfig{
// Keys can be anything except the empty string.
TimeKey: "T",
LevelKey: "L",
NameKey: "N",
CallerKey: "",
FunctionKey: zapcore.OmitKey,
MessageKey: "M",
StacktraceKey: "S",
LineEnding: zapcore.DefaultLineEnding,
EncodeLevel: zapcore.CapitalLevelEncoder,
EncodeDuration: zapcore.StringDurationEncoder,
EncodeCaller: nil,
ConsoleSeparator: "\t",
}
if cfg.EncodeColor {
encoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder
}
level, err := zap.ParseAtomicLevel(cfg.Level)
if err != nil {
level = zap.NewAtomicLevelAt(zap.InfoLevel)
}
encoder := zapcore.NewConsoleEncoder(encoderConfig)
switch cfg.EncodeType {
case "json":
encoder = zapcore.NewJSONEncoder(encoderConfig)
}
os.MkdirAll(filepath.Dir(cfg.OutputPath), 0600)
writer, _, err := zap.Open(cfg.OutputPath)
if err != nil {
return nil
}
return zapcore.NewCore(encoder, writer, level)
}

14
pkgs/log/log.go Normal file
View File

@@ -0,0 +1,14 @@
package mlog
import (
"go.uber.org/zap"
)
type Logger struct {
*zap.SugaredLogger
}
func New(name string) *Logger {
logger := zap.New(NewCore(), zap.AddCaller())
return &Logger{logger.Sugar().Named(name)}
}