初步实现HTTP3

internal/nodes/http3_manager_plus_test.go

@@ -0,0 +1,46 @@
+// Copyright 2023 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+//go:build plus
+
+package nodes_test
+
+import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
+	"github.com/TeaOSLab/EdgeNode/internal/nodes"
+	"testing"
+	"time"
+)
+
+func TestHTTP3Manager_Update(t *testing.T) {
+	var manager = nodes.NewHTTP3Manager()
+	err := manager.Update(map[int64]*nodeconfigs.HTTP3Policy{
+		1: {
+			IsOn: true,
+			Port: 443,
+		},
+		2: {
+			IsOn: true,
+			Port: 444,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	/**{
+		err = manager.Update(map[int64]*nodeconfigs.HTTP3Policy{
+			1: {
+				IsOn: false,
+				Port: 443,
+			},
+			2: {
+				IsOn: true,
+				Port: 445,
+			},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}**/
+
+	time.Sleep(1 * time.Minute)
+}

internal/nodes/http_request.go

@@ -46,6 +46,7 @@ type HTTPRequest struct {
 	ServerAddr string // 实际启动的服务器监听地址
 	IsHTTP     bool
 	IsHTTPS    bool
+	IsHTTP3    bool
 
 	// 共享参数
 	nodeConfig *nodeconfigs.NodeConfig
@@ -1828,6 +1829,11 @@ func (this *HTTPRequest) processResponseHeaders(responseHeader http.Header, stat
 		this.ReqServer.HTTPS.SSLPolicy.HSTS.Match(this.ReqHost) {
 		responseHeader.Set(this.ReqServer.HTTPS.SSLPolicy.HSTS.HeaderKey(), this.ReqServer.HTTPS.SSLPolicy.HSTS.HeaderValue())
 	}
+
+	// HTTP/3
+	if this.IsHTTPS && !this.IsHTTP3 && this.ReqServer.SupportsHTTP3() {
+		this.processHTTP3Headers(responseHeader)
+	}
 }
 
 // 添加错误信息
@@ -1897,7 +1903,7 @@ func (this *HTTPRequest) canIgnore(err error) bool {
 
 // 检查连接是否已关闭
 func (this *HTTPRequest) isConnClosed() bool {
-	requestConn := this.RawReq.Context().Value(HTTPConnContextKey)
+	var requestConn = this.RawReq.Context().Value(HTTPConnContextKey)
 	if requestConn == nil {
 		return true
 	}

internal/nodes/http_request_http3.go

@@ -0,0 +1,10 @@
+// Copyright 2023 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+//go:build !plus
+
+package nodes
+
+import "net/http"
+
+func (this *HTTPRequest) processHTTP3Headers(respHeader http.Header) {
+	// stub
+}

internal/nodes/listener_http.go

@@ -32,6 +32,7 @@ type HTTPListener struct {
 	addr       string
 	isHTTP     bool
 	isHTTPS    bool
+	isHTTP3    bool
 	httpServer *http.Server
 }
 
@@ -199,6 +200,7 @@ func (this *HTTPListener) ServeHTTP(rawWriter http.ResponseWriter, rawReq *http.
 		ServerAddr: this.addr,
 		IsHTTP:     this.isHTTP,
 		IsHTTPS:    this.isHTTPS,
+		IsHTTP3:    this.isHTTP3,
 
 		nodeConfig: sharedNodeConfig,
 	}

internal/nodes/listener_manager.go

@@ -37,6 +37,8 @@ func init() {
 // ListenerManager 端口监听管理器
 type ListenerManager struct {
 	listenersMap  map[string]*Listener // addr => *Listener
+	http3Listener *HTTPListener
+
 	locker     sync.Mutex
 	lastConfig *nodeconfigs.NodeConfig
 
@@ -73,7 +75,7 @@ func NewListenerManager() *ListenerManager {
 }
 
 // Start 启动监听
-func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
+func (this *ListenerManager) Start(nodeConfig *nodeconfigs.NodeConfig) error {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
@@ -84,12 +86,12 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
 	/**if this.lastConfig != nil && this.lastConfig.Version == node.Version {
 		return nil
 	}**/
-	this.lastConfig = node
+	this.lastConfig = nodeConfig
 
 	// 所有的新地址
-	groupAddrs := []string{}
+	var groupAddrs = []string{}
-	availableServerGroups := node.AvailableGroups()
+	var availableServerGroups = nodeConfig.AvailableGroups()
-	if !node.IsOn {
+	if !nodeConfig.IsOn {
 		availableServerGroups = []*serverconfigs.ServerAddressGroup{}
 	}
 
@@ -98,13 +100,13 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
 	}
 
 	for _, group := range availableServerGroups {
-		addr := group.FullAddr()
+		var addr = group.FullAddr()
 		groupAddrs = append(groupAddrs, addr)
 	}
 
 	// 停掉老的
 	for listenerKey, listener := range this.listenersMap {
-		addr := listener.FullAddr()
+		var addr = listener.FullAddr()
 		if !lists.ContainsString(groupAddrs, addr) {
 			remotelogs.Println("LISTENER_MANAGER", "close '"+addr+"'")
 			_ = listener.Close()
@@ -115,7 +117,7 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
 
 	// 启动新的或修改老的
 	for _, group := range availableServerGroups {
-		addr := group.FullAddr()
+		var addr = group.FullAddr()
 		listener, ok := this.listenersMap[addr]
 		if ok {
 			// 不需要打印reload信息，防止日志数量过多
@@ -129,7 +131,7 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
 				// 放入到重试队列中
 				this.retryListenerMap[addr] = listener
 
-				firstServer := group.FirstServer()
+				var firstServer = group.FirstServer()
 				if firstServer == nil {
 					remotelogs.Error("LISTENER_MANAGER", err.Error())
 				} else {
@@ -167,10 +169,15 @@ func (this *ListenerManager) TotalActiveConnections() int {
 	this.locker.Lock()
 	defer this.locker.Unlock()
 
-	total := 0
+	var total = 0
 	for _, listener := range this.listenersMap {
 		total += listener.listener.CountActiveConnections()
 	}
+
+	if this.http3Listener != nil {
+		total += this.http3Listener.CountActiveConnections()
+	}
+
 	return total
 }
 
@@ -239,6 +246,17 @@ func (this *ListenerManager) addToFirewalld(groupAddrs []string) {
 		return
 	}
 
+	// HTTP/3相关端口
+	var http3Ports = sharedNodeConfig.FindHTTP3Ports()
+	if len(http3Ports) > 0 {
+		for _, port := range http3Ports {
+			var groupAddr = "udp://:" + types.String(port)
+			if !lists.ContainsString(groupAddrs, groupAddr) {
+				groupAddrs = append(groupAddrs, groupAddr)
+			}
+		}
+	}
+
 	// 组合端口号
 	var portStrings = []string{}
 	var udpPorts = []int{}

internal/nodes/node_tasks.go

@@ -82,6 +82,8 @@ func (this *Node) execTask(rpcClient *rpc.RPCClient, task *pb.NodeTask) error {
 		err = this.execUAMPolicyChangedTask(rpcClient)
 	case "httpCCPolicyChanged":
 		err = this.execHTTPCCPolicyChangedTask(rpcClient)
+	case "http3PolicyChanged":
+		err = this.execHTTP3PolicyChangedTask(rpcClient)
 	case "httpPagesPolicyChanged":
 		err = this.execHTTPPagesPolicyChangedTask(rpcClient)
 	case "updatingServers":
@@ -128,15 +130,6 @@ func (this *Node) execNodeVersionChangedTask() error {
 	return nil
 }
 
-// 脚本库变更
-func (this *Node) execScriptsChangedTask() error {
-	err := this.reloadCommonScripts()
-	if err != nil {
-		return errors.New("reload common scripts failed: " + err.Error())
-	}
-	return nil
-}
-
 // 节点级别变更
 func (this *Node) execNodeLevelChangedTask(rpcClient *rpc.RPCClient) error {
 	levelInfoResp, err := rpcClient.NodeRPC.FindNodeLevelInfo(rpcClient.Context(), &pb.FindNodeLevelInfoRequest{})
@@ -163,90 +156,6 @@ func (this *Node) execNodeLevelChangedTask(rpcClient *rpc.RPCClient) error {
 	return nil
 }
 
-// UAM策略变更
-func (this *Node) execUAMPolicyChangedTask(rpcClient *rpc.RPCClient) error {
-	remotelogs.Println("NODE", "updating uam policies ...")
-	resp, err := rpcClient.NodeRPC.FindNodeUAMPolicies(rpcClient.Context(), &pb.FindNodeUAMPoliciesRequest{})
-	if err != nil {
-		return err
-	}
-	var uamPolicyMap = map[int64]*nodeconfigs.UAMPolicy{}
-	for _, policy := range resp.UamPolicies {
-		if len(policy.UamPolicyJSON) > 0 {
-			var uamPolicy = &nodeconfigs.UAMPolicy{}
-			err = json.Unmarshal(policy.UamPolicyJSON, uamPolicy)
-			if err != nil {
-				remotelogs.Error("NODE", "decode uam policy failed: "+err.Error())
-				continue
-			}
-			err = uamPolicy.Init()
-			if err != nil {
-				remotelogs.Error("NODE", "initialize uam policy failed: "+err.Error())
-				continue
-			}
-			uamPolicyMap[policy.NodeClusterId] = uamPolicy
-		}
-	}
-	sharedNodeConfig.UpdateUAMPolicies(uamPolicyMap)
-	return nil
-}
-
-// HTTP CC策略变更
-func (this *Node) execHTTPCCPolicyChangedTask(rpcClient *rpc.RPCClient) error {
-	remotelogs.Println("NODE", "updating http cc policies ...")
-	resp, err := rpcClient.NodeRPC.FindNodeHTTPCCPolicies(rpcClient.Context(), &pb.FindNodeHTTPCCPoliciesRequest{})
-	if err != nil {
-		return err
-	}
-	var httpCCPolicyMap = map[int64]*nodeconfigs.HTTPCCPolicy{}
-	for _, policy := range resp.HttpCCPolicies {
-		if len(policy.HttpCCPolicyJSON) > 0 {
-			var httpCCPolicy = nodeconfigs.NewHTTPCCPolicy()
-			err = json.Unmarshal(policy.HttpCCPolicyJSON, httpCCPolicy)
-			if err != nil {
-				remotelogs.Error("NODE", "decode http cc policy failed: "+err.Error())
-				continue
-			}
-			err = httpCCPolicy.Init()
-			if err != nil {
-				remotelogs.Error("NODE", "initialize http cc policy failed: "+err.Error())
-				continue
-			}
-			httpCCPolicyMap[policy.NodeClusterId] = httpCCPolicy
-		}
-	}
-	sharedNodeConfig.UpdateHTTPCCPolicies(httpCCPolicyMap)
-	return nil
-}
-
-// 自定义页面策略变更
-func (this *Node) execHTTPPagesPolicyChangedTask(rpcClient *rpc.RPCClient) error {
-	remotelogs.Println("NODE", "updating http pages policies ...")
-	resp, err := rpcClient.NodeRPC.FindNodeHTTPPagesPolicies(rpcClient.Context(), &pb.FindNodeHTTPPagesPoliciesRequest{})
-	if err != nil {
-		return err
-	}
-	var httpPagesPolicyMap = map[int64]*nodeconfigs.HTTPPagesPolicy{}
-	for _, policy := range resp.HttpPagesPolicies {
-		if len(policy.HttpPagesPolicyJSON) > 0 {
-			var httpPagesPolicy = nodeconfigs.NewHTTPPagesPolicy()
-			err = json.Unmarshal(policy.HttpPagesPolicyJSON, httpPagesPolicy)
-			if err != nil {
-				remotelogs.Error("NODE", "decode http pages policy failed: "+err.Error())
-				continue
-			}
-			err = httpPagesPolicy.Init()
-			if err != nil {
-				remotelogs.Error("NODE", "initialize http pages policy failed: "+err.Error())
-				continue
-			}
-			httpPagesPolicyMap[policy.NodeClusterId] = httpPagesPolicy
-		}
-	}
-	sharedNodeConfig.UpdateHTTPPagesPolicies(httpPagesPolicyMap)
-	return nil
-}
-
 // DDoS配置变更
 func (this *Node) execDDoSProtectionChangedTask(rpcClient *rpc.RPCClient) error {
 	resp, err := rpcClient.NodeRPC.FindNodeDDoSProtection(rpcClient.Context(), &pb.FindNodeDDoSProtectionRequest{})

internal/nodes/node_tasks_ext.go

@@ -0,0 +1,31 @@
+// Copyright 2023 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+//go:build !plus
+
+package nodes
+
+import "github.com/TeaOSLab/EdgeNode/internal/rpc"
+
+func (this *Node) execScriptsChangedTask() error {
+	// stub
+	return nil
+}
+
+func (this *Node) execUAMPolicyChangedTask(rpcClient *rpc.RPCClient) error {
+	// stub
+	return nil
+}
+
+func (this *Node) execHTTPCCPolicyChangedTask(rpcClient *rpc.RPCClient) error {
+	// stub
+	return nil
+}
+
+func (this *Node) execHTTP3PolicyChangedTask(rpcClient *rpc.RPCClient) error {
+	// stub
+	return nil
+}
+
+func (this *Node) execHTTPPagesPolicyChangedTask(rpcClient *rpc.RPCClient) error {
+	// stub
+	return nil
+}