From dd67efe0d53972292f9b4e9aed1d2f807dc19527 Mon Sep 17 00:00:00 2001 From: gjs Date: Fri, 21 Aug 2026 10:06:06 +0800 Subject: [PATCH] log watcher is done --- pkg/utils/log.go | 1 - pkg/utils/time.go | 9 + .../admin/internal/logic/adminstatuslogic.go | 1 + .../internal/logic/authoritystatuslogic.go | 1 + .../admin/internal/logic/rolestatuslogic.go | 1 + .../internal/logic/servicestatuslogic.go | 1 + services/lonelog/internal/dao/actionLogs.go | 12 +- services/lonelog/internal/dao/loginLogs.go | 25 +++ services/lonelog/internal/model/action.go | 11 +- services/lonelog/internal/model/login.go | 23 +++ .../lonelog/internal/service/logwatcher.go | 178 ++++++++++++++++++ services/lonelog/lonelog.go | 19 +- 12 files changed, 259 insertions(+), 23 deletions(-) create mode 100644 services/lonelog/internal/dao/loginLogs.go create mode 100644 services/lonelog/internal/model/login.go create mode 100644 services/lonelog/internal/service/logwatcher.go diff --git a/pkg/utils/log.go b/pkg/utils/log.go index 56b408e..43f8fd7 100644 --- a/pkg/utils/log.go +++ b/pkg/utils/log.go @@ -139,5 +139,4 @@ func SetLoginLog(adminInfo UserInfo, status bool, reason string) { return } logRedis.LogRedisClient.LPush(context.Background(), LogLoginQueueKey, string(logJson)).Result() - } diff --git a/pkg/utils/time.go b/pkg/utils/time.go index f389be1..8d883f1 100644 --- a/pkg/utils/time.go +++ b/pkg/utils/time.go @@ -203,3 +203,12 @@ func TimeToUnix(timeStr string, layout string) (int64, error) { // 转 11 位时间戳（秒） return t.Unix(), nil } +func ParseCustomTime(s string) (CustomTime, error) { + ct := CustomTime{} + // UnmarshalJSON 需要带双引号的json字符串，所以包装成json字符串格式 + err := ct.UnmarshalJSON([]byte(`"` + s + `"`)) + if err != nil { + return CustomTime{}, err + } + return ct, nil +} diff --git a/services/admin/internal/logic/adminstatuslogic.go b/services/admin/internal/logic/adminstatuslogic.go index a06c914..fe5a2e1 100644 --- a/services/admin/internal/logic/adminstatuslogic.go +++ b/services/admin/internal/logic/adminstatuslogic.go @@ -62,6 +62,7 @@ func (l *AdminStatusLogic) AdminStatus(in *admin.StatusRequest) (*admin.Response } if row > utils.NumberZero { action.NewContent = info + action.Reason = in.Reason action.Type = utils.LogActionTypeStatus action.ModuleName = utils.LogActionModuleAdmin utils.SetActionLog(adminInfo, action) diff --git a/services/admin/internal/logic/authoritystatuslogic.go b/services/admin/internal/logic/authoritystatuslogic.go index 1db9a28..6e7a0e3 100644 --- a/services/admin/internal/logic/authoritystatuslogic.go +++ b/services/admin/internal/logic/authoritystatuslogic.go @@ -80,6 +80,7 @@ func (l *AuthorityStatusLogic) AuthorityStatus(in *admin.StatusRequest) (*admin. if row > utils.NumberZero { action.NewContent = info + action.Reason = in.Reason action.Type = utils.LogActionTypeStatus action.ModuleName = utils.LogActionModuleAuthority utils.SetActionLog(adminInfo, action) diff --git a/services/admin/internal/logic/rolestatuslogic.go b/services/admin/internal/logic/rolestatuslogic.go index 5be74e9..0febe6a 100644 --- a/services/admin/internal/logic/rolestatuslogic.go +++ b/services/admin/internal/logic/rolestatuslogic.go @@ -65,6 +65,7 @@ func (l *RoleStatusLogic) RoleStatus(in *admin.StatusRequest) (*admin.Response, if row > utils.NumberZero { action.NewContent = info + action.Reason = in.Reason action.Type = utils.LogActionTypeStatus action.ModuleName = utils.LogActionModuleRole utils.SetActionLog(adminInfo, action) diff --git a/services/admin/internal/logic/servicestatuslogic.go b/services/admin/internal/logic/servicestatuslogic.go index 8e95bf0..1e9181a 100644 --- a/services/admin/internal/logic/servicestatuslogic.go +++ b/services/admin/internal/logic/servicestatuslogic.go @@ -64,6 +64,7 @@ func (l *ServiceStatusLogic) ServiceStatus(in *admin.StatusRequest) (*admin.Resp if row > utils.NumberZero { action.NewContent = info + action.Reason = in.Reason action.Type = utils.LogActionTypeStatus action.ModuleName = utils.LogActionModuleService utils.SetActionLog(adminInfo, action) diff --git a/services/lonelog/internal/dao/actionLogs.go b/services/lonelog/internal/dao/actionLogs.go index 5dbd675..240cdd8 100644 --- a/services/lonelog/internal/dao/actionLogs.go +++ b/services/lonelog/internal/dao/actionLogs.go @@ -6,7 +6,7 @@ import "lone-services/pkg/utils" type ActionLog struct { Id int `gorm:"column:id;type:int(11);primary_key;AUTO_INCREMENT;comment:ID" json:"id"` Content string `gorm:"column:content;type:text;comment:内容;NOT NULL" json:"content"` - AdminId int `gorm:"column:admin_id;type:int(11);comment:操作人;NOT NULL" json:"admin_id"` + AdminId int64 `gorm:"column:admin_id;type:int(11);comment:操作人;NOT NULL" json:"admin_id"` AdminName string `gorm:"column:admin_name;type:varchar(255);comment:真实姓名" json:"admin_name"` Type uint8 `gorm:"column:type;type:smallint(3);comment:类型，以常量配置文件为准，目前，1为添加，2为编辑，3为删除，4为强退，5为下载;NOT NULL" json:"type"` ModuleName string `gorm:"column:module_name;type:varchar(255);comment:操作模块名，如管理员;NOT NULL" json:"module_name"` @@ -14,7 +14,6 @@ type ActionLog struct { BrowserInfo string `gorm:"column:browser_info;type:varchar(255);comment:浏览器详细信息" json:"browser_info"` BrowserName string `gorm:"column:browser_name;type:varchar(50);comment:浏览器名称" json:"browser_name"` BrowserVersion string `gorm:"column:browser_version;type:varchar(50);comment:版本" json:"browser_version"` - ServiceId int `gorm:"column:service_id;type:int(11);default:0;comment:项目ID" json:"service_id"` Reason string `gorm:"column:reason;type:varchar(255);comment:原因" json:"reason"` CreateTime utils.CustomTime `gorm:"column:create_time;type:datetime;default:NULL;comment:添加时间;NOT NULL" json:"create_time"` } @@ -25,15 +24,6 @@ type ActionItems struct { Type []ActionType `json:"type"` } -type ActionAdd struct { - NewContent interface{} `json:"new_content"` - OldContent interface{} `json:"old_content"` - ServiceId int `json:"service_id"` - Reason string `json:"reason"` - Type uint8 `json:"type"` - ModuleName string `json:"module_name"` -} - type ActionType struct { Id int `json:"id"` Name string `json:"name"` diff --git a/services/lonelog/internal/dao/loginLogs.go b/services/lonelog/internal/dao/loginLogs.go new file mode 100644 index 0000000..58fc544 --- /dev/null +++ b/services/lonelog/internal/dao/loginLogs.go @@ -0,0 +1,25 @@ +package dao + +import "lone-services/pkg/utils" + +type LoginLog struct { + Id int64 `gorm:"column:id;type:int(11);primary_key;AUTO_INCREMENT;comment:ID" json:"id"` + AdminId int64 `gorm:"column:admin_id;type:int(11);default:0;comment:登录人ID;NOT NULL" json:"admin_id"` + AdminName string `gorm:"column:admin_name;type:varchar(255);comment:登录人姓名" json:"admin_name"` + Ip string `gorm:"column:ip;type:varchar(50);comment:IP;NOT NULL" json:"ip"` + BrowserInfo string `gorm:"column:browser_info;type:varchar(255);comment:浏览器详细信息" json:"browser_info"` + BrowserName string `gorm:"column:browser_name;type:varchar(50);comment:浏览器名称" json:"browser_name"` + BrowserVersion string `gorm:"column:browser_version;type:varchar(50);comment:浏览器版本" json:"browser_version"` + Status uint8 `gorm:"column:status;type:tinyint(1);default:1;comment:状态，1为成功，2为失败" json:"status"` + CreateTime utils.CustomTime `gorm:"column:create_time;type:datetime;comment:添加时间;NOT NULL" json:"create_time"` + Reason string `gorm:"column:reason;type:varchar(50);comment:失败原因" json:"reason"` +} + +type LoginLogItems struct { + Items []LoginLog `json:"items"` + Count int64 `json:"count"` +} + +type Count struct { + Id int `gorm:"column:id;type:int(11);primary_key;AUTO_INCREMENT;comment:ID" json:"id"` +} diff --git a/services/lonelog/internal/model/action.go b/services/lonelog/internal/model/action.go index 1413d3e..769f857 100644 --- a/services/lonelog/internal/model/action.go +++ b/services/lonelog/internal/model/action.go @@ -1,6 +1,9 @@ package model -import "lone-services/pkg/modelbase" +import ( + "lone-services/pkg/modelbase" + "lone-services/services/lonelog/internal/dao" +) type ActionModel struct { modelbase.Base @@ -15,6 +18,6 @@ func (m ActionModel) Init() ActionModel { return m } -//func (m ActionModel) Create(data *dao.Admin) error { -// return m.Base.Create(data) -//} +func (m ActionModel) Create(data *dao.ActionLog) error { + return m.Base.Create(data) +} diff --git a/services/lonelog/internal/model/login.go b/services/lonelog/internal/model/login.go new file mode 100644 index 0000000..db7c27e --- /dev/null +++ b/services/lonelog/internal/model/login.go @@ -0,0 +1,23 @@ +package model + +import ( + "lone-services/pkg/modelbase" + "lone-services/services/lonelog/internal/dao" +) + +type LoginModel struct { + modelbase.Base +} + +func (m LoginModel) TableName() string { + return modelbase.Prefix() + "admin_login" +} + +func (m LoginModel) Init() LoginModel { + m.Table = m.TableName() + return m +} + +func (m LoginModel) Create(data *dao.LoginLog) error { + return m.Base.Create(data) +} diff --git a/services/lonelog/internal/service/logwatcher.go b/services/lonelog/internal/service/logwatcher.go new file mode 100644 index 0000000..c567706 --- /dev/null +++ b/services/lonelog/internal/service/logwatcher.go @@ -0,0 +1,178 @@ +package service + +import ( + "context" + "encoding/json" + "lone-services/pkg/logRedis" + "lone-services/pkg/utils" + "lone-services/services/lonelog/internal/dao" + "lone-services/services/lonelog/internal/model" + "sync" + "time" + + "github.com/zeromicro/go-zero/core/logx" +) + +// LogWatcher 队列监听器，内部直接使用 logredis.LogRedisClient 包全局变量，不再传入redis参数 +type LogWatcher struct { + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup +} + +// GlobalLogWatcher 全局单例，main初始化之后其它包直接 service.GlobalLogWatcher 访问 +var GlobalLogWatcher *LogWatcher + +// NewLogWatcher 无入参，直接使用包全局 logredis.LogRedisClient +func NewLogWatcher() *LogWatcher { + ctx, cancel := context.WithCancel(context.Background()) + return &LogWatcher{ + ctx: ctx, + cancel: cancel, + } +} + +// StartWatch 启动两个消费协程 +func (w *LogWatcher) StartWatch() { + w.wg.Add(2) + go w.loopActionLog() + go w.loopLoginLog() + logx.Infof("LogWatcher start, watch %s , %s", utils.LogActionQueueKey, utils.LogLoginQueueKey) +} + +// loopActionLog 消费action日志队列，带recover自动重启 +func (w *LogWatcher) loopActionLog() { + defer func() { + if r := recover(); r != nil { + logx.Errorf("loopActionLog panic recovered, err=%v", r) + time.Sleep(2 * time.Second) + w.wg.Add(1) + go w.loopActionLog() + } + w.wg.Done() + }() + + for { + select { + case <-w.ctx.Done(): + logx.Info("loopActionLog received exit signal, quit") + return + default: + } + + res, err := logRedis.LogRedisClient.BRPop(w.ctx, 5*time.Second, utils.LogActionQueueKey).Result() + if err != nil { + if err.Error() == "redis: nil" { + continue + } + logx.Errorf("BRPop %s redis err: %v", utils.LogActionQueueKey, err) + time.Sleep(1 * time.Second) + continue + } + + if len(res) < 2 { + continue + } + payload := res[1] + + var action utils.ActionLog + if err := json.Unmarshal([]byte(payload), &action); err != nil { + logx.Errorf("unmarshal actionlog failed payload=%s err=%v", payload, err) + continue + } + + ctxBg := context.Background() + logTime, _ := utils.ParseCustomTime(action.CreateTime) + edit := dao.ActionLog{ + Content: action.Content, + AdminId: action.AdminId, + AdminName: action.AdminName, + Type: uint8(action.Type), + ModuleName: action.ModuleName, + Ip: action.Ip, + BrowserInfo: action.BrowserInfo, + BrowserName: action.BrowserName, + BrowserVersion: action.BrowserVersion, + Reason: action.Reason, + CreateTime: logTime, + } + + err = model.ActionModel{}.Init().Create(&edit) + if err != nil { + logx.Errorf("insert actionlog mysql err=%v payload=%s", err, payload) + _ = logRedis.LogRedisClient.LPush(ctxBg, utils.LogActionQueueKey, payload).Err() + time.Sleep(500 * time.Millisecond) + } + } +} + +// loopLoginLog 消费登录日志队列，带recover自动重启 +func (w *LogWatcher) loopLoginLog() { + defer func() { + if r := recover(); r != nil { + logx.Errorf("loopLoginLog panic recovered, err=%v", r) + time.Sleep(2 * time.Second) + w.wg.Add(1) + go w.loopLoginLog() + } + w.wg.Done() + }() + + for { + select { + case <-w.ctx.Done(): + logx.Info("loopLoginLog received exit signal, quit") + return + default: + } + + res, err := logRedis.LogRedisClient.BRPop(w.ctx, 5*time.Second, utils.LogLoginQueueKey).Result() + if err != nil { + if err.Error() == "redis: nil" { + continue + } + logx.Errorf("BRPop %s redis err: %v", utils.LogLoginQueueKey, err) + time.Sleep(1 * time.Second) + continue + } + + if len(res) < 2 { + continue + } + payload := res[1] + + var login utils.LoginLog + if err := json.Unmarshal([]byte(payload), &login); err != nil { + logx.Errorf("unmarshal loginlog failed payload=%s err=%v", payload, err) + continue + } + + ctxBg := context.Background() + + logTime, _ := utils.ParseCustomTime(login.CreateTime) + edit := dao.LoginLog{ + AdminId: login.AdminId, + AdminName: login.Name, + Ip: login.Ip, + BrowserInfo: login.BrowserInfo, + BrowserName: login.BrowserName, + BrowserVersion: login.BrowserVersion, + Reason: login.Reason, + CreateTime: logTime, + Status: uint8(login.Status), + } + err = model.LoginModel{}.Init().Create(&edit) + if err != nil { + logx.Errorf("insert loginlog mysql err=%v payload=%s", err, payload) + _ = logRedis.LogRedisClient.LPush(ctxBg, utils.LogLoginQueueKey, payload).Err() + time.Sleep(500 * time.Millisecond) + } + } +} + +// StopWatch 优雅停止消费协程 +func (w *LogWatcher) StopWatch() { + w.cancel() + w.wg.Wait() + logx.Info("LogWatcher stopped complete") +} diff --git a/services/lonelog/lonelog.go b/services/lonelog/lonelog.go index 9ea7567..343c61d 100644 --- a/services/lonelog/lonelog.go +++ b/services/lonelog/lonelog.go @@ -3,14 +3,15 @@ package main import ( "flag" "lone-services/pkg/discovery" + "lone-services/pkg/logRedis" "lone-services/pkg/modelbase" "lone-services/pkg/mysql" - "lone-services/pkg/redis" "lone-services/pkg/utils" "lone-services/pkg/validate" lonelog "lone-services/rpc/lonelog/pb" "lone-services/services/lonelog/internal/config" "lone-services/services/lonelog/internal/server" + service2 "lone-services/services/lonelog/internal/service" "lone-services/services/lonelog/internal/svc" "net" "os" @@ -108,13 +109,13 @@ func main() { os.Exit(1) } - if err := redis.Init(redis.Config{ - Host: utils.GetConfigString("redis.host"), - Port: utils.GetConfigInt("redis.port"), - Password: utils.GetConfigString("redis.password"), - DB: utils.GetConfigInt("redis.db"), + if err := logRedis.Init(logRedis.LogRedisConfig{ + Host: utils.GetConfigString("log-redis.host"), + Port: utils.GetConfigInt("log-redis.port"), + Password: utils.GetConfigString("log-redis.password"), + DB: utils.GetConfigInt("log-redis.db"), }); err != nil { - logx.Errorf("redis init: %v", err) + logx.Errorf("log redis init: %v", err) os.Exit(1) } @@ -138,6 +139,10 @@ func main() { s.AddUnaryInterceptors(validate.UnaryServerInterceptor(validate.MustNew())) + watcher := service2.NewLogWatcher() + watcher.StartWatch() + defer watcher.StopWatch() + logx.Infof("Starting rpc server at %s...", listenOn) s.Start() }