TRKJ/EdgeNode
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeNode.git synced 2025-11-04 07:40:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

在节点重新实现缓存策略和WAF策略

Browse Source
This commit is contained in:
刘祥超
2020-12-17 17:36:10 +08:00
parent 0e312f6707
commit 1ea63ab8c6
26 changed files with 151 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
internal/caches/manager.go
View File

@@ -2,7 +2,7 @@ package caches
import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/lists"
"strconv"
"sync"
@@ -36,7 +36,7 @@ func (this *Manager) UpdatePolicies(newPolicies []*serverconfigs.HTTPCachePolicy
// 停止旧有的
for _, oldPolicy := range this.policyMap {
if !lists.ContainsInt64(newPolicyIds, oldPolicy.Id) {
logs.Error("CACHE", "remove policy "+strconv.FormatInt(oldPolicy.Id, 10))
remotelogs.Error("CACHE", "remove policy "+strconv.FormatInt(oldPolicy.Id, 10))
delete(this.policyMap, oldPolicy.Id)
storage, ok := this.storageMap[oldPolicy.Id]
if ok {
@@ -50,13 +50,13 @@ func (this *Manager) UpdatePolicies(newPolicies []*serverconfigs.HTTPCachePolicy
for _, newPolicy := range newPolicies {
_, ok := this.policyMap[newPolicy.Id]
if !ok {
logs.Println("CACHE", "add policy "+strconv.FormatInt(newPolicy.Id, 10))
remotelogs.Println("CACHE", "add policy "+strconv.FormatInt(newPolicy.Id, 10))
}
// 初始化
err := newPolicy.Init()
if err != nil {
logs.Error("CACHE", "UpdatePolicies: init policy error: "+err.Error())
remotelogs.Error("CACHE", "UpdatePolicies: init policy error: "+err.Error())
continue
}
this.policyMap[newPolicy.Id] = newPolicy
@@ -68,19 +68,19 @@ func (this *Manager) UpdatePolicies(newPolicies []*serverconfigs.HTTPCachePolicy
if !ok {
storage := this.NewStorageWithPolicy(policy)
if storage == nil {
logs.Error("CACHE", "can not find storage type '"+policy.Type+"'")
remotelogs.Error("CACHE", "can not find storage type '"+policy.Type+"'")
continue
}
err := storage.Init()
if err != nil {
logs.Error("CACHE", "UpdatePolicies: init storage failed: "+err.Error())
remotelogs.Error("CACHE", "UpdatePolicies: init storage failed: "+err.Error())
continue
}
this.storageMap[policy.Id] = storage
} else {
// 检查policy是否有变化
if !storage.Policy().IsSame(policy) {
logs.Println("CACHE", "policy "+strconv.FormatInt(policy.Id, 10)+" changed")
remotelogs.Println("CACHE", "policy "+strconv.FormatInt(policy.Id, 10)+" changed")
// 停止老的
storage.Stop()
@@ -89,12 +89,12 @@ func (this *Manager) UpdatePolicies(newPolicies []*serverconfigs.HTTPCachePolicy
// 启动新的
storage := this.NewStorageWithPolicy(policy)
if storage == nil {
logs.Error("CACHE", "can not find storage type '"+policy.Type+"'")
remotelogs.Error("CACHE", "can not find storage type '"+policy.Type+"'")
continue
}
err := storage.Init()
if err != nil {
logs.Error("CACHE", "UpdatePolicies: init storage failed: "+err.Error())
remotelogs.Error("CACHE", "UpdatePolicies: init storage failed: "+err.Error())
continue
}
this.storageMap[policy.Id] = storage

10
internal/caches/storage_file.go
View File

@@ -7,7 +7,7 @@ import (
"fmt"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/types"
@@ -82,7 +82,7 @@ func (this *FileStorage) Init() error {
}
cost := time.Since(before).Seconds() * 1000
logs.Println("CACHE", "init policy "+strconv.FormatInt(this.policy.Id, 10)+", cost: "+fmt.Sprintf("%.2f", cost)+" ms, count: "+strconv.Itoa(count)+", size: "+fmt.Sprintf("%.3f", float64(size)/1024/1024)+" M")
remotelogs.Println("CACHE", "init policy "+strconv.FormatInt(this.policy.Id, 10)+", cost: "+fmt.Sprintf("%.2f", cost)+" ms, count: "+strconv.Itoa(count)+", size: "+fmt.Sprintf("%.3f", float64(size)/1024/1024)+" M")
}()
// 配置
@@ -546,7 +546,7 @@ func (this *FileStorage) initList() error {
item, err := this.decodeFile(path)
if err != nil {
if err != ErrNotFound {
logs.Error("CACHE", "decode path '"+path+"': "+err.Error())
remotelogs.Error("CACHE", "decode path '"+path+"': "+err.Error())
}
continue
}
@@ -559,7 +559,7 @@ func (this *FileStorage) initList() error {
// 启动定时清理任务
this.ticker = utils.NewTicker(30 * time.Second)
events.On(events.EventQuit, func() {
logs.Println("CACHE", "quit clean timer")
remotelogs.Println("CACHE", "quit clean timer")
var ticker = this.ticker
if ticker != nil {
ticker.Stop()
@@ -642,7 +642,7 @@ func (this *FileStorage) purgeLoop() {
path := this.hashPath(hash)
err := os.Remove(path)
if err != nil && !os.IsNotExist(err) {
logs.Error("CACHE", "purge '"+path+"' error: "+err.Error())
remotelogs.Error("CACHE", "purge '"+path+"' error: "+err.Error())
}
})
}

8
internal/iplibrary/manager_country.go
View File

@@ -6,7 +6,7 @@ import (
"fmt"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/Tea"
@@ -46,13 +46,13 @@ func (this *CountryManager) Start() {
// 从缓存中读取
err := this.load()
if err != nil {
logs.Error("COUNTRY_MANAGER", err.Error())
remotelogs.Error("COUNTRY_MANAGER", err.Error())
}
// 第一次更新
err = this.loop()
if err != nil {
logs.Error("COUNTRY_MANAGER", err.Error())
remotelogs.Error("COUNTRY_MANAGER", err.Error())
}
// 定时更新
@@ -63,7 +63,7 @@ func (this *CountryManager) Start() {
for range ticker.C {
err := this.loop()
if err != nil {
logs.Error("COUNTRY_MANAGER", err.Error())
remotelogs.Error("COUNTRY_MANAGER", err.Error())
}
}
}

6
internal/iplibrary/manager_ip_list.go
View File

@@ -3,7 +3,7 @@ package iplibrary
import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/iwind/TeaGo/Tea"
"sync"
@@ -46,7 +46,7 @@ func (this *IPListManager) Start() {
// 第一次读取
err := this.loop()
if err != nil {
logs.Println("IP_LIST_MANAGER", err.Error())
remotelogs.Println("IP_LIST_MANAGER", err.Error())
}
ticker := time.NewTicker(60 * time.Second) // TODO 未来改成可以手动触发IP变更事件
@@ -63,7 +63,7 @@ func (this *IPListManager) Start() {
if err != nil {
countErrors++
logs.Println("IP_LIST_MANAGER", err.Error())
remotelogs.Println("IP_LIST_MANAGER", err.Error())
// 连续错误小于3次的我们立即重试
if countErrors <= 3 {

8
internal/iplibrary/manager_province.go
View File

@@ -6,7 +6,7 @@ import (
"fmt"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/Tea"
@@ -50,13 +50,13 @@ func (this *ProvinceManager) Start() {
// 从缓存中读取
err := this.load()
if err != nil {
logs.Error("PROVINCE_MANAGER", err.Error())
remotelogs.Error("PROVINCE_MANAGER", err.Error())
}
// 第一次更新
err = this.loop()
if err != nil {
logs.Error("PROVINCE_MANAGER", err.Error())
remotelogs.Error("PROVINCE_MANAGER", err.Error())
}
// 定时更新
@@ -67,7 +67,7 @@ func (this *ProvinceManager) Start() {
for range ticker.C {
err := this.loop()
if err != nil {
logs.Error("PROVINCE_MANAGER", err.Error())
remotelogs.Error("PROVINCE_MANAGER", err.Error())
}
}
}

8
internal/nodes/api_stream.go
View File

@@ -10,7 +10,7 @@ import (
"github.com/TeaOSLab/EdgeNode/internal/errors"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/iplibrary"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"io"
"net/http"
@@ -39,7 +39,7 @@ func (this *APIStream) Start() {
}
err := this.loop()
if err != nil {
logs.Error("API_STREAM", err.Error())
remotelogs.Error("API_STREAM", err.Error())
time.Sleep(10 * time.Second)
continue
}
@@ -102,7 +102,7 @@ func (this *APIStream) loop() error {
err = this.handleUnknownMessage(message)
}
if err != nil {
logs.Error("API_STREAM", "handle message failed: "+err.Error())
remotelogs.Error("API_STREAM", "handle message failed: "+err.Error())
}
}
@@ -130,7 +130,7 @@ func (this *APIStream) handleConnectedAPINode(message *pb.NodeStreamMessage) err
if err != nil {
return errors.Wrap(err)
}
logs.Println("API_STREAM", "connected to api node '"+strconv.FormatInt(msg.APINodeId, 10)+"'")
remotelogs.Println("API_STREAM", "connected to api node '"+strconv.FormatInt(msg.APINodeId, 10)+"'")
return nil
}

4
internal/nodes/http_access_log_queue.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"time"
)
@@ -33,7 +33,7 @@ func (this *HTTPAccessLogQueue) Start() {
for range ticker.C {
err := this.loop()
if err != nil {
logs.Error("ACCESS_LOG_QUEUE", err.Error())
remotelogs.Error("ACCESS_LOG_QUEUE", err.Error())
}
}
}

4
internal/nodes/http_client_pool.go
View File

@@ -5,7 +5,7 @@ import (
"crypto/tls"
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"net"
"net/http"
"runtime"
@@ -107,7 +107,7 @@ func (this *HTTPClientPool) Client(req *http.Request, origin *serverconfigs.Orig
// TODO 思考是否支持X-Real-IP/X-Forwarded-IP
err := sharedTOAManager.SendMsg("add:" + strconv.Itoa(port) + ":" + req.RemoteAddr)
if err != nil {
logs.Error("TOA", "add failed: "+err.Error())
remotelogs.Error("TOA", "add failed: "+err.Error())
} else {
dialer := net.Dialer{
Timeout: connectionTimeout,

3
internal/nodes/http_request.go
View File

@@ -107,7 +107,7 @@ func (this *HTTPRequest) Do() {
}
// WAF
if this.web.FirewallRef != nil && this.web.FirewallRef.IsOn && this.web.FirewallPolicy != nil && this.web.FirewallPolicy.IsOn {
if this.web.FirewallRef != nil && this.web.FirewallRef.IsOn {
if this.doWAFRequest() {
this.doEnd()
return
@@ -292,7 +292,6 @@ func (this *HTTPRequest) configureWeb(web *serverconfigs.HTTPWebConfig, isTop bo
// waf
if web.FirewallRef != nil && (web.FirewallRef.IsPrior || isTop) {
this.web.FirewallRef = web.FirewallRef
this.web.FirewallPolicy = web.FirewallPolicy
}
// access log

6
internal/nodes/http_request_acme.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"net/http"
"path/filepath"
@@ -15,13 +15,13 @@ func (this *HTTPRequest) doACME() {
rpcClient, err := rpc.SharedRPC()
if err != nil {
logs.Error("RPC", "[ACME]rpc failed: "+err.Error())
remotelogs.Error("RPC", "[ACME]rpc failed: "+err.Error())
return
}
keyResp, err := rpcClient.ACMEAuthenticationRPC().FindACMEAuthenticationKeyWithToken(rpcClient.Context(), &pb.FindACMEAuthenticationKeyWithTokenRequest{Token: token})
if err != nil {
logs.Error("RPC", "[ACME]read key for token failed: "+err.Error())
remotelogs.Error("RPC", "[ACME]read key for token failed: "+err.Error())
return
}
if len(keyResp.Key) == 0 {

20
internal/nodes/http_request_cache.go
View File

@@ -3,7 +3,7 @@ package nodes
import (
"bytes"
"github.com/TeaOSLab/EdgeNode/internal/caches"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/types"
"net/http"
"strconv"
@@ -15,12 +15,14 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
return
}
cachePolicy := sharedNodeConfig.HTTPCachePolicy
if cachePolicy == nil || !cachePolicy.IsOn {
return
}
// 检查条件
for _, cacheRef := range this.web.Cache.CacheRefs {
if !cacheRef.IsOn ||
cacheRef.CachePolicyId == 0 ||
cacheRef.CachePolicy == nil ||
!cacheRef.CachePolicy.IsOn ||
cacheRef.Conds == nil ||
!cacheRef.Conds.HasRequestConds() {
continue
@@ -35,9 +37,9 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
}
// 相关变量
this.varMapping["cache.policy.name"] = this.cacheRef.CachePolicy.Name
this.varMapping["cache.policy.id"] = strconv.FormatInt(this.cacheRef.CachePolicy.Id, 10)
this.varMapping["cache.policy.type"] = this.cacheRef.CachePolicy.Type
this.varMapping["cache.policy.name"] = cachePolicy.Name
this.varMapping["cache.policy.id"] = strconv.FormatInt(cachePolicy.Id, 10)
this.varMapping["cache.policy.type"] = cachePolicy.Type
// Cache-Pragma
if this.cacheRef.EnableRequestCachePragma {
@@ -58,7 +60,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
this.cacheKey = key
// 读取缓存
storage := caches.SharedManager.FindStorageWithPolicy(this.cacheRef.CachePolicyId)
storage := caches.SharedManager.FindStorageWithPolicy(cachePolicy.Id)
if storage == nil {
this.cacheRef = nil
return
@@ -139,7 +141,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
return
}
logs.Error("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Error("REQUEST_CACHE", "read from cache failed: "+err.Error())
return
}

20
internal/nodes/http_request_reverse_proxy.go
View File

@@ -5,7 +5,7 @@ import (
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/shared"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"io"
"net/url"
@@ -35,7 +35,7 @@ func (this *HTTPRequest) doReverseProxy() {
origin := this.reverseProxy.NextOrigin(requestCall)
if origin == nil {
err := errors.New(this.requestPath() + ": no available backends for reverse proxy")
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
this.write502(err)
return
}
@@ -55,7 +55,7 @@ func (this *HTTPRequest) doReverseProxy() {
// 处理Scheme
if origin.Addr == nil {
err := errors.New(this.requestPath() + ": origin '" + strconv.FormatInt(origin.Id, 10) + "' does not has a address")
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
this.write502(err)
return
}
@@ -142,7 +142,7 @@ func (this *HTTPRequest) doReverseProxy() {
// 获取请求客户端
client, err := SharedHTTPClientPool.Client(this.RawReq, origin, originAddr)
if err != nil {
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
this.write502(err)
return
}
@@ -156,7 +156,7 @@ func (this *HTTPRequest) doReverseProxy() {
// TODO 如果超过最大失败次数，则下线
this.write502(err)
logs.Println("REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
remotelogs.Println("REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
} else {
// 是否为客户端方面的错误
isClientError := false
@@ -179,11 +179,11 @@ func (this *HTTPRequest) doReverseProxy() {
}
// WAF对出站进行检查
if this.web.FirewallRef != nil && this.web.FirewallRef.IsOn && this.web.FirewallPolicy != nil && this.web.FirewallPolicy.IsOn {
if this.web.FirewallRef != nil && this.web.FirewallRef.IsOn {
if this.doWAFResponse(resp) {
err = resp.Body.Close()
if err != nil {
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
}
return
}
@@ -195,7 +195,7 @@ func (this *HTTPRequest) doReverseProxy() {
if len(this.web.Pages) > 0 && this.doPage(resp.StatusCode) {
err = resp.Body.Close()
if err != nil {
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
}
return
}
@@ -250,11 +250,11 @@ func (this *HTTPRequest) doReverseProxy() {
err1 := resp.Body.Close()
if err1 != nil {
logs.Error("REQUEST_REVERSE_PROXY", err1.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err1.Error())
}
if err != nil && err != io.EOF {
logs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
this.addError(err)
}
}

31
internal/nodes/http_request_waf.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeNode/internal/iplibrary"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/waf"
"github.com/iwind/TeaGo/lists"
"github.com/iwind/TeaGo/types"
@@ -11,14 +11,16 @@ import (
// 调用WAF
func (this *HTTPRequest) doWAFRequest() (blocked bool) {
firewallPolicy := sharedNodeConfig.HTTPFirewallPolicy
// 检查配置是否为空
if this.web.FirewallPolicy == nil || this.web.FirewallPolicy.Inbound == nil || !this.web.FirewallPolicy.Inbound.IsOn {
if firewallPolicy == nil || !firewallPolicy.IsOn || firewallPolicy.Inbound == nil || !firewallPolicy.Inbound.IsOn {
return
}
// 检查IP白名单
remoteAddr := this.requestRemoteAddr()
inbound := this.web.FirewallPolicy.Inbound
inbound := firewallPolicy.Inbound
if inbound.WhiteListRef != nil && inbound.WhiteListRef.IsOn && inbound.WhiteListRef.ListId > 0 {
list := iplibrary.SharedIPListManager.FindList(inbound.WhiteListRef.ListId)
if list != nil && list.Contains(iplibrary.IP2Long(remoteAddr)) {
@@ -43,12 +45,12 @@ func (this *HTTPRequest) doWAFRequest() (blocked bool) {
// 检查地区封禁
if iplibrary.SharedLibrary != nil {
if this.web.FirewallPolicy.Inbound.Region != nil && this.web.FirewallPolicy.Inbound.Region.IsOn {
regionConfig := this.web.FirewallPolicy.Inbound.Region
if firewallPolicy.Inbound.Region != nil && firewallPolicy.Inbound.Region.IsOn {
regionConfig := firewallPolicy.Inbound.Region
if regionConfig.IsNotEmpty() {
result, err := iplibrary.SharedLibrary.Lookup(remoteAddr)
if err != nil {
logs.Error("REQUEST", "iplibrary lookup failed: "+err.Error())
remotelogs.Error("REQUEST", "iplibrary lookup failed: "+err.Error())
} else if result != nil {
// 检查国家级别封禁
if len(regionConfig.DenyCountryIds) > 0 && len(result.Country) > 0 {
@@ -85,19 +87,19 @@ func (this *HTTPRequest) doWAFRequest() (blocked bool) {
}
// 规则测试
w := sharedWAFManager.FindWAF(this.web.FirewallPolicy.Id)
w := sharedWAFManager.FindWAF(firewallPolicy.Id)
if w == nil {
return
}
goNext, ruleGroup, ruleSet, err := w.MatchRequest(this.RawReq, this.writer)
if err != nil {
logs.Error("REQUEST", this.rawURI+": "+err.Error())
remotelogs.Error("REQUEST", this.rawURI+": "+err.Error())
return
}
if ruleSet != nil {
if ruleSet.Action != waf.ActionAllow {
this.firewallPolicyId = this.web.FirewallPolicy.Id
this.firewallPolicyId = firewallPolicy.Id
this.firewallRuleGroupId = types.Int64(ruleGroup.Id)
this.firewallRuleSetId = types.Int64(ruleSet.Id)
}
@@ -110,20 +112,25 @@ func (this *HTTPRequest) doWAFRequest() (blocked bool) {
// call response waf
func (this *HTTPRequest) doWAFResponse(resp *http.Response) (blocked bool) {
w := sharedWAFManager.FindWAF(this.web.FirewallPolicy.Id)
firewallPolicy := sharedNodeConfig.HTTPFirewallPolicy
if firewallPolicy == nil || !firewallPolicy.IsOn || !firewallPolicy.Outbound.IsOn {
return
}
w := sharedWAFManager.FindWAF(firewallPolicy.Id)
if w == nil {
return
}
goNext, ruleGroup, ruleSet, err := w.MatchResponse(this.RawReq, resp, this.writer)
if err != nil {
logs.Error("REQUEST", this.rawURI+": "+err.Error())
remotelogs.Error("REQUEST", this.rawURI+": "+err.Error())
return
}
if ruleSet != nil {
if ruleSet.Action != waf.ActionAllow {
this.firewallPolicyId = this.web.FirewallPolicy.Id
this.firewallPolicyId = firewallPolicy.Id
this.firewallRuleGroupId = types.Int64(ruleGroup.Id)
this.firewallRuleSetId = types.Int64(ruleSet.Id)
}

24
internal/nodes/http_writer.go
View File

@@ -6,7 +6,7 @@ import (
"compress/gzip"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/caches"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/lists"
"net"
@@ -115,7 +115,7 @@ func (this *HTTPWriter) Write(data []byte) (n int, err error) {
if err != nil {
_ = this.cacheWriter.Discard()
this.cacheWriter = nil
logs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
}
}
} else {
@@ -127,7 +127,7 @@ func (this *HTTPWriter) Write(data []byte) (n int, err error) {
if this.gzipBodyWriter != nil {
_, err := this.gzipBodyWriter.Write(data)
if err != nil {
logs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("REQUEST_WRITER", err.Error())
}
} else {
this.body = append(this.body, data...)
@@ -281,7 +281,7 @@ func (this *HTTPWriter) prepareGzip(size int64) {
var err error = nil
this.gzipWriter, err = gzip.NewWriterLevel(this.writer, int(this.gzipConfig.Level))
if err != nil {
logs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("REQUEST_WRITER", err.Error())
return
}
@@ -290,7 +290,7 @@ func (this *HTTPWriter) prepareGzip(size int64) {
this.gzipBodyBuffer = bytes.NewBuffer([]byte{})
this.gzipBodyWriter, err = gzip.NewWriterLevel(this.gzipBodyBuffer, int(this.gzipConfig.Level))
if err != nil {
logs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("REQUEST_WRITER", err.Error())
}
}
@@ -307,12 +307,16 @@ func (this *HTTPWriter) prepareCache(size int64) {
return
}
cachePolicy := sharedNodeConfig.HTTPCachePolicy
if cachePolicy == nil || !cachePolicy.IsOn {
return
}
cacheRef := this.req.cacheRef
if cacheRef == nil ||
cacheRef.CachePolicy == nil ||
!cacheRef.IsOn ||
(cacheRef.MaxSizeBytes() > 0 && size > cacheRef.MaxSizeBytes()) ||
(cacheRef.CachePolicy.MaxSizeBytes() > 0 && size > cacheRef.CachePolicy.MaxSizeBytes()) {
(cachePolicy.MaxSizeBytes() > 0 && size > cachePolicy.MaxSizeBytes()) {
return
}
@@ -345,7 +349,7 @@ func (this *HTTPWriter) prepareCache(size int64) {
}
// 打开缓存写入
storage := caches.SharedManager.FindStorageWithPolicy(this.req.cacheRef.CachePolicyId)
storage := caches.SharedManager.FindStorageWithPolicy(cachePolicy.Id)
if storage == nil {
return
}
@@ -357,7 +361,7 @@ func (this *HTTPWriter) prepareCache(size int64) {
expiredAt := utils.UnixTime() + life
cacheWriter, err := storage.Open(this.req.cacheKey, expiredAt)
if err != nil {
logs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
return
}
this.cacheWriter = cacheWriter
@@ -369,7 +373,7 @@ func (this *HTTPWriter) prepareCache(size int64) {
headerData := this.HeaderData()
_, err = cacheWriter.Write(headerData)
if err != nil {
logs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
_ = this.cacheWriter.Discard()
this.cacheWriter = nil
return

6
internal/nodes/listener.go
View File

@@ -5,7 +5,7 @@ import (
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"net"
"sync"
)
@@ -49,7 +49,7 @@ func (this *Listener) Listen() error {
return err
}
events.On(events.EventQuit, func() {
logs.Println("LISTENER", "quit "+this.group.FullAddr())
remotelogs.Println("LISTENER", "quit "+this.group.FullAddr())
_ = netListener.Close()
})
@@ -100,7 +100,7 @@ func (this *Listener) Listen() error {
}
// 打印其他错误
logs.Error("LISTENER", err.Error())
remotelogs.Error("LISTENER", err.Error())
}
}()

4
internal/nodes/listener_http.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"golang.org/x/net/http2"
"net"
"net/http"
@@ -63,7 +63,7 @@ func (this *HTTPListener) Serve() error {
// support http/2
err := http2.ConfigureServer(this.httpServer, nil)
if err != nil {
logs.Error("HTTP_LISTENER", "configure http2 error: "+err.Error())
remotelogs.Error("HTTP_LISTENER", "configure http2 error: "+err.Error())
}
err = this.httpServer.ServeTLS(this.Listener, "", "")

12
internal/nodes/listener_manager.go
View File

@@ -3,7 +3,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/lists"
"net/url"
"regexp"
@@ -51,7 +51,7 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
}
if len(availableServerGroups) == 0 {
logs.Println("LISTENER_MANAGER", "no available servers to startup")
remotelogs.Println("LISTENER_MANAGER", "no available servers to startup")
}
for _, group := range availableServerGroups {
@@ -63,7 +63,7 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
for listenerKey, listener := range this.listenersMap {
addr := listener.FullAddr()
if !lists.ContainsString(groupAddrs, addr) {
logs.Println("LISTENER_MANAGER", "close '"+addr+"'")
remotelogs.Println("LISTENER_MANAGER", "close '"+addr+"'")
_ = listener.Close()
delete(this.listenersMap, listenerKey)
@@ -75,15 +75,15 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
addr := group.FullAddr()
listener, ok := this.listenersMap[addr]
if ok {
logs.Println("LISTENER_MANAGER", "reload '"+this.prettyAddress(addr)+"'")
remotelogs.Println("LISTENER_MANAGER", "reload '"+this.prettyAddress(addr)+"'")
listener.Reload(group)
} else {
logs.Println("LISTENER_MANAGER", "listen '"+this.prettyAddress(addr)+"'")
remotelogs.Println("LISTENER_MANAGER", "listen '"+this.prettyAddress(addr)+"'")
listener = NewListener()
listener.Reload(group)
err := listener.Listen()
if err != nil {
logs.Error("LISTENER_MANAGER", err.Error())
remotelogs.Error("LISTENER_MANAGER", err.Error())
continue
}
this.listenersMap[addr] = listener

6
internal/nodes/listener_tcp.go
View File

@@ -4,7 +4,7 @@ import (
"crypto/tls"
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"net"
"sync/atomic"
)
@@ -32,7 +32,7 @@ func (this *TCPListener) Serve() error {
go func(conn net.Conn) {
err = this.handleConn(conn)
if err != nil {
logs.Error("TCP_LISTENER", err.Error())
remotelogs.Error("TCP_LISTENER", err.Error())
}
atomic.AddInt64(&this.countActiveConnections, -1)
}(conn)
@@ -122,7 +122,7 @@ func (this *TCPListener) connectOrigin(reverseProxy *serverconfigs.ReverseProxyC
}
conn, err = OriginConnect(origin, remoteAddr)
if err != nil {
logs.Error("TCP_LISTENER", "unable to connect origin: "+origin.Addr.Host+":"+origin.Addr.PortRange+": "+err.Error())
remotelogs.Error("TCP_LISTENER", "unable to connect origin: "+origin.Addr.Host+":"+origin.Addr.PortRange+": "+err.Error())
continue
} else {
return

40
internal/nodes/node.go
View File

@@ -5,11 +5,13 @@ import (
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/firewallconfigs"
"github.com/TeaOSLab/EdgeNode/internal/apps"
"github.com/TeaOSLab/EdgeNode/internal/caches"
"github.com/TeaOSLab/EdgeNode/internal/configs"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/go-yaml/yaml"
@@ -62,14 +64,14 @@ func (this *Node) Start() {
// 本地Sock
err := this.listenSock()
if err != nil {
logs.Error("NODE", err.Error())
remotelogs.Error("NODE", err.Error())
return
}
// 读取API配置
err = this.syncConfig(false)
if err != nil {
logs.Error("NODE", err.Error())
remotelogs.Error("NODE", err.Error())
return
}
@@ -82,12 +84,12 @@ func (this *Node) Start() {
// 读取配置
nodeConfig, err := nodeconfigs.SharedNodeConfig()
if err != nil {
logs.Error("NODE", "start failed: read node config failed: "+err.Error())
remotelogs.Error("NODE", "start failed: read node config failed: "+err.Error())
return
}
err = nodeConfig.Init()
if err != nil {
logs.Error("NODE", "init node config failed: "+err.Error())
remotelogs.Error("NODE", "init node config failed: "+err.Error())
return
}
sharedNodeConfig = nodeConfig
@@ -104,14 +106,14 @@ func (this *Node) Start() {
// 启动端口
err = sharedListenerManager.Start(nodeConfig)
if err != nil {
logs.Error("NODE", "start failed: "+err.Error())
remotelogs.Error("NODE", "start failed: "+err.Error())
return
}
// 写入PID
err = apps.WritePid()
if err != nil {
logs.Error("NODE", "write pid failed: "+err.Error())
remotelogs.Error("NODE", "write pid failed: "+err.Error())
return
}
@@ -212,14 +214,22 @@ func (this *Node) syncConfig(isFirstTime bool) error {
// 刷新配置
if isFirstTime {
logs.Println("NODE", "reloading config ...")
remotelogs.Println("NODE", "reloading config ...")
} else {
logs.Println("NODE", "loading config ...")
remotelogs.Println("NODE", "loading config ...")
}
nodeconfigs.ResetNodeConfig(nodeConfig)
caches.SharedManager.UpdatePolicies(nodeConfig.AllCachePolicies())
sharedWAFManager.UpdatePolicies(nodeConfig.AllHTTPFirewallPolicies())
if nodeConfig.HTTPCachePolicy != nil {
caches.SharedManager.UpdatePolicies([]*serverconfigs.HTTPCachePolicy{nodeConfig.HTTPCachePolicy})
} else {
caches.SharedManager.UpdatePolicies([]*serverconfigs.HTTPCachePolicy{})
}
if nodeConfig.HTTPFirewallPolicy != nil {
sharedWAFManager.UpdatePolicies([]*firewallconfigs.HTTPFirewallPolicy{nodeConfig.HTTPFirewallPolicy})
} else {
sharedWAFManager.UpdatePolicies([]*firewallconfigs.HTTPFirewallPolicy{})
}
sharedNodeConfig = nodeConfig
// 发送事件
@@ -237,7 +247,7 @@ func (this *Node) startSyncTimer() {
// TODO 这个时间间隔可以自行设置
ticker := time.NewTicker(60 * time.Second)
events.On(events.EventQuit, func() {
logs.Println("NODE", "quit sync timer")
remotelogs.Println("NODE", "quit sync timer")
ticker.Stop()
})
go func() {
@@ -246,13 +256,13 @@ func (this *Node) startSyncTimer() {
case <-ticker.C:
err := this.syncConfig(false)
if err != nil {
logs.Error("NODE", "sync config error: "+err.Error())
remotelogs.Error("NODE", "sync config error: "+err.Error())
continue
}
case <-changeNotify:
err := this.syncConfig(false)
if err != nil {
logs.Error("NODE", "sync config error: "+err.Error())
remotelogs.Error("NODE", "sync config error: "+err.Error())
continue
}
}
@@ -333,7 +343,7 @@ func (this *Node) listenSock() error {
return err
}
events.On(events.EventQuit, func() {
logs.Println("NODE", "quit unix sock")
remotelogs.Println("NODE", "quit unix sock")
_ = listener.Close()
})

12
internal/nodes/node_status_executor.go
View File

@@ -6,7 +6,7 @@ import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/iwind/TeaGo/lists"
"github.com/shirou/gopsutil/cpu"
@@ -38,7 +38,7 @@ func (this *NodeStatusExecutor) Listen() {
ticker := time.NewTicker(30 * time.Second)
events.On(events.EventQuit, func() {
logs.Println("NODE_STATUS", "quit executor")
remotelogs.Println("NODE_STATUS", "quit executor")
ticker.Stop()
})
@@ -73,19 +73,19 @@ func (this *NodeStatusExecutor) update() {
// 发送数据
jsonData, err := json.Marshal(status)
if err != nil {
logs.Error("NODE_STATUS", "serial NodeStatus fail: "+err.Error())
remotelogs.Error("NODE_STATUS", "serial NodeStatus fail: "+err.Error())
return
}
rpcClient, err := rpc.SharedRPC()
if err != nil {
logs.Error("NODE_STATUS", "failed to open rpc: "+err.Error())
remotelogs.Error("NODE_STATUS", "failed to open rpc: "+err.Error())
return
}
_, err = rpcClient.NodeRPC().UpdateNodeStatus(rpcClient.Context(), &pb.UpdateNodeStatusRequest{
StatusJSON: jsonData,
})
if err != nil {
logs.Error("NODE_STATUS", "rpc UpdateNodeStatus() failed: "+err.Error())
remotelogs.Error("NODE_STATUS", "rpc UpdateNodeStatus() failed: "+err.Error())
return
}
}
@@ -131,7 +131,7 @@ func (this *NodeStatusExecutor) updateCPU(status *nodeconfigs.NodeStatus) {
func (this *NodeStatusExecutor) updateDisk(status *nodeconfigs.NodeStatus) {
partitions, err := disk.Partitions(false)
if err != nil {
logs.Error("NODE_STATUS", err.Error())
remotelogs.Error("NODE_STATUS", err.Error())
return
}
lists.Sort(partitions, func(i int, j int) bool {

4
internal/nodes/origin_utils.go
View File

@@ -4,7 +4,7 @@ import (
"crypto/tls"
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"net"
"strconv"
)
@@ -23,7 +23,7 @@ func OriginConnect(origin *serverconfigs.OriginConfig, remoteAddr string) (net.C
port := int(toaConfig.RandLocalPort())
err := sharedTOAManager.SendMsg("add:" + strconv.Itoa(port) + ":" + remoteAddr)
if err != nil {
logs.Error("TOA", "add failed: "+err.Error())
remotelogs.Error("TOA", "add failed: "+err.Error())
} else {
dialer := net.Dialer{
Timeout: origin.ConnTimeoutDuration(),

12
internal/nodes/toa_manager.go
View File

@@ -3,7 +3,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/Tea"
"net"
"os"
@@ -18,7 +18,7 @@ func init() {
events.On(events.EventReload, func() {
err := sharedTOAManager.Run(sharedNodeConfig.TOA)
if err != nil {
logs.Error("TOA", err.Error())
remotelogs.Error("TOA", err.Error())
}
})
}
@@ -37,10 +37,10 @@ func (this *TOAManager) Run(config *nodeconfigs.TOAConfig) error {
this.config = config
if this.pid > 0 {
logs.Println("TOA", "stopping ...")
remotelogs.Println("TOA", "stopping ...")
err := this.Quit()
if err != nil {
logs.Error("TOA", "quit error: "+err.Error())
remotelogs.Error("TOA", "quit error: "+err.Error())
}
_ = this.conn.Close()
this.conn = nil
@@ -56,8 +56,8 @@ func (this *TOAManager) Run(config *nodeconfigs.TOAConfig) error {
if err != nil {
return err
}
logs.Println("TOA", "starting ...")
logs.Println("TOA", "args: "+strings.Join(config.AsArgs(), " "))
remotelogs.Println("TOA", "starting ...")
remotelogs.Println("TOA", "args: "+strings.Join(config.AsArgs(), " "))
cmd := exec.Command(binPath, config.AsArgs()...)
err = cmd.Start()
if err != nil {

4
internal/nodes/traffic_stat_manager.go
View File

@@ -2,7 +2,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/rpc"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/Tea"
@@ -41,7 +41,7 @@ func (this *TrafficStatManager) Start() {
for range ticker.C {
err := this.Upload()
if err != nil {
logs.Error("TRAFFIC_STAT_MANAGER", "upload stats failed: "+err.Error())
remotelogs.Error("TRAFFIC_STAT_MANAGER", "upload stats failed: "+err.Error())
}
}
}

4
internal/nodes/waf_manager.go
View File

@@ -3,7 +3,7 @@ package nodes
import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/firewallconfigs"
"github.com/TeaOSLab/EdgeNode/internal/errors"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/waf"
"strconv"
"sync"
@@ -33,7 +33,7 @@ func (this *WAFManager) UpdatePolicies(policies []*firewallconfigs.HTTPFirewallP
for _, p := range policies {
w, err := this.convertWAF(p)
if err != nil {
logs.Error("WAF", "initialize policy '"+strconv.FormatInt(p.Id, 10)+"' failed: "+err.Error())
remotelogs.Error("WAF", "initialize policy '"+strconv.FormatInt(p.Id, 10)+"' failed: "+err.Error())
continue
}
if w == nil {

2
internal/logs/utils.go → internal/remotelogs/utils.go
View File

@@ -1,4 +1,4 @@
package logs
package remotelogs
import (
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"

4
internal/waf/rule.go
View File

@@ -7,7 +7,7 @@ import (
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/filterconfigs"
"github.com/TeaOSLab/EdgeNode/internal/logs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/waf/checkpoints"
"github.com/TeaOSLab/EdgeNode/internal/waf/requests"
"github.com/TeaOSLab/EdgeNode/internal/waf/utils"
@@ -637,7 +637,7 @@ func (this *Rule) execFilter(value interface{}) interface{} {
}
value, goNext, err = filterInstance.Do(value, filter.Options)
if err != nil {
logs.Println("WAF", "filter error: "+err.Error())
remotelogs.Println("WAF", "filter error: "+err.Error())
break
}
if !goNext {