实现网关监听并读取更新队列，定时发送更新数据到平台

README.md

@@ -1,15 +1,11 @@
 ## TODO
 
-- [x] 节点发送 open 请求后，如果是新节点则访问 geo 接口获取 geo 信息，并连带 host 和 port 信息提交到更新队列
-- [x] open 命令更新 status
-- [x] app 需要提供节点下线更新函数，节点断联后调用下线函数更新 status
-
-网关监听并读取更新队列，定时发送更新数据到平台
-
 节点每次发送心跳后提供或计算 loss 和 rtt 信息，并对比 host 和 status，将需要更新的数据提交到更新队列
 
 ### 长期
 
+实现一个接口以手动全量同步数据库中的节点信息，防止网关节点数据与数据库出现偏差
+
 将协议内容抽离出公共包，gateway 和 edge 节点共同调用
 
 ## 开发相关

cmd/gateway/main.go

@@ -5,8 +5,7 @@ import (
 )
 
 func main() {
-	var app = gateway.New()
+	var err = gateway.Start()
-	var err = app.Run()
 	if err != nil {
 		println(err.Error())
 	}

gateway/app/app.go

@@ -91,7 +91,7 @@ func OfflineEdge(id int32) error {
 		return fmt.Errorf("边缘节点 %d 不存在", id)
 	}
 
-	*edge.Status = core.EdgeOffline
+	edge.Status = &core.EdgeOffline
 	EdgeUpdates <- &core.Edge{
 		Id:     id,
 		Status: &core.EdgeOffline,
@@ -101,12 +101,14 @@ func OfflineEdge(id int32) error {
 }
 
 func StoreEdge(edge *core.Edge) error {
-	if edge == nil || edge.Id == 0 || edge.Port == nil {
+	if edge == nil || edge.Id == 0 {
 		return fmt.Errorf("无效的边缘节点: %+v", edge)
 	}
 
 	Edges.Store(edge.Id, edge)
-	Assigns.Store(*edge.Port, edge.Id)
+	if edge.Port != nil {
+		Assigns.Store(*edge.Port, edge.Id)
+	}
 	return nil
 }
 

gateway/app/geo.go

@@ -1,18 +1,13 @@
 package app
 
 import (
-	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"net"
-	"net/http"
 	g "proxy-server/gateway/globals"
 	"strings"
 )
 
-const base = "https://whois.pconline.com.cn/ipJson.jsp?json=true&ip="
-
 func IpGeo(ip net.IP) (info *IpGeoInfo, err error) {
 	defer func() {
 		var rs = recover()
@@ -26,50 +21,29 @@ func IpGeo(ip net.IP) (info *IpGeoInfo, err error) {
 	if err != nil {
 		return nil, err
 	}
+
 	if str != "" {
 		slog.Debug("本地解析归属地结果", "str", str)
 		values := strings.Split(str, "|")
 		if len(values) != 5 {
 			return nil, fmt.Errorf("本地归属地查询解析失败")
 		}
-		return &IpGeoInfo{
+
-			Prov: values[2],
+		var info = &IpGeoInfo{}
-			City: values[3],
+
-			Isp:  values[4],
+		if values[2] != "0" {
-		}, nil
+			info.Prov = values[2]
+		}
+		if values[3] != "0" {
+			info.City = values[3]
+		}
+		if values[4] != "0" {
+			info.Isp = values[4]
+		}
+		return info, nil
 	}
 
-	// 归属地查询
+	return nil, fmt.Errorf("本地归属地查询为空")
-	var url = base + ip.String()
-	resp, err := http.Get(url)
-	if err != nil {
-		return nil, fmt.Errorf("查询归属地失败: %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("查询归属地失败: %s", resp.Status)
-	}
-	defer func(Body io.ReadCloser) {
-		_ = Body.Close()
-	}(resp.Body)
-
-	// 返回节点信息
-	var bytes []byte
-	_, err = io.ReadFull(resp.Body, bytes)
-	if err != nil {
-		return nil, fmt.Errorf("读取归属地响应失败: %w", err)
-	}
-
-	var body = make(map[string]any)
-	err = json.Unmarshal(bytes, &body)
-	if err != nil {
-		return nil, fmt.Errorf("解析归属地响应失败: %w", err)
-	}
-
-	return &IpGeoInfo{
-		Prov: body["pro"].(string),
-		City: body["city"].(string),
-		Isp:  strings.Split(body["addr"].(string), " ")[1],
-	}, nil
 }
 
 type IpGeoInfo struct {

gateway/app/report.go

@@ -1,4 +1,4 @@
-package report
+package app
 
 import (
 	"encoding/base64"
@@ -7,7 +7,6 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
-	"proxy-server/gateway/app"
 	"proxy-server/gateway/core"
 	"proxy-server/gateway/env"
 	"strings"
@@ -34,13 +33,13 @@ func Online(name string) (err error) {
 		return err
 	}
 
-	app.Id = body.Id
+	Id = body.Id
-	app.PlatformSecret = body.Secret
+	PlatformSecret = body.Secret
 	for _, def := range body.Permits {
-		app.StorePermit(def)
+		StorePermit(def)
 	}
 	for _, edge := range body.Edges {
-		err := app.StoreEdge(edge)
+		err := StoreEdge(edge)
 		if err != nil {
 			slog.Error("存储边缘节点失败", "err", err, "edge", edge)
 		}
@@ -50,11 +49,22 @@ func Online(name string) (err error) {
 
 func Offline() (err error) {
 	_, err = call(env.EndpointOffline, map[string]any{
-		"id": app.Id,
+		"id": Id,
 	})
 	return err
 }
 
+func Update(data []*core.Edge) (err error) {
+	_, err = call(env.EndpointUpdate, map[string]any{
+		"id":    Id,
+		"edges": data,
+	})
+	if err != nil {
+		err = fmt.Errorf("更新节点数据失败：%w", err)
+	}
+	return err
+}
+
 func call(endpoint string, body any) (string, error) {
 	bodyStr, err := json.Marshal(body)
 	if err != nil {

gateway/env/env.go

@@ -35,9 +35,9 @@ var (
 	RedisDb   = 0
 	RedisPass = ""
 
-	EndpointOnline   string
+	EndpointOnline  string
-	EndpointOffline  string
+	EndpointOffline string
-	EndpointAssigned string
+	EndpointUpdate  string
 )
 
 func Init() {
@@ -209,10 +209,11 @@ func Init() {
 	} else {
 		panic("环境变量 ENDPOINT_OFFLINE 未设置")
 	}
-	value = os.Getenv("ENDPOINT_ASSIGNED")
+
+	value = os.Getenv("ENDPOINT_UPDATE")
 	if value != "" {
-		EndpointAssigned = value
+		EndpointUpdate = value
 	} else {
-		panic("环境变量 ENDPOINT_ASSIGNED 未设置")
+		panic("环境变量 ENDPOINT_UPDATE 未设置")
 	}
 }

gateway/gateway.go

@@ -13,7 +13,6 @@ import (
 	"proxy-server/gateway/fwd"
 	g "proxy-server/gateway/globals"
 	"proxy-server/gateway/log"
-	"proxy-server/gateway/report"
 	"proxy-server/gateway/web"
 	"proxy-server/utils"
 	"sync"
@@ -21,28 +20,16 @@ import (
 
 	"github.com/google/uuid"
 
-	"github.com/joho/godotenv"
-
 	_ "net/http/pprof"
 )
 
-type server struct {
+func Start() (err error) {
-}
-
-func New() *server {
-	return &server{}
-}
-
-func (s *server) Run() (err error) {
 
 	// 初始化
-	err = s.init()
+	setup()
-	if err != nil {
-		return err
-	}
 
 	// 恢复服务状态
-	err = s.restore()
+	err = restore()
 	if err != nil {
 		return err
 	}
@@ -59,18 +46,28 @@ func (s *server) Run() (err error) {
 	defer close(fwdQuit)
 	go func() {
 		defer wg.Done()
-		err = s.startFwd(ctx)
+		err = startFwd(ctx)
 		fwdQuit <- err
 	}()
 
 	// 接口服务
 	wg.Add(1)
-	apiQuit := make(chan error, 1)
+	webQuit := make(chan error, 1)
-	defer close(apiQuit)
+	defer close(webQuit)
 	go func() {
 		defer wg.Done()
-		err := s.startWeb(ctx)
+		err := startWeb(ctx)
-		apiQuit <- err
+		webQuit <- err
+	}()
+
+	// 调度任务服务
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := startTask(ctx)
+		if err != nil {
+			slog.Error("调度任务服务异常退出", "err", err)
+		}
 	}()
 
 	// debug
@@ -89,7 +86,7 @@ func (s *server) Run() (err error) {
 
 	// 报告上线
 	slog.Info("报告服务上线")
-	err = report.Online(app.Name)
+	err = app.Online(app.Name)
 	if err != nil {
 		return fmt.Errorf("服务上线失败: %w", err)
 	}
@@ -100,7 +97,7 @@ func (s *server) Run() (err error) {
 		if err != nil {
 			slog.Warn("fwd 服务异常退出", "err", err)
 		}
-	case err := <-apiQuit:
+	case err := <-webQuit:
 		if err != nil {
 			slog.Warn("web 服务异常退出", "err", err)
 		}
@@ -113,7 +110,7 @@ func (s *server) Run() (err error) {
 		defer wg.Done()
 		// 报告下线
 		slog.Debug("报告服务下线")
-		err = report.Offline()
+		err = app.Offline()
 		if err != nil {
 			slog.Error("服务下线失败", "err", err)
 		}
@@ -134,22 +131,14 @@ func (s *server) Run() (err error) {
 	return nil
 }
 
-func (s *server) init() error {
+func setup() {
-
-	err := godotenv.Load()
-	if err != nil {
-		println("没有本地环境变量文件")
-	}
-
 	log.Init()
 	env.Init()
 	g.InitRedis()
 	g.InitGeo()
-
-	return nil
 }
 
-func (s *server) restore() error {
+func restore() error {
 	var file = "proxy.lock"
 
 	bytes, err := os.ReadFile(file)
@@ -178,7 +167,7 @@ func (s *server) restore() error {
 	return nil
 }
 
-func (s *server) startFwd(ctx context.Context) error {
+func startFwd(ctx context.Context) error {
 	server := fwd.New()
 	go func() {
 		<-ctx.Done()
@@ -187,7 +176,7 @@ func (s *server) startFwd(ctx context.Context) error {
 	return server.Run()
 }
 
-func (s *server) startWeb(ctx context.Context) error {
+func startWeb(ctx context.Context) error {
 	server := web.New()
 	go func() {
 		<-ctx.Done()
@@ -198,3 +187,37 @@ func (s *server) startWeb(ctx context.Context) error {
 	}()
 	return server.Run()
 }
+
+func startTask(ctx context.Context) error {
+
+	var lock = sync.Mutex{}
+	var updates = make([]*core.Edge, 0)
+	go func() {
+		for data := range app.EdgeUpdates {
+			lock.Lock()
+			updates = append(updates, data)
+			lock.Unlock()
+		}
+	}()
+
+	// 每 30 秒批量提交一次更新
+	var scheduler = time.Tick(30 * time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-scheduler:
+			if len(updates) == 0 {
+				continue
+			}
+			err := app.Update(updates)
+			lock.Lock()
+			clear(updates)
+			updates = updates[:0]
+			lock.Unlock()
+			if err != nil {
+				slog.Error("调度更新任务失败", "err", err)
+			}
+		}
+	}
+}