tcjlj/flvx
1
0
Fork 0
mirror of https://github.com/Sagit-chu/flvx.git synced 2026-10-07 10:16:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: apply proxy protocol and max connection settings

Browse Source
This commit is contained in:
sagit
2026-04-27 10:54:25 +08:00
committed by GitHub
parent 58d2e89147
commit 2ca3849917
11 changed files with 478 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go-backend/internal/http/handler/control_plane.go
+6 -6
View File
@@ -1706,10 +1706,11 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
service["climiter"] = cLimiterName
}
if forward.ProxyProtocol > 0 {
if service["metadata"] == nil {
service["metadata"] = map[string]interface{}{}
handlerConfig := service["handler"].(map[string]interface{})
if handlerConfig["metadata"] == nil {
handlerConfig["metadata"] = map[string]interface{}{}
}
service["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol
handlerConfig["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol
}
if protocol == "udp" {
listenerMetadata := map[string]interface{}{
@@ -1831,12 +1832,12 @@ func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int)
func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxConn int) error {
limitStr := fmt.Sprintf("$ %d", maxConn)
payload := map[string]interface{}{
"name": limiterName,
"limits": []string{limitStr},
}
if _, err := h.sendNodeCommand(nodeID, "AddCLimiters", payload, false, false); err != nil {
if !isAlreadyExistsMessage(err.Error()) {
return fmt.Errorf("连接限制器下发失败: %w", err)
@@ -1852,7 +1853,6 @@ func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxC
return nil
}
func buildLimiterAddPayload(limiterID int64, speed int) (string, map[string]interface{}) {
rate := float64(speed) / 8.0
limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate)
go-backend/internal/http/handler/forward_proxy_protocol_test.go
+18 -6
View File
@@ -8,7 +8,7 @@ import (
"go-backend/internal/store/repo"
)
func TestBuildForwardServiceConfigsPreservesProxyProtocolWithInterfaceMetadata(t *testing.T) {
func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing.T) {
forward := &forwardRecord{
ID: 1,
UserID: 2,
@@ -30,15 +30,27 @@ func TestBuildForwardServiceConfigsPreservesProxyProtocolWithInterfaceMetadata(t
}
for _, service := range services {
metadata, ok := service["metadata"].(map[string]interface{})
serviceMetadata, ok := service["metadata"].(map[string]interface{})
if !ok {
t.Fatalf("expected metadata map, got %T", service["metadata"])
}
if metadata["interface"] != "eth0" {
t.Fatalf("expected interface metadata eth0, got %v", metadata["interface"])
if serviceMetadata["interface"] != "eth0" {
t.Fatalf("expected interface metadata eth0, got %v", serviceMetadata["interface"])
}
if metadata["proxyProtocol"] != 2 {
t.Fatalf("expected proxyProtocol 2, got %v", metadata["proxyProtocol"])
if _, ok := serviceMetadata["proxyProtocol"]; ok {
t.Fatalf("proxyProtocol should not be listener metadata: %v", serviceMetadata)
}
handlerConfig, ok := service["handler"].(map[string]interface{})
if !ok {
t.Fatalf("expected handler config map, got %T", service["handler"])
}
handlerMetadata, ok := handlerConfig["metadata"].(map[string]interface{})
if !ok {
t.Fatalf("expected handler metadata map, got %T", handlerConfig["metadata"])
}
if handlerMetadata["proxyProtocol"] != 2 {
t.Fatalf("expected handler proxyProtocol 2, got %v", handlerMetadata["proxyProtocol"])
}
}
}
go-backend/internal/http/handler/mutations.go
+40
View File
@@ -138,6 +138,15 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) {
response.WriteJSON(w, response.ErrDefault("请不要作死"))
return
}
oldUser, err := h.repo.GetUserByID(id)
if err != nil {
response.WriteJSON(w, response.Err(-2, err.Error()))
return
}
if oldUser == nil {
response.WriteJSON(w, response.ErrDefault("用户不存在"))
return
}
dup, err := h.repo.UserExistsExcluding(username, id)
if err != nil {
@@ -210,6 +219,17 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) {
}
}
}
if oldUser.MaxConn != maxConn {
warnings, syncErr := h.syncUserMaxConnForwards(id)
if syncErr != nil {
response.WriteJSON(w, response.ErrDefault(fmt.Sprintf("最大连接数下发失败: %v", syncErr)))
return
}
if len(warnings) > 0 {
response.WriteJSON(w, response.OK(map[string]interface{}{"warnings": warnings}))
return
}
}
response.WriteJSON(w, response.OKEmpty())
}
@@ -4254,6 +4274,26 @@ func (h *Handler) syncUserTunnelForwards(userID, tunnelID int64) error {
return nil
}
func (h *Handler) syncUserMaxConnForwards(userID int64) ([]string, error) {
forwards, err := h.listActiveForwardsByUser(userID)
if err != nil {
return nil, err
}
warnings := make([]string, 0)
for i := range forwards {
f := &forwards[i]
if f.MaxConn > 0 {
continue
}
syncWarnings, syncErr := h.syncForwardServicesWithWarnings(f, "UpdateService", true)
warnings = append(warnings, syncWarnings...)
if syncErr != nil {
return warnings, syncErr
}
}
return warnings, nil
}
// cleanupForwardsForUserTunnel deletes all forwarding rules belonging to a
// specific user+tunnel pair. It notifies nodes to remove the runtime services
// first, then deletes the DB records. This is best-effort: individual failures
go-backend/internal/store/repo/repository_control.go
+11 -12
View File
@@ -45,15 +45,17 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards {
rows = append(rows, model.ForwardRecord{
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
MaxConn: f.MaxConn,
ProxyProtocol: f.ProxyProtocol,
})
}
for i := range rows {
@@ -64,7 +66,6 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode
return rows, nil
}
func (r *Repository) ListActiveTunnelIDsByNode(nodeID int64) ([]int64, error) {
if r == nil || r.db == nil {
return nil, errors.New("repository not initialized")
@@ -142,7 +143,6 @@ func (r *Repository) ListForwardPortsTx(tx *gorm.DB, forwardID int64) ([]model.F
return rows, nil
}
func (r *Repository) HasOtherForwardOnNodePort(nodeID int64, port int, currentForwardID int64) (bool, error) {
if r == nil || r.db == nil {
return false, errors.New("repository not initialized")
@@ -169,7 +169,6 @@ func (r *Repository) HasOtherForwardOnNodePortTx(tx *gorm.DB, nodeID int64, port
return count > 0, nil
}
func (r *Repository) GetTunnelOutProtocol(tunnelID int64) (string, error) {
if r == nil || r.db == nil {
return "", errors.New("repository not initialized")
go-backend/internal/store/repo/repository_flow.go
+33 -27
View File
@@ -114,15 +114,17 @@ func (r *Repository) ListActiveForwardsByUser(userID int64) ([]model.ForwardReco
rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards {
rows = append(rows, model.ForwardRecord{
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
MaxConn: f.MaxConn,
ProxyProtocol: f.ProxyProtocol,
})
}
for i := range rows {
@@ -145,15 +147,17 @@ func (r *Repository) ListActiveForwardsByUserTunnel(userID, tunnelID int64) ([]m
rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards {
rows = append(rows, model.ForwardRecord{
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
MaxConn: f.MaxConn,
ProxyProtocol: f.ProxyProtocol,
})
}
for i := range rows {
@@ -176,15 +180,17 @@ func (r *Repository) ListForwardsByUserAndTunnel(userID, tunnelID int64) ([]mode
rows := make([]model.ForwardRecord, 0, len(forwards))
for _, f := range forwards {
rows = append(rows, model.ForwardRecord{
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
ID: f.ID,
UserID: f.UserID,
UserName: f.UserName,
Name: f.Name,
TunnelID: f.TunnelID,
RemoteAddr: f.RemoteAddr,
Strategy: f.Strategy,
Status: f.Status,
SpeedID: f.SpeedID,
MaxConn: f.MaxConn,
ProxyProtocol: f.ProxyProtocol,
})
}
for i := range rows {
go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go
+105
View File
@@ -46,6 +46,111 @@ func TestGetForwardRecordIncludesProxyProtocol(t *testing.T) {
}
}
func TestListForwardsByTunnelIncludesProxyProtocol(t *testing.T) {
r, err := Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
if err := r.DB().Create(&model.Forward{
UserID: 1,
UserName: "admin",
Name: "proxy-forward",
TunnelID: 7,
RemoteAddr: "1.1.1.1:443",
Strategy: "fifo",
CreatedTime: now,
UpdatedTime: now,
Status: 1,
ProxyProtocol: 2,
}).Error; err != nil {
t.Fatalf("create forward: %v", err)
}
records, err := r.ListForwardsByTunnel(7)
if err != nil {
t.Fatalf("ListForwardsByTunnel: %v", err)
}
if len(records) != 1 {
t.Fatalf("expected 1 forward record, got %d", len(records))
}
if records[0].ProxyProtocol != 2 {
t.Fatalf("expected proxyProtocol 2, got %d", records[0].ProxyProtocol)
}
}
func TestListForwardsByTunnelIncludesMaxConn(t *testing.T) {
r, err := Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
if err := r.DB().Create(&model.Forward{
UserID: 1,
UserName: "admin",
Name: "max-conn-forward",
TunnelID: 8,
RemoteAddr: "1.1.1.1:443",
Strategy: "fifo",
CreatedTime: now,
UpdatedTime: now,
Status: 1,
MaxConn: 42,
}).Error; err != nil {
t.Fatalf("create forward: %v", err)
}
records, err := r.ListForwardsByTunnel(8)
if err != nil {
t.Fatalf("ListForwardsByTunnel: %v", err)
}
if len(records) != 1 {
t.Fatalf("expected 1 forward record, got %d", len(records))
}
if records[0].MaxConn != 42 {
t.Fatalf("expected maxConn 42, got %d", records[0].MaxConn)
}
}
func TestListActiveForwardsByUserTunnelIncludesMaxConn(t *testing.T) {
r, err := Open(":memory:")
if err != nil {
t.Fatalf("open repo: %v", err)
}
defer r.Close()
now := time.Now().UnixMilli()
if err := r.DB().Create(&model.Forward{
UserID: 2,
UserName: "user",
Name: "active-max-conn-forward",
TunnelID: 9,
RemoteAddr: "1.1.1.1:443",
Strategy: "fifo",
CreatedTime: now,
UpdatedTime: now,
Status: 1,
MaxConn: 55,
}).Error; err != nil {
t.Fatalf("create forward: %v", err)
}
records, err := r.ListActiveForwardsByUserTunnel(2, 9)
if err != nil {
t.Fatalf("ListActiveForwardsByUserTunnel: %v", err)
}
if len(records) != 1 {
t.Fatalf("expected 1 forward record, got %d", len(records))
}
if records[0].MaxConn != 55 {
t.Fatalf("expected maxConn 55, got %d", records[0].MaxConn)
}
}
func mustRepoLastInsertID(t *testing.T, r *Repository) int64 {
t.Helper()
var id int64