Merge branch 'nezhahq:master' into master

pkg/tsdb/disk_guard_test.go

@@ -0,0 +1,124 @@
+package tsdb
+
+import (
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type testReadOnlyChecker struct {
+	value atomic.Bool
+}
+
+func (c *testReadOnlyChecker) IsReadOnly() bool {
+	return c.value.Load()
+}
+
+func newDiskGuardTestDB(t *testing.T) *TSDB {
+	t.Helper()
+	db, err := Open(&Config{
+		DataPath:           filepath.Join(t.TempDir(), "tsdb"),
+		RetentionDays:      1,
+		MinFreeDiskSpaceGB: 1,
+		DedupInterval:      time.Second,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, db.Close()) })
+	return db
+}
+
+func bufferedRows(w *bufferedWriter) int {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return len(w.buffer)
+}
+
+func TestReadOnlyStorageDropsWritesAndKeepsQueriesAvailable(t *testing.T) {
+	db := newDiskGuardTestDB(t)
+	checker := &testReadOnlyChecker{}
+	db.readOnly = checker
+	originalAddRows := db.addRowsFn
+	var addRowsCalls atomic.Int32
+	db.addRowsFn = func(rows []storage.MetricRow, precisionBits uint8) {
+		addRowsCalls.Add(1)
+		originalAddRows(rows, precisionBits)
+	}
+
+	firstTimestamp := time.Now().Add(-time.Minute)
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: firstTimestamp,
+		CPU:       10,
+	}))
+	db.Flush()
+	require.Equal(t, int32(1), addRowsCalls.Load())
+
+	before, err := db.QueryServerMetrics(1, MetricServerCPU, Period1Day)
+	require.NoError(t, err)
+	require.NotEmpty(t, before)
+
+	checker.value.Store(true)
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: firstTimestamp.Add(2 * time.Second),
+		CPU:       20,
+	}))
+	assert.True(t, db.WritesPaused())
+	assert.Zero(t, bufferedRows(db.writer), "read-only writes must not accumulate in memory")
+	assert.Equal(t, int32(1), addRowsCalls.Load(), "read-only samples must not reach VictoriaMetrics")
+
+	afterDrop, err := db.QueryServerMetrics(1, MetricServerCPU, Period1Day)
+	require.NoError(t, err)
+	assert.Equal(t, before, afterDrop, "read-only mode must preserve existing query access")
+
+	checker.value.Store(false)
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: firstTimestamp.Add(4 * time.Second),
+		CPU:       30,
+	}))
+	db.Flush()
+	assert.False(t, db.WritesPaused())
+	assert.Equal(t, int32(2), addRowsCalls.Load(), "writes must resume after storage leaves read-only mode")
+}
+
+func TestDiskFullAddRowsPanicPausesWrites(t *testing.T) {
+	db := newDiskGuardTestDB(t)
+	db.addRowsFn = func([]storage.MetricRow, uint8) {
+		panic("write data: no space left on device")
+	}
+
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: time.Now(),
+		CPU:       10,
+	}))
+	assert.NotPanics(t, db.Flush)
+	assert.True(t, db.WritesPaused())
+
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: time.Now().Add(time.Second),
+		CPU:       20,
+	}))
+	assert.Zero(t, bufferedRows(db.writer))
+}
+
+func TestNonDiskStoragePanicIsNotHidden(t *testing.T) {
+	db := newDiskGuardTestDB(t)
+	db.addRowsFn = func([]storage.MetricRow, uint8) {
+		panic("storage invariant failed")
+	}
+
+	require.NoError(t, db.WriteServerMetrics(&ServerMetrics{
+		ServerID:  1,
+		Timestamp: time.Now(),
+		CPU:       10,
+	}))
+	assert.PanicsWithValue(t, "storage invariant failed", db.Flush)
+}

pkg/tsdb/maintenance.go

@@ -10,8 +10,11 @@ func (db *TSDB) Maintenance() {
 	if db.closed {
 		return
 	}
+	if !db.acceptsWrites() {
+		return
+	}
 
 	log.Println("NEZHA>> TSDB starting maintenance (flush)...")
-	db.storage.DebugFlush()
+	db.debugFlushSafely()
 	log.Println("NEZHA>> TSDB maintenance completed")
 }

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()
 }

pkg/tsdb/writer.go

@@ -47,18 +47,29 @@ func (w *bufferedWriter) flushLoop() {
 }
 
 func (w *bufferedWriter) write(rows []storage.MetricRow) {
+	if !w.db.acceptsWrites() {
+		w.discard()
+		return
+	}
+
 	w.mu.Lock()
 	w.buffer = append(w.buffer, rows...)
 	if len(w.buffer) >= w.maxSize {
 		rows := w.buffer
 		w.buffer = make([]storage.MetricRow, 0, w.maxSize)
 		w.mu.Unlock()
-		w.db.storage.AddRows(rows, 64)
+		w.db.addRowsSafely(rows)
 		return
 	}
 	w.mu.Unlock()
 }
 
+func (w *bufferedWriter) discard() {
+	w.mu.Lock()
+	w.buffer = make([]storage.MetricRow, 0, w.maxSize)
+	w.mu.Unlock()
+}
+
 func (w *bufferedWriter) flush() {
 	w.mu.Lock()
 	if len(w.buffer) == 0 {
@@ -69,7 +80,7 @@ func (w *bufferedWriter) flush() {
 	w.buffer = make([]storage.MetricRow, 0, w.maxSize)
 	w.mu.Unlock()
 
-	w.db.storage.AddRows(rows, 64)
+	w.db.addRowsSafely(rows)
 }
 
 func (w *bufferedWriter) stop() {
@@ -171,7 +182,7 @@ func (db *TSDB) WriteServerMetrics(m *ServerMetrics) error {
 	if db.writer != nil {
 		db.writer.write(rows)
 	} else {
-		db.storage.AddRows(rows, 64)
+		db.addRowsSafely(rows)
 	}
 	return nil
 }
@@ -200,7 +211,7 @@ func (db *TSDB) WriteServiceMetrics(m *ServiceMetrics) error {
 	if db.writer != nil {
 		db.writer.write(rows)
 	} else {
-		db.storage.AddRows(rows, 64)
+		db.addRowsSafely(rows)
 	}
 	return nil
 }
@@ -265,7 +276,7 @@ func (db *TSDB) WriteBatchServerMetrics(metrics []*ServerMetrics) error {
 	if db.writer != nil {
 		db.writer.write(rows)
 	} else {
-		db.storage.AddRows(rows, 64)
+		db.addRowsSafely(rows)
 	}
 	return nil
 }
@@ -295,7 +306,7 @@ func (db *TSDB) WriteBatchServiceMetrics(metrics []*ServiceMetrics) error {
 	if db.writer != nil {
 		db.writer.write(rows)
 	} else {
-		db.storage.AddRows(rows, 64)
+		db.addRowsSafely(rows)
 	}
 	return nil
 }