修复统计指标数据上传不完整的问题

.gitignore

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

build/build.sh

@@ -7,6 +7,7 @@ function build() {
 	DIST=$ROOT/"../dist/${NAME}"
 	OS=${1}
 	ARCH=${2}
+	TAG=${3}
 
 	if [ -z $OS ]; then
 		echo "usage: build.sh OS ARCH"
@@ -16,6 +17,9 @@ function build() {
 		echo "usage: build.sh OS ARCH"
 		exit
 	fi
+	if [ -z $TAG ]; then
+		TAG="community"
+	fi
 
 	echo "checking ..."
 	ZIP_PATH=$(which zip)
@@ -24,8 +28,8 @@ function build() {
 		exit
 	fi
 
-	echo "building v${VERSION}/${OS}/${ARCH} ..."
+	echo "building v${VERSION}/${OS}/${ARCH}/${TAG} ..."
-	ZIP="${NAME}-${OS}-${ARCH}-v${VERSION}.zip"
+	ZIP="${NAME}-${OS}-${ARCH}-${TAG}-v${VERSION}.zip"
 
 	echo "copying ..."
 	if [ ! -d $DIST ]; then
@@ -80,9 +84,9 @@ function build() {
 		fi
 	fi
 	if [ ! -z $CC_PATH ]; then
-		env CC=$MUSL_DIR/$CC_PATH CXX=$MUSL_DIR/$CXX_PATH GOOS=${OS} GOARCH=${ARCH} CGO_ENABLED=1 go build -o $DIST/bin/${NAME} -ldflags "-linkmode external -extldflags -static -s -w" $ROOT/../cmd/edge-node/main.go
+		env CC=$MUSL_DIR/$CC_PATH CXX=$MUSL_DIR/$CXX_PATH GOOS=${OS} GOARCH=${ARCH} CGO_ENABLED=1 go build -tags $TAG -o $DIST/bin/${NAME} -ldflags "-linkmode external -extldflags -static -s -w" $ROOT/../cmd/edge-node/main.go
 	else
-		env GOOS=${OS} GOARCH=${ARCH} CGO_ENABLED=1 go build -o $DIST/bin/${NAME} -ldflags="-s -w" $ROOT/../cmd/edge-node/main.go
+		env GOOS=${OS} GOARCH=${ARCH} CGO_ENABLED=1 go build -tags $TAG -o $DIST/bin/${NAME} -ldflags="-s -w" $ROOT/../cmd/edge-node/main.go
 	fi
 
 	# delete hidden files
@@ -114,4 +118,4 @@ function lookup-version() {
 	fi
 }
 
-build $1 $2
+build $1 $2 $3

cmd/edge-node/main.go

@@ -5,15 +5,12 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/apps"
 	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
 	"github.com/TeaOSLab/EdgeNode/internal/nodes"
-	"github.com/iwind/TeaGo/Tea"
 	_ "github.com/iwind/TeaGo/bootstrap"
 	"github.com/iwind/TeaGo/logs"
-	"github.com/iwind/TeaGo/types"
+	"github.com/iwind/gosock/pkg/gosock"
-	"io/ioutil"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
-	"syscall"
 )
 
 func main() {
@@ -40,25 +37,13 @@ func main() {
 		fmt.Println("done")
 	})
 	app.On("quit", func() {
-		pidFile := Tea.Root + "/bin/pid"
+		var sock = gosock.NewTmpSock(teaconst.ProcessName)
-		data, err := ioutil.ReadFile(pidFile)
+		_, err := sock.Send(&gosock.Command{Code: "quit"})
 		if err != nil {
 			fmt.Println("[ERROR]quit failed: " + err.Error())
 			return
 		}
-		pid := types.Int(string(data))
+		fmt.Println("done")
-		if pid == 0 {
-			fmt.Println("[ERROR]quit failed: pid=0")
-			return
-		}
-
-		process, err := os.FindProcess(pid)
-		if err != nil {
-			return
-		}
-		if process != nil {
-			_ = process.Signal(syscall.SIGQUIT)
-		}
 	})
 	app.On("pprof", func() {
 		// TODO 自己指定端口

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/golang/protobuf v1.5.2
 	github.com/iwind/TeaGo v0.0.0-20210628135026-38575a4ab060
 	github.com/iwind/gofcgi v0.0.0-20210528023741-a92711d45f11
+	github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3
 	github.com/lionsoul2014/ip2region v2.2.0-release+incompatible
 	github.com/mattn/go-sqlite3 v2.0.3+incompatible
 	github.com/mssola/user_agent v0.5.2

go.sum

@@ -66,8 +66,9 @@ github.com/iwind/TeaGo v0.0.0-20210628135026-38575a4ab060 h1:qdLtK4PDXxk2vMKkTWl
 github.com/iwind/TeaGo v0.0.0-20210628135026-38575a4ab060/go.mod h1:KU4mS7QNiZ7QWEuDBk1zw0/Q2LrAPZv3tycEFBsuUwc=
 github.com/iwind/gofcgi v0.0.0-20210528023741-a92711d45f11 h1:DaQjoWZhLNxjhIXedVg4/vFEtHkZhK4IjIwsWdyzBLg=
 github.com/iwind/gofcgi v0.0.0-20210528023741-a92711d45f11/go.mod h1:JtbX20untAjUVjZs1ZBtq80f5rJWvwtQNRL6EnuYRnY=
+github.com/iwind/gosock v0.0.0-20210722083328-12b2d66abec3 h1:aBSonas7vFcgTj9u96/bWGILGv1ZbUSTLiOzcI1ZT6c=
+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/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
@@ -77,9 +78,7 @@ github.com/lionsoul2014/ip2region v2.2.0-release+incompatible/go.mod h1:+ZBN7PBo
 github.com/mattn/go-sqlite3 v2.0.3+incompatible h1:gXHsfypPkaMZrKbD5209QV9jbUTJKjyR5WD3HYQSd+U=
 github.com/mattn/go-sqlite3 v2.0.3+incompatible/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/modern-go/reflect2 v1.0.1/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=

internal/apps/app_cmd.go

@@ -2,8 +2,11 @@ package apps
 
 import (
 	"fmt"
-	"github.com/iwind/TeaGo/Tea"
+	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
 	"github.com/iwind/TeaGo/logs"
+	"github.com/iwind/TeaGo/maps"
+	"github.com/iwind/TeaGo/types"
+	"github.com/iwind/gosock/pkg/gosock"
 	"os"
 	"os/exec"
 	"runtime"
@@ -11,7 +14,7 @@ import (
 	"time"
 )
 
-// App命令帮助
+// AppCmd App命令帮助
 type AppCmd struct {
 	product       string
 	version       string
@@ -20,10 +23,14 @@ type AppCmd struct {
 	appendStrings []string
 
 	directives []*Directive
+
+	sock *gosock.Sock
 }
 
 func NewAppCmd() *AppCmd {
-	return &AppCmd{}
+	return &AppCmd{
+		sock: gosock.NewTmpSock(teaconst.ProcessName),
+	}
 }
 
 type CommandHelpOption struct {
@@ -31,25 +38,25 @@ type CommandHelpOption struct {
 	Description string
 }
 
-// 产品
+// Product 产品
 func (this *AppCmd) Product(product string) *AppCmd {
 	this.product = product
 	return this
 }
 
-// 版本
+// Version 版本
 func (this *AppCmd) Version(version string) *AppCmd {
 	this.version = version
 	return this
 }
 
-// 使用方法
+// Usage 使用方法
 func (this *AppCmd) Usage(usage string) *AppCmd {
 	this.usage = usage
 	return this
 }
 
-// 选项
+// Option 选项
 func (this *AppCmd) Option(code string, description string) *AppCmd {
 	this.options = append(this.options, &CommandHelpOption{
 		Code:        code,
@@ -58,13 +65,13 @@ func (this *AppCmd) Option(code string, description string) *AppCmd {
 	return this
 }
 
-// 附加内容
+// Append 附加内容
 func (this *AppCmd) Append(appendString string) *AppCmd {
 	this.appendStrings = append(this.appendStrings, appendString)
 	return this
 }
 
-// 打印
+// Print 打印
 func (this *AppCmd) Print() {
 	fmt.Println(this.product + " v" + this.version)
 
@@ -103,7 +110,7 @@ func (this *AppCmd) Print() {
 	}
 }
 
-// 添加指令
+// On 添加指令
 func (this *AppCmd) On(arg string, callback func()) {
 	this.directives = append(this.directives, &Directive{
 		Arg:      arg,
@@ -111,7 +118,7 @@ func (this *AppCmd) On(arg string, callback func()) {
 	})
 }
 
-// 运行
+// Run 运行
 func (this *AppCmd) Run(main func()) {
 	// 获取参数
 	args := os.Args[1:]
@@ -161,7 +168,7 @@ func (this *AppCmd) Run(main func()) {
 
 // 版本号
 func (this *AppCmd) runVersion() {
-	fmt.Println(this.product+" v"+this.version, "(build: "+runtime.Version(), runtime.GOOS, runtime.GOARCH+")")
+	fmt.Println(this.product+" v"+this.version, "(build: "+runtime.Version(), runtime.GOOS, runtime.GOARCH, teaconst.Tag+")")
 }
 
 // 帮助
@@ -171,9 +178,9 @@ func (this *AppCmd) runHelp() {
 
 // 启动
 func (this *AppCmd) runStart() {
-	proc := this.checkPid()
+	var pid = this.getPID()
-	if proc != nil {
+	if pid > 0 {
-		fmt.Println(this.product+" already started, pid:", proc.Pid)
+		fmt.Println(this.product+" already started, pid:", pid)
 		return
 	}
 
@@ -189,18 +196,15 @@ func (this *AppCmd) runStart() {
 
 // 停止
 func (this *AppCmd) runStop() {
-	proc := this.checkPid()
+	var pid = this.getPID()
-	if proc == nil {
+	if pid == 0 {
 		fmt.Println(this.product + " not started yet")
 		return
 	}
 
-	// 停止进程
+	_, _ = this.sock.Send(&gosock.Command{Code: "stop"})
-	_ = proc.Kill()
 
-	// 在Windows上经常不能及时释放资源
+	fmt.Println(this.product+" stopped ok, pid:", types.String(pid))
-	_ = DeletePid(Tea.Root + "/bin/pid")
-	fmt.Println(this.product+" stopped ok, pid:", proc.Pid)
 }
 
 // 重启
@@ -212,15 +216,24 @@ func (this *AppCmd) runRestart() {
 
 // 状态
 func (this *AppCmd) runStatus() {
-	proc := this.checkPid()
+	var pid = this.getPID()
-	if proc == nil {
+	if pid == 0 {
 		fmt.Println(this.product + " not started yet")
-	} else {
+		return
-		fmt.Println(this.product + " is running, pid: " + fmt.Sprintf("%d", proc.Pid))
 	}
+
+	fmt.Println(this.product + " is running, pid: " + types.String(pid))
 }
 
-// 检查PID
+// 获取当前的PID
-func (this *AppCmd) checkPid() *os.Process {
+func (this *AppCmd) getPID() int {
-	return CheckPid(Tea.Root + "/bin/pid")
+	if !this.sock.IsListening() {
+		return 0
+	}
+
+	reply, err := this.sock.Send(&gosock.Command{Code: "pid"})
+	if err != nil {
+		return 0
+	}
+	return maps.NewMap(reply.Params).GetInt("pid")
 }

internal/apps/file_others.go

@@ -1,17 +0,0 @@
-// +build !windows
-
-package apps
-
-import (
-	"os"
-	"syscall"
-)
-
-// lock file
-func LockFile(fp *os.File) error {
-	return syscall.Flock(int(fp.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
-}
-
-func UnlockFile(fp *os.File) error {
-	return syscall.Flock(int(fp.Fd()), syscall.LOCK_UN)
-}

internal/apps/file_windows.go

@@ -1,17 +0,0 @@
-// +build windows
-
-package apps
-
-import (
-	"errors"
-	"os"
-)
-
-// lock file
-func LockFile(fp *os.File) error {
-	return errors.New("not implemented on windows")
-}
-
-func UnlockFile(fp *os.File) error {
-	return errors.New("not implemented on windows")
-}

internal/apps/pid.go

@@ -1,119 +0,0 @@
-package apps
-
-import (
-	"fmt"
-	"github.com/TeaOSLab/EdgeNode/internal/events"
-	"github.com/iwind/TeaGo/Tea"
-	"github.com/iwind/TeaGo/types"
-	"io/ioutil"
-	"os"
-	"runtime"
-)
-
-var pidFileList = []*os.File{}
-
-// 检查Pid
-func CheckPid(path string) *os.Process {
-	// windows上打开的文件是不能删除的
-	if runtime.GOOS == "windows" {
-		if os.Remove(path) == nil {
-			return nil
-		}
-	}
-
-	file, err := os.Open(path)
-	if err != nil {
-		return nil
-	}
-
-	defer func() {
-		_ = file.Close()
-	}()
-
-	// 是否能取得Lock
-	err = LockFile(file)
-	if err == nil {
-		_ = UnlockFile(file)
-		return nil
-	}
-
-	pidBytes, err := ioutil.ReadAll(file)
-	if err != nil {
-		return nil
-	}
-	pid := types.Int(string(pidBytes))
-
-	if pid <= 0 {
-		return nil
-	}
-
-	proc, _ := os.FindProcess(pid)
-	return proc
-}
-
-// 写入Pid
-func WritePid() error {
-	path := Tea.Root + "/bin/pid"
-	fp, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY|os.O_RDONLY, 0666)
-	if err != nil {
-		return err
-	}
-	events.On(events.EventQuit, func() {
-		_ = fp.Close()
-	})
-
-	if runtime.GOOS != "windows" {
-		err = LockFile(fp)
-		if err != nil {
-			return err
-		}
-	}
-	pidFileList = append(pidFileList, fp) // hold the file pointers
-
-	_, err = fp.WriteString(fmt.Sprintf("%d", os.Getpid()))
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// 写入Ppid
-func WritePpid(path string) error {
-	fp, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY|os.O_RDONLY, 0666)
-	if err != nil {
-		return err
-	}
-
-	if runtime.GOOS != "windows" {
-		err = LockFile(fp)
-		if err != nil {
-			return err
-		}
-	}
-	pidFileList = append(pidFileList, fp) // hold the file pointers
-
-	_, err = fp.WriteString(fmt.Sprintf("%d", os.Getppid()))
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// 删除Pid
-func DeletePid(path string) error {
-	_, err := os.Stat(path)
-	if err != nil {
-		if !os.IsNotExist(err) {
-			return nil
-		}
-		return err
-	}
-
-	for _, fp := range pidFileList {
-		_ = UnlockFile(fp)
-		_ = fp.Close()
-	}
-	return os.Remove(path)
-}

internal/caches/list_file.go

@@ -54,7 +54,12 @@ func (this *FileList) Init() error {
 
 	this.itemsTableName = "cacheItems_v2"
 
-	db, err := sql.Open("sqlite3", "file:"+this.dir+"/index.db?cache=shared&mode=rwc&_journal_mode=WAL")
+	var dir = this.dir
+	if dir == "/" {
+		// 防止sqlite提示authority错误
+		dir = ""
+	}
+	db, err := sql.Open("sqlite3", "file:"+dir+"/index.db?cache=shared&mode=rwc&_journal_mode=WAL")
 	if err != nil {
 		return err
 	}

internal/const/build.go

@@ -0,0 +1,8 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+// +build community
+
+package teaconst
+
+const BuildCommunity = true
+const BuildPlus = false
+const Tag = "community"

internal/const/build_plus.go

@@ -0,0 +1,8 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+// +build plus
+
+package teaconst
+
+const BuildCommunity = false
+const BuildPlus = true
+const Tag = "plus"

internal/const/const.go

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

internal/iplibrary/library_interface.go

@@ -1,12 +1,13 @@
 package iplibrary
 
 type LibraryInterface interface {
-	// 加载数据库文件
+	// Load 加载数据库文件
 	Load(dbPath string) error
 
-	// 查询IP
+	// Lookup 查询IP
+	// 返回结果有可能为空
 	Lookup(ip string) (*Result, error)
 
-	// 关闭数据库文件
+	// Close 关闭数据库文件
 	Close()
 }

internal/metrics/task.go

@@ -131,7 +131,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
 	}
@@ -354,8 +354,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 +362,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/http_request.go

@@ -144,9 +144,10 @@ func (this *HTTPRequest) Do() {
 
 	// 自动跳转到HTTPS
 	if this.IsHTTP && this.web.RedirectToHttps != nil && this.web.RedirectToHttps.IsOn {
-		this.doRedirectToHTTPS(this.web.RedirectToHttps)
+		if this.doRedirectToHTTPS(this.web.RedirectToHttps) {
-		this.doEnd()
+			this.doEnd()
-		return
+			return
+		}
 	}
 
 	// Gzip
@@ -569,6 +570,12 @@ func (this *HTTPRequest) Format(source string) string {
 			return this.Host
 		case "referer":
 			return this.RawReq.Referer()
+		case "referer.host":
+			u, err := url.Parse(this.RawReq.Referer())
+			if err == nil {
+				return u.Host
+			}
+			return ""
 		case "userAgent":
 			return this.RawReq.UserAgent()
 		case "contentType":

internal/nodes/http_request_cache.go

@@ -12,7 +12,7 @@ import (
 
 // 读取缓存
 func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
-	cachePolicy := sharedNodeConfig.HTTPCachePolicy
+	cachePolicy := this.Server.HTTPCachePolicy
 	if cachePolicy == nil || !cachePolicy.IsOn {
 		return
 	}

internal/nodes/http_request_redirect_https.go

@@ -7,9 +7,14 @@ import (
 	"strings"
 )
 
-func (this *HTTPRequest) doRedirectToHTTPS(redirectToHTTPSConfig *serverconfigs.HTTPRedirectToHTTPSConfig) {
+func (this *HTTPRequest) doRedirectToHTTPS(redirectToHTTPSConfig *serverconfigs.HTTPRedirectToHTTPSConfig) (shouldBreak bool) {
 	host := this.RawReq.Host
 
+	// 检查域名是否匹配
+	if !redirectToHTTPSConfig.MatchDomain(host) {
+		return false
+	}
+
 	if len(redirectToHTTPSConfig.Host) > 0 {
 		if redirectToHTTPSConfig.Port > 0 && redirectToHTTPSConfig.Port != 443 {
 			host = redirectToHTTPSConfig.Host + ":" + strconv.Itoa(redirectToHTTPSConfig.Port)
@@ -38,4 +43,6 @@ func (this *HTTPRequest) doRedirectToHTTPS(redirectToHTTPSConfig *serverconfigs.
 
 	newURL := "https://" + host + this.RawReq.RequestURI
 	http.Redirect(this.writer, this.RawReq, newURL, statusCode)
+
+	return true
 }

internal/nodes/http_request_reverse_proxy.go

@@ -160,6 +160,9 @@ func (this *HTTPRequest) doReverseProxy() {
 		httpErr, ok := err.(*url.Error)
 		if !ok || httpErr.Err != context.Canceled {
 			// TODO 如果超过最大失败次数，则下线
+			SharedOriginStateManager.Fail(origin, this.reverseProxy, func() {
+				this.reverseProxy.ResetScheduling()
+			})
 
 			this.write502(err)
 			remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
@@ -183,6 +186,11 @@ func (this *HTTPRequest) doReverseProxy() {
 		}
 		return
 	}
+	if !origin.IsOk {
+		SharedOriginStateManager.Success(origin, func() {
+			this.reverseProxy.ResetScheduling()
+		})
+	}
 
 	// WAF对出站进行检查
 	if this.web.FirewallRef != nil && this.web.FirewallRef.IsOn {

internal/nodes/http_request_waf.go

@@ -28,8 +28,8 @@ func (this *HTTPRequest) doWAFRequest() (blocked bool) {
 	}
 
 	// 公用的防火墙设置
-	if sharedNodeConfig.HTTPFirewallPolicy != nil {
+	if this.Server.HTTPFirewallPolicy != nil && this.Server.HTTPFirewallPolicy.IsOn {
-		blocked, breakChecking := this.checkWAFRequest(sharedNodeConfig.HTTPFirewallPolicy)
+		blocked, breakChecking := this.checkWAFRequest(this.Server.HTTPFirewallPolicy)
 		if blocked {
 			return true
 		}
@@ -201,8 +201,8 @@ func (this *HTTPRequest) doWAFResponse(resp *http.Response) (blocked bool) {
 	}
 
 	// 公用的防火墙设置
-	if sharedNodeConfig.HTTPFirewallPolicy != nil {
+	if this.Server.HTTPFirewallPolicy != nil && this.Server.HTTPFirewallPolicy.IsOn {
-		blocked := this.checkWAFResponse(sharedNodeConfig.HTTPFirewallPolicy, resp)
+		blocked := this.checkWAFResponse(this.Server.HTTPFirewallPolicy, resp)
 		if blocked {
 			return true
 		}

internal/nodes/http_writer.go

@@ -342,7 +342,7 @@ func (this *HTTPWriter) prepareCache(size int64) {
 		return
 	}
 
-	cachePolicy := sharedNodeConfig.HTTPCachePolicy
+	cachePolicy := this.req.Server.HTTPCachePolicy
 	if cachePolicy == nil || !cachePolicy.IsOn {
 		return
 	}

internal/nodes/node.go

@@ -6,7 +6,6 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
-	"github.com/TeaOSLab/EdgeNode/internal/apps"
 	"github.com/TeaOSLab/EdgeNode/internal/caches"
 	"github.com/TeaOSLab/EdgeNode/internal/configs"
 	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
@@ -21,14 +20,13 @@ import (
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/lists"
 	"github.com/iwind/TeaGo/logs"
+	"github.com/iwind/TeaGo/maps"
+	"github.com/iwind/gosock/pkg/gosock"
 	"io/ioutil"
 	"log"
-	"net"
 	"os"
 	"os/exec"
-	"os/signal"
 	"runtime"
-	"syscall"
 	"time"
 )
 
@@ -40,10 +38,13 @@ var DaemonPid = 0
 // Node 节点
 type Node struct {
 	isLoaded bool
+	sock     *gosock.Sock
 }
 
 func NewNode() *Node {
-	return &Node{}
+	return &Node{
+		sock: gosock.NewTmpSock(teaconst.ProcessName),
+	}
 }
 
 // Test 检查配置
@@ -73,9 +74,6 @@ func (this *Node) Start() {
 	// 启动事件
 	events.Notify(events.EventStart)
 
-	// 处理信号
-	this.listenSignals()
-
 	// 本地Sock
 	err := this.listenSock()
 	if err != nil {
@@ -152,24 +150,16 @@ func (this *Node) Start() {
 		return
 	}
 
-	// 写入PID
-	err = apps.WritePid()
-	if err != nil {
-		remotelogs.Error("NODE", "write pid failed: "+err.Error())
-		return
-	}
-
 	// hold住进程
 	select {}
 }
 
 // Daemon 实现守护进程
 func (this *Node) Daemon() {
-	path := os.TempDir() + "/edge-node.sock"
 	isDebug := lists.ContainsString(os.Args, "debug")
 	isDebug = true
 	for {
-		conn, err := net.DialTimeout("unix", path, 1*time.Second)
+		conn, err := this.sock.Dial()
 		if err != nil {
 			if isDebug {
 				log.Println("[DAEMON]starting ...")
@@ -229,32 +219,6 @@ func (this *Node) InstallSystemService() error {
 	return nil
 }
 
-// 处理信号
-func (this *Node) listenSignals() {
-	signals := make(chan os.Signal)
-	signal.Notify(signals, syscall.SIGQUIT)
-	go func() {
-		for s := range signals {
-			switch s {
-			case syscall.SIGQUIT:
-				events.Notify(events.EventQuit)
-
-				// 监控连接数，如果连接数为0，则退出进程
-				go func() {
-					for {
-						countActiveConnections := sharedListenerManager.TotalActiveConnections()
-						if countActiveConnections <= 0 {
-							os.Exit(0)
-							return
-						}
-						time.Sleep(1 * time.Second)
-					}
-				}()
-			}
-		}
-	}()
-}
-
 // 循环
 func (this *Node) loop() error {
 	// 检查api.yaml是否存在
@@ -387,8 +351,8 @@ func (this *Node) syncConfig() error {
 	nodeconfigs.ResetNodeConfig(nodeConfig)
 	caches.SharedManager.MaxDiskCapacity = nodeConfig.MaxCacheDiskCapacity
 	caches.SharedManager.MaxMemoryCapacity = nodeConfig.MaxCacheMemoryCapacity
-	if nodeConfig.HTTPCachePolicy != nil {
+	if len(nodeConfig.HTTPCachePolicies) > 0 {
-		caches.SharedManager.UpdatePolicies([]*serverconfigs.HTTPCachePolicy{nodeConfig.HTTPCachePolicy})
+		caches.SharedManager.UpdatePolicies(nodeConfig.HTTPCachePolicies)
 	} else {
 		caches.SharedManager.UpdatePolicies([]*serverconfigs.HTTPCachePolicy{})
 	}
@@ -493,37 +457,63 @@ func (this *Node) checkClusterConfig() error {
 
 // 监听本地sock
 func (this *Node) listenSock() error {
-	path := os.TempDir() + "/edge-node.sock"
+	// 检查是否在运行
-
+	if this.sock.IsListening() {
-	// 检查是否已经存在
+		reply, err := this.sock.Send(&gosock.Command{Code: "pid"})
-	_, err := os.Stat(path)
+		if err == nil {
-	if err == nil {
+			return errors.New("error: the process is already running, pid: " + maps.NewMap(reply.Params).GetString("pid"))
-		conn, err := net.Dial("unix", path)
-		if err != nil {
-			_ = os.Remove(path)
 		} else {
-			_ = conn.Close()
+			return errors.New("error: the process is already running")
 		}
 	}
 
-	// 新的监听任务
+	// 启动监听
-	listener, err := net.Listen("unix", path)
-	if err != nil {
-		return err
-	}
-	events.On(events.EventQuit, func() {
-		remotelogs.Println("NODE", "quit unix sock")
-		_ = listener.Close()
-	})
-
 	go func() {
-		for {
+		this.sock.OnCommand(func(cmd *gosock.Command) {
-			_, err := listener.Accept()
+			switch cmd.Code {
-			if err != nil {
+			case "pid":
-				return
+				_ = cmd.Reply(&gosock.Command{
+					Code: "pid",
+					Params: map[string]interface{}{
+						"pid": os.Getpid(),
+					},
+				})
+			case "stop":
+				_ = cmd.ReplyOk()
+
+				// 退出主进程
+				events.Notify(events.EventQuit)
+				os.Exit(0)
+			case "quit":
+				_ = cmd.ReplyOk()
+				_ = this.sock.Close()
+
+				events.Notify(events.EventQuit)
+
+				// 监控连接数，如果连接数为0，则退出进程
+				go func() {
+					for {
+						countActiveConnections := sharedListenerManager.TotalActiveConnections()
+						if countActiveConnections <= 0 {
+							os.Exit(0)
+							return
+						}
+						time.Sleep(1 * time.Second)
+					}
+				}()
 			}
+		})
+
+		err := this.sock.Listen()
+		if err != nil {
+			logs.Println("NODE", err.Error())
 		}
 	}()
 
+	events.On(events.EventQuit, func() {
+		logs.Println("NODE", "quit unix sock")
+		_ = this.sock.Close()
+	})
+
 	return nil
 }

internal/nodes/origin_state.go

@@ -0,0 +1,12 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package nodes
+
+import "github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+
+type OriginState struct {
+	CountFails   int64
+	UpdatedAt    int64
+	Config       *serverconfigs.OriginConfig
+	ReverseProxy *serverconfigs.ReverseProxyConfig
+}

internal/nodes/origin_state_manager.go

@@ -0,0 +1,174 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package nodes
+
+import (
+	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeNode/internal/events"
+	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
+	"github.com/iwind/TeaGo/Tea"
+	"sync"
+	"time"
+)
+
+var SharedOriginStateManager = NewOriginStateManager()
+
+func init() {
+	events.On(events.EventLoaded, func() {
+		go SharedOriginStateManager.Start()
+	})
+}
+
+// OriginStateManager 源站状态管理
+type OriginStateManager struct {
+	stateMap map[int64]*OriginState // originId => *OriginState
+
+	ticker *time.Ticker
+	locker sync.RWMutex
+}
+
+// NewOriginStateManager 获取新管理对象
+func NewOriginStateManager() *OriginStateManager {
+	return &OriginStateManager{
+		stateMap: map[int64]*OriginState{},
+		ticker:   time.NewTicker(60 * time.Second),
+	}
+}
+
+// Start 启动
+func (this *OriginStateManager) Start() {
+	events.On(events.EventReload, func() {
+		this.locker.Lock()
+		this.stateMap = map[int64]*OriginState{}
+		this.locker.Unlock()
+	})
+
+	if Tea.IsTesting() {
+		this.ticker = time.NewTicker(10 * time.Second)
+	}
+	for range this.ticker.C {
+		err := this.Loop()
+		if err != nil {
+			remotelogs.Error("ORIGIN_MANAGER", err.Error())
+		}
+	}
+}
+
+// Loop 单次循环检查
+func (this *OriginStateManager) Loop() error {
+	if sharedNodeConfig == nil {
+		return nil
+	}
+
+	var currentStates = []*OriginState{}
+	this.locker.Lock()
+	for originId, state := range this.stateMap {
+		// 检查Origin是否正在使用
+		config := sharedNodeConfig.FindOrigin(originId)
+		if config == nil || !config.IsOn {
+			delete(this.stateMap, originId)
+			continue
+		}
+		state.Config = config
+		currentStates = append(currentStates, state)
+	}
+	this.locker.Unlock()
+
+	if len(currentStates) == 0 {
+		return nil
+	}
+
+	var count = len(currentStates)
+	wg := &sync.WaitGroup{}
+	wg.Add(count)
+	for _, state := range currentStates {
+		go func(state *OriginState) {
+			defer wg.Done()
+			conn, err := OriginConnect(state.Config, "")
+			if err == nil {
+				_ = conn.Close()
+
+				// 已经恢复正常
+				this.locker.Lock()
+				state.Config.IsOk = true
+				delete(this.stateMap, state.Config.Id)
+				this.locker.Unlock()
+
+				var reverseProxy = state.ReverseProxy
+				if reverseProxy != nil {
+					reverseProxy.ResetScheduling()
+				}
+			}
+		}(state)
+	}
+	wg.Wait()
+
+	return nil
+}
+
+// Fail 添加失败的源站
+func (this *OriginStateManager) Fail(origin *serverconfigs.OriginConfig, reverseProxy *serverconfigs.ReverseProxyConfig, callback func()) {
+	if origin == nil {
+		return
+	}
+	this.locker.Lock()
+	state, ok := this.stateMap[origin.Id]
+	var timestamp = time.Now().Unix()
+	if ok {
+		if state.UpdatedAt < timestamp-300 { // N 分钟之后重新计数
+			state.CountFails = 0
+			state.Config.IsOk = true
+		}
+
+		state.CountFails++
+		state.Config = origin
+		state.ReverseProxy = reverseProxy
+		state.UpdatedAt = timestamp
+
+		if origin.IsOk {
+			origin.IsOk = state.CountFails > 5 // 超过 N 次之后认为是异常
+
+			if !origin.IsOk {
+				if callback != nil {
+					callback()
+				}
+			}
+		}
+	} else {
+		this.stateMap[origin.Id] = &OriginState{
+			CountFails:   1,
+			Config:       origin,
+			ReverseProxy: reverseProxy,
+			UpdatedAt:    timestamp,
+		}
+		origin.IsOk = true
+	}
+	this.locker.Unlock()
+}
+
+// Success 添加成功的源站
+func (this *OriginStateManager) Success(origin *serverconfigs.OriginConfig, callback func()) {
+	if origin == nil {
+		return
+	}
+
+	if !origin.IsOk {
+		if callback != nil {
+			defer callback()
+		}
+	}
+
+	origin.IsOk = true
+	this.locker.Lock()
+	delete(this.stateMap, origin.Id)
+	this.locker.Unlock()
+}
+
+// IsAvailable 检查是否正常
+func (this *OriginStateManager) IsAvailable(originId int64) bool {
+	this.locker.RLock()
+	_, ok := this.stateMap[originId]
+	this.locker.RUnlock()
+
+	return !ok
+}

internal/nodes/origin_state_manager_test.go

@@ -0,0 +1,15 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package nodes
+
+import "testing"
+
+func TestOriginManager_Loop(t *testing.T) {
+	var manager = NewOriginStateManager()
+	err := manager.Loop()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Log(manager.stateMap)
+}

internal/nodes/origin_utils.go

@@ -16,41 +16,44 @@ func OriginConnect(origin *serverconfigs.OriginConfig, remoteAddr string) (net.C
 	}
 
 	// 支持TOA的连接
-	toaConfig := sharedTOAManager.Config()
+	// 这个条件很重要，如果没有传递remoteAddr，表示不使用TOA
-	if toaConfig != nil && toaConfig.IsOn {
+	if len(remoteAddr) > 0 {
-		retries := 3
+		toaConfig := sharedTOAManager.Config()
-		for i := 1; i <= retries; i++ {
+		if toaConfig != nil && toaConfig.IsOn {
-			port := int(toaConfig.RandLocalPort())
+			retries := 3
-			err := sharedTOAManager.SendMsg("add:" + strconv.Itoa(port) + ":" + remoteAddr)
+			for i := 1; i <= retries; i++ {
-			if err != nil {
+				port := int(toaConfig.RandLocalPort())
-				remotelogs.Error("TOA", "add failed: "+err.Error())
+				err := sharedTOAManager.SendMsg("add:" + strconv.Itoa(port) + ":" + remoteAddr)
-			} else {
+				if err != nil {
-				dialer := net.Dialer{
+					remotelogs.Error("TOA", "add failed: "+err.Error())
-					Timeout: origin.ConnTimeoutDuration(),
+				} else {
-					LocalAddr: &net.TCPAddr{
+					dialer := net.Dialer{
-						Port: port,
+						Timeout: origin.ConnTimeoutDuration(),
-					},
+						LocalAddr: &net.TCPAddr{
-				}
+							Port: port,
-				var conn net.Conn
+						},
-				switch origin.Addr.Protocol {
+					}
-				case "", serverconfigs.ProtocolTCP, serverconfigs.ProtocolHTTP:
+					var conn net.Conn
-					// TODO 支持TCP4/TCP6
+					switch origin.Addr.Protocol {
-					// TODO 支持指定特定网卡
+					case "", serverconfigs.ProtocolTCP, serverconfigs.ProtocolHTTP:
-					// TODO Addr支持端口范围，如果有多个端口时，随机一个端口使用
+						// TODO 支持TCP4/TCP6
-					conn, err = dialer.Dial("tcp", origin.Addr.Host+":"+origin.Addr.PortRange)
+						// TODO 支持指定特定网卡
-				case serverconfigs.ProtocolTLS, serverconfigs.ProtocolHTTPS:
+						// TODO Addr支持端口范围，如果有多个端口时，随机一个端口使用
-					// TODO 支持TCP4/TCP6
+						conn, err = dialer.Dial("tcp", origin.Addr.Host+":"+origin.Addr.PortRange)
-					// TODO 支持指定特定网卡
+					case serverconfigs.ProtocolTLS, serverconfigs.ProtocolHTTPS:
-					// TODO Addr支持端口范围，如果有多个端口时，随机一个端口使用
+						// TODO 支持TCP4/TCP6
-					// TODO 支持使用证书
+						// TODO 支持指定特定网卡
-					conn, err = tls.DialWithDialer(&dialer, "tcp", origin.Addr.Host+":"+origin.Addr.PortRange, &tls.Config{
+						// TODO Addr支持端口范围，如果有多个端口时，随机一个端口使用
-						InsecureSkipVerify: true,
+						// TODO 支持使用证书
-					})
+						conn, err = tls.DialWithDialer(&dialer, "tcp", origin.Addr.Host+":"+origin.Addr.PortRange, &tls.Config{
-				}
+							InsecureSkipVerify: true,
+						})
+					}
 
-				// TODO 需要在合适的时机删除TOA记录
+					// TODO 需要在合适的时机删除TOA记录
-				if err == nil || i == retries {
+					if err == nil || i == retries {
-					return conn, err
+						return conn, err
+					}
 				}
 			}
 		}

internal/stats/http_request_stat_manager.go

@@ -165,7 +165,7 @@ Loop:
 			ip := ipString[atIndex+1:]
 			if iplibrary.SharedLibrary != nil {
 				result, err := iplibrary.SharedLibrary.Lookup(ip)
-				if err == nil {
+				if err == nil && result != nil {
 					this.cityMap[serverId+"@"+result.Country+"@"+result.Province+"@"+result.City]  ++
 
 					if len(result.ISP) > 0 {

internal/waf/action_get_302.go

@@ -65,7 +65,9 @@ func (this *Get302Action) Perform(waf *WAF, group *RuleGroup, set *RuleSet, requ
 	http.Redirect(writer, request.WAFRaw(), Get302Path+"?info="+url.QueryEscape(info), http.StatusFound)
 
 	// 关闭连接
-	_ = this.CloseConn(writer)
+	if request.WAFRaw().ProtoMajor == 1 {
+		_ = this.CloseConn(writer)
+	}
 
 	return true
 }

internal/waf/action_post_307.go

@@ -81,8 +81,9 @@ func (this *Post307Action) Perform(waf *WAF, group *RuleGroup, set *RuleSet, req
 
 	http.Redirect(writer, request.WAFRaw(), request.WAFRaw().URL.String(), http.StatusTemporaryRedirect)
 
-	// 关闭连接
+	if request.WAFRaw().ProtoMajor == 1 {
-	_ = this.CloseConn(writer)
+		_ = this.CloseConn(writer)
+	}
 
 	return true
 }