重构通道创建逻辑
This commit is contained in:
@@ -1,9 +1,7 @@
|
|||||||
## todo
|
## todo
|
||||||
|
|
||||||
- channel 接口
|
- channel 接口
|
||||||
- 每个用户-节点为一条数据,联查白名单
|
- 待验证
|
||||||
- 重新梳理逻辑流程,简化循环
|
|
||||||
- 端口分配时加锁
|
|
||||||
- 长效业务接入
|
- 长效业务接入
|
||||||
- 微信支付
|
- 微信支付
|
||||||
- 页面 账户总览
|
- 页面 账户总览
|
||||||
@@ -11,6 +9,8 @@
|
|||||||
- 页面 使用记录
|
- 页面 使用记录
|
||||||
- 代理数据表的 secret 字段 aes 加密存储
|
- 代理数据表的 secret 字段 aes 加密存储
|
||||||
|
|
||||||
|
- globals 合并到 services 或者反之
|
||||||
|
|
||||||
### 下阶段
|
### 下阶段
|
||||||
|
|
||||||
- 支付回调处理
|
- 支付回调处理
|
||||||
|
|||||||
@@ -97,13 +97,23 @@ type CreateChannelReq struct {
|
|||||||
Isp string `json:"isp"`
|
Isp string `json:"isp"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type CreateChannelRespItem struct {
|
||||||
|
Proto s.ChannelProtocol `json:"-"`
|
||||||
|
Host string `json:"host"`
|
||||||
|
Port int32 `json:"port"`
|
||||||
|
Username *string `json:"username,omitempty"`
|
||||||
|
Password *string `json:"password,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
func CreateChannel(c *fiber.Ctx) error {
|
func CreateChannel(c *fiber.Ctx) error {
|
||||||
|
|
||||||
// 检查权限
|
// 检查权限
|
||||||
authContext, err := auth.Protect(c, []auth.PayloadType{auth.PayloadUser}, []string{})
|
authContext, err := auth.Protect(c, []auth.PayloadType{auth.PayloadUser}, []string{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// 获取用户信息
|
|
||||||
|
// 检查用户其他权限
|
||||||
user, err := q.User.
|
user, err := q.User.
|
||||||
Where(q.User.ID.Eq(authContext.Payload.Id)).
|
Where(q.User.ID.Eq(authContext.Payload.Id)).
|
||||||
Take()
|
Take()
|
||||||
@@ -140,6 +150,7 @@ func CreateChannel(c *fiber.Ctx) error {
|
|||||||
isp = "移动"
|
isp = "移动"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 创建通道
|
||||||
result, err := s.Channel.CreateChannel(
|
result, err := s.Channel.CreateChannel(
|
||||||
c.Context(),
|
c.Context(),
|
||||||
authContext,
|
authContext,
|
||||||
@@ -157,7 +168,20 @@ func CreateChannel(c *fiber.Ctx) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.JSON(result)
|
// 返回结果
|
||||||
|
var resp = make([]*CreateChannelRespItem, len(result))
|
||||||
|
for i, channel := range result {
|
||||||
|
resp[i] = &CreateChannelRespItem{
|
||||||
|
Proto: req.Protocol,
|
||||||
|
Host: channel.ProxyHost,
|
||||||
|
Port: channel.ProxyPort,
|
||||||
|
}
|
||||||
|
if req.AuthType == s.ChannelAuthTypePass {
|
||||||
|
resp[i].Username = &channel.Username
|
||||||
|
resp[i].Password = &channel.Password
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return c.JSON(resp)
|
||||||
}
|
}
|
||||||
|
|
||||||
type CreateChannelResultType string
|
type CreateChannelResultType string
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ import (
|
|||||||
"platform/web/auth"
|
"platform/web/auth"
|
||||||
"platform/web/core"
|
"platform/web/core"
|
||||||
g "platform/web/globals"
|
g "platform/web/globals"
|
||||||
"platform/web/models"
|
m "platform/web/models"
|
||||||
q "platform/web/queries"
|
q "platform/web/queries"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -32,23 +32,6 @@ var Channel = &channelService{}
|
|||||||
type channelService struct {
|
type channelService struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type ChannelAuthType int
|
|
||||||
|
|
||||||
const (
|
|
||||||
ChannelAuthTypeAll ChannelAuthType = iota
|
|
||||||
ChannelAuthTypeIp
|
|
||||||
ChannelAuthTypePass
|
|
||||||
)
|
|
||||||
|
|
||||||
type ChannelProtocol int32
|
|
||||||
|
|
||||||
const (
|
|
||||||
ProtocolAll ChannelProtocol = iota
|
|
||||||
ProtocolHTTP
|
|
||||||
ProtocolHttps
|
|
||||||
ProtocolSocks5
|
|
||||||
)
|
|
||||||
|
|
||||||
type ResourceInfo struct {
|
type ResourceInfo struct {
|
||||||
Id int32
|
Id int32
|
||||||
UserId int32
|
UserId int32
|
||||||
@@ -135,7 +118,7 @@ func (s *channelService) RemoveChannels(ctx context.Context, authCtx *auth.Conte
|
|||||||
|
|
||||||
// 组织数据
|
// 组织数据
|
||||||
var configMap = make(map[int32][]g.PortConfigsReq, len(proxies))
|
var configMap = make(map[int32][]g.PortConfigsReq, len(proxies))
|
||||||
var proxyMap = make(map[int32]*models.Proxy, len(proxies))
|
var proxyMap = make(map[int32]*m.Proxy, len(proxies))
|
||||||
for _, proxy := range proxies {
|
for _, proxy := range proxies {
|
||||||
configMap[proxy.ID] = make([]g.PortConfigsReq, 0)
|
configMap[proxy.ID] = make([]g.PortConfigsReq, 0)
|
||||||
proxyMap[proxy.ID] = proxy
|
proxyMap[proxy.ID] = proxy
|
||||||
@@ -245,23 +228,81 @@ func (s *channelService) CreateChannel(
|
|||||||
authType ChannelAuthType,
|
authType ChannelAuthType,
|
||||||
count int,
|
count int,
|
||||||
nodeFilter ...NodeFilterConfig,
|
nodeFilter ...NodeFilterConfig,
|
||||||
) ([]*PortInfo, error) {
|
) (newChannels []*m.Channel, err error) {
|
||||||
var now = time.Now()
|
var now = time.Now()
|
||||||
|
|
||||||
var step = time.Now()
|
|
||||||
var rid = ctx.Value(requestid.ConfigDefault.ContextKey).(string)
|
var rid = ctx.Value(requestid.ConfigDefault.ContextKey).(string)
|
||||||
|
var filter = NodeFilterConfig{}
|
||||||
filter := NodeFilterConfig{}
|
|
||||||
if len(nodeFilter) > 0 {
|
if len(nodeFilter) > 0 {
|
||||||
filter = nodeFilter[0]
|
filter = nodeFilter[0]
|
||||||
}
|
}
|
||||||
|
|
||||||
var addr []*PortInfo
|
err = q.Q.Transaction(func(q *q.Query) (err error) {
|
||||||
err := q.Q.Transaction(func(q *q.Query) error {
|
|
||||||
|
|
||||||
// 查找套餐
|
// 查找套餐
|
||||||
step = time.Now()
|
resource, err := findResource(q, rid, resourceId, authCtx, count)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 查找网关
|
||||||
|
proxies, err := findProxies(q, rid)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 查找已使用的节点
|
||||||
|
channels, err := findChannels(q, rid, proxies)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 查找白名单
|
||||||
|
var whitelist *[]string
|
||||||
|
if authType == ChannelAuthTypeIp {
|
||||||
|
whitelist, err = findWhitelist(q, rid, authCtx.Payload.Id)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 分配节点
|
||||||
|
var expire = now.Add(time.Duration(resource.Live) * time.Second)
|
||||||
|
newChannels, err = calcChannels(proxies, channels, whitelist, count, authCtx.Payload.Id, protocol, authType, expire, filter)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 更新套餐使用记录
|
||||||
|
err = updateResource(rid, resource, count, now)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 保存通道
|
||||||
|
err = saveChannels(newChannels)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 缓存通道数据
|
||||||
|
err = cacheChannels(ctx, rid, newChannels)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return newChannels, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func findResource(q *q.Query, rid string, resourceId int32, authCtx *auth.Context, count int) (*ResourceInfo, error) {
|
||||||
|
var step = time.Now()
|
||||||
|
|
||||||
|
// 查找套餐
|
||||||
var resource = new(ResourceInfo)
|
var resource = new(ResourceInfo)
|
||||||
data := q.Resource.As("data")
|
data := q.Resource.As("data")
|
||||||
pss := q.ResourcePss.As("pss")
|
pss := q.ResourcePss.As("pss")
|
||||||
@@ -271,159 +312,68 @@ func (s *channelService) CreateChannel(
|
|||||||
pss.Type, pss.Live, pss.DailyUsed, pss.DailyLimit, pss.DailyLast, pss.Quota, pss.Used, pss.Expire,
|
pss.Type, pss.Live, pss.DailyUsed, pss.DailyLimit, pss.DailyLast, pss.Quota, pss.Used, pss.Expire,
|
||||||
).
|
).
|
||||||
LeftJoin(q.ResourcePss.As("pss"), pss.ResourceID.EqCol(data.ID)).
|
LeftJoin(q.ResourcePss.As("pss"), pss.ResourceID.EqCol(data.ID)).
|
||||||
Where(data.ID.Eq(resourceId)).
|
Where(
|
||||||
|
data.ID.Eq(resourceId),
|
||||||
|
data.UserID.Eq(authCtx.Payload.Id),
|
||||||
|
).
|
||||||
Scan(&resource)
|
Scan(&resource)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||||
// 禁止 id 猜测
|
return nil, ErrResourceNotExist // 防止 id 猜测
|
||||||
return ChannelServiceErr("无权限访问")
|
|
||||||
}
|
}
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
slog.Debug("查找套餐", "rid", rid, "step", time.Since(step))
|
|
||||||
|
|
||||||
// 检查用户权限
|
|
||||||
err = checkUser(authCtx, resource, count)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// 申请节点
|
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
edgeAssigns, err := assignEdge(q, count, filter)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
slog.Debug("申请节点", "rid", rid, "total", time.Since(step))
|
|
||||||
|
|
||||||
// 分配端口
|
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
expiration := core.LocalDateTime(now.Add(time.Duration(resource.Live) * time.Second))
|
|
||||||
_addr, channels, err := assignPort(q, edgeAssigns, authCtx.Payload.Id, protocol, authType, expiration, filter)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
addr = _addr
|
|
||||||
|
|
||||||
slog.Debug("分配端口", "rid", rid, "total", time.Since(step))
|
|
||||||
|
|
||||||
// 更新套餐使用记录
|
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
toUpdate := models.ResourcePss{
|
|
||||||
Used: resource.Used + int32(count),
|
|
||||||
DailyLast: core.LocalDateTime(now),
|
|
||||||
}
|
|
||||||
var last = time.Time(resource.DailyLast)
|
|
||||||
if now.Year() != last.Year() || now.Month() != last.Month() || now.Day() != last.Day() {
|
|
||||||
toUpdate.DailyUsed = int32(count)
|
|
||||||
} else {
|
|
||||||
toUpdate.DailyUsed = resource.DailyUsed + int32(count)
|
|
||||||
}
|
|
||||||
_, err = q.ResourcePss.
|
|
||||||
Where(q.ResourcePss.ResourceID.Eq(resourceId)).
|
|
||||||
Select(
|
|
||||||
q.ResourcePss.Used,
|
|
||||||
q.ResourcePss.DailyUsed,
|
|
||||||
q.ResourcePss.DailyLast,
|
|
||||||
).
|
|
||||||
Updates(toUpdate)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
slog.Debug("更新套餐使用记录", "rid", rid, "step", time.Since(step))
|
|
||||||
|
|
||||||
// 缓存通道数据
|
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
err = cache(ctx, channels)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
slog.Debug("缓存通道数据", "rid", rid, "step", time.Since(step))
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return addr, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func checkUser(authCtx *auth.Context, resource *ResourceInfo, count int) error {
|
|
||||||
|
|
||||||
// 检查使用人
|
|
||||||
if authCtx.Payload.Type == auth.PayloadUser && authCtx.Payload.Id != resource.UserId {
|
|
||||||
return core.ForbiddenErr("无权限访问")
|
|
||||||
}
|
|
||||||
|
|
||||||
// 检查套餐状态
|
// 检查套餐状态
|
||||||
if !resource.Active {
|
if !resource.Active {
|
||||||
return ChannelServiceErr("套餐已失效")
|
return nil, ErrResourceInvalid
|
||||||
}
|
}
|
||||||
|
|
||||||
// 检查每日限额
|
// 检查每日限额
|
||||||
today := time.Now().Format("2006-01-02") == time.Time(resource.DailyLast).Format("2006-01-02")
|
today := time.Now().Format("2006-01-02") == time.Time(resource.DailyLast).Format("2006-01-02")
|
||||||
dailyRemain := int(math.Max(float64(resource.DailyLimit-resource.DailyUsed), 0))
|
dailyRemain := int(math.Max(float64(resource.DailyLimit-resource.DailyUsed), 0))
|
||||||
if today && dailyRemain < count {
|
if today && dailyRemain < count {
|
||||||
return ChannelServiceErr("套餐每日配额不足")
|
return nil, ErrResourceDailyLimit
|
||||||
}
|
}
|
||||||
|
|
||||||
// 检查时间或配额
|
// 检查时间或配额
|
||||||
if resource.Type == 1 { // 包时
|
if resource.Type == 1 { // 包时
|
||||||
if time.Time(resource.Expire).Before(time.Now()) {
|
if time.Time(resource.Expire).Before(time.Now()) {
|
||||||
return ChannelServiceErr("套餐已过期")
|
return nil, ErrResourceExpired
|
||||||
}
|
}
|
||||||
} else { // 包量
|
} else { // 包量
|
||||||
remain := int(math.Max(float64(resource.Quota-resource.Used), 0))
|
remain := int(math.Max(float64(resource.Quota-resource.Used), 0))
|
||||||
if remain < count {
|
if remain < count {
|
||||||
return ChannelServiceErr("套餐配额不足")
|
return nil, ErrResourceExhausted
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
slog.Debug("查找套餐", "rid", rid, "step", time.Since(step))
|
||||||
|
return resource, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// assignEdge 分配边缘节点数量
|
func findProxies(q *q.Query, rid string) (proxies []*m.Proxy, err error) {
|
||||||
func assignEdge(q *q.Query, count int, filter NodeFilterConfig) (*AssignEdgeResult, error) {
|
|
||||||
|
|
||||||
// 查询可以使用的网关
|
|
||||||
var step = time.Now()
|
var step = time.Now()
|
||||||
|
|
||||||
proxies, err := q.Proxy.
|
proxies, err = q.Proxy.
|
||||||
Where(q.Proxy.Type.Eq(1)).
|
Where(q.Proxy.Type.Eq(1)).
|
||||||
Find()
|
Find()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Debug("查找网关", "step", time.Since(step))
|
slog.Debug("查找网关", "rid", rid, "step", time.Since(step))
|
||||||
|
return proxies, nil
|
||||||
|
}
|
||||||
|
|
||||||
// 查询已配置的节点
|
func findChannels(q *q.Query, rid string, proxies []*m.Proxy) (channels []*m.Channel, err error) {
|
||||||
step = time.Now()
|
var step = time.Now()
|
||||||
|
|
||||||
rProxyConfigs, err := g.Cloud.CloudAutoQuery()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
slog.Debug("查询已配置节点 (remote)", "step", time.Since(step))
|
|
||||||
|
|
||||||
// 查询已使用的节点
|
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
var proxyIds = make([]int32, len(proxies))
|
var proxyIds = make([]int32, len(proxies))
|
||||||
for i, proxy := range proxies {
|
for i, proxy := range proxies {
|
||||||
proxyIds[i] = proxy.ID
|
proxyIds[i] = proxy.ID
|
||||||
}
|
}
|
||||||
channels, err := q.Channel.Debug().
|
channels, err = q.Channel.Debug().
|
||||||
Select(
|
Select(
|
||||||
q.Channel.ProxyID,
|
q.Channel.ProxyID,
|
||||||
q.Channel.ProxyPort).
|
q.Channel.ProxyPort).
|
||||||
@@ -437,74 +387,111 @@ func assignEdge(q *q.Query, count int, filter NodeFilterConfig) (*AssignEdgeResu
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
var proxyUses = make(map[int32]int, len(channels))
|
|
||||||
for _, channel := range channels {
|
slog.Debug("查找已使用节点", "rid", rid, "step", time.Since(step))
|
||||||
|
return channels, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func findWhitelist(q *q.Query, rid string, userId int32) (*[]string, error) {
|
||||||
|
var step = time.Now()
|
||||||
|
|
||||||
|
// 按需查找用户白名单
|
||||||
|
var whitelist []string
|
||||||
|
err := q.Whitelist.
|
||||||
|
Where(q.Whitelist.UserID.Eq(userId)).
|
||||||
|
Select(q.Whitelist.Host).
|
||||||
|
Scan(&whitelist)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if len(whitelist) == 0 {
|
||||||
|
return nil, ChannelServiceErr("用户没有白名单")
|
||||||
|
}
|
||||||
|
|
||||||
|
slog.Debug("查找用户白名单", "rid", rid, "step", time.Since(step))
|
||||||
|
return &whitelist, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func calcChannels(
|
||||||
|
proxies []*m.Proxy,
|
||||||
|
allChannels []*m.Channel,
|
||||||
|
whitelist *[]string,
|
||||||
|
count int,
|
||||||
|
userId int32,
|
||||||
|
protocol ChannelProtocol,
|
||||||
|
authType ChannelAuthType,
|
||||||
|
expiration time.Time,
|
||||||
|
filter NodeFilterConfig,
|
||||||
|
) ([]*m.Channel, error) {
|
||||||
|
var step = time.Now()
|
||||||
|
|
||||||
|
// 查询已配置的节点
|
||||||
|
remoteConfigs, err := g.Cloud.CloudAutoQuery()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// 统计已用节点量与端口查找表
|
||||||
|
var proxyUses = make(map[int32]int, len(allChannels))
|
||||||
|
var portsMap = make(map[uint64]struct{})
|
||||||
|
for _, channel := range allChannels {
|
||||||
proxyUses[channel.ProxyID]++
|
proxyUses[channel.ProxyID]++
|
||||||
|
key := uint64(channel.ProxyID)<<32 | uint64(channel.ProxyPort)
|
||||||
|
portsMap[key] = struct{}{}
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Debug("查找已使用节点", "step", time.Since(step))
|
// 计算分配额度
|
||||||
|
var total = len(allChannels) + count
|
||||||
|
var avg = int(math.Ceil(float64(total) / float64(len(proxies))))
|
||||||
|
|
||||||
// 组织数据
|
// 分配节点
|
||||||
var infos = make([]*ProxyInfo, len(proxies))
|
var newChannels []*m.Channel
|
||||||
for i, proxy := range proxies {
|
for _, proxy := range proxies {
|
||||||
infos[i] = &ProxyInfo{
|
|
||||||
proxy: proxy,
|
|
||||||
used: proxyUses[proxy.ID],
|
|
||||||
}
|
|
||||||
|
|
||||||
rConfigs, ok := rProxyConfigs[proxy.Name]
|
// 分配前后的节点量
|
||||||
if !ok {
|
var prev = proxyUses[proxy.ID]
|
||||||
infos[i].count = 0
|
var next = int(math.Max(float64(prev), float64(int(math.Min(float64(avg), float64(total))))))
|
||||||
continue
|
total -= next
|
||||||
}
|
|
||||||
|
|
||||||
for _, rConfig := range rConfigs {
|
// 网关配置的节点量
|
||||||
if rConfig.Isp == filter.Isp && rConfig.City == filter.City && rConfig.Province == filter.Prov {
|
var count = 0
|
||||||
infos[i].count = rConfig.Count
|
remoteConfig, ok := remoteConfigs[proxy.Name]
|
||||||
|
if ok {
|
||||||
|
for _, config := range remoteConfig {
|
||||||
|
if config.Isp == filter.Isp && config.City == filter.City && config.Province == filter.Prov {
|
||||||
|
count = config.Count
|
||||||
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 分配新增的节点
|
if env.DebugExternalChange && next > count {
|
||||||
var configs = make([]*ProxyConfig, len(proxies))
|
|
||||||
var needed = len(channels) + count
|
|
||||||
avg := int(math.Ceil(float64(needed) / float64(len(proxies))))
|
|
||||||
for i, info := range infos {
|
|
||||||
var prev = info.used
|
|
||||||
var next = int(math.Min(float64(avg), float64(needed)))
|
|
||||||
|
|
||||||
info.used = int(math.Max(float64(prev), float64(next)))
|
|
||||||
needed -= info.used
|
|
||||||
|
|
||||||
if env.DebugExternalChange && info.used > info.count {
|
|
||||||
step = time.Now()
|
step = time.Now()
|
||||||
|
|
||||||
slog.Debug("新增新节点", "proxy", info.proxy.Name, "used", info.used, "count", info.count)
|
var multiple float64 = 2 // 扩张倍数
|
||||||
|
|
||||||
rConfigs := rProxyConfigs[info.proxy.Name]
|
|
||||||
|
|
||||||
var newConfig = g.AutoConfig{
|
var newConfig = g.AutoConfig{
|
||||||
Province: filter.Prov,
|
Province: filter.Prov,
|
||||||
City: filter.City,
|
City: filter.City,
|
||||||
Isp: filter.Isp,
|
Isp: filter.Isp,
|
||||||
Count: int(math.Ceil(float64(info.used) * 2)),
|
Count: int(math.Ceil(float64(next) * multiple)),
|
||||||
}
|
}
|
||||||
|
|
||||||
var newConfigs []g.AutoConfig
|
var newConfigs []g.AutoConfig
|
||||||
var update = false
|
if count == 0 {
|
||||||
for _, rConfig := range rConfigs {
|
|
||||||
if rConfig.Isp == filter.Isp && rConfig.City == filter.City && rConfig.Province == filter.Prov {
|
|
||||||
newConfigs = append(newConfigs, newConfig)
|
newConfigs = append(newConfigs, newConfig)
|
||||||
update = true
|
|
||||||
} else {
|
} else {
|
||||||
newConfigs = append(newConfigs, rConfig)
|
newConfigs = make([]g.AutoConfig, len(remoteConfig))
|
||||||
|
for i, config := range remoteConfig {
|
||||||
|
if config.Isp == filter.Isp && config.City == filter.City && config.Province == filter.Prov {
|
||||||
|
count = config.Count
|
||||||
|
break
|
||||||
}
|
}
|
||||||
|
newConfigs[i] = config
|
||||||
}
|
}
|
||||||
if !update {
|
|
||||||
newConfigs = append(newConfigs, newConfig)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
err := g.Cloud.CloudConnect(g.CloudConnectReq{
|
err := g.Cloud.CloudConnect(g.CloudConnectReq{
|
||||||
Uuid: info.proxy.Name,
|
Uuid: proxy.Name,
|
||||||
Edge: nil,
|
Edge: nil,
|
||||||
AutoConfig: newConfigs,
|
AutoConfig: newConfigs,
|
||||||
})
|
})
|
||||||
@@ -512,86 +499,23 @@ func assignEdge(q *q.Query, count int, filter NodeFilterConfig) (*AssignEdgeResu
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Debug("分配新增的节点", "step", time.Since(step))
|
slog.Debug("提交节点配置",
|
||||||
|
slog.Duration("step", time.Since(step)),
|
||||||
|
slog.String("proxy", proxy.Name),
|
||||||
|
slog.Int("used", prev),
|
||||||
|
slog.Int("count", next),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
configs[i] = &ProxyConfig{
|
// 节点增量
|
||||||
proxy: info.proxy,
|
var acc = next - prev
|
||||||
count: int(math.Max(float64(next-prev), 0)),
|
if acc <= 0 {
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
return &AssignEdgeResult{
|
|
||||||
configs: configs,
|
|
||||||
channels: channels,
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
type ProxyInfo struct {
|
|
||||||
proxy *models.Proxy
|
|
||||||
used int
|
|
||||||
count int
|
|
||||||
}
|
|
||||||
|
|
||||||
type AssignEdgeResult struct {
|
|
||||||
configs []*ProxyConfig
|
|
||||||
channels []*models.Channel
|
|
||||||
}
|
|
||||||
|
|
||||||
type ProxyConfig struct {
|
|
||||||
proxy *models.Proxy
|
|
||||||
count int
|
|
||||||
}
|
|
||||||
|
|
||||||
// assignPort 分配指定数量的端口
|
|
||||||
func assignPort(
|
|
||||||
q *q.Query,
|
|
||||||
proxies *AssignEdgeResult,
|
|
||||||
userId int32,
|
|
||||||
protocol ChannelProtocol,
|
|
||||||
authType ChannelAuthType,
|
|
||||||
expiration core.LocalDateTime,
|
|
||||||
filter NodeFilterConfig,
|
|
||||||
) ([]*PortInfo, []*models.Channel, error) {
|
|
||||||
var step time.Time
|
|
||||||
|
|
||||||
var configs = proxies.configs
|
|
||||||
var exists = proxies.channels
|
|
||||||
|
|
||||||
// 端口查找表
|
|
||||||
var portsMap = make(map[uint64]struct{})
|
|
||||||
for _, channel := range exists {
|
|
||||||
key := uint64(channel.ProxyID)<<32 | uint64(channel.ProxyPort)
|
|
||||||
portsMap[key] = struct{}{}
|
|
||||||
}
|
|
||||||
println(len(portsMap))
|
|
||||||
|
|
||||||
// 查找用户白名单
|
|
||||||
var whitelist []string
|
|
||||||
if authType == ChannelAuthTypeIp {
|
|
||||||
err := q.Whitelist.
|
|
||||||
Where(q.Whitelist.UserID.Eq(userId)).
|
|
||||||
Select(q.Whitelist.Host).
|
|
||||||
Scan(&whitelist)
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
if len(whitelist) == 0 {
|
|
||||||
return nil, nil, ChannelServiceErr("用户没有白名单")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 配置启用代理
|
|
||||||
var result []*PortInfo
|
|
||||||
var channels []*models.Channel
|
|
||||||
for _, config := range configs {
|
|
||||||
var err error
|
|
||||||
var proxy = config.proxy
|
|
||||||
var count = config.count
|
|
||||||
|
|
||||||
// 筛选可用端口
|
// 筛选可用端口
|
||||||
var configs = make([]g.PortConfigsReq, 0, count)
|
var portConfigs = make([]g.PortConfigsReq, 0, acc)
|
||||||
for port := 10000; port < 20000 && len(configs) < count; port++ {
|
for port := 10000; port < 20000 && len(portConfigs) < acc; port++ {
|
||||||
// 跳过存在的端口
|
// 跳过存在的端口
|
||||||
key := uint64(proxy.ID)<<32 | uint64(port)
|
key := uint64(proxy.ID)<<32 | uint64(port)
|
||||||
_, ok := portsMap[key]
|
_, ok := portsMap[key]
|
||||||
@@ -600,8 +524,7 @@ func assignPort(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 配置新端口
|
// 配置新端口
|
||||||
var i = len(configs)
|
var portConf = g.PortConfigsReq{
|
||||||
configs = append(configs, g.PortConfigsReq{
|
|
||||||
Port: port,
|
Port: port,
|
||||||
Edge: nil,
|
Edge: nil,
|
||||||
Status: true,
|
Status: true,
|
||||||
@@ -612,78 +535,42 @@ func assignPort(
|
|||||||
Count: u.P(1),
|
Count: u.P(1),
|
||||||
PacketLoss: 30,
|
PacketLoss: 30,
|
||||||
},
|
},
|
||||||
})
|
}
|
||||||
|
var newChannel = &m.Channel{
|
||||||
|
UserID: userId,
|
||||||
|
ProxyID: proxy.ID,
|
||||||
|
ProxyHost: proxy.Host,
|
||||||
|
ProxyPort: int32(port),
|
||||||
|
Protocol: int32(protocol),
|
||||||
|
Expiration: core.LocalDateTime(expiration),
|
||||||
|
}
|
||||||
|
|
||||||
switch authType {
|
switch authType {
|
||||||
|
|
||||||
case ChannelAuthTypeIp:
|
case ChannelAuthTypeIp:
|
||||||
configs[i].Whitelist = &whitelist
|
portConf.Whitelist = whitelist
|
||||||
configs[i].Userpass = u.P("")
|
portConf.Userpass = u.P("")
|
||||||
for _, item := range whitelist {
|
newChannel.AuthIP = true
|
||||||
channels = append(channels, &models.Channel{
|
|
||||||
UserID: userId,
|
|
||||||
ProxyID: proxy.ID,
|
|
||||||
UserHost: item,
|
|
||||||
ProxyHost: proxy.Host,
|
|
||||||
ProxyPort: int32(port),
|
|
||||||
AuthIP: true,
|
|
||||||
AuthPass: false,
|
|
||||||
Protocol: int32(protocol),
|
|
||||||
Expiration: expiration,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
result = append(result, &PortInfo{
|
|
||||||
Proto: protocol,
|
|
||||||
Host: proxy.Host,
|
|
||||||
Port: port,
|
|
||||||
})
|
|
||||||
case ChannelAuthTypePass:
|
case ChannelAuthTypePass:
|
||||||
username, password := genPassPair()
|
username, password := genPassPair()
|
||||||
configs[i].Whitelist = &[]string{}
|
portConf.Whitelist = &[]string{}
|
||||||
configs[i].Userpass = u.P(fmt.Sprintf("%s:%s", username, password))
|
portConf.Userpass = u.P(fmt.Sprintf("%s:%s", username, password))
|
||||||
channels = append(channels, &models.Channel{
|
newChannel.AuthPass = true
|
||||||
UserID: userId,
|
newChannel.Username = username
|
||||||
ProxyID: proxy.ID,
|
newChannel.Password = password
|
||||||
ProxyHost: proxy.Host,
|
|
||||||
ProxyPort: int32(port),
|
|
||||||
AuthIP: false,
|
|
||||||
AuthPass: true,
|
|
||||||
Username: username,
|
|
||||||
Password: password,
|
|
||||||
Protocol: int32(protocol),
|
|
||||||
Expiration: expiration,
|
|
||||||
})
|
|
||||||
result = append(result, &PortInfo{
|
|
||||||
Proto: protocol,
|
|
||||||
Host: proxy.Host,
|
|
||||||
Port: port,
|
|
||||||
Username: &username,
|
|
||||||
Password: &password,
|
|
||||||
})
|
|
||||||
default:
|
default:
|
||||||
return nil, nil, ChannelServiceErr("不支持的通道认证方式")
|
return nil, ChannelServiceErr("不支持的通道认证方式")
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(configs) < count {
|
portConfigs = append(portConfigs, portConf)
|
||||||
return nil, nil, ChannelServiceErr("网关端口数量到达上限,无法分配")
|
newChannels = append(newChannels, newChannel)
|
||||||
}
|
}
|
||||||
|
if len(portConfigs) < acc {
|
||||||
// 保存到数据库
|
return nil, ChannelServiceErr("网关端口数量到达上限,无法分配")
|
||||||
step = time.Now()
|
|
||||||
|
|
||||||
err = q.Channel.
|
|
||||||
Omit(
|
|
||||||
q.Channel.NodeID,
|
|
||||||
q.Channel.NodeHost,
|
|
||||||
q.Channel.DeletedAt,
|
|
||||||
).
|
|
||||||
Create(channels...)
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Debug("保存到数据库", "step", time.Since(step))
|
|
||||||
|
|
||||||
// 提交端口配置并更新节点列表
|
// 提交端口配置并更新节点列表
|
||||||
if env.DebugExternalChange {
|
if env.DebugExternalChange {
|
||||||
step = time.Now()
|
step = time.Now()
|
||||||
@@ -694,24 +581,77 @@ func assignPort(
|
|||||||
secret[0],
|
secret[0],
|
||||||
secret[1],
|
secret[1],
|
||||||
)
|
)
|
||||||
err = gateway.GatewayPortConfigs(configs)
|
err = gateway.GatewayPortConfigs(portConfigs)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
slog.Debug("提交端口配置", "step", time.Since(step))
|
slog.Debug("提交端口配置", "step", time.Since(step))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return result, channels, nil
|
slog.Debug("申请节点", "rid", step, "total", time.Since(step))
|
||||||
|
return newChannels, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type PortInfo struct {
|
func updateResource(rid string, resource *ResourceInfo, count int, now time.Time) (err error) {
|
||||||
Proto ChannelProtocol `json:"-"`
|
var step = time.Now()
|
||||||
Host string `json:"host"`
|
|
||||||
Port int `json:"port"`
|
toUpdate := m.ResourcePss{
|
||||||
Username *string `json:"username,omitempty"`
|
Used: resource.Used + int32(count),
|
||||||
Password *string `json:"password,omitempty"`
|
DailyLast: core.LocalDateTime(now),
|
||||||
|
}
|
||||||
|
var last = time.Time(resource.DailyLast)
|
||||||
|
if now.Year() != last.Year() || now.Month() != last.Month() || now.Day() != last.Day() {
|
||||||
|
toUpdate.DailyUsed = int32(count)
|
||||||
|
} else {
|
||||||
|
toUpdate.DailyUsed = resource.DailyUsed + int32(count)
|
||||||
|
}
|
||||||
|
_, err = q.ResourcePss.
|
||||||
|
Where(q.ResourcePss.ResourceID.Eq(resource.Id)).
|
||||||
|
Select(
|
||||||
|
q.ResourcePss.Used,
|
||||||
|
q.ResourcePss.DailyUsed,
|
||||||
|
q.ResourcePss.DailyLast,
|
||||||
|
).
|
||||||
|
Updates(toUpdate)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
slog.Debug("更新套餐使用记录", "rid", rid, "step", time.Since(step))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func saveChannels(channels []*m.Channel) (err error) {
|
||||||
|
// 保存到数据库
|
||||||
|
var step = time.Now()
|
||||||
|
|
||||||
|
err = q.Channel.
|
||||||
|
Omit(
|
||||||
|
q.Channel.NodeID,
|
||||||
|
q.Channel.NodeHost,
|
||||||
|
q.Channel.DeletedAt,
|
||||||
|
).
|
||||||
|
Create(channels...)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
slog.Debug("保存到数据库", "step", time.Since(step))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func cacheChannels(ctx context.Context, rid string, channels []*m.Channel) (err error) {
|
||||||
|
var step = time.Now()
|
||||||
|
|
||||||
|
err = cache(ctx, channels)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
slog.Debug("缓存通道数据", "rid", rid, "step", time.Since(step))
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// endregion
|
// endregion
|
||||||
@@ -735,11 +675,11 @@ func genPassPair() (string, string) {
|
|||||||
return string(username), string(password)
|
return string(username), string(password)
|
||||||
}
|
}
|
||||||
|
|
||||||
func chKey(channel *models.Channel) string {
|
func chKey(channel *m.Channel) string {
|
||||||
return fmt.Sprintf("channel:%d", channel.ID)
|
return fmt.Sprintf("channel:%d", channel.ID)
|
||||||
}
|
}
|
||||||
|
|
||||||
func cache(ctx context.Context, channels []*models.Channel) error {
|
func cache(ctx context.Context, channels []*m.Channel) error {
|
||||||
if len(channels) == 0 {
|
if len(channels) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -769,7 +709,7 @@ func cache(ctx context.Context, channels []*models.Channel) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func deleteCache(ctx context.Context, channels []*models.Channel) error {
|
func deleteCache(ctx context.Context, channels []*m.Channel) error {
|
||||||
if len(channels) == 0 {
|
if len(channels) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -786,8 +726,33 @@ func deleteCache(ctx context.Context, channels []*models.Channel) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type ChannelAuthType int
|
||||||
|
|
||||||
|
const (
|
||||||
|
ChannelAuthTypeAll ChannelAuthType = iota
|
||||||
|
ChannelAuthTypeIp
|
||||||
|
ChannelAuthTypePass
|
||||||
|
)
|
||||||
|
|
||||||
|
type ChannelProtocol int32
|
||||||
|
|
||||||
|
const (
|
||||||
|
ProtocolAll ChannelProtocol = iota
|
||||||
|
ProtocolHTTP
|
||||||
|
ProtocolHttps
|
||||||
|
ProtocolSocks5
|
||||||
|
)
|
||||||
|
|
||||||
type ChannelServiceErr string
|
type ChannelServiceErr string
|
||||||
|
|
||||||
func (c ChannelServiceErr) Error() string {
|
func (c ChannelServiceErr) Error() string {
|
||||||
return string(c)
|
return string(c)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const (
|
||||||
|
ErrResourceNotExist = ChannelServiceErr("套餐不存在")
|
||||||
|
ErrResourceInvalid = ChannelServiceErr("套餐不可用")
|
||||||
|
ErrResourceExhausted = ChannelServiceErr("套餐已用完")
|
||||||
|
ErrResourceExpired = ChannelServiceErr("套餐已过期")
|
||||||
|
ErrResourceDailyLimit = ChannelServiceErr("套餐每日配额已用完")
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user