实现节点看板（仅对企业版开放）

internal/monitor/value_queue.go

@@ -72,7 +72,8 @@ func (this *ValueQueue) Loop() error {
 			CreatedAt: value.CreatedAt,
 		})
 		if err != nil {
-			return err
+			remotelogs.Error("MONITOR", err.Error())
+			continue
 		}
 	}
 	return nil

internal/stats/http_request_stat_manager.go

@@ -1,12 +1,15 @@
 package stats
 
 import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeNode/internal/events"
 	"github.com/TeaOSLab/EdgeNode/internal/iplibrary"
+	"github.com/TeaOSLab/EdgeNode/internal/monitor"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
 	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/maps"
 	"github.com/iwind/TeaGo/types"
 	timeutil "github.com/iwind/TeaGo/utils/time"
 	"github.com/mssola/user_agent"
@@ -30,6 +33,8 @@ type HTTPRequestStatManager struct {
 	browserMap  map[string]int64 // serverId@browser@version => count
 
 	dailyFirewallRuleGroupMap map[string]int64 // serverId@firewallRuleGroupId@action => count
+
+	totalAttackRequests int64
 }
 
 // NewHTTPRequestStatManager 获取新对象
@@ -48,6 +53,19 @@ func NewHTTPRequestStatManager() *HTTPRequestStatManager {
 
 // Start 启动
 func (this *HTTPRequestStatManager) Start() {
+	// 上传请求总数
+	go func() {
+		ticker := time.NewTicker(1 * time.Minute)
+		go func() {
+			for range ticker.C {
+				if this.totalAttackRequests > 0 {
+					monitor.SharedValueQueue.Add(nodeconfigs.NodeValueItemAttackRequests, maps.Map{"total": this.totalAttackRequests})
+					this.totalAttackRequests = 0
+				}
+			}
+		}()
+	}()
+
 	loopTicker := time.NewTicker(1 * time.Second)
 	uploadTicker := time.NewTicker(30 * time.Minute)
 	if Tea.IsTesting() {
@@ -118,6 +136,9 @@ func (this *HTTPRequestStatManager) AddFirewallRuleGroupId(serverId int64, firew
 	if firewallRuleGroupId <= 0 {
 		return
 	}
+
+	this.totalAttackRequests ++
+
 	select {
 	case this.firewallRuleGroupChan <- strconv.FormatInt(serverId, 10) + "@" + strconv.FormatInt(firewallRuleGroupId, 10) + "@" + action:
 	default:

internal/stats/traffic_stat_manager.go

@@ -4,10 +4,12 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeNode/internal/events"
+	"github.com/TeaOSLab/EdgeNode/internal/monitor"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/maps"
 	"github.com/iwind/TeaGo/types"
 	"strconv"
 	"strings"
@@ -30,6 +32,8 @@ type TrafficStatManager struct {
 	domainsMap map[string]*TrafficItem // timestamp @ serverId @ domain => *TrafficItem
 	locker     sync.Mutex
 	configFunc func() *nodeconfigs.NodeConfig
+
+	totalRequests int64
 }
 
 // NewTrafficStatManager 获取新对象
@@ -46,6 +50,20 @@ func NewTrafficStatManager() *TrafficStatManager {
 func (this *TrafficStatManager) Start(configFunc func() *nodeconfigs.NodeConfig) {
 	this.configFunc = configFunc
 
+	// 上传请求总数
+	go func() {
+		ticker := time.NewTicker(1 * time.Minute)
+		go func() {
+			for range ticker.C {
+				if this.totalRequests > 0 {
+					monitor.SharedValueQueue.Add(nodeconfigs.NodeValueItemRequests, maps.Map{"total": this.totalRequests})
+					this.totalRequests = 0
+				}
+			}
+		}()
+	}()
+
+	// 上传统计数据
 	duration := 5 * time.Minute
 	if Tea.IsTesting() {
 		// 测试环境缩短上传时间，方便我们调试
@@ -71,6 +89,8 @@ func (this *TrafficStatManager) Add(serverId int64, domain string, bytes int64,
 		return
 	}
 
+	this.totalRequests++
+
 	timestamp := utils.UnixTime() / 300 * 300
 
 	key := strconv.FormatInt(timestamp, 10) + strconv.FormatInt(serverId, 10)

internal/stats/traffic_stat_manager_test.go

@@ -8,7 +8,7 @@ import (
 func TestTrafficStatManager_Add(t *testing.T) {
 	manager := NewTrafficStatManager()
 	for i := 0; i < 100; i++ {
-		manager.Add(1, 10, 1, 0)
+		manager.Add(1, "goedge.cn", 1, 0, 0, 0)
 	}
 	t.Log(manager.itemMap)
 }
@@ -16,7 +16,7 @@ func TestTrafficStatManager_Add(t *testing.T) {
 func TestTrafficStatManager_Upload(t *testing.T) {
 	manager := NewTrafficStatManager()
 	for i := 0; i < 100; i++ {
-		manager.Add(1, 10, 1, 0)
+		manager.Add(1, "goedge.cn", 1, 0, 0, 0)
 	}
 	err := manager.Upload()
 	if err != nil {
@@ -30,6 +30,6 @@ func BenchmarkTrafficStatManager_Add(b *testing.B) {
 
 	manager := NewTrafficStatManager()
 	for i := 0; i < b.N; i++ {
-		manager.Add(1, 1024, 1, 0)
+		manager.Add(1, "goedge.cn", 1024, 1, 0, 0)
 	}
 }