323 lines
8.1 KiB
Go
323 lines
8.1 KiB
Go
package steamcache
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"runtime"
|
|
"s1d3sw1ped/SteamCache2/steamcache/avgcachestate"
|
|
"s1d3sw1ped/SteamCache2/steamcache/logger"
|
|
"s1d3sw1ped/SteamCache2/version"
|
|
"s1d3sw1ped/SteamCache2/vfs"
|
|
"s1d3sw1ped/SteamCache2/vfs/cache"
|
|
"s1d3sw1ped/SteamCache2/vfs/cachestate"
|
|
"s1d3sw1ped/SteamCache2/vfs/disk"
|
|
"s1d3sw1ped/SteamCache2/vfs/gc"
|
|
"s1d3sw1ped/SteamCache2/vfs/memory"
|
|
syncfs "s1d3sw1ped/SteamCache2/vfs/sync"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
pprof "net/http/pprof"
|
|
|
|
"github.com/docker/go-units"
|
|
)
|
|
|
|
type SteamCache struct {
|
|
pprof bool
|
|
address string
|
|
upstream string
|
|
|
|
vfs vfs.VFS
|
|
|
|
memory *memory.MemoryFS
|
|
disk *disk.DiskFS
|
|
|
|
memorygc *gc.GCFS
|
|
diskgc *gc.GCFS
|
|
|
|
hits *avgcachestate.AvgCacheState
|
|
|
|
dirty bool
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func New(address string, memorySize string, memoryMultiplier int, diskSize string, diskMultiplier int, diskPath, upstream string, pprof bool) *SteamCache {
|
|
memorysize, err := units.FromHumanSize(memorySize)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
disksize, err := units.FromHumanSize(diskSize)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
c := cache.New(
|
|
cachehandler,
|
|
)
|
|
|
|
var m *memory.MemoryFS
|
|
var mgc *gc.GCFS
|
|
if memorysize > 0 {
|
|
m = memory.New(memorysize)
|
|
mgc = gc.New(m, memoryMultiplier, randomgc)
|
|
}
|
|
|
|
var d *disk.DiskFS
|
|
var dgc *gc.GCFS
|
|
if disksize > 0 {
|
|
d = disk.New(diskPath, disksize)
|
|
dgc = gc.New(d, diskMultiplier, randomgc)
|
|
}
|
|
|
|
// configure the cache to match the specified mode (memory only, disk only, or memory and disk) based on the provided sizes
|
|
if disksize == 0 && memorysize != 0 {
|
|
//memory only mode - no disk
|
|
|
|
c.SetSlow(mgc)
|
|
} else if disksize != 0 && memorysize == 0 {
|
|
// disk only mode
|
|
|
|
c.SetSlow(dgc)
|
|
} else if disksize != 0 && memorysize != 0 {
|
|
// memory and disk mode
|
|
|
|
c.SetFast(mgc)
|
|
c.SetSlow(dgc)
|
|
} else {
|
|
// no memory or disk isn't a valid configuration
|
|
logger.Logger.Error().Bool("memory", false).Bool("disk", false).Msg("configuration invalid :( exiting")
|
|
os.Exit(1)
|
|
}
|
|
|
|
sc := &SteamCache{
|
|
pprof: pprof,
|
|
upstream: upstream,
|
|
address: address,
|
|
vfs: syncfs.New(c),
|
|
|
|
memory: m,
|
|
disk: d,
|
|
|
|
memorygc: mgc,
|
|
diskgc: dgc,
|
|
|
|
hits: avgcachestate.New(100),
|
|
}
|
|
|
|
if d != nil {
|
|
if d.Size() > d.Capacity() {
|
|
randomgc(d, uint(d.Size()-d.Capacity()))
|
|
}
|
|
}
|
|
|
|
return sc
|
|
}
|
|
|
|
func (sc *SteamCache) Run() {
|
|
if sc.upstream != "" {
|
|
_, err := http.Get(sc.upstream)
|
|
if err != nil {
|
|
logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed to connect to upstream server")
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
sc.mu.Lock()
|
|
sc.dirty = true
|
|
sc.mu.Unlock()
|
|
|
|
sc.LogStats()
|
|
t := time.NewTicker(1 * time.Second)
|
|
go func() {
|
|
for range t.C {
|
|
sc.LogStats()
|
|
}
|
|
}()
|
|
|
|
err := http.ListenAndServe(sc.address, sc)
|
|
if err != nil {
|
|
if err == http.ErrServerClosed {
|
|
logger.Logger.Info().Msg("shutdown")
|
|
return
|
|
}
|
|
logger.Logger.Error().Err(err).Msg("Failed to start SteamCache2")
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func (sc *SteamCache) LogStats() {
|
|
sc.mu.Lock()
|
|
defer sc.mu.Unlock()
|
|
if sc.dirty {
|
|
|
|
up := sc.upstream
|
|
if up == "" {
|
|
up = "{host in request}"
|
|
}
|
|
|
|
logger.Logger.Info().Str("address", sc.address).Str("version", version.Version).Str("upstream", up).Msg("listening")
|
|
if sc.memory != nil { // only log memory if memory is enabled
|
|
lifetimeBytes, lifetimeFiles, reclaimedBytes, deletedFiles, gcTime := sc.memorygc.Stats()
|
|
|
|
logger.Logger.Info().
|
|
Str("size", units.HumanSize(float64(sc.memory.Size()))).
|
|
Str("capacity", units.HumanSize(float64(sc.memory.Capacity()))).
|
|
Str("files", fmt.Sprintf("%d", len(sc.memory.StatAll()))).
|
|
Msg("memory")
|
|
|
|
logger.Logger.Info().
|
|
Str("data_total", units.HumanSize(float64(lifetimeBytes))).
|
|
Uint("files_total", lifetimeFiles).
|
|
Str("data", units.HumanSize(float64(reclaimedBytes))).
|
|
Uint("files", deletedFiles).
|
|
Str("gc_time", gcTime.String()).
|
|
Msg("memory_gc")
|
|
}
|
|
|
|
if sc.disk != nil { // only log disk if disk is enabled
|
|
lifetimeBytes, lifetimeFiles, reclaimedBytes, deletedFiles, gcTime := sc.diskgc.Stats()
|
|
|
|
logger.Logger.Info().
|
|
Str("size", units.HumanSize(float64(sc.disk.Size()))).
|
|
Str("capacity", units.HumanSize(float64(sc.disk.Capacity()))).
|
|
Str("files", fmt.Sprintf("%d", len(sc.disk.StatAll()))).
|
|
Msg("disk")
|
|
|
|
logger.Logger.Info().
|
|
Str("data_total", units.HumanSize(float64(lifetimeBytes))).
|
|
Uint("files_total", lifetimeFiles).
|
|
Str("data", units.HumanSize(float64(reclaimedBytes))).
|
|
Uint("files", deletedFiles).
|
|
Str("gc_time", gcTime.String()).
|
|
Msg("disk_gc")
|
|
}
|
|
|
|
// log golang Garbage Collection stats
|
|
var m runtime.MemStats
|
|
runtime.ReadMemStats(&m)
|
|
|
|
logger.Logger.Info().
|
|
Str("alloc", units.HumanSize(float64(m.Alloc))).
|
|
Str("sys", units.HumanSize(float64(m.Sys))).
|
|
Msg("app_gc")
|
|
|
|
logger.Logger.Info().
|
|
Str("hitrate", fmt.Sprintf("%.2f%%", sc.hits.Avg()*100)).
|
|
Msg("cache")
|
|
|
|
logger.Logger.Info().Msg("") // empty line to separate log entries for better readability
|
|
sc.dirty = false
|
|
}
|
|
}
|
|
|
|
func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if sc.pprof && r.URL.Path == "/debug/pprof/" {
|
|
pprof.Index(w, r)
|
|
return
|
|
} else if sc.pprof && strings.HasPrefix(r.URL.Path, "/debug/pprof/") {
|
|
pprof.Handler(strings.TrimPrefix(r.URL.Path, "/debug/pprof/")).ServeHTTP(w, r)
|
|
return
|
|
}
|
|
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "Only GET method is supported", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
if r.URL.String() == "/lancache-heartbeat" {
|
|
w.Header().Add("X-LanCache-Processed-By", "SteamCache2")
|
|
w.WriteHeader(http.StatusNoContent)
|
|
w.Write(nil)
|
|
return
|
|
}
|
|
|
|
sc.mu.Lock()
|
|
sc.dirty = true
|
|
sc.mu.Unlock()
|
|
|
|
w.Header().Add("X-LanCache-Processed-By", "SteamCache2") // SteamPrefill uses this header to determine if the request was processed by the cache maybe steam uses it too
|
|
|
|
cacheKey := strings.ReplaceAll(r.URL.String()[1:], "\\", "/") // replace all backslashes with forward slashes shouldn't be necessary but just in case
|
|
if cacheKey == "" {
|
|
http.Error(w, "Invalid URL", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
data, err := sc.vfs.Get(cacheKey)
|
|
if err == nil {
|
|
sc.hits.Add(cachestate.CacheStateHit)
|
|
w.Header().Add("X-LanCache-Status", "HIT")
|
|
w.Write(data)
|
|
logger.Logger.Debug().Str("key", r.URL.String()).Msg("cache")
|
|
return
|
|
}
|
|
|
|
var req *http.Request
|
|
if sc.upstream != "" { // if an upstream server is configured, proxy the request to the upstream server
|
|
ur, err := url.JoinPath(sc.upstream, r.URL.String())
|
|
if err != nil {
|
|
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
req, err = http.NewRequest(http.MethodGet, ur, nil)
|
|
if err != nil {
|
|
http.Error(w, "Failed to create request", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
req.Host = r.Host
|
|
logger.Logger.Debug().Str("key", cacheKey).Str("host", sc.upstream).Msg("upstream")
|
|
} else { // if no upstream server is configured, proxy the request to the host specified in the request
|
|
host := r.Host
|
|
if r.Header.Get("X-Sls-Https") == "enable" {
|
|
host = "https://" + host
|
|
} else {
|
|
host = "http://" + host
|
|
}
|
|
|
|
ur, err := url.JoinPath(host, r.URL.String())
|
|
if err != nil {
|
|
http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
req, err = http.NewRequest(http.MethodGet, ur, nil)
|
|
if err != nil {
|
|
http.Error(w, "Failed to create request", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
logger.Logger.Debug().Str("key", cacheKey).Str("host", host).Msg("forward")
|
|
}
|
|
|
|
req.Header.Add("X-Sls-Https", r.Header.Get("X-Sls-Https"))
|
|
req.Header.Add("User-Agent", r.Header.Get("User-Agent"))
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
http.Error(w, "Failed to fetch the requested URL", resp.StatusCode)
|
|
return
|
|
}
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
http.Error(w, "Failed to read response body", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
sc.vfs.Set(cacheKey, body)
|
|
sc.hits.Add(cachestate.CacheStateMiss)
|
|
w.Header().Add("X-LanCache-Status", "MISS")
|
|
w.Write(body)
|
|
}
|