shanshanzhong147
3d1a31a19f
- getServerUserListLogic.getExpiredUsers 之前完全忽略 user_subscribe.speed_limit， 现在带出用户级覆盖并与过期节点组 speed_limit 取更严（mergeSpeedLimit：0 视为无限制） - node_group.SpeedLimit 注释从 "KB/s" 修正为 "Mbps"（旧注释是笔误，实际下发节点的 ServerUser.SpeedLimit 字段语义就是 Mbps，节点端 ppanel-node 按 *1e6/8 换算为 Byte/s） - apis/node/node.api 给 ServerUser.SpeedLimit 加 Mbps 单位注释 - 新增 TestMergeSpeedLimit 表驱动测试覆盖 8 种边界 主链路（活跃用户、套餐 traffic_limit 阶梯）行为不变，已在生产 (server_id=52) 验证 147 个限速用户下发正确，与 DB 完全对应。
619 lines
18 KiB
Go
619 lines
18 KiB
Go
package server
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/perfect-panel/server/internal/model/group"
|
|
"github.com/perfect-panel/server/internal/model/node"
|
|
"github.com/perfect-panel/server/internal/model/subscribe"
|
|
"github.com/perfect-panel/server/internal/model/user"
|
|
|
|
"github.com/perfect-panel/server/internal/svc"
|
|
"github.com/perfect-panel/server/internal/types"
|
|
"github.com/perfect-panel/server/pkg/logger"
|
|
"github.com/perfect-panel/server/pkg/speedlimit"
|
|
"github.com/perfect-panel/server/pkg/tool"
|
|
"github.com/perfect-panel/server/pkg/uuidx"
|
|
"github.com/perfect-panel/server/pkg/xerr"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
type GetServerUserListLogic struct {
|
|
logger.Logger
|
|
ctx *gin.Context
|
|
svcCtx *svc.ServiceContext
|
|
}
|
|
|
|
type serverUserListPerfStats struct {
|
|
serverID int64
|
|
protocol string
|
|
cacheHit bool
|
|
nodesCount int
|
|
subsCount int
|
|
usersCount int
|
|
speedLimitZeroCount int
|
|
speedLimitPositiveCount int
|
|
trafficCalcMS int64
|
|
startedAt time.Time
|
|
}
|
|
|
|
type serverUserSpeedLimitCandidate struct {
|
|
userID int64
|
|
userSubscribeID int64
|
|
baseSpeed int64
|
|
trafficLimit string
|
|
}
|
|
|
|
type serverUserTrafficWindow struct {
|
|
statType string
|
|
statValue int64
|
|
start time.Time
|
|
end time.Time
|
|
}
|
|
|
|
type serverUserTrafficUsageKey struct {
|
|
userID int64
|
|
userSubscribeID int64
|
|
statType string
|
|
statValue int64
|
|
}
|
|
|
|
type serverUserTrafficUsage struct {
|
|
userID int64
|
|
userSubscribeID int64
|
|
usedGB float64
|
|
}
|
|
|
|
// NewGetServerUserListLogic Get user list
|
|
func NewGetServerUserListLogic(ctx *gin.Context, svcCtx *svc.ServiceContext) *GetServerUserListLogic {
|
|
return &GetServerUserListLogic{
|
|
Logger: logger.WithContext(ctx.Request.Context()),
|
|
ctx: ctx,
|
|
svcCtx: svcCtx,
|
|
}
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) GetServerUserList(req *types.GetServerUserListRequest) (resp *types.GetServerUserListResponse, err error) {
|
|
startedAt := time.Now()
|
|
protocolRequest := normalizeServerUserListProtocol(req.Protocol)
|
|
stats := serverUserListPerfStats{
|
|
serverID: req.ServerId,
|
|
protocol: protocolRequest,
|
|
startedAt: startedAt,
|
|
cacheHit: false,
|
|
}
|
|
|
|
cacheKey := fmt.Sprintf("%s%d:%s", node.ServerUserListCacheKey, req.ServerId, protocolRequest)
|
|
cache, err := l.svcCtx.Redis.Get(l.ctx, cacheKey).Result()
|
|
if err != nil && err != redis.Nil {
|
|
l.Errorw("[ServerUserListCacheKey] redis get error", logger.Field("error", err.Error()))
|
|
}
|
|
if cache != "" {
|
|
stats.cacheHit = true
|
|
etag := tool.GenerateETag([]byte(cache))
|
|
resp = &types.GetServerUserListResponse{}
|
|
// Check If-None-Match header
|
|
if match := l.ctx.GetHeader("If-None-Match"); match == etag {
|
|
l.logServerUserListPerf(stats)
|
|
return nil, xerr.StatusNotModified
|
|
}
|
|
l.ctx.Header("ETag", etag)
|
|
err = json.Unmarshal([]byte(cache), resp)
|
|
if err != nil {
|
|
l.Errorw("[ServerUserListCacheKey] json unmarshal error", logger.Field("error", err.Error()))
|
|
return nil, err
|
|
}
|
|
stats.usersCount = len(resp.Users)
|
|
stats.recordSpeedLimits(resp.Users)
|
|
l.logServerUserListPerf(stats)
|
|
return resp, nil
|
|
}
|
|
server, err := l.svcCtx.NodeModel.FindOneServer(l.ctx, req.ServerId)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// 查询该服务器上该协议的所有节点（包括属于节点组的节点）
|
|
_, nodes, err := l.svcCtx.NodeModel.FilterNodeList(l.ctx, &node.FilterNodeParams{
|
|
Page: 1,
|
|
Size: 1000,
|
|
ServerId: []int64{server.Id},
|
|
Protocol: protocolRequest,
|
|
})
|
|
if err != nil {
|
|
l.Errorw("FilterNodeList error", logger.Field("error", err.Error()))
|
|
return nil, err
|
|
}
|
|
stats.nodesCount = len(nodes)
|
|
|
|
if len(nodes) == 0 {
|
|
l.Errorw("[ServerUserList] fallback: no nodes matched server+protocol, returning placeholder without cache",
|
|
logger.Field("server_id", req.ServerId),
|
|
logger.Field("protocol", req.Protocol),
|
|
)
|
|
stats.usersCount = 1
|
|
stats.speedLimitZeroCount = 1
|
|
l.logServerUserListPerf(stats)
|
|
return &types.GetServerUserListResponse{
|
|
Users: []types.ServerUser{
|
|
{
|
|
Id: 1,
|
|
UUID: uuidx.NewUUID().String(),
|
|
},
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// 收集所有唯一的节点组 ID
|
|
nodeGroupMap := make(map[int64]bool) // nodeGroupId -> true
|
|
var nodeIds []int64
|
|
var nodeTags []string
|
|
|
|
for _, n := range nodes {
|
|
nodeIds = append(nodeIds, n.Id)
|
|
if n.Tags != "" {
|
|
nodeTags = append(nodeTags, strings.Split(n.Tags, ",")...)
|
|
}
|
|
// 收集节点组 ID
|
|
if len(n.NodeGroupIds) > 0 {
|
|
for _, gid := range n.NodeGroupIds {
|
|
if gid > 0 {
|
|
nodeGroupMap[gid] = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 获取所有节点组 ID
|
|
nodeGroupIds := make([]int64, 0, len(nodeGroupMap))
|
|
for gid := range nodeGroupMap {
|
|
nodeGroupIds = append(nodeGroupIds, gid)
|
|
}
|
|
|
|
// 查询订阅：
|
|
// 1. 如果有节点组，查询匹配这些节点组的订阅
|
|
// 2. 如果没有节点组，查询使用节点 ID 或 tags 的订阅
|
|
var subs []*subscribe.Subscribe
|
|
if len(nodeGroupIds) > 0 {
|
|
// 节点组模式：查询 node_group_id 或 node_group_ids 匹配的订阅
|
|
_, subs, err = l.svcCtx.SubscribeModel.FilterListByNodeGroups(l.ctx, &subscribe.FilterByNodeGroupsParams{
|
|
Page: 1,
|
|
Size: 9999,
|
|
NodeGroupIds: nodeGroupIds,
|
|
})
|
|
if err != nil {
|
|
l.Errorw("FilterListByNodeGroups error", logger.Field("error", err.Error()))
|
|
return nil, err
|
|
}
|
|
} else {
|
|
// 传统模式：查询匹配节点 ID 或 tags 的订阅
|
|
nodeTags = tool.RemoveDuplicateElements(nodeTags...)
|
|
_, subs, err = l.svcCtx.SubscribeModel.FilterList(l.ctx, &subscribe.FilterParams{
|
|
Page: 1,
|
|
Size: 9999,
|
|
Node: nodeIds,
|
|
Tags: nodeTags,
|
|
})
|
|
if err != nil {
|
|
l.Errorw("FilterList error", logger.Field("error", err.Error()))
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
if len(subs) == 0 {
|
|
l.Errorw("[ServerUserList] fallback: no subscriptions matched node group/tags, returning placeholder without cache",
|
|
logger.Field("server_id", req.ServerId),
|
|
logger.Field("protocol", req.Protocol),
|
|
)
|
|
stats.usersCount = 1
|
|
stats.speedLimitZeroCount = 1
|
|
l.logServerUserListPerf(stats)
|
|
return &types.GetServerUserListResponse{
|
|
Users: []types.ServerUser{
|
|
{
|
|
Id: 1,
|
|
UUID: uuidx.NewUUID().String(),
|
|
},
|
|
},
|
|
}, nil
|
|
}
|
|
stats.subsCount = len(subs)
|
|
users := make([]types.ServerUser, 0)
|
|
speedCandidates := make(map[int64]serverUserSpeedLimitCandidate)
|
|
for _, sub := range subs {
|
|
data, err := l.svcCtx.UserModel.FindUsersSubscribeBySubscribeId(l.ctx, sub.Id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, datum := range data {
|
|
if !l.shouldIncludeServerUser(datum, nodeGroupIds) {
|
|
continue
|
|
}
|
|
|
|
baseSpeed, trafficLimit := serverUserSpeedLimitInputs(sub, datum)
|
|
speedCandidates[datum.Id] = serverUserSpeedLimitCandidate{
|
|
userID: datum.UserId,
|
|
userSubscribeID: datum.Id,
|
|
baseSpeed: baseSpeed,
|
|
trafficLimit: trafficLimit,
|
|
}
|
|
users = append(users, types.ServerUser{
|
|
Id: datum.Id,
|
|
UUID: datum.UUID,
|
|
SpeedLimit: baseSpeed,
|
|
DeviceLimit: sub.DeviceLimit,
|
|
})
|
|
}
|
|
}
|
|
trafficCalcStartedAt := time.Now()
|
|
speedLimits := l.calculateServerUserSpeedLimits(speedCandidates)
|
|
stats.trafficCalcMS = time.Since(trafficCalcStartedAt).Milliseconds()
|
|
for i := range users {
|
|
if speedLimit, ok := speedLimits[users[i].Id]; ok {
|
|
users[i].SpeedLimit = speedLimit
|
|
}
|
|
}
|
|
|
|
// 处理过期订阅用户：如果当前节点属于过期节点组，添加符合条件的过期用户。
|
|
// 用户级 speed_limit (user_subscribe.speed_limit) 与过期节点组 speed_limit
|
|
// 取更严格的一个 — 0 视为"无限制"，正值优先于 0。
|
|
if len(nodeGroupIds) > 0 {
|
|
expiredUsers, expiredSpeedLimit := l.getExpiredUsers(nodeGroupIds)
|
|
for i := range expiredUsers {
|
|
expiredUsers[i].SpeedLimit = mergeSpeedLimit(expiredUsers[i].SpeedLimit, expiredSpeedLimit)
|
|
}
|
|
users = append(users, expiredUsers...)
|
|
}
|
|
|
|
if len(users) == 0 {
|
|
l.Errorw("[ServerUserList] fallback: matched subs returned zero eligible users, returning placeholder without cache",
|
|
logger.Field("server_id", req.ServerId),
|
|
logger.Field("protocol", req.Protocol),
|
|
)
|
|
stats.usersCount = 1
|
|
stats.speedLimitZeroCount = 1
|
|
l.logServerUserListPerf(stats)
|
|
return &types.GetServerUserListResponse{
|
|
Users: []types.ServerUser{
|
|
{
|
|
Id: 1,
|
|
UUID: uuidx.NewUUID().String(),
|
|
},
|
|
},
|
|
}, nil
|
|
}
|
|
resp = &types.GetServerUserListResponse{
|
|
Users: users,
|
|
}
|
|
stats.usersCount = len(users)
|
|
stats.recordSpeedLimits(users)
|
|
val, _ := json.Marshal(resp)
|
|
etag := tool.GenerateETag(val)
|
|
l.ctx.Header("ETag", etag)
|
|
err = l.svcCtx.Redis.Set(l.ctx, cacheKey, string(val), l.serverUserListCacheTTL()).Err()
|
|
if err != nil {
|
|
l.Errorw("[ServerUserListCacheKey] redis set error", logger.Field("error", err.Error()))
|
|
}
|
|
// Check If-None-Match header
|
|
if match := l.ctx.GetHeader("If-None-Match"); match == etag {
|
|
l.logServerUserListPerf(stats)
|
|
return nil, xerr.StatusNotModified
|
|
}
|
|
l.logServerUserListPerf(stats)
|
|
return resp, nil
|
|
}
|
|
|
|
// normalizeServerUserListProtocol 将客户端可能携带的 hysteria2 兼容字段映射回
|
|
// DB 中存储的规范名 "hysteria"。其它协议原样返回。
|
|
// 缓存 key 与 FilterNodeList 查询都必须用归一化后的值，
|
|
// 否则 hysteria2 永远查不到节点，永远走兜底。
|
|
func normalizeServerUserListProtocol(protocol string) string {
|
|
if protocol == Hysteria2 {
|
|
return Hysteria
|
|
}
|
|
return protocol
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) serverUserListCacheTTL() time.Duration {
|
|
pullInterval := l.svcCtx.Config.Node.NodePullInterval
|
|
if pullInterval <= 0 {
|
|
pullInterval = 60
|
|
}
|
|
ttl := time.Duration(pullInterval*2) * time.Second
|
|
if ttl < time.Minute {
|
|
return time.Minute
|
|
}
|
|
return ttl
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) shouldIncludeServerUser(userSub *user.Subscribe, serverNodeGroupIds []int64) bool {
|
|
if userSub == nil {
|
|
return false
|
|
}
|
|
|
|
if userSub.ExpireTime.Unix() == 0 || userSub.ExpireTime.After(time.Now()) {
|
|
return true
|
|
}
|
|
|
|
return l.canUseExpiredNodeGroup(userSub, serverNodeGroupIds)
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) getExpiredUsers(serverNodeGroupIds []int64) ([]types.ServerUser, int64) {
|
|
var expiredGroup group.NodeGroup
|
|
if err := l.svcCtx.DB.Where("is_expired_group = ?", true).First(&expiredGroup).Error; err != nil {
|
|
return nil, 0
|
|
}
|
|
|
|
if !tool.Contains(serverNodeGroupIds, expiredGroup.Id) {
|
|
return nil, 0
|
|
}
|
|
|
|
var expiredSubs []*user.Subscribe
|
|
if err := l.svcCtx.DB.Where("status = ?", 3).Find(&expiredSubs).Error; err != nil {
|
|
l.Errorw("query expired subscriptions failed", logger.Field("error", err.Error()))
|
|
return nil, 0
|
|
}
|
|
|
|
users := make([]types.ServerUser, 0)
|
|
seen := make(map[int64]bool)
|
|
for _, userSub := range expiredSubs {
|
|
if !l.checkExpiredUserEligibility(userSub, &expiredGroup) {
|
|
continue
|
|
}
|
|
if seen[userSub.Id] {
|
|
continue
|
|
}
|
|
seen[userSub.Id] = true
|
|
users = append(users, types.ServerUser{
|
|
Id: userSub.Id,
|
|
UUID: userSub.UUID,
|
|
SpeedLimit: userSub.SpeedLimit,
|
|
})
|
|
}
|
|
|
|
return users, int64(expiredGroup.SpeedLimit)
|
|
}
|
|
|
|
// mergeSpeedLimit 返回两个速度限制（Mbps）中更严格的一个。
|
|
// 0 视为"无限制"，因此会被任意正值覆盖；都为 0 时返回 0。
|
|
// 用于用户级 speed_limit 与节点组级 speed_limit 的合并：
|
|
// - both 0 → 0 (不限速)
|
|
// - 仅一个 > 0 → 取该值
|
|
// - both > 0 → 取较小者（更严格）
|
|
func mergeSpeedLimit(a, b int64) int64 {
|
|
if a <= 0 {
|
|
return b
|
|
}
|
|
if b <= 0 {
|
|
return a
|
|
}
|
|
if a < b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) checkExpiredUserEligibility(userSub *user.Subscribe, expiredGroup *group.NodeGroup) bool {
|
|
expiredDays := int(time.Since(userSub.ExpireTime).Hours() / 24)
|
|
if expiredDays > expiredGroup.ExpiredDaysLimit {
|
|
return false
|
|
}
|
|
|
|
if expiredGroup.MaxTrafficGBExpired != nil && *expiredGroup.MaxTrafficGBExpired > 0 {
|
|
usedTrafficGB := (userSub.ExpiredDownload + userSub.ExpiredUpload) / (1024 * 1024 * 1024)
|
|
if usedTrafficGB >= *expiredGroup.MaxTrafficGBExpired {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) canUseExpiredNodeGroup(userSub *user.Subscribe, serverNodeGroupIds []int64) bool {
|
|
var expiredGroup group.NodeGroup
|
|
if err := l.svcCtx.DB.Where("is_expired_group = ?", true).First(&expiredGroup).Error; err != nil {
|
|
return false
|
|
}
|
|
|
|
if !tool.Contains(serverNodeGroupIds, expiredGroup.Id) {
|
|
return false
|
|
}
|
|
|
|
expiredDays := int(time.Since(userSub.ExpireTime).Hours() / 24)
|
|
if expiredDays > expiredGroup.ExpiredDaysLimit {
|
|
return false
|
|
}
|
|
|
|
if expiredGroup.MaxTrafficGBExpired != nil && *expiredGroup.MaxTrafficGBExpired > 0 {
|
|
usedTrafficGB := (userSub.ExpiredDownload + userSub.ExpiredUpload) / (1024 * 1024 * 1024)
|
|
if usedTrafficGB >= *expiredGroup.MaxTrafficGBExpired {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func serverUserSpeedLimitInputs(sub *subscribe.Subscribe, userSub *user.Subscribe) (int64, string) {
|
|
baseSpeed := sub.SpeedLimit
|
|
if userSub.SpeedLimit > 0 {
|
|
baseSpeed = userSub.SpeedLimit
|
|
}
|
|
|
|
trafficLimit := sub.TrafficLimit
|
|
if userSub.TrafficLimit != nil && *userSub.TrafficLimit != "" {
|
|
trafficLimit = *userSub.TrafficLimit
|
|
}
|
|
|
|
return baseSpeed, trafficLimit
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) calculateServerUserSpeedLimits(candidates map[int64]serverUserSpeedLimitCandidate) map[int64]int64 {
|
|
speedLimits := make(map[int64]int64, len(candidates))
|
|
rulesByUserSubscribeID := make(map[int64][]speedlimit.TrafficLimitRule)
|
|
windowByKey := make(map[string]serverUserTrafficWindow)
|
|
now := time.Now()
|
|
|
|
for userSubscribeID, candidate := range candidates {
|
|
speedLimits[userSubscribeID] = candidate.baseSpeed
|
|
if candidate.trafficLimit == "" {
|
|
continue
|
|
}
|
|
|
|
var rules []speedlimit.TrafficLimitRule
|
|
if err := json.Unmarshal([]byte(candidate.trafficLimit), &rules); err != nil || len(rules) == 0 {
|
|
continue
|
|
}
|
|
rulesByUserSubscribeID[userSubscribeID] = rules
|
|
for _, rule := range rules {
|
|
window, ok := serverUserTrafficRuleWindow(rule, now)
|
|
if !ok {
|
|
continue
|
|
}
|
|
windowByKey[serverUserTrafficWindowKey(rule.StatType, rule.StatValue)] = window
|
|
}
|
|
}
|
|
|
|
if len(rulesByUserSubscribeID) == 0 || len(windowByKey) == 0 {
|
|
return speedLimits
|
|
}
|
|
|
|
usageByKey := make(map[serverUserTrafficUsageKey]float64)
|
|
for _, window := range windowByKey {
|
|
trafficUsage, err := l.queryServerUserTrafficUsage(candidates, window)
|
|
if err != nil {
|
|
l.Errorw("[ServerUserList] batch traffic usage query failed",
|
|
logger.Field("error", err.Error()),
|
|
logger.Field("stat_type", window.statType),
|
|
logger.Field("stat_value", window.statValue),
|
|
)
|
|
continue
|
|
}
|
|
for _, usage := range trafficUsage {
|
|
usageByKey[serverUserTrafficUsageKey{
|
|
userID: usage.userID,
|
|
userSubscribeID: usage.userSubscribeID,
|
|
statType: window.statType,
|
|
statValue: window.statValue,
|
|
}] = usage.usedGB
|
|
}
|
|
}
|
|
|
|
for userSubscribeID, rules := range rulesByUserSubscribeID {
|
|
candidate := candidates[userSubscribeID]
|
|
for _, rule := range rules {
|
|
if rule.SpeedLimit <= 0 {
|
|
continue
|
|
}
|
|
if _, ok := serverUserTrafficRuleWindow(rule, now); !ok {
|
|
continue
|
|
}
|
|
usedGB := usageByKey[serverUserTrafficUsageKey{
|
|
userID: candidate.userID,
|
|
userSubscribeID: candidate.userSubscribeID,
|
|
statType: rule.StatType,
|
|
statValue: rule.StatValue,
|
|
}]
|
|
if usedGB < float64(rule.TrafficUsage) {
|
|
continue
|
|
}
|
|
current := speedLimits[userSubscribeID]
|
|
if current == 0 || rule.SpeedLimit < current {
|
|
speedLimits[userSubscribeID] = rule.SpeedLimit
|
|
}
|
|
}
|
|
}
|
|
|
|
return speedLimits
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) queryServerUserTrafficUsage(
|
|
candidates map[int64]serverUserSpeedLimitCandidate,
|
|
window serverUserTrafficWindow,
|
|
) ([]serverUserTrafficUsage, error) {
|
|
userSubscribeIDs := make([]int64, 0, len(candidates))
|
|
for userSubscribeID := range candidates {
|
|
userSubscribeIDs = append(userSubscribeIDs, userSubscribeID)
|
|
}
|
|
|
|
var rows []struct {
|
|
UserID int64
|
|
UserSubscribeID int64
|
|
Upload int64
|
|
Download int64
|
|
}
|
|
err := l.svcCtx.DB.WithContext(l.ctx.Request.Context()).
|
|
Table("traffic_log").
|
|
Select("user_id, subscribe_id AS user_subscribe_id, COALESCE(SUM(upload), 0) AS upload, COALESCE(SUM(download), 0) AS download").
|
|
Where("subscribe_id IN ? AND timestamp >= ? AND timestamp < ?", userSubscribeIDs, window.start, window.end).
|
|
Group("user_id, subscribe_id").
|
|
Scan(&rows).Error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
trafficUsage := make([]serverUserTrafficUsage, 0, len(rows))
|
|
for _, row := range rows {
|
|
trafficUsage = append(trafficUsage, serverUserTrafficUsage{
|
|
userID: row.UserID,
|
|
userSubscribeID: row.UserSubscribeID,
|
|
usedGB: float64(row.Upload+row.Download) / (1024 * 1024 * 1024),
|
|
})
|
|
}
|
|
return trafficUsage, nil
|
|
}
|
|
|
|
func serverUserTrafficRuleWindow(rule speedlimit.TrafficLimitRule, now time.Time) (serverUserTrafficWindow, bool) {
|
|
if rule.StatValue <= 0 {
|
|
return serverUserTrafficWindow{}, false
|
|
}
|
|
|
|
window := serverUserTrafficWindow{
|
|
statType: rule.StatType,
|
|
statValue: rule.StatValue,
|
|
end: now,
|
|
}
|
|
switch rule.StatType {
|
|
case "hour":
|
|
window.start = now.Add(-time.Duration(rule.StatValue) * time.Hour)
|
|
case "day":
|
|
window.start = now.AddDate(0, 0, -int(rule.StatValue))
|
|
default:
|
|
return serverUserTrafficWindow{}, false
|
|
}
|
|
return window, true
|
|
}
|
|
|
|
func serverUserTrafficWindowKey(statType string, statValue int64) string {
|
|
return fmt.Sprintf("%s:%d", statType, statValue)
|
|
}
|
|
|
|
func (s *serverUserListPerfStats) recordSpeedLimits(users []types.ServerUser) {
|
|
for _, user := range users {
|
|
if user.SpeedLimit > 0 {
|
|
s.speedLimitPositiveCount++
|
|
continue
|
|
}
|
|
s.speedLimitZeroCount++
|
|
}
|
|
}
|
|
|
|
func (l *GetServerUserListLogic) logServerUserListPerf(stats serverUserListPerfStats) {
|
|
l.Infow("[ServerUserList] performance",
|
|
logger.Field("server_id", stats.serverID),
|
|
logger.Field("protocol", stats.protocol),
|
|
logger.Field("cache_hit", stats.cacheHit),
|
|
logger.Field("nodes_count", stats.nodesCount),
|
|
logger.Field("subs_count", stats.subsCount),
|
|
logger.Field("users_count", stats.usersCount),
|
|
logger.Field("speed_limit_0_count", stats.speedLimitZeroCount),
|
|
logger.Field("speed_limit_positive_count", stats.speedLimitPositiveCount),
|
|
logger.Field("traffic_calc_ms", stats.trafficCalcMS),
|
|
logger.Field("total_ms", time.Since(stats.startedAt).Milliseconds()),
|
|
)
|
|
}
|