TRKJ/EdgeNode
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeNode.git synced 2025-12-26 06:56:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化HTTP缓存，主要是并发冲突、缓存写入不全等问题

Browse Source
This commit is contained in:
GoEdgeLab
2021-06-06 23:42:11 +08:00
parent 5c0e2fa6ca
commit bcd561f52f
18 changed files with 299 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
internal/nodes/http_request.go
View File

@@ -1142,6 +1142,12 @@ func (this *HTTPRequest) canIgnore(err error) bool {
return true
}
// 网络错误
_, ok := err.(*net.OpError)
if ok {
return true
}
// 客户端主动取消
if err == context.Canceled {
return true

10
internal/nodes/http_request_cache.go
View File

@@ -108,7 +108,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
}
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Warn("HTTP_REQUEST_CACHE", "read from cache failed: "+err.Error())
}
return
}
@@ -142,7 +142,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
})
if err != nil {
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Warn("HTTP_REQUEST_CACHE", "read from cache failed: "+err.Error())
}
return
}
@@ -234,7 +234,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
return true
}
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Warn("HTTP_REQUEST_CACHE", "read from cache failed: "+err.Error())
}
return
}
@@ -277,7 +277,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
})
if err != nil {
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Warn("HTTP_REQUEST_CACHE", "read from cache failed: "+err.Error())
}
return true
}
@@ -300,7 +300,7 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
})
if err != nil {
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_CACHE", "read from cache failed: "+err.Error())
remotelogs.Warn("HTTP_REQUEST_CACHE", "read from cache failed: "+err.Error())
}
return
}

14
internal/nodes/http_request_fastcgi.go
View File

@@ -200,14 +200,20 @@ func (this *HTTPRequest) doFastcgi() (shouldStop bool) {
_, err = io.CopyBuffer(this.writer, resp.Body, buf)
pool.Put(buf)
err1 := resp.Body.Close()
if err1 != nil {
remotelogs.Warn("REQUEST_FASTCGI", err1.Error())
closeErr := resp.Body.Close()
if closeErr != nil {
remotelogs.Warn("HTTP_REQUEST_FASTCGI", closeErr.Error())
}
if err != nil && err != io.EOF {
remotelogs.Warn("REQUEST_FASTCGI", err.Error())
remotelogs.Warn("HTTP_REQUEST_FASTCGI", err.Error())
this.addError(err)
}
// 是否成功结束
if err == nil && closeErr == nil {
this.writer.SetOk()
}
return
}

7
internal/nodes/http_request_page.go
View File

@@ -1,6 +1,7 @@
package nodes
import (
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/logs"
"io"
@@ -50,7 +51,11 @@ func (this *HTTPRequest) doPage(status int) (shouldStop bool) {
_, err = io.CopyBuffer(this.writer, fp, buf)
bytePool1k.Put(buf)
if err != nil {
logs.Error(err)
if !this.canIgnore(err) {
remotelogs.Warn("HTTP_REQUEST_PAGE", "write to client failed: "+err.Error())
}
} else {
this.writer.SetOk()
}
err = fp.Close()
if err != nil {

25
internal/nodes/http_request_reverse_proxy.go
View File

@@ -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")
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("HTTP_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")
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("HTTP_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 {
remotelogs.Error("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
this.write502(err)
return
}
@@ -162,7 +162,7 @@ func (this *HTTPRequest) doReverseProxy() {
// TODO 如果超过最大失败次数，则下线
this.write502(err)
remotelogs.Println("REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
remotelogs.Println("HTTP_REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
} else {
// 是否为客户端方面的错误
isClientError := false
@@ -189,7 +189,7 @@ func (this *HTTPRequest) doReverseProxy() {
if this.doWAFResponse(resp) {
err = resp.Body.Close()
if err != nil {
remotelogs.Warn("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", err.Error())
}
return
}
@@ -201,7 +201,7 @@ func (this *HTTPRequest) doReverseProxy() {
if len(this.web.Pages) > 0 && this.doPage(resp.StatusCode) {
err = resp.Body.Close()
if err != nil {
remotelogs.Warn("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", err.Error())
}
return
}
@@ -254,17 +254,22 @@ func (this *HTTPRequest) doReverseProxy() {
}
pool.Put(buf)
err1 := resp.Body.Close()
if err1 != nil {
closeErr := resp.Body.Close()
if closeErr != nil {
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_REVERSE_PROXY", err1.Error())
remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", closeErr.Error())
}
}
if err != nil && err != io.EOF {
if !this.canIgnore(err) {
remotelogs.Warn("REQUEST_REVERSE_PROXY", err.Error())
remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", err.Error())
this.addError(err)
}
}
// 是否成功结束
if err == nil && closeErr == nil {
this.writer.SetOk()
}
}

3
internal/nodes/http_request_root.go
View File

@@ -382,6 +382,9 @@ func (this *HTTPRequest) doRoot() (isBreak bool) {
}
}
// 设置成功
this.writer.SetOk()
return true
}

11
internal/nodes/http_request_shutdown.go
View File

@@ -1,6 +1,7 @@
package nodes
import (
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/logs"
"io"
@@ -65,8 +66,16 @@ func (this *HTTPRequest) doShutdown() {
buf := bytePool1k.Get()
_, err = io.CopyBuffer(this.writer, fp, buf)
bytePool1k.Put(buf)
if err != nil {
if !this.canIgnore(err) {
remotelogs.Warn("HTTP_REQUEST_SHUTDOWN", "write to client failed: "+err.Error())
}
} else {
this.writer.SetOk()
}
err = fp.Close()
if err != nil {
logs.Error(err)
remotelogs.Warn("HTTP_REQUEST_SHUTDOWN", "close file failed: "+err.Error())
}
}

9
internal/nodes/http_request_url.go
View File

@@ -2,6 +2,7 @@ package nodes
import (
"errors"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"github.com/TeaOSLab/EdgeNode/internal/utils"
"github.com/iwind/TeaGo/logs"
"io"
@@ -68,4 +69,12 @@ func (this *HTTPRequest) doURL(method string, url string, host string, statusCod
buf := pool.Get()
_, err = io.CopyBuffer(this.writer, resp.Body, buf)
pool.Put(buf)
if err != nil {
if !this.canIgnore(err) {
remotelogs.Warn("HTTP_REQUEST_URL", "write to client failed: "+err.Error())
}
} else {
this.writer.SetOk()
}
}

8
internal/nodes/http_request_waf.go
View File

@@ -70,7 +70,7 @@ func (this *HTTPRequest) checkWAFRequest(firewallPolicy *firewallconfigs.HTTPFir
for _, action := range actions {
goNext, err := action.DoHTTP(this.RawReq, this.RawWriter)
if err != nil {
remotelogs.Error("REQUEST", "do action '"+err.Error()+"' failed: "+err.Error())
remotelogs.Error("HTTP_REQUEST_WAF", "do action '"+err.Error()+"' failed: "+err.Error())
return true, false
}
if !goNext {
@@ -101,7 +101,7 @@ func (this *HTTPRequest) checkWAFRequest(firewallPolicy *firewallconfigs.HTTPFir
for _, remoteAddr := range remoteAddrs {
result, err := iplibrary.SharedLibrary.Lookup(remoteAddr)
if err != nil {
remotelogs.Error("REQUEST", "iplibrary lookup failed: "+err.Error())
remotelogs.Error("HTTP_REQUEST_WAF", "iplibrary lookup failed: "+err.Error())
} else if result != nil {
// 检查国家级别封禁
if len(regionConfig.DenyCountryIds) > 0 && len(result.Country) > 0 {
@@ -147,7 +147,7 @@ func (this *HTTPRequest) checkWAFRequest(firewallPolicy *firewallconfigs.HTTPFir
}
goNext, ruleGroup, ruleSet, err := w.MatchRequest(this.RawReq, this.writer)
if err != nil {
remotelogs.Error("REQUEST", this.rawURI+": "+err.Error())
remotelogs.Error("HTTP_REQUEST_WAF", this.rawURI+": "+err.Error())
return
}
@@ -181,7 +181,7 @@ func (this *HTTPRequest) doWAFResponse(resp *http.Response) (blocked bool) {
goNext, ruleGroup, ruleSet, err := w.MatchResponse(this.RawReq, resp, this.writer)
if err != nil {
remotelogs.Error("REQUEST", this.rawURI+": "+err.Error())
remotelogs.Error("HTTP_REQUEST_WAF", this.rawURI+": "+err.Error())
return
}

83
internal/nodes/http_writer.go
View File

@@ -14,7 +14,7 @@ import (
"strings"
)
// 响应Writer
// HTTPWriter 响应Writer
type HTTPWriter struct {
req *HTTPRequest
writer http.ResponseWriter
@@ -32,9 +32,11 @@ type HTTPWriter struct {
cacheWriter caches.Writer // 缓存写入
cacheStorage caches.StorageInterface
isOk bool // 是否完全成功
}
// 包装对象
// NewHTTPWriter 包装对象
func NewHTTPWriter(req *HTTPRequest, httpResponseWriter http.ResponseWriter) *HTTPWriter {
return &HTTPWriter{
req: req,
@@ -42,7 +44,7 @@ func NewHTTPWriter(req *HTTPRequest, httpResponseWriter http.ResponseWriter) *HT
}
}
// 重置
// Reset 重置
func (this *HTTPWriter) Reset(httpResponseWriter http.ResponseWriter) {
this.writer = httpResponseWriter
@@ -58,12 +60,12 @@ func (this *HTTPWriter) Reset(httpResponseWriter http.ResponseWriter) {
this.gzipBodyWriter = nil
}
// 设置Gzip
// Gzip 设置Gzip
func (this *HTTPWriter) Gzip(config *serverconfigs.HTTPGzipConfig) {
this.gzipConfig = config
}
// 准备输出
// Prepare 准备输出
func (this *HTTPWriter) Prepare(size int64, status int) {
this.statusCode = status
@@ -71,12 +73,12 @@ func (this *HTTPWriter) Prepare(size int64, status int) {
this.prepareCache(size)
}
// 包装前的原始的Writer
// Raw 包装前的原始的Writer
func (this *HTTPWriter) Raw() http.ResponseWriter {
return this.writer
}
// 获取Header
// Header 获取Header
func (this *HTTPWriter) Header() http.Header {
if this.writer == nil {
return http.Header{}
@@ -84,7 +86,7 @@ func (this *HTTPWriter) Header() http.Header {
return this.writer.Header()
}
// 添加一组Header
// AddHeaders 添加一组Header
func (this *HTTPWriter) AddHeaders(header http.Header) {
if this.writer == nil {
return
@@ -99,7 +101,7 @@ func (this *HTTPWriter) AddHeaders(header http.Header) {
}
}
// 写入数据
// Write 写入数据
func (this *HTTPWriter) Write(data []byte) (n int, err error) {
if this.writer != nil {
if this.gzipWriter != nil {
@@ -115,8 +117,9 @@ func (this *HTTPWriter) Write(data []byte) (n int, err error) {
if this.cacheWriter != nil {
_, err = this.cacheWriter.Write(data)
if err != nil {
_ = this.cacheWriter.Discard()
this.cacheWriter = nil
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("HTTP_WRITER", "write cache failed: "+err.Error())
}
}
} else {
@@ -128,7 +131,7 @@ func (this *HTTPWriter) Write(data []byte) (n int, err error) {
if this.gzipBodyWriter != nil {
_, err := this.gzipBodyWriter.Write(data)
if err != nil {
remotelogs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("HTTP_WRITER", err.Error())
}
} else {
this.body = append(this.body, data...)
@@ -137,17 +140,17 @@ func (this *HTTPWriter) Write(data []byte) (n int, err error) {
return
}
// 写入字符串
// WriteString 写入字符串
func (this *HTTPWriter) WriteString(s string) (n int, err error) {
return this.Write([]byte(s))
}
// 读取发送的字节数
// SentBodyBytes 读取发送的字节数
func (this *HTTPWriter) SentBodyBytes() int64 {
return this.sentBodyBytes
}
// 写入状态码
// WriteHeader 写入状态码
func (this *HTTPWriter) WriteHeader(statusCode int) {
if this.writer != nil {
this.writer.WriteHeader(statusCode)
@@ -155,7 +158,7 @@ func (this *HTTPWriter) WriteHeader(statusCode int) {
this.statusCode = statusCode
}
// 读取状态码
// StatusCode 读取状态码
func (this *HTTPWriter) StatusCode() int {
if this.statusCode == 0 {
return http.StatusOK
@@ -163,22 +166,22 @@ func (this *HTTPWriter) StatusCode() int {
return this.statusCode
}
// 设置拷贝Body数据
// SetBodyCopying 设置拷贝Body数据
func (this *HTTPWriter) SetBodyCopying(b bool) {
this.bodyCopying = b
}
// 判断是否在拷贝Body数据
// BodyIsCopying 判断是否在拷贝Body数据
func (this *HTTPWriter) BodyIsCopying() bool {
return this.bodyCopying
}
// 读取拷贝的Body数据
// Body 读取拷贝的Body数据
func (this *HTTPWriter) Body() []byte {
return this.body
}
// 读取Header二进制数据
// HeaderData 读取Header二进制数据
func (this *HTTPWriter) HeaderData() []byte {
if this.writer == nil {
return nil
@@ -200,7 +203,12 @@ func (this *HTTPWriter) HeaderData() []byte {
return writer.Bytes()
}
// 关闭
// SetOk 设置成功
func (this *HTTPWriter) SetOk() {
this.isOk = true
}
// Close 关闭
func (this *HTTPWriter) Close() {
// gzip writer
if this.gzipWriter != nil {
@@ -214,20 +222,24 @@ func (this *HTTPWriter) Close() {
// cache writer
if this.cacheWriter != nil {
err := this.cacheWriter.Close()
if err == nil {
this.cacheStorage.AddToList(&caches.Item{
Type: this.cacheWriter.ItemType(),
Key: this.cacheWriter.Key(),
ExpiredAt: this.cacheWriter.ExpiredAt(),
HeaderSize: this.cacheWriter.HeaderSize(),
BodySize: this.cacheWriter.BodySize(),
})
if this.isOk {
err := this.cacheWriter.Close()
if err == nil {
this.cacheStorage.AddToList(&caches.Item{
Type: this.cacheWriter.ItemType(),
Key: this.cacheWriter.Key(),
ExpiredAt: this.cacheWriter.ExpiredAt(),
HeaderSize: this.cacheWriter.HeaderSize(),
BodySize: this.cacheWriter.BodySize(),
})
}
} else {
_ = this.cacheWriter.Discard()
}
}
}
// Hijack
// Hijack Hijack
func (this *HTTPWriter) Hijack() (conn net.Conn, buf *bufio.ReadWriter, err error) {
hijack, ok := this.writer.(http.Hijacker)
if ok {
@@ -236,7 +248,7 @@ func (this *HTTPWriter) Hijack() (conn net.Conn, buf *bufio.ReadWriter, err erro
return
}
// Flush
// Flush Flush
func (this *HTTPWriter) Flush() {
flusher, ok := this.writer.(http.Flusher)
if ok {
@@ -284,7 +296,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 {
remotelogs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("HTTP_WRITER", err.Error())
return
}
@@ -293,7 +305,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 {
remotelogs.Error("REQUEST_WRITER", err.Error())
remotelogs.Error("HTTP_WRITER", err.Error())
}
}
@@ -376,7 +388,7 @@ func (this *HTTPWriter) prepareCache(size int64) {
cacheWriter, err := storage.OpenWriter(this.req.cacheKey, expiredAt, this.StatusCode())
if err != nil {
if err != caches.ErrFileIsWriting {
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("HTTP_WRITER", "write cache failed: "+err.Error())
}
return
}
@@ -390,7 +402,8 @@ func (this *HTTPWriter) prepareCache(size int64) {
for _, v1 := range v {
_, err = cacheWriter.WriteHeader([]byte(k + ":" + v1 + "\n"))
if err != nil {
remotelogs.Error("REQUEST_WRITER", "write cache failed: "+err.Error())
remotelogs.Error("HTTP_WRITER", "write cache failed: "+err.Error())
_ = this.cacheWriter.Discard()
this.cacheWriter = nil
return
}

5
internal/nodes/listener_http.go
View File

@@ -4,6 +4,8 @@ import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
"golang.org/x/net/http2"
"io"
"log"
"net"
"net/http"
"strings"
@@ -11,6 +13,8 @@ import (
"time"
)
var httpErrorLogger = log.New(io.Discard, "", 0)
type HTTPListener struct {
BaseListener
@@ -37,6 +41,7 @@ func (this *HTTPListener) Serve() error {
Handler: handler,
ReadHeaderTimeout: 3 * time.Second, // TODO 改成可以配置
IdleTimeout: 2 * time.Minute, // TODO 改成可以配置
ErrorLog: httpErrorLogger,
ConnState: func(conn net.Conn, state http.ConnState) {
switch state {
case http.StateNew: