指标数据队列增加最大数量限制，防止过载

.gitignore

@@ -0,0 +1,2 @@
+*_plus.go
+*-plus.sh

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/TeaOSLab/EdgeCommon v0.0.0-00010101000000-000000000000
 	github.com/cespare/xxhash v1.1.0
 	github.com/dchest/captcha v0.0.0-20200903113550-03f5f0333e1f
+	github.com/dop251/goja v0.0.0-20210804101310-32956a348b49
 	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/go-yaml/yaml v2.1.0+incompatible
 	github.com/golang/protobuf v1.5.2

go.sum

@@ -16,12 +16,18 @@ github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dchest/captcha v0.0.0-20200903113550-03f5f0333e1f h1:q/DpyjJjZs94bziQ7YkBmIlpqbVP7yw179rnzoNVX1M=
 github.com/dchest/captcha v0.0.0-20200903113550-03f5f0333e1f/go.mod h1:QGrK8vMWWHQYQ3QU9bw9Y9OPNfxccGzfb41qjvVeXtY=
 github.com/dgryski/go-rendezvous v0.0.0-20200624174652-8d2f3be8b2d9/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91 h1:Izz0+t1Z5nI16/II7vuEo/nHjodOg0p7+OiDpjX5t1E=
+github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
+github.com/dop251/goja v0.0.0-20210804101310-32956a348b49 h1:CtSi0QlA2Hy+nOh8JAZoiEBLW5pliAiKJ3l1Iq1472I=
+github.com/dop251/goja v0.0.0-20210804101310-32956a348b49/go.mod h1:R9ET47fwRVRPZnOGvHxxhuZcbrMCuiqOz3Rlrh4KSnk=
+github.com/dop251/goja_nodejs v0.0.0-20210225215109-d91c329300e7/go.mod h1:hn7BA7c8pLvoGndExHudxTDKZ84Pyvv+90pbBjbTz0Y=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -33,6 +39,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2
 github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI=
 github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM=
 github.com/go-redis/redis/v8 v8.0.0-beta.7/go.mod h1:FGJAWDWFht1sQ4qxyJHZZbVyvnVcKQN0E3u5/5lRz+g=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
+github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gGcHOs=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-yaml/yaml v2.1.0+incompatible h1:RYi2hDdss1u4YE7GwixGzWwVo47T8UQwnTLB6vQiq+o=
@@ -70,9 +78,12 @@ github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3 h1:aBSonas7vFcgTj9u96
 github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3/go.mod h1:H5Q7SXwbx3a97ecJkaS2sD77gspzE7HFUafBO0peEyA=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lionsoul2014/ip2region v2.2.0-release+incompatible h1:1qp9iks+69h7IGLazAplzS9Ca14HAxuD5c0rbFdPGy4=
 github.com/lionsoul2014/ip2region v2.2.0-release+incompatible/go.mod h1:+ZBN7PBoh5gG6/y0ZQ85vJDBe21WnfbRrQQwTfliJJI=
 github.com/mattn/go-sqlite3 v2.0.3+incompatible h1:gXHsfypPkaMZrKbD5209QV9jbUTJKjyR5WD3HYQSd+U=
@@ -81,7 +92,6 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJ
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/mssola/user_agent v0.5.2 h1:CZkTUahjL1+OcZ5zv3kZr8QiJ8jy2H08vZIEkBeRbxo=
 github.com/mssola/user_agent v0.5.2/go.mod h1:TTPno8LPY3wAIEKRpAtkdMT0f8SE24pLRGPahjCH4uw=
-github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -134,6 +144,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210614182718-04defd469f4e h1:XpT3nA5TvE525Ne3hInMh6+GETgn27Zfm9dxsThnX2Q=
 golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
@@ -155,6 +166,7 @@ golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210303074136-134d130e1a04/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210316164454-77fc1eacc6aa/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -212,8 +224,9 @@ google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/l
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

internal/caches/list_file.go

@@ -5,6 +5,7 @@ package caches
 import (
 	"database/sql"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeNode/internal/ttlcache"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/lists"
 	_ "github.com/mattn/go-sqlite3"
@@ -35,10 +36,15 @@ type FileList struct {
 	itemsTableName string
 
 	isClosed bool
+
+	memoryCache *ttlcache.Cache
 }
 
 func NewFileList(dir string) ListInterface {
-	return &FileList{dir: dir}
+	return &FileList{
+		dir:         dir,
+		memoryCache: ttlcache.NewCache(),
+	}
 }
 
 func (this *FileList) Init() error {
@@ -100,7 +106,7 @@ func (this *FileList) Init() error {
 	this.total = total
 
 	// 常用语句
-	this.existsByHashStmt, err = this.db.Prepare(`SELECT "bodySize" FROM "` + this.itemsTableName + `" WHERE "hash"=? AND expiredAt>? LIMIT 1`)
+	this.existsByHashStmt, err = this.db.Prepare(`SELECT "expiredAt" FROM "` + this.itemsTableName + `" WHERE "hash"=? AND expiredAt>? LIMIT 1`)
 	if err != nil {
 		return err
 	}
@@ -166,6 +172,11 @@ func (this *FileList) Exist(hash string) (bool, error) {
 		return false, nil
 	}
 
+	item := this.memoryCache.Read(hash)
+	if item != nil {
+		return true, nil
+	}
+
 	rows, err := this.existsByHashStmt.Query(hash, time.Now().Unix())
 	if err != nil {
 		return false, err
@@ -174,6 +185,12 @@ func (this *FileList) Exist(hash string) (bool, error) {
 		_ = rows.Close()
 	}()
 	if rows.Next() {
+		var expiredAt int64
+		err = rows.Scan(&expiredAt)
+		if err != nil {
+			return true, nil
+		}
+		this.memoryCache.Write(hash, 1, expiredAt)
 		return true, nil
 	}
 	return false, nil
@@ -189,6 +206,10 @@ func (this *FileList) CleanPrefix(prefix string) error {
 		return nil
 	}
 
+	defer func() {
+		this.memoryCache.Clean()
+	}()
+
 	var count = int64(10000)
 	for {
 		result, err := this.db.Exec(`UPDATE "`+this.itemsTableName+`" SET expiredAt=0 WHERE id IN (SELECT id FROM "`+this.itemsTableName+`" WHERE expiredAt>0 AND createdAt<=? AND INSTR("key", ?)=1 LIMIT `+strconv.FormatInt(count, 10)+`)`, utils.UnixTime(), prefix)
@@ -210,6 +231,9 @@ func (this *FileList) Remove(hash string) error {
 		return nil
 	}
 
+	// 从缓存中删除
+	this.memoryCache.Delete(hash)
+
 	row := this.selectByHashStmt.QueryRow(hash)
 	if row.Err() != nil {
 		return row.Err()
@@ -288,6 +312,8 @@ func (this *FileList) CleanAll() error {
 		return nil
 	}
 
+	this.memoryCache.Clean()
+
 	_, err := this.deleteAllStmt.Exec()
 	if err != nil {
 		return err
@@ -334,6 +360,8 @@ func (this *FileList) OnRemove(f func(item *Item)) {
 func (this *FileList) Close() error {
 	this.isClosed = true
 
+	this.memoryCache.Destroy()
+
 	if this.db != nil {
 		_ = this.existsByHashStmt.Close()
 		_ = this.insertStmt.Close()

internal/caches/list_file_test.go

@@ -50,7 +50,7 @@ func TestFileList_Add_Many(t *testing.T) {
 	}
 	before := time.Now()
 	for i := 0; i < 2000_0000; i++ {
-		u := "http://edge.teaos.cn/123456" + strconv.Itoa(i)
+		u := "https://edge.teaos.cn/123456" + strconv.Itoa(i)
 		_ = list.Add(stringutil.Md5(u), &Item{
 			Key:        u,
 			ExpiredAt:  time.Now().Unix() + 3600,

internal/caches/reader_file.go

@@ -34,13 +34,7 @@ func (this *FileReader) Init() error {
 		}
 	}()
 
-	// 读取状态
-	_, err := this.fp.Seek(SizeExpiresAt, io.SeekStart)
-	if err != nil {
-		_ = this.discard()
-		return err
-	}
-	buf := make([]byte, 3)
+	var buf = make([]byte, SizeMeta)
 	ok, err := this.readToBuff(this.fp, buf)
 	if err != nil {
 		return err
@@ -48,37 +42,18 @@ func (this *FileReader) Init() error {
 	if !ok {
 		return ErrNotFound
 	}
-	status := types.Int(string(buf))
+
+	status := types.Int(string(buf[SizeExpiresAt : SizeExpiresAt+SizeStatus]))
 	if status < 100 || status > 999 {
 		return errors.New("invalid status")
 	}
 	this.status = status
 
 	// URL
-	_, err = this.fp.Seek(SizeExpiresAt+SizeStatus, io.SeekStart)
-	if err != nil {
-		return err
-	}
-
-	bytes4 := make([]byte, 4)
-	ok, err = this.readToBuff(this.fp, bytes4)
-	if err != nil {
-		return err
-	}
-	if !ok {
-		return ErrNotFound
-	}
-	urlLength := binary.BigEndian.Uint32(bytes4)
+	urlLength := binary.BigEndian.Uint32(buf[SizeExpiresAt+SizeStatus : SizeExpiresAt+SizeStatus+SizeURLLength])
 
 	// header
-	ok, err = this.readToBuff(this.fp, bytes4)
-	if err != nil {
-		return err
-	}
-	if !ok {
-		return ErrNotFound
-	}
-	headerSize := int(binary.BigEndian.Uint32(bytes4))
+	headerSize := int(binary.BigEndian.Uint32(buf[SizeExpiresAt+SizeStatus+SizeURLLength : SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength]))
 	if headerSize == 0 {
 		return nil
 	}
@@ -86,15 +61,7 @@ func (this *FileReader) Init() error {
 	this.headerOffset = int64(SizeMeta) + int64(urlLength)
 
 	// body
-	bytes8 := make([]byte, 8)
-	ok, err = this.readToBuff(this.fp, bytes8)
-	if err != nil {
-		return err
-	}
-	if !ok {
-		return ErrNotFound
-	}
-	bodySize := int(binary.BigEndian.Uint64(bytes8))
+	bodySize := int(binary.BigEndian.Uint64(buf[SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength : SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength+SizeBodyLength]))
 	if bodySize == 0 {
 		return nil
 	}

internal/caches/reader_file_test.go

@@ -49,6 +49,9 @@ func TestFileReader(t *testing.T) {
 		t.Log("body:", string(buf[:n]))
 		return true, nil
 	})
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func TestFileReader_Range(t *testing.T) {

internal/const/const.go

@@ -1,7 +1,7 @@
 package teaconst
 
 const (
-	Version = "0.2.6"
+	Version = "0.3.0"
 
 	ProductName = "Edge Node"
 	ProcessName = "edge-node"

internal/js/console.go

@@ -0,0 +1,50 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"encoding/json"
+	"github.com/iwind/TeaGo/logs"
+	"reflect"
+)
+
+type Console struct {
+}
+
+func (this *Console) Log(args ...interface{}) {
+	for index, arg := range args {
+		if arg != nil {
+			switch arg.(type) {
+			case bool, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, string:
+			default:
+				var argType = reflect.TypeOf(arg)
+
+				// 是否有String()方法，如果有直接调用
+				method, ok := argType.MethodByName("String")
+				if ok && method.Type.NumIn() == 1 && method.Type.NumOut() == 1 && method.Type.Out(0).Kind() == reflect.String {
+					args[index] = method.Func.Call([]reflect.Value{reflect.ValueOf(arg)})[0].String()
+					continue
+				}
+
+				// 转为JSON
+				argJSON, err := this.toJSON(arg)
+				if err != nil {
+					if argType.Kind() == reflect.Func {
+						args[index] = "[function]"
+					} else {
+						args[index] = "[object]"
+					}
+				} else {
+					args[index] = string(argJSON)
+				}
+			}
+		} else {
+			args[index] = "null"
+		}
+	}
+	logs.Println(append([]interface{}{"[js][console]"}, args...)...)
+}
+
+func (this *Console) toJSON(o interface{}) ([]byte, error) {
+	return json.Marshal(o)
+}

internal/js/console_test.go

@@ -0,0 +1,38 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"testing"
+)
+
+func TestConsole_Log(t *testing.T) {
+	{
+		vm := NewVM()
+		_, err := vm.RunString("console.log('Hello', 'world')")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	{
+		vm := NewVM()
+		_, err := vm.RunString("console.log(null, true, false, 10, 10.123)")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	{
+		vm := NewVM()
+		_, err := vm.RunString("console.log({ a:1, b:2 })")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	{
+		vm := NewVM()
+		_, err := vm.RunString("console.log(console.log)")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}

internal/js/http.go

@@ -0,0 +1,36 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+type HTTP struct {
+	r RequestInterface
+
+	req  *Request
+	resp *Response
+
+	onRequest func(req *Request, resp *Response)
+}
+
+func NewHTTP(r RequestInterface) *HTTP {
+	return &HTTP{
+		req:  NewRequest(r),
+		resp: NewResponse(r),
+	}
+}
+
+func (this *HTTP) OnRequest(callback func(req *Request, resp *Response)) {
+	// TODO 考虑是否支持多个callback
+	this.onRequest = callback
+}
+
+func (this *HTTP) OnData(callback func(req *Request, resp *Response)) {
+	// TODO
+}
+
+func (this *HTTP) OnResponse(callback func(req *Request, resp *Response)) {
+	// TODO
+}
+
+func (this *HTTP) TriggerRequest() {
+	this.onRequest(this.req, this.resp)
+}

internal/js/request.go

@@ -0,0 +1,82 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"bytes"
+	"io/ioutil"
+	"net"
+)
+
+type Request struct {
+	r RequestInterface
+}
+
+func NewRequest(r RequestInterface) *Request {
+	return &Request{
+		r: r,
+	}
+}
+
+func (this *Request) Proto() string {
+	return this.r.JSRequest().Proto
+}
+
+func (this *Request) Method() string {
+	return this.r.JSRequest().Method
+}
+
+func (this *Request) Header() map[string][]string {
+	return this.r.JSRequest().Header
+}
+
+func (this *Request) AddHeader(name string, value string) {
+	this.r.JSRequest().Header[name] = append(this.r.JSRequest().Header[name], value)
+}
+
+func (this *Request) SetHeader(name string, value string) {
+	this.r.JSRequest().Header[name] = []string{value}
+}
+
+func (this *Request) RemoteAddr() string {
+	var remoteAddr = this.r.JSRequest().RemoteAddr
+	host, _, err := net.SplitHostPort(remoteAddr)
+	if err == nil {
+		return host
+	}
+	return remoteAddr
+}
+
+func (this *Request) Url() *URL {
+	return NewURL(this.r.JSRequest().URL)
+}
+
+func (this *Request) ContentLength() int64 {
+	return this.r.JSRequest().ContentLength
+}
+
+func (this *Request) Body() []byte {
+	var bodyReader = this.r.JSRequest().Body
+	if bodyReader == nil {
+		return []byte{}
+	}
+	data, err := ioutil.ReadAll(bodyReader)
+	if err != nil {
+		this.r.JSLog("read body failed: " + err.Error())
+	}
+	return data
+}
+
+func (this *Request) CopyBody() []byte {
+	var bodyReader = this.r.JSRequest().Body
+	if bodyReader == nil {
+		return []byte{}
+	}
+
+	data, err := ioutil.ReadAll(bodyReader)
+	if err != nil {
+		this.r.JSLog("read body failed: " + err.Error())
+	}
+	this.r.JSRequest().Body = ioutil.NopCloser(bytes.NewReader(data))
+	return data
+}

internal/js/request_interface.go

@@ -0,0 +1,19 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import "net/http"
+
+type RequestInterface interface {
+	// JSRequest 请求
+	JSRequest() *http.Request
+
+	// JSWriter 响应
+	JSWriter() http.ResponseWriter
+
+	// JSStop 中止请求
+	JSStop()
+
+	// JSLog 打印日志
+	JSLog(msg ...interface{})
+}

internal/js/request_test.go

@@ -0,0 +1,124 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js_test
+
+import (
+	"bytes"
+	"github.com/TeaOSLab/EdgeNode/internal/js"
+	"github.com/iwind/TeaGo/logs"
+	"io/ioutil"
+	"net/http"
+	"testing"
+)
+
+type testRequest struct {
+	rawRequest  *http.Request
+	rawResponse *testResponse
+}
+
+func (this *testRequest) JSRequest() *http.Request {
+	if this.rawRequest != nil {
+		return this.rawRequest
+	}
+	req, _ := http.NewRequest(http.MethodGet, "https://iwind:123456@goedge.cn/docs?name=Libai&age=20", nil)
+	req.Header.Set("Server", "edgejs/1.0")
+	req.Header.Set("Content-Type", "application/json")
+	req.Body = ioutil.NopCloser(bytes.NewReader([]byte("123456")))
+	this.rawRequest = req
+	return req
+}
+
+func (this *testRequest) JSWriter() http.ResponseWriter {
+	if this.rawResponse != nil {
+		return this.rawResponse
+	}
+	this.rawResponse = &testResponse{}
+	return this.rawResponse
+}
+
+func (this *testRequest) JSStop() {
+
+}
+
+func (this *testRequest) JSLog(msg ...interface{}) {
+	logs.Println(msg...)
+}
+
+type testResponse struct {
+	statusCode int
+	header     http.Header
+}
+
+func (this *testResponse) Header() http.Header {
+	if this.header == nil {
+		this.header = http.Header{}
+	}
+	return this.header
+}
+
+func (this *testResponse) Write(p []byte) (int, error) {
+	return len(p), nil
+}
+
+func (this *testResponse) WriteHeader(statusCode int) {
+	this.statusCode = statusCode
+}
+
+func TestRequest(t *testing.T) {
+	vm := js.NewVM()
+	vm.SetRequest(&testRequest{})
+
+	// 事件监听
+	_, err := vm.RunString(`
+	http.onRequest(function (req, resp) {
+		console.log(req.proto())
+
+		let url = req.url()
+		console.log(url, "port:", url.port(), "args:", url.args())
+		console.log("username:", url.username(), "password:", url.password())
+		console.log("uri:", url.uri(), "path:", url.path())
+
+		req.addHeader("Server", "1.0")
+
+		
+		resp.write("this is response")
+		console.log(resp)
+
+		console.log(req.body()) 
+	})
+`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// 触发事件
+	_, err = vm.RunString(`http.triggerRequest()`)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestRequest_Header(t *testing.T) {
+	var req = js.NewRequest(&testRequest{})
+	logs.PrintAsJSON(req.Header(), t)
+
+	req.AddHeader("Content-Length", "10")
+	req.AddHeader("Vary", "1.0")
+	req.AddHeader("Vary", "2.0")
+	logs.PrintAsJSON(req.Header(), t)
+
+	req.SetHeader("Vary", "3.0")
+	logs.PrintAsJSON(req.Header(), t)
+}
+
+func TestRequest_Body(t *testing.T) {
+	var req = js.NewRequest(&testRequest{})
+	t.Log(string(req.Body()))
+	t.Log(string(req.Body()))
+}
+
+func TestRequest_CopyBody(t *testing.T) {
+	var req = js.NewRequest(&testRequest{})
+	t.Log(string(req.CopyBody()))
+	t.Log(string(req.CopyBody()))
+}

internal/js/response.go

@@ -0,0 +1,39 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+type Response struct {
+	r RequestInterface
+}
+
+func NewResponse(r RequestInterface) *Response {
+	return &Response{
+		r: r,
+	}
+}
+
+func (this *Response) Write(s string) error {
+	_, err := this.r.JSWriter().Write([]byte(s))
+	return err
+}
+
+func (this *Response) Reply(status int) {
+	this.SetStatus(status)
+	this.r.JSStop()
+}
+
+func (this *Response) Header() map[string][]string {
+	return this.r.JSWriter().Header()
+}
+
+func (this *Response) AddHeader(name string, value string) {
+	this.r.JSWriter().Header()[name] = append(this.r.JSWriter().Header()[name], value)
+}
+
+func (this *Response) SetHeader(name string, value string) {
+	this.r.JSWriter().Header()[name] = []string{value}
+}
+
+func (this *Response) SetStatus(statusCode int) {
+	this.r.JSWriter().WriteHeader(statusCode)
+}

internal/js/response_test.go

@@ -0,0 +1,16 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js_test
+
+import (
+	"github.com/TeaOSLab/EdgeNode/internal/js"
+	"testing"
+)
+
+func TestNewResponse(t *testing.T) {
+	var resp = js.NewResponse(&testRequest{})
+	resp.AddHeader("Vary", "1.0")
+	resp.AddHeader("Vary", "2.0")
+	resp.SetHeader("Server", "edgejs/1.0")
+	t.Logf("%#v", resp.Header())
+}

internal/js/url.go

@@ -0,0 +1,90 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"github.com/dop251/goja"
+	"github.com/iwind/TeaGo/types"
+	"net/url"
+)
+
+type URL struct {
+	u *url.URL
+}
+
+func NewURL(u *url.URL) *URL {
+	return &URL{
+		u: u,
+	}
+}
+
+func (this *URL) JSNew(args []goja.Value) *URL {
+	var urlString = ""
+	if len(args) == 1 {
+		urlString = args[0].String()
+	}
+	u, _ := url.Parse(urlString)
+	if u == nil {
+		u = &url.URL{}
+	}
+	return NewURL(u)
+}
+
+func (this *URL) Port() int {
+	return types.Int(this.u.Port())
+}
+
+func (this *URL) Args() map[string][]string {
+	return this.u.Query()
+}
+
+func (this *URL) Arg(name string) string {
+	return this.u.Query().Get(name)
+}
+
+func (this *URL) Username() string {
+	if this.u.User != nil {
+		return this.u.User.Username()
+	}
+	return ""
+}
+
+func (this *URL) Password() string {
+	if this.u.User != nil {
+		password, _ := this.u.User.Password()
+		return password
+	}
+	return ""
+}
+
+func (this *URL) Uri() string {
+	return this.u.RequestURI()
+}
+
+func (this *URL) Path() string {
+	return this.u.Path
+}
+
+func (this *URL) Host() string {
+	return this.u.Host
+}
+
+func (this *URL) Fragment() string {
+	return this.u.Fragment
+}
+
+func (this *URL) Hash() string {
+	if len(this.u.Fragment) > 0 {
+		return "#" + this.u.Fragment
+	} else {
+		return ""
+	}
+}
+
+func (this *URL) Scheme() string {
+	return this.u.Scheme
+}
+
+func (this *URL) String() string {
+	return this.u.String()
+}

internal/js/url_test.go

@@ -0,0 +1,18 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"net/url"
+	"testing"
+)
+
+func TestURL(t *testing.T) {
+	raw, err := url.Parse("https://iwind:123456@goedge.cn/docs?name=Libai&age=20#a=b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var u = NewURL(raw)
+	t.Log("host:", u.Host())
+	t.Log("hash:", u.Hash())
+}

internal/js/vm.go

@@ -0,0 +1,153 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"errors"
+	"github.com/dop251/goja"
+	"github.com/iwind/TeaGo/logs"
+	"reflect"
+	"strings"
+)
+
+var sharedPrograms []*goja.Program
+var sharedConsole = &Console{}
+
+func init() {
+	// compile programs
+}
+
+type VM struct {
+	vm *goja.Runtime
+}
+
+func NewVM() *VM {
+	vm := goja.New()
+	vm.SetFieldNameMapper(goja.TagFieldNameMapper("json", true))
+
+	// programs
+	for _, program := range sharedPrograms {
+		_, _ = vm.RunProgram(program)
+	}
+
+	v := &VM{vm: vm}
+	v.initVM()
+	return v
+}
+
+func (this *VM) Set(name string, obj interface{}) error {
+	return this.vm.Set(name, obj)
+}
+
+func (this *VM) AddConstructor(name string, instance interface{}) error {
+	objType := reflect.TypeOf(instance)
+
+	if objType.Kind() != reflect.Ptr {
+		return errors.New("instance should be pointer")
+	}
+
+	// construct
+	newMethod, ok := objType.MethodByName("JSNew")
+	if !ok {
+		return errors.New("can not find 'JSNew()' method in '" + objType.Elem().Name() + "'")
+	}
+
+	var err = this.Set(name, func(call goja.ConstructorCall) *goja.Object {
+		if newMethod.Type.NumIn() != 2 {
+			this.throw(errors.New(objType.Elem().Name() + ".JSNew() should accept a '[]goja.Value' argument"))
+			return nil
+		}
+		if newMethod.Type.In(1).String() != "[]goja.Value" {
+			this.throw(errors.New(objType.Elem().Name() + ".JSNew() should accept a '[]goja.Value' argument"))
+			return nil
+		}
+
+		// new
+		var results = newMethod.Func.Call([]reflect.Value{reflect.ValueOf(instance), reflect.ValueOf(call.Arguments)})
+		if len(results) == 0 {
+			this.throw(errors.New(objType.Elem().Name() + ".JSNew() should return a valid instance"))
+			return nil
+		}
+		var result = results[0]
+		if result.Type() != objType {
+			this.throw(errors.New(objType.Elem().Name() + ".JSNew() should return a same instance"))
+			return nil
+		}
+
+		// methods
+		var resultType = result.Type()
+		var numMethod = result.NumMethod()
+		for i := 0; i < numMethod; i++ {
+			var method = resultType.Method(i)
+			var methodName = strings.ToLower(method.Name[:1]) + method.Name[1:]
+			err := call.This.Set(methodName, result.MethodByName(method.Name).Interface())
+			if err != nil {
+				this.throw(err)
+				continue
+			}
+		}
+
+		//  支持属性
+		var numField = result.Elem().Type().NumField()
+		for i := 0; i < numField; i++ {
+			var field = result.Elem().Field(i)
+			if !field.CanInterface() {
+				continue
+			}
+			var fieldType = objType.Elem().Field(i)
+			tag, ok := fieldType.Tag.Lookup("json")
+			if !ok {
+				tag = fieldType.Name
+				tag = strings.ToLower(tag[:1]) + tag[1:]
+			} else {
+				// TODO 校验tag是否符合变量语法
+			}
+			err := call.This.Set(tag, field.Interface())
+			if err != nil {
+				this.throw(err)
+				continue
+			}
+		}
+
+		return nil
+	})
+	return err
+}
+
+func (this *VM) RunString(str string) (goja.Value, error) {
+	defer func() {
+		e := recover()
+		if e != nil {
+			// TODO 需要打印trace
+			logs.Println("panic:", e)
+		}
+	}()
+	return this.vm.RunString(str)
+}
+
+func (this *VM) SetRequest(req RequestInterface) {
+	{
+		err := this.vm.Set("http", NewHTTP(req))
+		if err != nil {
+			this.throw(err)
+		}
+	}
+}
+
+func (this *VM) initVM() {
+	{
+		err := this.vm.Set("console", sharedConsole)
+		if err != nil {
+			this.throw(err)
+		}
+	}
+}
+
+func (this *VM) throw(err error) {
+	if err == nil {
+		return
+	}
+
+	// TODO
+	logs.Println("js:VM:error: " + err.Error())
+}

internal/js/vm_test.go

@@ -0,0 +1,158 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package js
+
+import (
+	"github.com/dop251/goja"
+	"testing"
+	"time"
+)
+
+func TestNewVM(t *testing.T) {
+	before := time.Now()
+	defer func() {
+		t.Log(time.Since(before).Seconds()*1000, "ms")
+	}()
+
+	vm := NewVM()
+	{
+		v, err := vm.RunString("JSON.stringify({\"a\":\"b\"})")
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("JSON.stringify():", v)
+	}
+	{
+		v, err := vm.RunString(`JSON.parse('{\"a\":\"b\"}')`)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log("JSON.parse():", v)
+	}
+	{
+		err := vm.AddConstructor("Url", &URL{})
+		if err != nil {
+			t.Fatal("add constructor error:", err)
+		}
+		_, err = vm.RunString(`
+{
+	let u = new Url("https://goedge.cn/docs?v=1")
+	console.log("host:", u.host(), u.uri())
+}
+{
+	let u = new Url("https://teaos.cn/downloads?v=1")
+	console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url()
+	console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url("a", "b", "c")
+	console.log("host:", u.host(), u.uri())
+}
+`)
+		if err != nil {
+			t.Fatal("add constructor error:" + err.Error())
+		}
+	}
+}
+
+func TestVM_Program(t *testing.T) {
+	var s = `
+{
+	let u = new Url("https://goedge.cn/docs?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+{
+	let u = new Url("https://teaos.cn/downloads?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url()
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url("a", "b", "c")
+	//console.log("host:", u.host(), u.uri())
+}
+`
+	program := goja.MustCompile("s", s, true)
+
+	before := time.Now()
+	defer func() {
+		t.Log(time.Since(before).Seconds()*1000, "ms")
+	}()
+
+	vm := NewVM()
+	err := vm.AddConstructor("Url", &URL{})
+	if err != nil {
+		t.Fatal("add constructor error:", err)
+	}
+	//_, err = vm.RunString(s)
+	_, err = vm.vm.RunProgram(program)
+	if err != nil {
+		t.Fatal("add constructor error:" + err.Error())
+	}
+}
+
+func Benchmark_Program(b *testing.B) {
+	var s = `
+{
+	let u = new Url("https://goedge.cn/docs?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+{
+	let u = new Url("https://teaos.cn/downloads?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url()
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url("a", "b", "c")
+	//console.log("host:", u.host(), u.uri())
+}
+{
+	let u = new Url("https://goedge.cn/docs?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+{
+	let u = new Url("https://teaos.cn/downloads?v=1")
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url()
+	//console.log("host:", u.host(), u.uri())
+}
+
+{
+	let u = new Url("a", "b", "c")
+	//console.log("host:", u.host(), u.uri())
+}
+`
+	program := goja.MustCompile("s", s, true)
+
+	vm := NewVM()
+
+	err := vm.AddConstructor("Url", &URL{})
+	if err != nil {
+		b.Fatal("add constructor error:", err)
+	}
+
+	for i := 0; i < b.N; i++ {
+		//_, err = vm.RunString(s)
+		_, err = vm.vm.RunProgram(program)
+		if err != nil {
+			b.Fatal("add constructor error:" + err.Error())
+		}
+	}
+}

internal/metrics/stat.go

@@ -3,6 +3,7 @@
 package metrics
 
 import (
+	"encoding/json"
 	"github.com/cespare/xxhash"
 	"strconv"
 )
@@ -13,10 +14,9 @@ type Stat struct {
 	Hash     string
 	Value    int64
 	Time     string
-
-	keysData []byte
 }
 
-func (this *Stat) Sum(version int32, itemId int64) {
-	this.Hash = strconv.FormatUint(xxhash.Sum64String(strconv.FormatInt(this.ServerId, 10)+"@"+string(this.keysData)+"@"+this.Time+"@"+strconv.Itoa(int(version))+"@"+strconv.FormatInt(itemId, 10)), 10)
+func SumStat(serverId int64, keys []string, time string, version int32, itemId int64) string {
+	keysData, _ := json.Marshal(keys)
+	return strconv.FormatUint(xxhash.Sum64String(strconv.FormatInt(serverId, 10)+"@"+string(keysData)+"@"+time+"@"+strconv.Itoa(int(version))+"@"+strconv.FormatInt(itemId, 10)), 10)
 }

internal/metrics/task.go

@@ -19,6 +19,8 @@ import (
 	"time"
 )
 
+const MaxQueueSize = 10240
+
 // Task 单个指标任务
 // 数据库存储：
 //  data/
@@ -35,7 +37,6 @@ type Task struct {
 
 	db            *sql.DB
 	statTableName string
-	statsChan     chan *Stat
 	isStopped     bool
 
 	cleanTicker  *utils.Ticker
@@ -52,15 +53,19 @@ type Task struct {
 	serverIdMap       map[int64]bool  // 所有的服务Ids
 	timeMap           map[string]bool // time => bool
 	serverIdMapLocker sync.Mutex
+
+	statsMap    map[string]*Stat
+	statsLocker sync.Mutex
+	statsTicker *utils.Ticker
 }
 
 // NewTask 获取新任务
 func NewTask(item *serverconfigs.MetricItemConfig) *Task {
 	return &Task{
 		item:        item,
-		statsChan:   make(chan *Stat, 40960),
 		serverIdMap: map[int64]bool{},
 		timeMap:     map[string]bool{},
+		statsMap:    map[string]*Stat{},
 	}
 }
 
@@ -131,7 +136,7 @@ ON "` + this.statTableName + `" (
 	}
 
 	// select topN stmt
-	this.selectTopStmt, err = db.Prepare(`SELECT "id", "hash", "keys", "value", "isUploaded" FROM "` + this.statTableName + `" WHERE "serverId"=? AND "version"=? AND time=? ORDER BY "value" DESC LIMIT 100`)
+	this.selectTopStmt, err = db.Prepare(`SELECT "id", "hash", "keys", "value", "isUploaded" FROM "` + this.statTableName + `" WHERE "serverId"=? AND "version"=? AND time=? ORDER BY "value" DESC LIMIT 20`)
 	if err != nil {
 		return err
 	}
@@ -156,14 +161,19 @@ ON "` + this.statTableName + `" (
 // Start 启动任务
 func (this *Task) Start() error {
 	// 读取数据
+	this.statsTicker = utils.NewTicker(1 * time.Minute)
 	go func() {
-		for stat := range this.statsChan {
-			if stat == nil {
-				return
-			}
-			err := this.InsertStat(stat)
-			if err != nil {
-				remotelogs.Error("METRIC", "insert stat failed: "+err.Error())
+		for this.statsTicker.Next() {
+			this.statsLocker.Lock()
+			var statsMap = this.statsMap
+			this.statsMap = map[string]*Stat{}
+			this.statsLocker.Unlock()
+
+			for _, stat := range statsMap {
+				err := this.InsertStat(stat)
+				if err != nil {
+					remotelogs.Error("METRIC", "insert stat failed: "+err.Error())
+				}
 			}
 		}
 	}()
@@ -210,18 +220,25 @@ func (this *Task) Add(obj MetricInterface) {
 		return
 	}
 
-	var stat = &Stat{
-		ServerId: obj.MetricServerId(),
-		Keys:     keys,
-		Value:    v,
-		Time:     this.item.CurrentTime(),
-	}
-
-	select {
-	case this.statsChan <- stat:
-	default:
-		// 丢弃
+	var hash = SumStat(obj.MetricServerId(), keys, this.item.CurrentTime(), this.item.Version, this.item.Id)
+	this.statsLocker.Lock()
+	oldStat, ok := this.statsMap[hash]
+	if ok {
+		oldStat.Value += v
+		oldStat.Hash = hash
+	} else {
+		// 防止过载
+		if len(this.statsMap) < MaxQueueSize {
+			this.statsMap[hash] = &Stat{
+				ServerId: obj.MetricServerId(),
+				Keys:     keys,
+				Value:    v,
+				Time:     this.item.CurrentTime(),
+				Hash:     hash,
+			}
+		}
 	}
+	this.statsLocker.Unlock()
 }
 
 // Stop 停止任务
@@ -234,6 +251,9 @@ func (this *Task) Stop() error {
 	if this.uploadTicker != nil {
 		this.uploadTicker.Stop()
 	}
+	if this.statsTicker != nil {
+		this.statsTicker.Stop()
+	}
 
 	_ = this.insertStatStmt.Close()
 	_ = this.deleteByVersionStmt.Close()
@@ -241,14 +261,6 @@ func (this *Task) Stop() error {
 	_ = this.selectTopStmt.Close()
 	_ = this.sumStmt.Close()
 
-	if this.statsChan != nil {
-		go func() {
-			// 延时关闭，防止关闭时写入
-			time.Sleep(5 * time.Second)
-			close(this.statsChan)
-		}()
-	}
-
 	if this.db != nil {
 		_ = this.db.Close()
 	}
@@ -274,10 +286,8 @@ func (this *Task) InsertStat(stat *Stat) error {
 	if err != nil {
 		return err
 	}
-	stat.keysData = keyData
-	stat.Sum(this.item.Version, this.item.Id)
 
-	_, err = this.insertStatStmt.Exec(stat.ServerId, stat.Hash, stat.keysData, stat.Value, stat.Time, this.item.Version, stat.Value)
+	_, err = this.insertStatStmt.Exec(stat.ServerId, stat.Hash, keyData, stat.Value, stat.Time, this.item.Version, stat.Value)
 	if err != nil {
 		return err
 	}
@@ -354,8 +364,7 @@ func (this *Task) Upload(pauseDuration time.Duration) error {
 
 				var pbStats []*pb.UploadingMetricStat
 				for rows.Next() {
-					var pbStat = &pb.UploadingMetricStat{
-					}
+					var pbStat = &pb.UploadingMetricStat{}
 					// "id", "hash", "keys", "value", "isUploaded"
 					var isUploaded int
 					var keysData []byte
@@ -363,9 +372,11 @@ func (this *Task) Upload(pauseDuration time.Duration) error {
 					if err != nil {
 						return nil, err
 					}
-					if isUploaded == 1 {
+
+					// TODO 先不判断是否已经上传，需要改造API进行配合
+					/**if isUploaded == 1 {
 						continue
-					}
+					}**/
 					if len(keysData) > 0 {
 						err = json.Unmarshal(keysData, &pbStat.Keys)
 						if err != nil {

internal/nodes/api_stream.go

@@ -2,6 +2,7 @@ package nodes
 
 import (
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"github.com/TeaOSLab/EdgeCommon/pkg/messageconfigs"
@@ -16,6 +17,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/logs"
 	"io"
+	"net"
 	"net/http"
 	"os/exec"
 	"strconv"
@@ -367,7 +369,28 @@ func (this *APIStream) handlePreheatCache(message *pb.NodeStreamMessage) error {
 
 	wg := sync.WaitGroup{}
 	wg.Add(len(msg.Keys))
-	client := http.Client{} // TODO 可以设置请求超时事件
+	client := &http.Client{
+		Timeout: 30 * time.Second, // TODO 可以设置请求超时时间
+		Transport: &http.Transport{
+			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+				_, port, err := net.SplitHostPort(addr)
+				if err != nil {
+					return nil, err
+				}
+				return net.Dial(network, "127.0.0.1:"+port)
+			},
+			MaxIdleConns:          4096,
+			MaxIdleConnsPerHost:   32,
+			MaxConnsPerHost:       32,
+			IdleConnTimeout:       2 * time.Minute,
+			ExpectContinueTimeout: 1 * time.Second,
+			TLSHandshakeTimeout:   0,
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		},
+	}
+	defer client.CloseIdleConnections()
 	errorMessages := []string{}
 	locker := sync.Mutex{}
 	for _, key := range msg.Keys {
@@ -381,7 +404,9 @@ func (this *APIStream) handlePreheatCache(message *pb.NodeStreamMessage) error {
 				locker.Unlock()
 				return
 			}
+
 			// TODO 可以在管理界面自定义Header
+			req.Header.Set("X-Cache-Action", "preheat")
 			req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.121 Safari/537.36")
 			req.Header.Set("Accept-Encoding", "gzip, deflate, br") // TODO 这里需要记录下缓存是否为gzip的
 			resp, err := client.Do(req)

internal/nodes/http_access_log_queue.go

@@ -9,14 +9,14 @@ import (
 
 var sharedHTTPAccessLogQueue = NewHTTPAccessLogQueue()
 
-// HTTP访问日志队列
+// HTTPAccessLogQueue HTTP访问日志队列
 type HTTPAccessLogQueue struct {
 	queue chan *pb.HTTPAccessLog
 }
 
-// 获取新对象
+// NewHTTPAccessLogQueue 获取新对象
 func NewHTTPAccessLogQueue() *HTTPAccessLogQueue {
-	// 队列中最大的值，超出此数量的访问日志会被抛弃
+	// 队列中最大的值，超出此数量的访问日志会被丢弃
 	// TODO 需要可以在界面中设置
 	maxSize := 10000
 	queue := &HTTPAccessLogQueue{
@@ -27,7 +27,7 @@ func NewHTTPAccessLogQueue() *HTTPAccessLogQueue {
 	return queue
 }
 
-// 开始处理访问日志
+// Start 开始处理访问日志
 func (this *HTTPAccessLogQueue) Start() {
 	ticker := time.NewTicker(1 * time.Second)
 	for range ticker.C {
@@ -38,7 +38,7 @@ func (this *HTTPAccessLogQueue) Start() {
 	}
 }
 
-// 加入新访问日志
+// Push 加入新访问日志
 func (this *HTTPAccessLogQueue) Push(accessLog *pb.HTTPAccessLog) {
 	select {
 	case this.queue <- accessLog:

internal/nodes/http_request.go

@@ -121,7 +121,7 @@ func (this *HTTPRequest) Do() {
 	// Web配置
 	err := this.configureWeb(this.Server.Web, true, 0)
 	if err != nil {
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		this.doEnd()
 		return
 	}
@@ -167,6 +167,16 @@ func (this *HTTPRequest) Do() {
 
 // 开始调用
 func (this *HTTPRequest) doBegin() {
+	// 特殊URL处理
+	if len(this.rawURI) > 1 && this.rawURI[1] == '.' {
+		// ACME
+		// TODO 需要配置是否启用ACME检测
+		if strings.HasPrefix(this.rawURI, "/.well-known/acme-challenge/") {
+			this.doACME()
+			return
+		}
+	}
+
 	// 统计
 	if this.web.StatRef != nil && this.web.StatRef.IsOn {
 		this.doStat()
@@ -179,16 +189,6 @@ func (this *HTTPRequest) doBegin() {
 		}
 	}
 
-	// 特殊URL处理
-	if len(this.rawURI) > 1 && this.rawURI[1] == '.' {
-		// ACME
-		// TODO 需要配置是否启用ACME检测
-		if strings.HasPrefix(this.rawURI, "/.well-known/acme-challenge/") {
-			this.doACME()
-			return
-		}
-	}
-
 	// 临时关闭页面
 	if this.web.Shutdown != nil && this.web.Shutdown.IsOn {
 		this.doShutdown()

internal/nodes/http_request_acme.go

@@ -12,6 +12,10 @@ func (this *HTTPRequest) doACME() {
 	// TODO 对请求进行校验，防止恶意攻击
 
 	token := filepath.Base(this.RawReq.URL.Path)
+	if token == "acme-challenge" || len(token) <= 32 {
+		this.writer.WriteHeader(http.StatusNotFound)
+		return
+	}
 
 	rpcClient, err := rpc.SharedRPC()
 	if err != nil {

internal/nodes/http_request_auth.go

@@ -33,7 +33,7 @@ func (this *HTTPRequest) doAuth() (shouldStop bool) {
 			return writer.StatusCode(), nil
 		}, this.Format)
 		if err != nil {
-			this.write502(err)
+			this.write50x(err, http.StatusInternalServerError)
 			return
 		}
 		if b {

internal/nodes/http_request_cache.go

@@ -7,6 +7,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -20,6 +21,12 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
 	if this.web.Cache == nil || !this.web.Cache.IsOn || (len(cachePolicy.CacheRefs) == 0 && len(this.web.Cache.CacheRefs) == 0) {
 		return
 	}
+
+	// 判断是否在预热
+	if strings.HasPrefix(this.RawReq.RemoteAddr, "127.") && this.RawReq.Header.Get("X-Cache-Action") == "preheat" {
+		return
+	}
+
 	var addStatusHeader = this.web.Cache.AddStatusHeader
 	if addStatusHeader {
 		defer func() {

internal/nodes/http_request_error.go

@@ -1,6 +1,7 @@
 package nodes
 
 import (
+	"github.com/iwind/TeaGo/types"
 	"net/http"
 )
 
@@ -17,30 +18,15 @@ func (this *HTTPRequest) write404() {
 	_, _ = this.writer.Write([]byte(msg))
 }
 
-func (this *HTTPRequest) write500(err error) {
+func (this *HTTPRequest) write50x(err error, statusCode int) {
 	if err != nil {
 		this.addError(err)
 	}
 
-	statusCode := http.StatusInternalServerError
 	if this.doPage(statusCode) {
 		return
 	}
 	this.processResponseHeaders(statusCode)
 	this.writer.WriteHeader(statusCode)
-	_, _ = this.writer.Write([]byte(http.StatusText(statusCode)))
-}
-
-func (this *HTTPRequest) write502(err error) {
-	if err != nil {
-		this.addError(err)
-	}
-
-	statusCode := http.StatusBadGateway
-	if this.doPage(statusCode) {
-		return
-	}
-	this.processResponseHeaders(statusCode)
-	this.writer.WriteHeader(statusCode)
-	_, _ = this.writer.Write([]byte("502 Bad Gateway"))
+	_, _ = this.writer.Write([]byte(types.String(statusCode) + " " + http.StatusText(statusCode)))
 }

internal/nodes/http_request_fastcgi.go

@@ -15,6 +15,7 @@ import (
 	"github.com/iwind/gofcgi/pkg/fcgi"
 	"io"
 	"net"
+	"net/http"
 	"net/url"
 	"path/filepath"
 	"strings"
@@ -80,7 +81,7 @@ func (this *HTTPRequest) doFastcgi() (shouldStop bool) {
 
 	client, err := fcgi.SharedPool(fastcgi.Network(), fastcgi.RealAddress(), uint(poolSize)).Client()
 	if err != nil {
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
 
@@ -158,13 +159,13 @@ func (this *HTTPRequest) doFastcgi() (shouldStop bool) {
 
 	resp, stderr, err := client.Call(fcgiReq)
 	if err != nil {
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
 
 	if len(stderr) > 0 {
 		err := errors.New("Fastcgi Error: " + strings.TrimSpace(string(stderr)) + " script: " + maps.NewMap(params).GetString("SCRIPT_FILENAME"))
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
 

internal/nodes/http_request_reverse_proxy.go

@@ -8,6 +8,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"io"
+	"net/http"
 	"net/url"
 	"strconv"
 	"strings"
@@ -36,7 +37,7 @@ func (this *HTTPRequest) doReverseProxy() {
 	if origin == nil {
 		err := errors.New(this.requestPath() + ": no available backends for reverse proxy")
 		remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
-		this.write502(err)
+		this.write50x(err, http.StatusBadGateway)
 		return
 	}
 	this.origin = origin // 设置全局变量是为了日志等处理
@@ -56,7 +57,7 @@ func (this *HTTPRequest) doReverseProxy() {
 	if origin.Addr == nil {
 		err := errors.New(this.requestPath() + ": origin '" + strconv.FormatInt(origin.Id, 10) + "' does not has a address")
 		remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
-		this.write502(err)
+		this.write50x(err, http.StatusBadGateway)
 		return
 	}
 	this.RawReq.URL.Scheme = origin.Addr.Protocol.Primary().Scheme()
@@ -143,7 +144,7 @@ func (this *HTTPRequest) doReverseProxy() {
 	client, err := SharedHTTPClientPool.Client(this.RawReq, origin, originAddr)
 	if err != nil {
 		remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
-		this.write502(err)
+		this.write50x(err, http.StatusBadGateway)
 		return
 	}
 
@@ -158,13 +159,23 @@ func (this *HTTPRequest) doReverseProxy() {
 	if err != nil {
 		// 客户端取消请求，则不提示
 		httpErr, ok := err.(*url.Error)
-		if !ok || httpErr.Err != context.Canceled {
-			// TODO 如果超过最大失败次数，则下线
+		if !ok {
 			SharedOriginStateManager.Fail(origin, this.reverseProxy, func() {
 				this.reverseProxy.ResetScheduling()
 			})
-
-			this.write502(err)
+			this.write50x(err, http.StatusBadGateway)
+			remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
+		} else if httpErr.Err != context.Canceled {
+			SharedOriginStateManager.Fail(origin, this.reverseProxy, func() {
+				this.reverseProxy.ResetScheduling()
+			})
+			if httpErr.Timeout() {
+				this.write50x(err, http.StatusGatewayTimeout)
+			} else if httpErr.Temporary() {
+				this.write50x(err, http.StatusServiceUnavailable)
+			} else {
+				this.write50x(err, http.StatusBadGateway)
+			}
 			remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
 		} else {
 			// 是否为客户端方面的错误
@@ -178,7 +189,7 @@ func (this *HTTPRequest) doReverseProxy() {
 			}
 
 			if !isClientError {
-				this.write502(err)
+				this.write50x(err, http.StatusBadGateway)
 			}
 		}
 		if resp != nil && resp.Body != nil {

internal/nodes/http_request_root.go

@@ -109,7 +109,7 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
 			}
 			return
 		} else {
-			this.write500(err)
+			this.write50x(err, http.StatusInternalServerError)
 			logs.Error(err)
 			return true
 		}
@@ -138,7 +138,7 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
 					}
 					return
 				} else {
-					this.write500(err)
+					this.write50x(err, http.StatusInternalServerError)
 					logs.Error(err)
 					return true
 				}
@@ -283,8 +283,7 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
 
 	reader, err := os.OpenFile(filePath, os.O_RDONLY, 0444)
 	if err != nil {
-		this.write500(err)
-		logs.Error(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return true
 	}
 

internal/nodes/http_request_url.go

@@ -36,7 +36,7 @@ func (this *HTTPRequest) doURL(method string, url string, host string, statusCod
 	resp, err := client.Do(req)
 	if err != nil {
 		logs.Error(errors.New(req.URL.String() + ": " + err.Error()))
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
 	defer func() {

internal/nodes/http_request_websocket.go

@@ -2,7 +2,6 @@ package nodes
 
 import (
 	"errors"
-	"github.com/iwind/TeaGo/logs"
 	"io"
 	"net/http"
 	"net/url"
@@ -43,8 +42,7 @@ func (this *HTTPRequest) doWebsocket() {
 	// TODO 增加N次错误重试，重试的时候需要尝试不同的源站
 	originConn, err := OriginConnect(this.origin, this.RawReq.RemoteAddr)
 	if err != nil {
-		logs.Error(err)
-		this.write500(err)
+		this.write50x(err, http.StatusBadGateway)
 		return
 	}
 	defer func() {
@@ -53,15 +51,13 @@ func (this *HTTPRequest) doWebsocket() {
 
 	err = this.RawReq.Write(originConn)
 	if err != nil {
-		logs.Error(err)
-		this.write500(err)
+		this.write50x(err, http.StatusBadGateway)
 		return
 	}
 
 	clientConn, _, err := this.writer.Hijack()
 	if err != nil {
-		logs.Error(err)
-		this.write500(err)
+		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
 	defer func() {

internal/nodes/traffic_listener.go

@@ -25,7 +25,7 @@ func (this *TrafficListener) Accept() (net.Conn, error) {
 	ip, _, err := net.SplitHostPort(conn.RemoteAddr().String())
 	if err == nil {
 		if !waf.SharedIPWhiteList.Contains(waf.IPTypeAll, ip) && waf.SharedIPBlackLIst.Contains(waf.IPTypeAll, ip) {
-			go func() {
+			defer func() {
 				_ = conn.Close()
 			}()
 			return conn, nil

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, "goedge.cn", 1, 0, 0, 0)
+		manager.Add(1, "goedge.cn", 1, 0, 0, 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, "goedge.cn", 1, 0, 0, 0)
+		manager.Add(1, "goedge.cn", 1, 0, 0, 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, "goedge.cn", 1024, 1, 0, 0)
+		manager.Add(1, "goedge.cn", 1024, 1, 0, 0, 0, 0)
 	}
 }

internal/ttlcache/cache.go

@@ -5,7 +5,7 @@ import (
 	"time"
 )
 
-// TTL缓存
+// Cache TTL缓存
 // 最大的缓存时间为30 * 86400
 // Piece数据结构：
 //      Piece1          |  Piece2 | Piece3 | ...
@@ -136,6 +136,12 @@ func (this *Cache) GC() {
 	this.gcPieceIndex = newIndex
 }
 
+func (this *Cache) Clean() {
+	for _, piece := range this.pieces {
+		piece.Clean()
+	}
+}
+
 func (this *Cache) Destroy() {
 	this.isDestroyed = true
 

internal/ttlcache/piece.go

@@ -17,7 +17,7 @@ func NewPiece(maxItems int) *Piece {
 	return &Piece{m: map[uint64]*Item{}, maxItems: maxItems}
 }
 
-func (this *Piece) Add(key uint64, item *Item) () {
+func (this *Piece) Add(key uint64, item *Item) {
 	this.locker.Lock()
 	if len(this.m) >= this.maxItems {
 		this.locker.Unlock()
@@ -82,6 +82,12 @@ func (this *Piece) GC() {
 	this.locker.Unlock()
 }
 
+func (this *Piece) Clean() {
+	this.locker.Lock()
+	this.m = map[uint64]*Item{}
+	this.locker.Unlock()
+}
+
 func (this *Piece) Destroy() {
 	this.locker.Lock()
 	this.m = nil

internal/utils/http.go

@@ -16,6 +16,9 @@ var timeoutClientLocker = sync.Mutex{}
 // 导出响应
 func DumpResponse(resp *http.Response) (header []byte, body []byte, err error) {
 	header, err = httputil.DumpResponse(resp, false)
+	if err != nil {
+		return
+	}
 	body, err = ioutil.ReadAll(resp.Body)
 	return
 }

internal/waf/checkpoints/request_upload_test.go

@@ -17,46 +17,67 @@ func TestRequestUploadCheckpoint_RequestValue(t *testing.T) {
 	{
 		part, err := writer.CreateFormField("name")
 		if err == nil {
-			part.Write([]byte("lu"))
+			_, err := part.Write([]byte("lu"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
 	{
 		part, err := writer.CreateFormField("age")
 		if err == nil {
-			part.Write([]byte("20"))
+			_, err = part.Write([]byte("20"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
 	{
 		part, err := writer.CreateFormFile("myFile", "hello.txt")
 		if err == nil {
-			part.Write([]byte("Hello, World!"))
+			_, err = part.Write([]byte("Hello, World!"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
 	{
 		part, err := writer.CreateFormFile("myFile2", "hello.PHP")
 		if err == nil {
-			part.Write([]byte("Hello, World, PHP!"))
+			_, err = part.Write([]byte("Hello, World, PHP!"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
 	{
 		part, err := writer.CreateFormFile("myFile3", "hello.asp")
 		if err == nil {
-			part.Write([]byte("Hello, World, ASP Pages!"))
+			_, err = part.Write([]byte("Hello, World, ASP Pages!"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
 	{
 		part, err := writer.CreateFormFile("myFile4", "hello.asp")
 		if err == nil {
-			part.Write([]byte("Hello, World, ASP Pages!"))
+			_, err = part.Write([]byte("Hello, World, ASP Pages!"))
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 
-	writer.Close()
+	err := writer.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	rawReq, err := http.NewRequest(http.MethodPost, "http://teaos.cn/", body)
 	if err != nil {