318 lines
10 KiB
Go
318 lines
10 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/polling"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// MonitoringService 轮询监控服务
|
||
type MonitoringService struct {
|
||
redis *redis.Client
|
||
queueMgr *polling.PollingQueueManager
|
||
}
|
||
|
||
// NewMonitoringService 创建轮询监控服务实例
|
||
// queueMgr 可选（nil 时降级到旧非分片 Key 兼容模式）
|
||
func NewMonitoringService(redis *redis.Client, opts ...interface{}) *MonitoringService {
|
||
svc := &MonitoringService{redis: redis}
|
||
for _, opt := range opts {
|
||
if qm, ok := opt.(*polling.PollingQueueManager); ok {
|
||
svc.queueMgr = qm
|
||
}
|
||
}
|
||
return svc
|
||
}
|
||
|
||
// NewMonitoringServiceWithQueueMgr 创建注入了 PollingQueueManager 的监控服务（推荐）
|
||
func NewMonitoringServiceWithQueueMgr(redis *redis.Client, queueMgr *polling.PollingQueueManager, _ *zap.Logger) *MonitoringService {
|
||
return &MonitoringService{redis: redis, queueMgr: queueMgr}
|
||
}
|
||
|
||
// OverviewStats 总览统计
|
||
type OverviewStats struct {
|
||
TotalCards int64 `json:"total_cards"` // 总卡数
|
||
InitializedCards int64 `json:"initialized_cards"` // 已初始化卡数
|
||
InitProgress float64 `json:"init_progress"` // 初始化进度 (0-100)
|
||
IsInitializing bool `json:"is_initializing"` // 是否正在初始化
|
||
RealnameQueueSize int64 `json:"realname_queue_size"` // 实名检查队列大小
|
||
CarddataQueueSize int64 `json:"carddata_queue_size"` // 流量检查队列大小
|
||
PackageQueueSize int64 `json:"package_queue_size"` // 套餐检查队列大小
|
||
}
|
||
|
||
// QueueStatus 队列状态
|
||
type QueueStatus struct {
|
||
TaskType string `json:"task_type"` // 任务类型
|
||
TaskTypeName string `json:"task_type_name"` // 任务类型名称
|
||
QueueSize int64 `json:"queue_size"` // 队列大小
|
||
ManualPending int64 `json:"manual_pending"` // 手动触发待处理数
|
||
DueCount int64 `json:"due_count"` // 到期待处理数
|
||
AvgWaitTime float64 `json:"avg_wait_time_s"` // 平均等待时间(秒)
|
||
}
|
||
|
||
// TaskStats 任务统计
|
||
type TaskStats struct {
|
||
TaskType string `json:"task_type"` // 任务类型
|
||
TaskTypeName string `json:"task_type_name"` // 任务类型名称
|
||
SuccessCount1h int64 `json:"success_count_1h"` // 1小时成功数
|
||
FailureCount1h int64 `json:"failure_count_1h"` // 1小时失败数
|
||
TotalCount1h int64 `json:"total_count_1h"` // 1小时总数
|
||
SuccessRate float64 `json:"success_rate"` // 成功率 (0-100)
|
||
AvgDurationMs float64 `json:"avg_duration_ms"` // 平均耗时(毫秒)
|
||
}
|
||
|
||
// InitProgress 初始化进度
|
||
type InitProgress struct {
|
||
TotalCards int64 `json:"total_cards"` // 总卡数
|
||
InitializedCards int64 `json:"initialized_cards"` // 已初始化卡数
|
||
Progress float64 `json:"progress"` // 进度百分比 (0-100)
|
||
IsComplete bool `json:"is_complete"` // 是否完成
|
||
StartedAt time.Time `json:"started_at"` // 开始时间
|
||
EstimatedETA string `json:"estimated_eta"` // 预计完成时间
|
||
}
|
||
|
||
// GetOverview 获取总览统计
|
||
func (s *MonitoringService) GetOverview(ctx context.Context) (*OverviewStats, error) {
|
||
stats := &OverviewStats{}
|
||
|
||
// 获取初始化进度
|
||
progressKey := constants.RedisPollingInitProgressKey()
|
||
progressData, err := s.redis.HGetAll(ctx, progressKey).Result()
|
||
if err != nil && err != redis.Nil {
|
||
return nil, errors.Wrap(errors.CodeRedisError, err, "获取初始化进度失败")
|
||
}
|
||
|
||
if total, ok := progressData["total"]; ok {
|
||
stats.TotalCards = parseInt64(total)
|
||
}
|
||
if initialized, ok := progressData["initialized"]; ok {
|
||
stats.InitializedCards = parseInt64(initialized)
|
||
}
|
||
if stats.TotalCards > 0 {
|
||
stats.InitProgress = float64(stats.InitializedCards) / float64(stats.TotalCards) * 100
|
||
}
|
||
stats.IsInitializing = stats.InitializedCards < stats.TotalCards && stats.TotalCards > 0
|
||
|
||
if s.queueMgr != nil {
|
||
stats.RealnameQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingRealname)
|
||
stats.CarddataQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingCarddata)
|
||
stats.PackageQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingPackage)
|
||
} else {
|
||
stats.RealnameQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueueRealnameKey()).Result()
|
||
stats.CarddataQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueueCarddataKey()).Result()
|
||
stats.PackageQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueuePackageKey()).Result()
|
||
}
|
||
|
||
return stats, nil
|
||
}
|
||
|
||
// GetQueueStatuses 获取所有队列状态
|
||
func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatus, error) {
|
||
taskTypes := pollingMonitorTaskTypes()
|
||
|
||
result := make([]*QueueStatus, 0, len(taskTypes))
|
||
now := time.Now()
|
||
|
||
for _, taskType := range taskTypes {
|
||
status := &QueueStatus{
|
||
TaskType: taskType,
|
||
TaskTypeName: s.getTaskTypeName(taskType),
|
||
}
|
||
|
||
// 获取队列 key
|
||
var queueKey string
|
||
switch taskType {
|
||
case constants.TaskTypePollingRealname:
|
||
queueKey = constants.RedisPollingQueueRealnameKey()
|
||
case constants.TaskTypePollingCarddata:
|
||
queueKey = constants.RedisPollingQueueCarddataKey()
|
||
case constants.TaskTypePollingPackage:
|
||
queueKey = constants.RedisPollingQueuePackageKey()
|
||
}
|
||
|
||
if s.queueMgr != nil {
|
||
status.QueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, taskType)
|
||
status.DueCount, _ = s.queueMgr.GetTotalDueCount(ctx, taskType, now)
|
||
status.AvgWaitTime, _ = s.queueMgr.GetAverageWaitTime(ctx, taskType, now, 10)
|
||
} else {
|
||
status.QueueSize, _ = s.redis.ZCard(ctx, queueKey).Result()
|
||
status.DueCount, _ = s.redis.ZCount(ctx, queueKey, "-inf", formatInt64(now.Unix())).Result()
|
||
status.AvgWaitTime = s.getLegacyAverageWaitTime(ctx, queueKey, now.Unix())
|
||
}
|
||
|
||
// 获取手动触发队列待处理数
|
||
manualKey := constants.RedisPollingManualQueueKey(taskType)
|
||
status.ManualPending, _ = s.redis.LLen(ctx, manualKey).Result()
|
||
|
||
result = append(result, status)
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// getLegacyAverageWaitTime 从旧版非分片队列计算平均等待时间。
|
||
func (s *MonitoringService) getLegacyAverageWaitTime(ctx context.Context, queueKey string, now int64) float64 {
|
||
earliest, err := s.redis.ZRangeWithScores(ctx, queueKey, 0, 9).Result()
|
||
if err != nil || len(earliest) == 0 {
|
||
return 0
|
||
}
|
||
var totalWait float64
|
||
for _, z := range earliest {
|
||
waitTime := float64(now) - z.Score
|
||
if waitTime > 0 {
|
||
totalWait += waitTime
|
||
}
|
||
}
|
||
return totalWait / float64(len(earliest))
|
||
}
|
||
|
||
// GetTaskStatuses 获取所有任务统计
|
||
func (s *MonitoringService) GetTaskStatuses(ctx context.Context) ([]*TaskStats, error) {
|
||
taskTypes := pollingMonitorTaskTypes()
|
||
|
||
result := make([]*TaskStats, 0, len(taskTypes))
|
||
|
||
for _, taskType := range taskTypes {
|
||
stats := &TaskStats{
|
||
TaskType: taskType,
|
||
TaskTypeName: s.getTaskTypeName(taskType),
|
||
}
|
||
|
||
// 获取统计数据
|
||
statsKey := constants.RedisPollingStatsKey(taskType)
|
||
data, err := s.redis.HGetAll(ctx, statsKey).Result()
|
||
if err != nil && err != redis.Nil {
|
||
continue
|
||
}
|
||
|
||
if success, ok := data["success_count_1h"]; ok {
|
||
stats.SuccessCount1h = parseInt64(success)
|
||
}
|
||
if failure, ok := data["failure_count_1h"]; ok {
|
||
stats.FailureCount1h = parseInt64(failure)
|
||
}
|
||
stats.TotalCount1h = stats.SuccessCount1h + stats.FailureCount1h
|
||
|
||
if stats.TotalCount1h > 0 {
|
||
stats.SuccessRate = float64(stats.SuccessCount1h) / float64(stats.TotalCount1h) * 100
|
||
|
||
if duration, ok := data["total_duration_1h"]; ok {
|
||
totalDuration := parseInt64(duration)
|
||
stats.AvgDurationMs = float64(totalDuration) / float64(stats.TotalCount1h)
|
||
}
|
||
}
|
||
|
||
result = append(result, stats)
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// GetInitProgress 获取初始化进度详情
|
||
func (s *MonitoringService) GetInitProgress(ctx context.Context) (*InitProgress, error) {
|
||
progressKey := constants.RedisPollingInitProgressKey()
|
||
data, err := s.redis.HGetAll(ctx, progressKey).Result()
|
||
if err != nil && err != redis.Nil {
|
||
return nil, errors.Wrap(errors.CodeRedisError, err, "获取初始化进度失败")
|
||
}
|
||
|
||
progress := &InitProgress{}
|
||
|
||
if total, ok := data["total"]; ok {
|
||
progress.TotalCards = parseInt64(total)
|
||
}
|
||
if initialized, ok := data["initialized"]; ok {
|
||
progress.InitializedCards = parseInt64(initialized)
|
||
}
|
||
if startedAt, ok := data["started_at"]; ok {
|
||
if t, err := time.Parse(time.RFC3339, startedAt); err == nil {
|
||
progress.StartedAt = t
|
||
}
|
||
}
|
||
|
||
if progress.TotalCards > 0 {
|
||
progress.Progress = float64(progress.InitializedCards) / float64(progress.TotalCards) * 100
|
||
}
|
||
progress.IsComplete = progress.InitializedCards >= progress.TotalCards && progress.TotalCards > 0
|
||
|
||
// 估算完成时间
|
||
if !progress.IsComplete && progress.InitializedCards > 0 && !progress.StartedAt.IsZero() {
|
||
elapsed := time.Since(progress.StartedAt)
|
||
remaining := progress.TotalCards - progress.InitializedCards
|
||
rate := float64(progress.InitializedCards) / elapsed.Seconds()
|
||
if rate > 0 {
|
||
etaSeconds := float64(remaining) / rate
|
||
eta := time.Now().Add(time.Duration(etaSeconds) * time.Second)
|
||
progress.EstimatedETA = eta.Format("15:04:05")
|
||
}
|
||
}
|
||
|
||
return progress, nil
|
||
}
|
||
|
||
// getTaskTypeName 获取任务类型的中文名称
|
||
func (s *MonitoringService) getTaskTypeName(taskType string) string {
|
||
switch taskType {
|
||
case constants.TaskTypePollingRealname:
|
||
return "实名检查"
|
||
case constants.TaskTypePollingCarddata:
|
||
return "流量检查"
|
||
case constants.TaskTypePollingPackage:
|
||
return "套餐检查"
|
||
case constants.TaskTypePollingProtect:
|
||
return "保护期检查"
|
||
case constants.TaskTypePollingCardStatus:
|
||
return "卡状态检查"
|
||
default:
|
||
return taskType
|
||
}
|
||
}
|
||
|
||
// pollingMonitorTaskTypes 返回后台监控需要展示的全部轮询任务类型。
|
||
func pollingMonitorTaskTypes() []string {
|
||
return []string{
|
||
constants.TaskTypePollingRealname,
|
||
constants.TaskTypePollingCarddata,
|
||
constants.TaskTypePollingPackage,
|
||
constants.TaskTypePollingProtect,
|
||
constants.TaskTypePollingCardStatus,
|
||
}
|
||
}
|
||
|
||
// parseInt64 解析字符串为 int64
|
||
func parseInt64(s string) int64 {
|
||
var result int64
|
||
for _, c := range s {
|
||
if c >= '0' && c <= '9' {
|
||
result = result*10 + int64(c-'0')
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// formatInt64 格式化 int64 为字符串
|
||
func formatInt64(n int64) string {
|
||
if n == 0 {
|
||
return "0"
|
||
}
|
||
var result []byte
|
||
negative := n < 0
|
||
if negative {
|
||
n = -n
|
||
}
|
||
for n > 0 {
|
||
result = append([]byte{byte('0' + n%10)}, result...)
|
||
n /= 10
|
||
}
|
||
if negative {
|
||
result = append([]byte{'-'}, result...)
|
||
}
|
||
return string(result)
|
||
}
|