fix(tsdb): survive low disk space

pkg/tsdb/tsdb.go

@@ -1,15 +1,26 @@
 package tsdb
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"path/filepath"
+	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
+// ErrDiskFull is returned when VictoriaMetrics cannot initialize because the
+// filesystem containing the TSDB has run out of space.
+var ErrDiskFull = errors.New("TSDB disk is full")
+
+type readOnlyChecker interface {
+	IsReadOnly() bool
+}
+
 // TSDB 封装 VictoriaMetrics 存储
 type TSDB struct {
 	storage *storage.Storage
@@ -18,6 +29,13 @@ type TSDB struct {
 	closed  bool
 
 	writer *bufferedWriter
+
+	readOnly     readOnlyChecker
+	addRowsFn    func([]storage.MetricRow, uint8)
+	debugFlushFn func()
+
+	readOnlyObserved atomic.Bool
+	diskFullPaused   atomic.Bool
 }
 
 // InitGlobalSettings 初始化 VictoriaMetrics 包级别的全局设置。
@@ -35,7 +53,18 @@ func InitGlobalSettings(config *Config) {
 }
 
 // Open 打开或创建 TSDB 存储
-func Open(config *Config) (*TSDB, error) {
+func Open(config *Config) (db *TSDB, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			if diskErr := diskFullPanicError(recovered); diskErr != nil {
+				db = nil
+				err = diskErr
+				return
+			}
+			panic(recovered)
+		}
+	}()
+
 	if config == nil {
 		config = DefaultConfig()
 	}
@@ -59,9 +88,12 @@ func Open(config *Config) (*TSDB, error) {
 
 	stor := storage.MustOpenStorage(dataPath, opts)
 
-	db := &TSDB{
-		storage: stor,
-		config:  config,
+	db = &TSDB{
+		storage:      stor,
+		config:       config,
+		readOnly:     stor,
+		addRowsFn:    stor.AddRows,
+		debugFlushFn: stor.DebugFlush,
 	}
 
 	db.writer = newBufferedWriter(db, config.WriteBufferSize, config.WriteBufferFlushInterval)
@@ -72,6 +104,86 @@ func Open(config *Config) (*TSDB, error) {
 	return db, nil
 }
 
+func diskFullPanicError(recovered any) error {
+	message := strings.ToLower(fmt.Sprint(recovered))
+	for _, marker := range []string{
+		"no space left on device",
+		"disk quota exceeded",
+		"not enough space on the disk",
+	} {
+		if strings.Contains(message, marker) {
+			return fmt.Errorf("%w: %v", ErrDiskFull, recovered)
+		}
+	}
+	return nil
+}
+
+// WritesPaused reports whether new samples are currently being discarded.
+// Queries remain available while VictoriaMetrics is in read-only mode.
+func (db *TSDB) WritesPaused() bool {
+	if db.diskFullPaused.Load() {
+		return true
+	}
+	return db.readOnly != nil && db.readOnly.IsReadOnly()
+}
+
+func (db *TSDB) acceptsWrites() bool {
+	if db.diskFullPaused.Load() {
+		return false
+	}
+
+	readOnly := db.readOnly != nil && db.readOnly.IsReadOnly()
+	if readOnly {
+		if db.readOnlyObserved.CompareAndSwap(false, true) {
+			log.Println("NEZHA>> TSDB writes paused because free disk space is below the configured limit; dashboard, queries, and alerts remain available")
+		}
+		return false
+	}
+
+	if db.readOnlyObserved.CompareAndSwap(true, false) {
+		log.Println("NEZHA>> TSDB writes resumed after free disk space recovered")
+	}
+	return true
+}
+
+func (db *TSDB) pauseWritesAfterDiskFull(recovered any) {
+	if db.diskFullPaused.CompareAndSwap(false, true) {
+		log.Printf("NEZHA>> TSDB disk is full; disabling TSDB writes until dashboard restart while keeping dashboard, queries, and alerts running: %v", recovered)
+	}
+}
+
+func (db *TSDB) addRowsSafely(rows []storage.MetricRow) {
+	if len(rows) == 0 || !db.acceptsWrites() {
+		return
+	}
+
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			if diskFullPanicError(recovered) == nil {
+				panic(recovered)
+			}
+			db.pauseWritesAfterDiskFull(recovered)
+		}
+	}()
+	db.addRowsFn(rows, 64)
+}
+
+func (db *TSDB) debugFlushSafely() {
+	if !db.acceptsWrites() {
+		return
+	}
+
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			if diskFullPanicError(recovered) == nil {
+				panic(recovered)
+			}
+			db.pauseWritesAfterDiskFull(recovered)
+		}
+	}()
+	db.debugFlushFn()
+}
+
 // Close 关闭 TSDB 存储
 func (db *TSDB) Close() error {
 	db.mu.Lock()
@@ -110,8 +222,14 @@ func (db *TSDB) IsClosed() bool {
 
 // Flush 强制刷盘（主要用于测试）
 func (db *TSDB) Flush() {
+	db.mu.RLock()
+	defer db.mu.RUnlock()
+	if db.closed {
+		return
+	}
+
 	if db.writer != nil {
 		db.writer.flush()
 	}
-	db.storage.DebugFlush()
+	db.debugFlushSafely()
 }