TRKJ/EdgeNode
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeNode.git synced 2025-11-03 23:20:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

实现stale cache读取

Browse Source
This commit is contained in:
GoEdgeLab
2021-12-16 17:27:21 +08:00
parent a15b15004b
commit b0cddd1b23
16 changed files with 164 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/caches/item.go
View File

@@ -22,6 +22,7 @@ type Item struct {
Type ItemType `json:"type"`
Key string `json:"key"`
ExpiredAt int64 `json:"expiredAt"`
StaleAt int64 `json:"staleAt"`
HeaderSize int64 `json:"headerSize"`
BodySize int64 `json:"bodySize"`
MetaSize int64 `json:"metaSize"`

48
internal/caches/list_file.go
View File

@@ -13,6 +13,7 @@ import (
_ "github.com/mattn/go-sqlite3"
"os"
"strconv"
"strings"
"sync/atomic"
"time"
)
@@ -104,6 +105,12 @@ func (this *FileList) Init() error {
return err
}
// 检查staleAt字段
err = this.checkStaleAtField()
if err != nil {
return err
}
// 读取总数量
row := this.db.QueryRow(`SELECT COUNT(*) FROM "` + this.itemsTableName + `"`)
if row.Err() != nil {
@@ -122,7 +129,7 @@ func (this *FileList) Init() error {
return err
}
this.insertStmt, err = this.db.Prepare(`INSERT INTO "` + this.itemsTableName + `" ("hash", "key", "headerSize", "bodySize", "metaSize", "expiredAt", "host", "serverId", "createdAt") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`)
this.insertStmt, err = this.db.Prepare(`INSERT INTO "` + this.itemsTableName + `" ("hash", "key", "headerSize", "bodySize", "metaSize", "expiredAt", "staleAt", "host", "serverId", "createdAt") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
if err != nil {
return err
}
@@ -142,7 +149,7 @@ func (this *FileList) Init() error {
return err
}
this.purgeStmt, err = this.db.Prepare(`SELECT "hash" FROM "` + this.itemsTableName + `" WHERE expiredAt<=? LIMIT ?`)
this.purgeStmt, err = this.db.Prepare(`SELECT "hash" FROM "` + this.itemsTableName + `" WHERE staleAt<=? LIMIT ?`)
if err != nil {
return err
}
@@ -182,7 +189,11 @@ func (this *FileList) Add(hash string, item *Item) error {
return nil
}
_, err := this.insertStmt.Exec(hash, item.Key, item.HeaderSize, item.BodySize, item.MetaSize, item.ExpiredAt, item.Host, item.ServerId, utils.UnixTime())
if item.StaleAt == 0 {
item.StaleAt = item.ExpiredAt
}
_, err := this.insertStmt.Exec(hash, item.Key, item.HeaderSize, item.BodySize, item.MetaSize, item.ExpiredAt, item.StaleAt, item.Host, item.ServerId, utils.UnixTime())
if err != nil {
return err
}
@@ -474,6 +485,8 @@ func (this *FileList) Close() error {
// 初始化
func (this *FileList) initTables(db *sql.DB, times int) error {
{
// expiredAt - 过期时间，用来判断有无过期
// staleAt - 陈旧最大时间，用来清理缓存
_, err := db.Exec(`CREATE TABLE IF NOT EXISTS "` + this.itemsTableName + `" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"hash" varchar(32),
@@ -482,6 +495,7 @@ func (this *FileList) initTables(db *sql.DB, times int) error {
"bodySize" integer DEFAULT 0,
"metaSize" integer DEFAULT 0,
"expiredAt" integer DEFAULT 0,
"staleAt" integer DEFAULT 0,
"createdAt" integer DEFAULT 0,
"host" varchar(128),
"serverId" integer
@@ -497,6 +511,11 @@ ON "` + this.itemsTableName + `" (
"expiredAt" ASC
);
CREATE INDEX IF NOT EXISTS "staleAt"
ON "` + this.itemsTableName + `" (
"staleAt" ASC
);
CREATE UNIQUE INDEX IF NOT EXISTS "hash"
ON "` + this.itemsTableName + `" (
"hash" ASC
@@ -579,3 +598,26 @@ func (this *FileList) removeOldTables() error {
}
return nil
}
func (this *FileList) checkStaleAtField() error {
rows, err := this.db.Query(`SELECT staleAt FROM "` + this.itemsTableName + `"`)
if err != nil {
if strings.Contains(err.Error(), "no such column: staleAt") { // 暂时没有更好的判断方法
_, err = this.db.Exec(`ALTER TABLE "` + this.itemsTableName + `" ADD COLUMN staleAt integer DEFAULT 0`)
if err != nil {
return err
}
_, err = this.db.Exec(`UPDATE "` + this.itemsTableName + `" SET staleAt=expiredAt`)
if err != nil {
return err
}
} else {
return err
}
} else {
_ = rows.Close()
}
return nil
}

37
internal/caches/storage_file.go
View File

@@ -204,14 +204,19 @@ func (this *FileStorage) Init() error {
return nil
}
func (this *FileStorage) OpenReader(key string) (Reader, error) {
return this.openReader(key, true)
func (this *FileStorage) OpenReader(key string, useStale bool) (Reader, error) {
return this.openReader(key, true, useStale)
}
func (this *FileStorage) openReader(key string, allowMemory bool) (Reader, error) {
func (this *FileStorage) openReader(key string, allowMemory bool, useStale bool) (Reader, error) {
// 使用陈旧缓存的时候，我们认为是短暂的，只需要从文件里检查即可
if useStale {
allowMemory = false
}
// 先尝试内存缓存
if allowMemory && this.memoryStorage != nil {
reader, err := this.memoryStorage.OpenReader(key)
reader, err := this.memoryStorage.OpenReader(key, useStale)
if err == nil {
return reader, err
}
@@ -219,6 +224,17 @@ func (this *FileStorage) openReader(key string, allowMemory bool) (Reader, error
hash, path := this.keyPath(key)
// 检查文件记录是否已过期
if !useStale {
exists, err := this.list.Exist(hash)
if err != nil {
return nil, err
}
if !exists {
return nil, ErrNotFound
}
}
// TODO 尝试使用mmap加快读取速度
var isOk = false
fp, err := os.OpenFile(path, os.O_RDONLY, 0444)
@@ -235,15 +251,6 @@ func (this *FileStorage) openReader(key string, allowMemory bool) (Reader, error
}
}()
// 检查文件记录是否已过期
exists, err := this.list.Exist(hash)
if err != nil {
return nil, err
}
if !exists {
return nil, ErrNotFound
}
reader := NewFileReader(fp)
if err != nil {
return nil, err
@@ -923,7 +930,7 @@ func (this *FileStorage) hotLoop() {
this.hotMap = map[string]*HotItem{}
this.hotMapLocker.Unlock()
// 取Top10
// 取Top10写入内存
if len(result) > 0 {
sort.Slice(result, func(i, j int) bool {
return result[i].Hits > result[j].Hits
@@ -937,7 +944,7 @@ func (this *FileStorage) hotLoop() {
var buf = make([]byte, 32*1024)
for _, item := range result[:size] {
reader, err := this.openReader(item.Key, false)
reader, err := this.openReader(item.Key, false, false)
if err != nil {
continue
}

2
internal/caches/storage_interface.go
View File

@@ -10,7 +10,7 @@ type StorageInterface interface {
Init() error
// OpenReader 读取缓存
OpenReader(key string) (Reader, error)
OpenReader(key string, useStale bool) (reader Reader, err error)
// OpenWriter 打开缓存写入器等待写入
OpenWriter(key string, expiredAt int64, status int) (Writer, error)

4
internal/caches/storage_memory.go
View File

@@ -105,7 +105,7 @@ func (this *MemoryStorage) Init() error {
}
// OpenReader 读取缓存
func (this *MemoryStorage) OpenReader(key string) (Reader, error) {
func (this *MemoryStorage) OpenReader(key string, useStale bool) (Reader, error) {
hash := this.hash(key)
this.locker.RLock()
@@ -115,7 +115,7 @@ func (this *MemoryStorage) OpenReader(key string) (Reader, error) {
return nil, ErrNotFound
}
if item.ExpiredAt > utils.UnixTime() {
if useStale || (item.ExpiredAt > utils.UnixTime()) {
reader := NewMemoryReader(item)
err := reader.Init()
if err != nil {

2
internal/nodes/api_stream.go
View File

@@ -236,7 +236,7 @@ func (this *APIStream) handleReadCache(message *pb.NodeStreamMessage) error {
}()
}
reader, err := storage.OpenReader(msg.Key)
reader, err := storage.OpenReader(msg.Key, false)
if err != nil {
if err == caches.ErrNotFound {
this.replyFail(message.RequestId, "key not found")

9
internal/nodes/http_request.go
View File

@@ -69,9 +69,12 @@ type HTTPRequest struct {
rewriteRule *serverconfigs.HTTPRewriteRule // 匹配到的重写规则
rewriteReplace string // 重写规则的目标
rewriteIsExternalURL bool // 重写目标是否为外部URL
cacheRef *serverconfigs.HTTPCacheRef // 缓存设置
cacheKey string // 缓存使用的Key
isCached bool // 是否已经被缓存
cacheCanTryStale bool // 是否可以尝试使用Stale缓存
isAttack bool // 是否是攻击请求
requestBodyData []byte // 读取的Body内容
@@ -137,7 +140,7 @@ func (this *HTTPRequest) Do() {
// Web配置
err := this.configureWeb(this.Server.Web, true, 0)
if err != nil {
this.write50x(err, http.StatusInternalServerError)
this.write50x(err, http.StatusInternalServerError, false)
this.doEnd()
return
}
@@ -224,7 +227,7 @@ func (this *HTTPRequest) doBegin() {
var err error
this.requestBodyData, err = ioutil.ReadAll(io.LimitReader(this.RawReq.Body, AccessLogMaxRequestBodySize))
if err != nil {
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, false)
return
}
this.RawReq.Body = ioutil.NopCloser(io.MultiReader(bytes.NewBuffer(this.requestBodyData), this.RawReq.Body))
@@ -253,7 +256,7 @@ func (this *HTTPRequest) doBegin() {
// 缓存
if this.web.Cache != nil && this.web.Cache.IsOn {
if this.doCacheRead() {
if this.doCacheRead(false) {
return
}
}

2
internal/nodes/http_request_auth.go
View File

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

21
internal/nodes/http_request_cache.go
View File

@@ -17,7 +17,7 @@ import (
)
// 读取缓存
func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
cachePolicy := this.Server.HTTPCachePolicy
if cachePolicy == nil || !cachePolicy.IsOn {
return
@@ -148,11 +148,6 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
return true
}
buf := bytePool32k.Get()
defer func() {
bytePool32k.Put(buf)
}()
var reader caches.Reader
var err error
@@ -161,16 +156,20 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
this.web.WebP.IsOn &&
this.web.WebP.MatchRequest(filepath.Ext(this.requestPath()), this.Format) &&
this.web.WebP.MatchAccept(this.requestHeader("Accept")) {
reader, _ = storage.OpenReader(key + webpSuffix)
reader, _ = storage.OpenReader(key+webpSuffix, useStale)
}
// 检查正常的文件
if reader == nil {
reader, err = storage.OpenReader(key)
reader, err = storage.OpenReader(key, useStale)
if err != nil {
if err == caches.ErrNotFound {
// cache相关变量
this.varMapping["cache.status"] = "MISS"
if this.web.Cache.Stale != nil && this.web.Cache.Stale.IsOn {
this.cacheCanTryStale = true
}
return
}
@@ -187,6 +186,12 @@ func (this *HTTPRequest) doCacheRead() (shouldStop bool) {
this.varMapping["cache.status"] = "HIT"
this.logAttrs["cache.status"] = "HIT"
// 准备Buffer
buf := bytePool32k.Get()
defer func() {
bytePool32k.Put(buf)
}()
// 读取Header
headerBuf := []byte{}
err = reader.ReadHeader(buf, func(n int) (goNext bool, err error) {

16
internal/nodes/http_request_error.go
View File

@@ -1,6 +1,7 @@
package nodes
import (
"github.com/iwind/TeaGo/lists"
"github.com/iwind/TeaGo/types"
"net/http"
)
@@ -25,11 +26,24 @@ func (this *HTTPRequest) writeCode(code int) {
_, _ = this.writer.Write([]byte(types.String(code) + " " + http.StatusText(code) + ": '" + this.requestFullURL() + "'" + " (Request Id: " + this.requestId + ")"))
}
func (this *HTTPRequest) write50x(err error, statusCode int) {
func (this *HTTPRequest) write50x(err error, statusCode int, canTryStale bool) {
if err != nil {
this.addError(err)
}
// 尝试从缓存中恢复
if canTryStale &&
this.cacheCanTryStale &&
this.web.Cache.Stale != nil &&
this.web.Cache.Stale.IsOn &&
(len(this.web.Cache.Stale.Status) == 0 || lists.ContainsInt(this.web.Cache.Stale.Status, statusCode)) {
ok := this.doCacheRead(true)
if ok {
return
}
}
// 显示自定义页面
if this.doPage(statusCode) {
return
}

6
internal/nodes/http_request_fastcgi.go
View File

@@ -81,7 +81,7 @@ func (this *HTTPRequest) doFastcgi() (shouldStop bool) {
client, err := fcgi.SharedPool(fastcgi.Network(), fastcgi.RealAddress(), uint(poolSize)).Client()
if err != nil {
this.write50x(err, http.StatusInternalServerError)
this.write50x(err, http.StatusInternalServerError, false)
return
}
@@ -159,13 +159,13 @@ func (this *HTTPRequest) doFastcgi() (shouldStop bool) {
resp, stderr, err := client.Call(fcgiReq)
if err != nil {
this.write50x(err, http.StatusInternalServerError)
this.write50x(err, http.StatusInternalServerError, false)
return
}
if len(stderr) > 0 {
err := errors.New("Fastcgi Error: " + strings.TrimSpace(string(stderr)) + " script: " + maps.NewMap(params).GetString("SCRIPT_FILENAME"))
this.write50x(err, http.StatusInternalServerError)
this.write50x(err, http.StatusInternalServerError, false)
return
}

16
internal/nodes/http_request_reverse_proxy.go
View File

@@ -41,7 +41,7 @@ func (this *HTTPRequest) doReverseProxy() {
if origin == nil {
err := errors.New(this.requestFullURL() + ": no available origin sites for reverse proxy")
remotelogs.ServerError(this.Server.Id, "HTTP_REQUEST_REVERSE_PROXY", err.Error(), "", nil)
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, true)
return
}
this.origin = origin // 设置全局变量是为了日志等处理
@@ -61,7 +61,7 @@ func (this *HTTPRequest) doReverseProxy() {
if origin.Addr == nil {
err := errors.New(this.requestFullURL() + ": origin '" + strconv.FormatInt(origin.Id, 10) + "' does not has a address")
remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, false)
return
}
this.RawReq.URL.Scheme = origin.Addr.Protocol.Primary().Scheme()
@@ -156,7 +156,7 @@ func (this *HTTPRequest) doReverseProxy() {
client, err := SharedHTTPClientPool.Client(this, origin, originAddr, this.reverseProxy.ProxyProtocol)
if err != nil {
remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, true)
return
}
@@ -175,18 +175,18 @@ func (this *HTTPRequest) doReverseProxy() {
SharedOriginStateManager.Fail(origin, this.reverseProxy, func() {
this.reverseProxy.ResetScheduling()
})
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, true)
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)
this.write50x(err, http.StatusGatewayTimeout, true)
} else if httpErr.Temporary() {
this.write50x(err, http.StatusServiceUnavailable)
this.write50x(err, http.StatusServiceUnavailable, true)
} else {
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, true)
}
remotelogs.Warn("HTTP_REQUEST_REVERSE_PROXY", this.RawReq.URL.String()+"': "+err.Error())
} else {
@@ -207,7 +207,7 @@ func (this *HTTPRequest) doReverseProxy() {
}
if !isClientError {
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, true)
}
}
if resp != nil && resp.Body != nil {

6
internal/nodes/http_request_root.go
View File

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

2
internal/nodes/http_request_url.go
View File

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

6
internal/nodes/http_request_websocket.go
View File

@@ -43,7 +43,7 @@ func (this *HTTPRequest) doWebsocket() {
// TODO 增加N次错误重试，重试的时候需要尝试不同的源站
originConn, err := OriginConnect(this.origin, this.RawReq.RemoteAddr)
if err != nil {
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, false)
return
}
defer func() {
@@ -52,13 +52,13 @@ func (this *HTTPRequest) doWebsocket() {
err = this.RawReq.Write(originConn)
if err != nil {
this.write50x(err, http.StatusBadGateway)
this.write50x(err, http.StatusBadGateway, false)
return
}
clientConn, _, err := this.writer.Hijack()
if err != nil || clientConn == nil {
this.write50x(err, http.StatusInternalServerError)
this.write50x(err, http.StatusInternalServerError, false)
return
}
defer func() {

35
internal/nodes/http_writer.go
View File

@@ -1,3 +1,5 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
package nodes
import (
@@ -413,10 +415,12 @@ func (this *HTTPWriter) Close() {
if this.isOk {
err := this.cacheWriter.Close()
if err == nil {
var expiredAt = this.cacheWriter.ExpiredAt()
this.cacheStorage.AddToList(&caches.Item{
Type: this.cacheWriter.ItemType(),
Key: this.cacheWriter.Key(),
ExpiredAt: this.cacheWriter.ExpiredAt(),
ExpiredAt: expiredAt,
StaleAt: expiredAt + int64(this.calculateStaleLife()),
HeaderSize: this.cacheWriter.HeaderSize(),
BodySize: this.cacheWriter.BodySize(),
Host: this.req.Host,
@@ -690,3 +694,32 @@ func (this *HTTPWriter) prepareCache(size int64) {
}
}
}
// 计算stale时长
func (this *HTTPWriter) calculateStaleLife() int {
var staleLife = 600 // TODO 可以在缓存策略里设置此时间
var staleConfig = this.req.web.Cache.Stale
if staleConfig != nil && staleConfig.IsOn {
// 从Header中读取stale-if-error
var isDefinedInHeader = false
if staleConfig.SupportStaleIfErrorHeader {
var cacheControl = this.Header().Get("Cache-Control")
var pieces = strings.Split(cacheControl, ",")
for _, piece := range pieces {
var eqIndex = strings.Index(piece, "=")
if eqIndex > 0 && strings.TrimSpace(piece[:eqIndex]) == "stale-if-error" {
// 这里预示着如果stale-if-error=0，可以关闭stale功能
staleLife = types.Int(strings.TrimSpace(piece[eqIndex+1:]))
isDefinedInHeader = true
break
}
}
}
// 自定义
if !isDefinedInHeader && staleConfig.Life != nil {
staleLife = types.Int(staleConfig.Life.Duration().Seconds())
}
}
return staleLife
}