lanhu/proxy
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增代理服务与边缘节点注册功能

Browse Source
This commit is contained in:
luorijun
2025-05-13 18:48:17 +08:00
parent 536f36ae02
commit d69a77df38
17 changed files with 573 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
server/pkg/env/env.go vendored
View File

@@ -2,104 +2,139 @@ package env
import (
"fmt"
"log/slog"
"os"
"strconv"
"github.com/joho/godotenv"
)
var (
AppCtrlPort uint16
AppDataPort uint16
AppWebPort uint16
AppLogMode string
AppCtrlPort uint16 = 18080
AppDataPort uint16 = 18081
AppWebPort uint16 = 8848
AppLogMode string = "dev"
ClientId string
ClientSecret string
DbHost string
DbPort uint16
DbPort uint16 = 5432
DbDatabase string
DbUsername string
DbPassword string
DbTimezone string
DbTimezone string = "Asia/Shanghai"
EndpointOnline string
EndpointOffline string
)
func Init() {
// AppCtrlPort
appCtrlPortStr := os.Getenv("APP_CTRL_PORT")
if appCtrlPortStr == "" {
panic("环境变量 APP_CTRL_PORT 未设置")
}
appCtrlPort, err := strconv.ParseUint(appCtrlPortStr, 10, 16)
var err = godotenv.Load()
if err != nil {
panic(fmt.Sprintf("环境变量 APP_CTRL_PORT 格式错误: %v", err))
slog.Debug("没有本地环境变量文件")
}
AppCtrlPort = uint16(appCtrlPort)
var value string
// AppDataPort
appDataPortStr := os.Getenv("APP_DATA_PORT")
if appDataPortStr == "" {
panic("环境变量 APP_DATA_PORT 未设置")
value = os.Getenv("APP_CTRL_PORT")
if value != "" {
appCtrlPort, err := strconv.Atoi(value)
if err != nil {
panic(fmt.Sprintf("环境变量 APP_CTRL_PORT 格式错误: %v", err))
}
AppCtrlPort = uint16(appCtrlPort)
}
appDataPort, err := strconv.ParseUint(appDataPortStr, 10, 16)
if err != nil {
panic(fmt.Sprintf("环境变量 APP_DATA_PORT 格式错误: %v", err))
}
AppDataPort = uint16(appDataPort)
// AppWebPort
appWebPortStr := os.Getenv("APP_WEB_PORT")
if appWebPortStr == "" {
appWebPortStr = "8848"
value = os.Getenv("APP_DATA_PORT")
if value != "" {
appDataPort, err := strconv.Atoi(value)
if err != nil {
panic(fmt.Sprintf("环境变量 APP_DATA_PORT 格式错误: %v", err))
}
AppDataPort = uint16(appDataPort)
}
appWebPort, err := strconv.ParseUint(appWebPortStr, 10, 16)
if err != nil {
panic(fmt.Sprintf("环境变量 APP_WEB_PORT 格式错误: %v", err))
}
AppWebPort = uint16(appWebPort)
// AppLogMode
appLogMode := os.Getenv("APP_LOG_MODE")
if appLogMode == "" {
AppLogMode = "dev"
value = os.Getenv("APP_WEB_PORT")
if value != "" {
appWebPort, err := strconv.Atoi(value)
if err != nil {
panic(fmt.Sprintf("环境变量 APP_WEB_PORT 格式错误: %v", err))
}
AppWebPort = uint16(appWebPort)
}
AppLogMode = appLogMode
// DbHost
DbHost = os.Getenv("DB_HOST")
if DbHost == "" {
value = os.Getenv("APP_LOG_MODE")
if value != "" {
AppLogMode = value
}
value = os.Getenv("CLIENT_ID")
if value != "" {
ClientId = value
} else {
panic("环境变量 CLIENT_ID 未设置")
}
value = os.Getenv("CLIENT_SECRET")
if value != "" {
ClientSecret = value
} else {
panic("环境变量 CLIENT_SECRET 未设置")
}
value = os.Getenv("DB_HOST")
if value != "" {
DbHost = os.Getenv("DB_HOST")
} else {
panic("环境变量 DB_HOST 未设置")
}
// DbPort
dbPortStr := os.Getenv("DB_PORT")
if dbPortStr == "" {
dbPortStr = "5432"
value = os.Getenv("DB_PORT")
if value != "" {
dbPort, err := strconv.Atoi(value)
if err != nil {
panic(fmt.Sprintf("环境变量 DB_PORT 格式错误: %v", err))
}
DbPort = uint16(dbPort)
}
dbPort, err := strconv.ParseUint(dbPortStr, 10, 16)
if err != nil {
panic(fmt.Sprintf("环境变量 DB_PORT 格式错误: %v", err))
}
DbPort = uint16(dbPort)
// DbDatabase
DbDatabase = os.Getenv("DB_DATABASE")
if DbDatabase == "" {
value = os.Getenv("DB_DATABASE")
if value != "" {
DbDatabase = value
} else {
panic("环境变量 DB_DATABASE 未设置")
}
// DbUsername
DbUsername = os.Getenv("DB_USERNAME")
if DbUsername == "" {
value = os.Getenv("DB_USERNAME")
if value != "" {
DbUsername = value
} else {
panic("环境变量 DB_USERNAME 未设置")
}
// DbPassword
DbPassword = os.Getenv("DB_PASSWORD")
if DbPassword == "" {
value = os.Getenv("DB_PASSWORD")
if value != "" {
DbPassword = value
} else {
panic("环境变量 DB_PASSWORD 未设置")
}
// DbTimezone
DbTimezone = os.Getenv("DB_TIMEZONE")
if DbTimezone == "" {
DbTimezone = "Asia/Shanghai"
value = os.Getenv("DB_TIMEZONE")
if value != "" {
DbTimezone = value
}
value = os.Getenv("ENDPOINT_ONLINE")
if value != "" {
EndpointOnline = value
} else {
panic("环境变量 ENDPOINT_ONLINE 未设置")
}
value = os.Getenv("ENDPOINT_OFFLINE")
if value != "" {
EndpointOffline = value
} else {
panic("环境变量 ENDPOINT_OFFLINE 未设置")
}
}

166
server/server.go
View File

@@ -2,6 +2,8 @@ package server
import (
"context"
"encoding/base64"
"encoding/json"
"log/slog"
"net/http"
"os"
@@ -13,43 +15,52 @@ import (
"proxy-server/server/pkg/log"
"proxy-server/server/pkg/orm"
"proxy-server/server/web"
"runtime"
"strings"
"sync"
"syscall"
"time"
"github.com/google/uuid"
"github.com/joho/godotenv"
_ "net/http/pprof"
)
import _ "net/http/pprof"
const (
Version = 1
RestoreMagic = 0x72
)
type Context struct {
context.Context
log *slog.Logger
type server struct {
name string
}
func Start() {
func New() *server {
return &server{}
}
func (s *server) Run() (err error) {
// 初始化
err := godotenv.Load()
err = s.init()
if err != nil {
println("没有本地环境变量文件")
return err
}
log.Init()
env.Init()
orm.Init()
// 恢复服务状态
err = s.restore()
if err != nil {
return err
}
// 退出信号
osQuit := make(chan os.Signal)
signal.Notify(osQuit, os.Interrupt, syscall.SIGTERM)
// 启动服务
slog.Info("启动服务")
// 准备子服务
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
wg := sync.WaitGroup{}
// 转发服务
wg.Add(1)
fwdQuit := make(chan struct{}, 1)
go func() {
@@ -57,19 +68,19 @@ func Start() {
defer close(fwdQuit)
err := startFwdServer(ctx)
if err != nil {
slog.Error("代理服务发生错误", "err", err)
slog.Error("转发服务发生错误", "err", err)
}
fwdQuit <- struct{}{}
}()
// 启动 web 服务
// 接口服务
wg.Add(1)
apiQuit := make(chan struct{}, 1)
go func() {
defer wg.Done()
err := startWebServer(ctx)
if err != nil {
slog.Error("web 服务发生错误", "err", err)
slog.Error("接口服务发生错误", "err", err)
}
apiQuit <- struct{}{}
}()
@@ -80,15 +91,22 @@ func Start() {
}()
// 性能监控
go func() {
runtime.SetBlockProfileRate(1)
err := http.ListenAndServe(":6060", nil)
if err != nil {
slog.Error("性能监控服务发生错误", "err", err)
}
}()
// go func() {
// runtime.SetBlockProfileRate(1)
// err := http.ListenAndServe(":6060", nil)
// if err != nil {
// slog.Error("性能监控服务发生错误", "err", err)
// }
// }()
// 报告上线
slog.Debug("报告服务上线")
go reportOnline(ctx, s.name)
// 等待退出信号
osQuit := make(chan os.Signal, 1)
signal.Notify(osQuit, os.Interrupt, syscall.SIGTERM)
select {
case <-osQuit:
slog.Info("服务主动退出")
@@ -97,6 +115,11 @@ func Start() {
case <-apiQuit:
slog.Warn("web 服务异常退出")
}
// 报告下线
slog.Debug("报告服务下线")
go reportOffline(ctx, s.name)
// 退出其他服务
cancel()
@@ -109,6 +132,51 @@ func Start() {
case <-timeout.Done():
slog.Warn("退出超时，强制退出")
}
return nil
}
func (s *server) restore() (err error) {
var file = "proxy.lock"
bytes, err := os.ReadFile(file)
if err != nil {
return err
}
if len(bytes) == 17 && bytes[0] == RestoreMagic {
s.name = uuid.UUID(bytes[1:]).String()
slog.Info("恢复服务名称", "name", s.name)
} else {
var u = uuid.New()
s.name = u.String()
bytes = make([]byte, 17)
bytes[0] = RestoreMagic
copy(bytes[1:], u[:])
err := os.WriteFile(file, bytes, 0644)
if err != nil {
return err
}
slog.Info("生成服务名称", "name", s.name)
}
return nil
}
func (s *server) init() error {
err := godotenv.Load()
if err != nil {
println("没有本地环境变量文件")
}
log.Init()
env.Init()
orm.Init()
return nil
}
func startFwdServer(ctx context.Context) error {
@@ -124,3 +192,47 @@ func startFwdServer(ctx context.Context) error {
func startWebServer(ctx context.Context) error {
return web.Start(ctx)
}
func reportOnline(ctx context.Context, name string) {
reportRepeat(ctx, env.EndpointOnline, map[string]any{
"name": name,
"version": Version,
})
}
func reportOffline(ctx context.Context, name string) {
reportRepeat(ctx, env.EndpointOffline, map[string]any{
"name": name,
"version": Version,
})
}
func reportRepeat(ctx context.Context, endpoint string, body any) {
bodyStr, err := json.Marshal(body)
if err != nil {
panic(err)
}
for {
req, err := http.NewRequest("POST", endpoint, strings.NewReader(string(bodyStr)))
if err != nil {
panic(err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Basic "+base64.RawURLEncoding.EncodeToString([]byte("proxy:proxy")))
resp, err := http.DefaultClient.Do(req)
if resp != nil && resp.StatusCode == http.StatusOK {
return
}
select {
case <-ctx.Done():
return
default:
}
slog.Warn("服务注册失败，五秒后重试", "err", err)
time.Sleep(5 * time.Second)
}
}