统计数据上传时如果遇到invalid utf-8，则自动过滤非法字符/统计数据上传失败时，仍然丢弃已有的统计数据，防止数据堆积

internal/nodes/http_access_log_queue.go

@@ -6,6 +6,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/goman"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"strings"
 	"time"
 )
@@ -108,37 +109,33 @@ Loop:
 }
 
 func (this *HTTPAccessLogQueue) toValidUTF8(accessLog *pb.HTTPAccessLog) {
-	accessLog.RemoteUser = this.toValidUTF8string(accessLog.RemoteUser)
-	accessLog.RequestURI = this.toValidUTF8string(accessLog.RequestURI)
-	accessLog.RequestPath = this.toValidUTF8string(accessLog.RequestPath)
-	accessLog.RequestFilename = this.toValidUTF8string(accessLog.RequestFilename)
+	accessLog.RemoteUser = utils.ToValidUTF8string(accessLog.RemoteUser)
+	accessLog.RequestURI = utils.ToValidUTF8string(accessLog.RequestURI)
+	accessLog.RequestPath = utils.ToValidUTF8string(accessLog.RequestPath)
+	accessLog.RequestFilename = utils.ToValidUTF8string(accessLog.RequestFilename)
 	accessLog.RequestBody = bytes.ToValidUTF8(accessLog.RequestBody, []byte{})
 
 	for _, v := range accessLog.SentHeader {
 		for index, s := range v.Values {
-			v.Values[index] = this.toValidUTF8string(s)
+			v.Values[index] = utils.ToValidUTF8string(s)
 		}
 	}
 
-	accessLog.Referer = this.toValidUTF8string(accessLog.Referer)
-	accessLog.UserAgent = this.toValidUTF8string(accessLog.UserAgent)
-	accessLog.Request = this.toValidUTF8string(accessLog.Request)
-	accessLog.ContentType = this.toValidUTF8string(accessLog.ContentType)
+	accessLog.Referer = utils.ToValidUTF8string(accessLog.Referer)
+	accessLog.UserAgent = utils.ToValidUTF8string(accessLog.UserAgent)
+	accessLog.Request = utils.ToValidUTF8string(accessLog.Request)
+	accessLog.ContentType = utils.ToValidUTF8string(accessLog.ContentType)
 
 	for k, c := range accessLog.Cookie {
-		accessLog.Cookie[k] = this.toValidUTF8string(c)
+		accessLog.Cookie[k] = utils.ToValidUTF8string(c)
 	}
 
-	accessLog.Args = this.toValidUTF8string(accessLog.Args)
-	accessLog.QueryString = this.toValidUTF8string(accessLog.QueryString)
+	accessLog.Args = utils.ToValidUTF8string(accessLog.Args)
+	accessLog.QueryString = utils.ToValidUTF8string(accessLog.QueryString)
 
 	for _, v := range accessLog.Header {
 		for index, s := range v.Values {
-			v.Values[index] = this.toValidUTF8string(s)
+			v.Values[index] = utils.ToValidUTF8string(s)
 		}
 	}
 }
-
-func (this *HTTPAccessLogQueue) toValidUTF8string(v string) string {
-	return strings.ToValidUTF8(v, "")
-}

internal/stats/http_request_stat_manager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
 	"github.com/TeaOSLab/EdgeNode/internal/trackers"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/TeaOSLab/EdgeNode/internal/waf"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/maps"
@@ -319,6 +320,35 @@ func (this *HTTPRequestStatManager) Upload() error {
 		})
 	}
 
+	// 重置数据
+	// 这里需要放到上传数据之前，防止因上传失败而导致统计数据堆积
+	this.cityMap = map[string]*StatItem{}
+	this.providerMap = map[string]int64{}
+	this.systemMap = map[string]int64{}
+	this.browserMap = map[string]int64{}
+	this.dailyFirewallRuleGroupMap = map[string]int64{}
+
+	// 上传数据
+	_, err = rpcClient.ServerRPC().UploadServerHTTPRequestStat(rpcClient.Context(), &pb.UploadServerHTTPRequestStatRequest{
+		Month:                  timeutil.Format("Ym"),
+		Day:                    timeutil.Format("Ymd"),
+		RegionCities:           pbCities,
+		RegionProviders:        pbProviders,
+		Systems:                pbSystems,
+		Browsers:               pbBrowsers,
+		HttpFirewallRuleGroups: pbFirewallRuleGroups,
+	})
+	if err != nil {
+		// 是否包含了invalid UTF-8
+		if strings.Contains(err.Error(), "string field contains invalid UTF-8") {
+			for _, system := range pbSystems {
+				system.Name = utils.ToValidUTF8string(system.Name)
+			}
+			for _, browser := range pbBrowsers {
+				browser.Name = utils.ToValidUTF8string(browser.Name)
+			}
+
+			// 再次尝试
 			_, err = rpcClient.ServerRPC().UploadServerHTTPRequestStat(rpcClient.Context(), &pb.UploadServerHTTPRequestStatRequest{
 				Month:                  timeutil.Format("Ym"),
 				Day:                    timeutil.Format("Ymd"),
@@ -331,12 +361,8 @@ func (this *HTTPRequestStatManager) Upload() error {
 			if err != nil {
 				return err
 			}
+		}
+	}
 
-	// 重置数据
-	this.cityMap = map[string]*StatItem{}
-	this.providerMap = map[string]int64{}
-	this.systemMap = map[string]int64{}
-	this.browserMap = map[string]int64{}
-	this.dailyFirewallRuleGroupMap = map[string]int64{}
 	return nil
 }

internal/utils/string.go

@@ -5,17 +5,17 @@ import (
 	"unsafe"
 )
 
-// convert bytes to string
+// UnsafeBytesToString convert bytes to string
 func UnsafeBytesToString(bs []byte) string {
 	return *(*string)(unsafe.Pointer(&bs))
 }
 
-// convert string to bytes
+// UnsafeStringToBytes convert string to bytes
 func UnsafeStringToBytes(s string) []byte {
 	return *(*[]byte)(unsafe.Pointer(&s))
 }
 
-// format address
+// FormatAddress format address
 func FormatAddress(addr string) string {
 	if strings.HasSuffix(addr, "unix:") {
 		return addr
@@ -27,7 +27,7 @@ func FormatAddress(addr string) string {
 	return addr
 }
 
-// format address list
+// FormatAddressList format address list
 func FormatAddressList(addrList []string) []string {
 	result := []string{}
 	for _, addr := range addrList {
@@ -35,3 +35,7 @@ func FormatAddressList(addrList []string) []string {
 	}
 	return result
 }
+
+func ToValidUTF8string(v string) string {
+	return strings.ToValidUTF8(v, "")
+}