ops: Signal disk-tier attach pending vs ready #45
@@ -83,6 +83,7 @@ curl -s -i http://localhost/lancache-heartbeat
|
||||
| `cache_hits` / `cache_misses` / `hit_rate` | Whether later requests were served from cache |
|
||||
| `memory_cache_hits` / `disk_cache_hits` | Which tier served the hits |
|
||||
| `total_requests` / `errors` | Volume and failures |
|
||||
| `disk_tier_ready` | `0` while disk slow-tier attach pending; `1` when attached, or when no disk configured (N/A — not waiting) |
|
||||
|
||||
A first pass through new content is mostly misses (`hit_rate` near 0). Repeat the same content and `cache_hits` / `hit_rate` should rise.
|
||||
|
||||
@@ -94,6 +95,8 @@ curl -s -i http://localhost/lancache-heartbeat
|
||||
|
||||
Use GET (`curl -i`), not HEAD (`curl -I`): the server only accepts GET.
|
||||
|
||||
Heartbeat also returns `X-SteamCache-Disk-Tier: pending|ready|disabled` (`disabled` = memory-only / no disk; `pending`/`ready` = disk configured attach state).
|
||||
|
||||
These are the cache process's own `/metrics` and `/lancache-heartbeat` endpoints. There is no separate metrics daemon.
|
||||
|
||||
If you changed `listen_address`, point curl at that host:port instead. For a full SteamPrefill validation workflow (small caches, coalescing, GC), see [Validating Full Functionality](#validating-full-functionality-with-external-tools).
|
||||
@@ -287,6 +290,9 @@ See `config.Validate()` and `steamcache.New` error paths. This ensures the LAN a
|
||||
- The explicit startup guard (reduce size if pre-existing on-disk > cap) runs as the literal last step of bg init, before the barrier opens.
|
||||
- Add a note for operators: very large disk caches (tens/hundreds GB with millions files) may show extended "memory-only or no-cache" behavior at startup (seconds to minutes depending on storage speed); this is by design for responsiveness.
|
||||
- Godoc on `disk.New` and `DiskFS.Size` expanded with the barrier/attach behavior.
|
||||
- Startup logs: Info "Disk slow tier attach pending..." then later "Disk slow tier attached (...)" for disk-only and mixed modes.
|
||||
- `/metrics` exposes `disk_tier_ready` 0/1 and stays responsive during attach (GetMetrics does not block on Size while pending).
|
||||
- `/lancache-heartbeat` header `X-SteamCache-Disk-Tier` mirrors that state.
|
||||
|
||||
#### Garbage Collection Algorithms
|
||||
|
||||
|
||||
@@ -56,6 +56,15 @@ func (sc *SteamCache) handleSpecialEndpoints(w http.ResponseWriter, r *http.Requ
|
||||
logger.Logger.Debug().
|
||||
Str("client_ip", clientIP).
|
||||
Msg("LanCache heartbeat request")
|
||||
diskTier := "disabled"
|
||||
if sc.disk != nil {
|
||||
if sc.metrics.GetDiskTierReady() == 1 {
|
||||
diskTier = "ready"
|
||||
} else {
|
||||
diskTier = "pending"
|
||||
}
|
||||
}
|
||||
w.Header().Add("X-SteamCache-Disk-Tier", diskTier)
|
||||
w.Header().Add("X-LanCache-Processed-By", "SteamCache2")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
_, _ = w.Write(nil) // client write error ignored (heartbeat path; nil write is no-op)
|
||||
|
||||
@@ -31,6 +31,7 @@ type Metrics struct {
|
||||
DiskCacheHits int64
|
||||
Promotions int64
|
||||
Evictions int64
|
||||
DiskTierReady int64 // 0=pending (or unset), 1=ready or no-disk (N/A)
|
||||
|
||||
// Expanded observability (upstream breakdowns, cache write failures, per-service errors)
|
||||
UpstreamErrors int64
|
||||
@@ -114,6 +115,17 @@ func (m *Metrics) SetDiskCacheSize(size int64) {
|
||||
atomic.StoreInt64(&m.DiskCacheSize, size)
|
||||
}
|
||||
|
||||
// SetDiskTierReady sets whether the disk slow tier is attached (1) or still pending (0).
|
||||
// Memory-only (no disk) also uses 1 — meaning "not waiting on disk attach". Reset does not clear this.
|
||||
func (m *Metrics) SetDiskTierReady(ready int64) {
|
||||
atomic.StoreInt64(&m.DiskTierReady, ready)
|
||||
}
|
||||
|
||||
// GetDiskTierReady returns 1 if disk tier is ready (or no disk configured), else 0 while attach pending.
|
||||
func (m *Metrics) GetDiskTierReady() int64 {
|
||||
return atomic.LoadInt64(&m.DiskTierReady)
|
||||
}
|
||||
|
||||
// IncrementMemoryCacheHits increments memory cache hits
|
||||
func (m *Metrics) IncrementMemoryCacheHits() {
|
||||
atomic.AddInt64(&m.MemoryCacheHits, 1)
|
||||
@@ -196,6 +208,7 @@ func (m *Metrics) GetStats() *Stats {
|
||||
TotalBytesSaved: atomic.LoadInt64(&m.TotalBytesSaved),
|
||||
MemoryCacheSize: atomic.LoadInt64(&m.MemoryCacheSize),
|
||||
DiskCacheSize: atomic.LoadInt64(&m.DiskCacheSize),
|
||||
DiskTierReady: atomic.LoadInt64(&m.DiskTierReady),
|
||||
MemoryCacheHits: atomic.LoadInt64(&m.MemoryCacheHits),
|
||||
DiskCacheHits: atomic.LoadInt64(&m.DiskCacheHits),
|
||||
Promotions: atomic.LoadInt64(&m.Promotions),
|
||||
@@ -253,6 +266,7 @@ type Stats struct {
|
||||
MemoryCacheSize int64
|
||||
|
||||
DiskCacheSize int64
|
||||
DiskTierReady int64
|
||||
MemoryCacheHits int64
|
||||
DiskCacheHits int64
|
||||
Promotions int64
|
||||
@@ -297,5 +311,6 @@ func WriteText(w http.ResponseWriter, stats *Stats) {
|
||||
|
||||
_, _ = fmt.Fprintf(w, "memory_cache_size %d\n", stats.MemoryCacheSize)
|
||||
_, _ = fmt.Fprintf(w, "disk_cache_size %d\n", stats.DiskCacheSize)
|
||||
_, _ = fmt.Fprintf(w, "disk_tier_ready %d\n", stats.DiskTierReady)
|
||||
_, _ = fmt.Fprintf(w, "uptime_seconds %.2f\n", stats.Uptime.Seconds())
|
||||
}
|
||||
|
||||
@@ -200,23 +200,33 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
|
||||
|
||||
if disksize == 0 && memorysize != 0 {
|
||||
// memory only mode - no disk
|
||||
sc.metrics.SetDiskTierReady(1) // no disk — N/A / not pending
|
||||
c.SetSlow(mgc)
|
||||
} else if disksize != 0 && memorysize == 0 {
|
||||
// disk only mode: delay attach until disk ready (pure-proxy during scan; Create returns ErrNotFound until slow tier Set)
|
||||
sc.metrics.SetDiskTierReady(0)
|
||||
logger.Logger.Info().Msg("Disk slow tier attach pending; Size barrier in progress")
|
||||
sc.wg.Add(1)
|
||||
go func() {
|
||||
defer sc.wg.Done()
|
||||
t0 := time.Now()
|
||||
_ = d.Size() // block on barrier per design (all Size callers during window do this; documented)
|
||||
select {
|
||||
case <-sc.shutdownCh:
|
||||
return // Shutdown raced; do not attach or SetSlow after stop
|
||||
default:
|
||||
c.SetSlow(dgc)
|
||||
sc.metrics.SetDiskTierReady(1)
|
||||
logger.Logger.Info().
|
||||
Dur("attach_delay", time.Since(t0)).
|
||||
Msg("Disk slow tier attached (disk-only mode); prior traffic had no disk tier")
|
||||
}
|
||||
}()
|
||||
} else if disksize != 0 && memorysize != 0 {
|
||||
// memory and disk mode: fast mem immediate, disk delayed (mem-only during scan)
|
||||
c.SetFast(mgc)
|
||||
sc.metrics.SetDiskTierReady(0)
|
||||
logger.Logger.Info().Msg("Disk slow tier attach pending; Size barrier in progress")
|
||||
sc.wg.Add(1)
|
||||
go func() {
|
||||
defer sc.wg.Done()
|
||||
@@ -227,6 +237,7 @@ func New(address string, memorySize string, diskSize string, diskPath, upstream,
|
||||
return
|
||||
default:
|
||||
c.SetSlow(dgc)
|
||||
sc.metrics.SetDiskTierReady(1)
|
||||
logger.Logger.Info().
|
||||
Dur("attach_delay", time.Since(t0)).
|
||||
Msg("Disk slow tier attached (mixed mode); prior traffic was memory-only")
|
||||
@@ -326,15 +337,13 @@ func (sc *SteamCache) Shutdown() {
|
||||
|
||||
// GetMetrics returns current metrics
|
||||
func (sc *SteamCache) GetMetrics() *metrics.Stats {
|
||||
// Update cache sizes
|
||||
if sc.memory != nil {
|
||||
sc.metrics.SetMemoryCacheSize(sc.memory.Size())
|
||||
}
|
||||
if sc.disk != nil {
|
||||
// Note: blocks on initDone (post-eviction state) for accurate post-attach size during long disk init window.
|
||||
// Skip disk.Size() while attach pending — Size() blocks on initDone and would hang /metrics.
|
||||
if sc.disk != nil && sc.metrics.GetDiskTierReady() == 1 {
|
||||
sc.metrics.SetDiskCacheSize(sc.disk.Size())
|
||||
}
|
||||
|
||||
return sc.metrics.GetStats()
|
||||
}
|
||||
|
||||
|
||||
@@ -1057,6 +1057,12 @@ func TestDiskOnlyDelayedAttach(t *testing.T) {
|
||||
t.Errorf("during init window, expected ErrNotFound from disk-only tiered Create (no slow), got %v", err)
|
||||
}
|
||||
|
||||
// Disk tier is pending while the attach goroutine is in the Size barrier.
|
||||
// GetMetrics must return quickly (it skips disk.Size() while pending) and report 0.
|
||||
if got := sc.GetMetrics().DiskTierReady; got != 0 {
|
||||
t.Errorf("during pending attach, DiskTierReady=%d, want 0", got)
|
||||
}
|
||||
|
||||
// Wait the barrier (exercises the attach go's Size wait)
|
||||
_ = sc.disk.Size()
|
||||
|
||||
@@ -1083,6 +1089,91 @@ func TestDiskOnlyDelayedAttach(t *testing.T) {
|
||||
} else {
|
||||
rc.Close()
|
||||
}
|
||||
|
||||
// After attach, the disk tier must be marked ready (1)
|
||||
if got := sc.GetMetrics().DiskTierReady; got != 1 {
|
||||
t.Errorf("post-attach DiskTierReady=%d, want 1 (ready)", got)
|
||||
}
|
||||
|
||||
// /metrics text output includes the disk_tier_ready line
|
||||
rec := httptest.NewRecorder()
|
||||
metrics.WriteText(rec, sc.GetMetrics())
|
||||
if !bytes.Contains(rec.Body.Bytes(), []byte("disk_tier_ready 1")) {
|
||||
t.Errorf("WriteText output missing \"disk_tier_ready 1\": %q", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestDiskTierSignalMemoryOnly covers memory-only mode: DiskTierReady=1 (N/A, not
|
||||
// waiting on disk attach) and heartbeat header X-SteamCache-Disk-Tier: disabled.
|
||||
func TestDiskTierSignalMemoryOnly(t *testing.T) {
|
||||
sc, err := New("127.0.0.1:0", "1MB", "0", t.TempDir(), "", "lru", "lru", 10, 5, "0", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New memory-only: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { sc.Shutdown() })
|
||||
|
||||
if got := sc.GetMetrics().DiskTierReady; got != 1 {
|
||||
t.Errorf("DiskTierReady=%d, want 1 (memory-only = N/A/not pending)", got)
|
||||
}
|
||||
|
||||
req := httptest.NewRequest("GET", "/lancache-heartbeat", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
sc.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusNoContent {
|
||||
t.Errorf("heartbeat status=%d, want 204", rec.Code)
|
||||
}
|
||||
if got := rec.Header().Get("X-SteamCache-Disk-Tier"); got != "disabled" {
|
||||
t.Errorf("X-SteamCache-Disk-Tier=%q, want disabled", got)
|
||||
}
|
||||
if got := rec.Header().Get("X-LanCache-Processed-By"); got != "SteamCache2" {
|
||||
t.Errorf("X-LanCache-Processed-By=%q, want SteamCache2", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDiskTierSignalMixedPendingReady covers mixed mode: DiskTierReady=0 (header
|
||||
// pending) while the disk attach is in the Size barrier, then DiskTierReady=1
|
||||
// (header ready) after the barrier opens and the attach goroutine sets SetSlow.
|
||||
func TestDiskTierSignalMixedPendingReady(t *testing.T) {
|
||||
td := t.TempDir()
|
||||
diskPath := filepath.Join(td, "disk")
|
||||
if err := os.MkdirAll(diskPath, 0755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sc, err := New("127.0.0.1:0", "1MB", "10MB", diskPath, "", "lru", "lru", 10, 1, "0", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New mixed: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { sc.Shutdown() })
|
||||
|
||||
// Immediately in the pending window
|
||||
if got := sc.GetMetrics().DiskTierReady; got != 0 {
|
||||
t.Errorf("immediate DiskTierReady=%d, want 0 (pending)", got)
|
||||
}
|
||||
req := httptest.NewRequest("GET", "/lancache-heartbeat", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
sc.ServeHTTP(rec, req)
|
||||
if got := rec.Header().Get("X-SteamCache-Disk-Tier"); got != "pending" {
|
||||
t.Errorf("heartbeat header=%q, want pending", got)
|
||||
}
|
||||
|
||||
// Wait the barrier, then retry until the attach goroutine flips the flag
|
||||
_ = sc.disk.Size()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if sc.GetMetrics().DiskTierReady == 1 {
|
||||
break
|
||||
}
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
}
|
||||
if got := sc.GetMetrics().DiskTierReady; got != 1 {
|
||||
t.Fatalf("DiskTierReady=%d after barrier, want 1 (ready)", got)
|
||||
}
|
||||
|
||||
rec2 := httptest.NewRecorder()
|
||||
sc.ServeHTTP(rec2, httptest.NewRequest("GET", "/lancache-heartbeat", nil))
|
||||
if got := rec2.Header().Get("X-SteamCache-Disk-Tier"); got != "ready" {
|
||||
t.Errorf("heartbeat header=%q, want ready", got)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Phase 2: narrow black-box tests for the new wrapper types ---
|
||||
|
||||
Reference in New Issue
Block a user