lanhu/platform
2
0
Fork 0
Code Issues Actions Packages Releases 12 Activity

重构代理和边缘节点接口，更新请求和响应结构，添加全量节点信息返回，引入全局锁以防止并发注册；代理服务下线后相关节点也标记下线

Browse Source
This commit is contained in:
luorijun
2025-05-27 15:08:18 +08:00
parent c08d625975
commit 933c947b3e
9 changed files with 1222 additions and 649 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
web/domains/edge/types.go
View File

@@ -23,7 +23,7 @@ func ISPFromStr(str string) ISP {
return IspUnknown
}
func ISPToStr(isp ISP) string {
func (isp ISP) String() string {
switch isp {
case IspChinaTelecom:
return "电信"

11
web/globals/redis.go
View File

@@ -1,20 +1,29 @@
package globals
import (
"github.com/go-redsync/redsync/v4/redis/goredis/v9"
"net"
"platform/pkg/env"
"github.com/go-redsync/redsync/v4"
"github.com/redis/go-redis/v9"
)
var Redis *redis.Client
var Redsync *redsync.Redsync
func initRedis() {
Redis = redis.NewClient(&redis.Options{
client := redis.NewClient(&redis.Options{
Addr: net.JoinHostPort(env.RedisHost, env.RedisPort),
DB: env.RedisDb,
Password: env.RedisPass,
})
pool := goredis.NewPool(client)
sync := redsync.New(pool)
Redis = client
Redsync = sync
}
func ExitRedis() error {

124
web/handlers/edge.go
View File

@@ -1,6 +1,10 @@
package handlers
import (
"errors"
"gorm.io/gen/field"
"gorm.io/gorm"
"log/slog"
"platform/pkg/u"
"platform/web/auth"
edge2 "platform/web/domains/edge"
@@ -11,16 +15,11 @@ import (
s "platform/web/services"
"github.com/gofiber/fiber/v2"
"gorm.io/gen/field"
"gorm.io/gorm/clause"
)
type RegisterEdgeReq struct {
Name string `json:"name" validate:"required"`
Version int `json:"version" validate:"required"`
ISP edge2.ISP `json:"isp"`
Prov string `json:"prov"`
City string `json:"city"`
Name string `json:"name" validate:"required"`
Version int `json:"version" validate:"required"`
}
type RegisterEdgeResp struct {
@@ -28,7 +27,7 @@ type RegisterEdgeResp struct {
Host string `json:"host"`
}
func OnlineEdge(c *fiber.Ctx) (err error) {
func AssignEdge(c *fiber.Ctx) (err error) {
// 验证请求参数
var req = new(RegisterEdgeReq)
@@ -37,36 +36,64 @@ func OnlineEdge(c *fiber.Ctx) (err error) {
return err
}
// 挑选合适的转发服务
var fwd *m.Proxy
fwd, err = q.Proxy.
LeftJoin(q.Edge, q.Edge.ProxyID.EqCol(q.Proxy.ID), q.Edge.Status.Eq(1)).
Select(q.Proxy.ALL, q.Edge.ALL.Count().As("count")).
Where(q.Proxy.Type.Eq(int32(proxy2.TypeSelfHosted))).
Group(q.Proxy.ID).
Order(field.NewField("", "count").Desc()).
First()
if err != nil {
return err
// 全局锁，防止并发注册
var mutex = g.Redsync.NewMutex("edge:discovery")
if err := mutex.Lock(); err != nil {
return errors.New("服务繁忙，请稍后重试")
}
defer func() {
if ok, err := mutex.Unlock(); err != nil {
slog.Error("解锁失败", slog.Bool("ok", ok), slog.Any("err", err))
}
}()
// 保存节点信息
var edge = &m.Edge{
Name: req.Name,
Version: int32(req.Version),
Host: c.Context().RemoteIP().String(),
Isp: int32(req.ISP),
Prov: req.Prov,
City: req.City,
ProxyID: &fwd.ID,
Type: int32(edge2.TypeSelfHosted),
Status: 1,
}
err = q.Edge.Clauses(clause.OnConflict{
UpdateAll: true,
Columns: []clause.Column{{Name: "name"}},
}).Create(edge)
if err != nil {
// 检查节点
var fwd *m.Proxy
var edge *m.Edge
edge, err = q.Edge.
Where(q.Edge.Name.Eq(req.Name)).
Take()
if errors.Is(err, gorm.ErrRecordNotFound) {
// 挑选合适的转发服务
fwd, err = q.Proxy.
LeftJoin(q.Edge, q.Edge.ProxyID.EqCol(q.Proxy.ID), q.Edge.Status.Eq(1)).
Select(q.Proxy.ALL, q.Edge.ALL.Count().As("count")).
Where(q.Proxy.Type.Eq(int32(proxy2.TypeSelfHosted))).
Group(q.Proxy.ID).
Order(field.NewField("", "count").Desc()).
First()
if err != nil {
return err
}
// 保存节点信息
edge = &m.Edge{
Name: req.Name,
Version: int32(req.Version),
Type: int32(edge2.TypeSelfHosted),
ProxyID: &fwd.ID,
}
err = q.Edge.Create(edge)
if err != nil {
return err
}
} else if err == nil {
// 获取已配置的转发服务
fwd, err = q.Proxy.
Where(q.Proxy.ID.Eq(*edge.ProxyID)).
Take()
if err != nil {
return err
}
// 节点已存在，更新版本号
if edge.Version < int32(req.Version) {
_, err = q.Edge.
Where(q.Edge.ID.Eq(edge.ID)).
UpdateSimple(q.Edge.Version.Value(int32(req.Version)))
if err != nil {
return err
}
}
} else {
return err
}
@@ -77,29 +104,6 @@ func OnlineEdge(c *fiber.Ctx) (err error) {
})
}
type OfflineEdgeReq struct {
Name string `json:"name" validate:"required"`
}
func OfflineEdge(c *fiber.Ctx) (err error) {
// 验证请求参数
var req = new(OfflineEdgeReq)
err = g.Validator.Validate(c, req)
if err != nil {
return err
}
// 下线转发服务
_, err = q.Edge.
Where(q.Edge.Name.Eq(req.Name)).
UpdateSimple(q.Edge.Status.Value(0))
if err != nil {
return err
}
return c.SendStatus(fiber.StatusOK)
}
type AllEdgesAvailableReq struct {
s.EdgeFilter
Count int `json:"count"`
@@ -140,7 +144,7 @@ func AllEdgesAvailable(c *fiber.Ctx) (err error) {
edges[i] = AllEdgesAvailableRespItem{
Ip: info.Host,
Port: u.Z(info.ProxyPort),
Isp: edge2.ISPToStr(edge2.ISP(info.Isp)),
Isp: edge2.ISP(info.Isp).String(),
Prov: info.Prov,
City: info.City,
Status: info.Status,

239
web/handlers/proxy.go
View File

@@ -3,11 +3,12 @@ package handlers
import (
"crypto/rand"
"encoding/base32"
"github.com/gofiber/fiber/v2"
"fmt"
"log/slog"
"platform/pkg/u"
auth2 "platform/web/auth"
"platform/web/core"
edge2 "platform/web/domains/edge"
proxy2 "platform/web/domains/proxy"
g "platform/web/globals"
"platform/web/globals/orm"
@@ -16,23 +17,27 @@ import (
"strings"
"time"
"github.com/gofiber/fiber/v2"
"gorm.io/gen/field"
"gorm.io/gorm/clause"
)
// region OnlineProxy
// region 报告上线
type OnlineProxyReq struct {
type ProxyReportOnlineReq struct {
Name string `json:"name" validate:"required"`
Version int `json:"version" validate:"required"`
}
type OnlineProxyResp struct {
Id int32 `json:"id"`
Secret string `json:"secret"`
Permits []ProxyPermit `json:"permits"`
type ProxyReportOnlineResp struct {
Id int32 `json:"id"`
Secret string `json:"secret"`
Permits []*ProxyPermit `json:"permits"`
Edges []*ProxyEdge `json:"edges"`
}
func OnlineProxy(c *fiber.Ctx) (err error) {
func ProxyReportOnline(c *fiber.Ctx) (err error) {
// 检查接口权限
_, err = auth2.NewProtect(c).Payload(
@@ -43,7 +48,7 @@ func OnlineProxy(c *fiber.Ctx) (err error) {
}
// 验证请求参数
var req = new(OnlineProxyReq)
var req = new(ProxyReportOnlineReq)
err = g.Validator.Validate(c, req)
if err != nil {
return err
@@ -81,6 +86,28 @@ func OnlineProxy(c *fiber.Ctx) (err error) {
return err
}
// 获取边缘节点信息
data, err := q.Edge.Where(
q.Edge.ProxyID.Eq(proxy.ID),
).Find()
if err != nil {
return err
}
edges := make([]*ProxyEdge, len(data))
for i, edge := range data {
edges[i] = &ProxyEdge{
Id: edge.ID,
Port: edge.ProxyPort,
Prov: &edge.Prov,
City: &edge.City,
Isp: u.P(edge2.ISP(edge.Isp).String()),
Status: &edge.Status,
Loss: edge.Loss,
Rtt: edge.Rtt,
}
}
// 获取许可配置
channels, err := q.Channel.Where(
q.Channel.ProxyID.Eq(proxy.ID),
q.Channel.Expiration.Gt(orm.LocalDateTime(time.Now())),
@@ -89,38 +116,38 @@ func OnlineProxy(c *fiber.Ctx) (err error) {
return err
}
var permits []ProxyPermit
for _, channel := range channels {
var permits = make([]*ProxyPermit, len(channels))
for i, channel := range channels {
if channel.EdgeID == nil {
return core.NewBizErr("通道未分配边缘节点")
return core.NewBizErr(fmt.Sprintf("权限解析异常，通道缺少边缘节点ID %d", channel.ID))
}
permit := ProxyPermit{
permits[i] = &ProxyPermit{
Id: *channel.EdgeID,
Expire: time.Time(channel.Expiration),
Whitelists: u.P(strings.Split(u.Z(channel.Whitelists), ",")),
Username: channel.Username,
Password: channel.Password,
}
permits = append(permits, permit)
}
slog.Debug("注册转发服务", "ip", ip, "id", proxy.ID)
return c.JSON(&OnlineProxyResp{
return c.JSON(&ProxyReportOnlineResp{
Id: proxy.ID,
Secret: secret,
Edges: edges,
Permits: permits,
})
}
// endregion
// region OfflineProxy
// region 报告下线
type OfflineProxyReq struct {
Name string `json:"name" validate:"required"`
type ProxyReportOfflineReq struct {
Id int32 `json:"id" validate:"required"`
}
func OfflineProxy(c *fiber.Ctx) (err error) {
func ProxyReportOffline(c *fiber.Ctx) (err error) {
// 检查接口权限
_, err = auth2.NewProtect(c).Payload(
auth2.PayloadInternalServer,
@@ -130,7 +157,7 @@ func OfflineProxy(c *fiber.Ctx) (err error) {
}
// 验证请求参数
var req = new(OfflineProxyReq)
var req = new(ProxyReportOfflineReq)
err = g.Validator.Validate(c, req)
if err != nil {
return err
@@ -138,26 +165,33 @@ func OfflineProxy(c *fiber.Ctx) (err error) {
// 下线转发服务
_, err = q.Proxy.
Where(q.Proxy.Name.Eq(req.Name)).
Where(q.Proxy.ID.Eq(req.Id)).
UpdateSimple(q.Proxy.Status.Value(0))
if err != nil {
return err
}
// 下线所有相关的边缘节点
_, err = q.Edge.
Where(q.Edge.ProxyID.Eq(req.Id)).
UpdateSimple(q.Edge.Status.Value(0))
if err != nil {
return err
}
return nil
}
// endregion
// region AssignProxyFwdPort
// region 报告更新
type AssignProxyFwdPortReq struct {
Proxy int32 `json:"proxy" validate:"required"`
Edge int32 `json:"edge" validate:"required"`
Port uint16 `json:"port" validate:"required"`
type ProxyReportUpdateReq struct {
Id int32 `json:"id" validate:"required"`
Edges []*ProxyEdge `json:"edges" validate:"required"`
}
func AssignProxyFwdPort(c *fiber.Ctx) (err error) {
func ProxyReportUpdate(c *fiber.Ctx) (err error) {
// 检查接口权限
_, err = auth2.NewProtect(c).Payload(
auth2.PayloadInternalServer,
@@ -167,19 +201,142 @@ func AssignProxyFwdPort(c *fiber.Ctx) (err error) {
}
// 验证请求参数
var req = new(AssignProxyFwdPortReq)
var req = new(ProxyReportUpdateReq)
err = g.Validator.Validate(c, req)
if err != nil {
return err
}
// 更新边缘节点端口分配状态
_, err = q.Edge.
Where(q.Edge.ID.Eq(req.Edge)).
UpdateSimple(
q.Edge.ProxyID.Value(req.Proxy),
q.Edge.ProxyPort.Value(int32(req.Port)),
)
// 更新节点信息
var idsActive = make([]int32, 0, len(req.Edges))
var idsInactive = make([]int32, 0, len(req.Edges))
var idsIspUnknown = make([]int32, 0, len(req.Edges))
var idsIspTelecom = make([]int32, 0, len(req.Edges))
var idsIspUnicom = make([]int32, 0, len(req.Edges))
var idsIspMobile = make([]int32, 0, len(req.Edges))
var otherEdges = make([]*ProxyEdge, 0, len(req.Edges))
for _, edge := range req.Edges {
// 跳过不方便批量操作的边缘节点
if edge.Port != nil || edge.Prov != nil || edge.City != nil {
otherEdges = append(otherEdges, edge)
continue
}
// 检查更新ISP
if edge.Isp != nil {
switch edge2.ISPFromStr(*edge.Isp) {
case edge2.IspUnknown:
idsIspUnknown = append(idsIspUnknown, edge.Id)
case edge2.IspChinaTelecom:
idsIspTelecom = append(idsIspTelecom, edge.Id)
case edge2.IspChinaUnicom:
idsIspUnicom = append(idsIspUnicom, edge.Id)
case edge2.IspChinaMobile:
idsIspMobile = append(idsIspMobile, edge.Id)
}
}
// 检查更新状态
if edge.Status != nil {
if *edge.Status == 1 {
idsActive = append(idsActive, edge.Id)
} else {
idsInactive = append(idsInactive, edge.Id)
}
}
}
slog.Debug("更新边缘节点信息",
"active", len(idsActive),
"inactive", len(idsInactive),
"isp_unknown", len(idsIspUnknown),
"isp_china_telecom", len(idsIspTelecom),
"isp_china_unicom", len(idsIspUnicom),
"isp_china_mobile", len(idsIspMobile),
"other_edges", len(otherEdges),
)
err = q.Q.Transaction(func(q *q.Query) error {
// 更新边缘节点状态
if len(idsActive) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsActive...)).
UpdateSimple(q.Edge.Status.Value(1))
if err != nil {
return err
}
}
if len(idsInactive) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsInactive...)).
UpdateSimple(q.Edge.Status.Value(0))
if err != nil {
return err
}
}
// 更新边缘节点ISP
if len(idsIspUnknown) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsIspUnknown...)).
UpdateSimple(q.Edge.Isp.Value(int32(edge2.IspUnknown)))
if err != nil {
return err
}
}
if len(idsIspTelecom) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsIspTelecom...)).
UpdateSimple(q.Edge.Isp.Value(int32(edge2.IspChinaTelecom)))
if err != nil {
return err
}
}
if len(idsIspUnicom) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsIspUnicom...)).
UpdateSimple(q.Edge.Isp.Value(int32(edge2.IspChinaUnicom)))
if err != nil {
return err
}
}
if len(idsIspMobile) > 0 {
_, err = q.Edge.Debug().
Where(q.Edge.ID.In(idsIspMobile...)).
UpdateSimple(q.Edge.Isp.Value(int32(edge2.IspChinaMobile)))
if err != nil {
return err
}
}
// 更新其他边缘节点信息
for _, edge := range otherEdges {
do := q.Edge.Debug().Where(q.Edge.ID.Eq(edge.Id))
var assigns = make([]field.AssignExpr, 0, 5)
if edge.Port != nil {
assigns = append(assigns, q.Edge.ProxyPort.Value(*edge.Port))
}
if edge.Prov != nil {
assigns = append(assigns, q.Edge.Prov.Value(*edge.Prov))
}
if edge.City != nil {
assigns = append(assigns, q.Edge.City.Value(*edge.City))
}
// 更新边缘节点
_, err := do.UpdateSimple(assigns...)
if err != nil {
return fmt.Errorf("更新边缘节点 %d 失败: %w", edge.Id, err)
}
}
return nil
})
if err != nil {
return err
}
@@ -196,3 +353,15 @@ type ProxyPermit struct {
Username *string `json:"username"`
Password *string `json:"password"`
}
type ProxyEdge struct {
Id int32 `json:"id"`
Host *string `json:"host,omitempty"` // 边缘节点地址
Port *int32 `json:"port,omitempty"` // 边缘节点代理端口
Prov *string `json:"prov,omitempty"`
City *string `json:"city,omitempty"`
Isp *string `json:"isp,omitempty"`
Status *int32 `json:"status,omitempty"`
Loss *int32 `json:"loss,omitempty"` // 丢包率
Rtt *int32 `json:"latency,omitempty"` // 延迟
}

9
web/router.go
View File

@@ -67,14 +67,13 @@ func ApplyRouters(app *fiber.App) {
// 网关
proxy := api.Group("/proxy")
proxy.Post("/online", handlers.OnlineProxy)
proxy.Post("/offline", handlers.OfflineProxy)
proxy.Post("/assign", handlers.AssignProxyFwdPort)
proxy.Post("/online", handlers.ProxyReportOnline)
proxy.Post("/offline", handlers.ProxyReportOffline)
proxy.Post("/update", handlers.ProxyReportUpdate)
// 节点
edge := api.Group("/edge")
edge.Post("/online", handlers.OnlineEdge)
edge.Post("/offline", handlers.OfflineEdge)
edge.Post("/assign", handlers.AssignEdge)
edge.Post("/all", handlers.AllEdgesAvailable)
// 临时