diff --git a/pkg/tsdb/disk_guard_test.go b/pkg/tsdb/disk_guard_test.go new file mode 100644 index 00000000..2fa2109a --- /dev/null +++ b/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) +} diff --git a/pkg/tsdb/maintenance.go b/pkg/tsdb/maintenance.go index fe9c73e1..4f2c4a66 100644 --- a/pkg/tsdb/maintenance.go +++ b/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") } diff --git a/pkg/tsdb/tsdb.go b/pkg/tsdb/tsdb.go index d3627b61..9ae4a16b 100644 --- a/pkg/tsdb/tsdb.go +++ b/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() } diff --git a/pkg/tsdb/writer.go b/pkg/tsdb/writer.go index 7896c094..d6471e2e 100644 --- a/pkg/tsdb/writer.go +++ b/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 } diff --git a/service/singleton/tsdb.go b/service/singleton/tsdb.go index 745e83d6..e5311bb1 100644 --- a/service/singleton/tsdb.go +++ b/service/singleton/tsdb.go @@ -1,6 +1,7 @@ package singleton import ( + "errors" "log" "time" @@ -10,6 +11,8 @@ import ( var TSDBShared *tsdb.TSDB +var openTSDB = tsdb.Open + func InitTSDB() error { config := &tsdb.Config{ RetentionDays: 30, @@ -44,11 +47,21 @@ func InitTSDB() error { return nil } - var err error - TSDBShared, err = tsdb.Open(config) + TSDBShared = nil + db, err := openTSDB(config) if err != nil { + if errors.Is(err, tsdb.ErrDiskFull) { + log.Printf("NEZHA>> Warning: TSDB is unavailable because its disk is full; dashboard and alerts will continue without TSDB writes: %v", err) + if DB != nil { + if migrateErr := DB.AutoMigrate(model.ServiceHistory{}); migrateErr != nil { + log.Printf("NEZHA>> Warning: failed to prepare SQLite service history fallback: %v", migrateErr) + } + } + return nil + } return err } + TSDBShared = db log.Println("NEZHA>> TSDB initialized successfully") diff --git a/service/singleton/tsdb_disk_guard_test.go b/service/singleton/tsdb_disk_guard_test.go new file mode 100644 index 00000000..50f71f3f --- /dev/null +++ b/service/singleton/tsdb_disk_guard_test.go @@ -0,0 +1,58 @@ +package singleton + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/model" + "github.com/nezhahq/nezha/pkg/tsdb" +) + +func TestInitTSDBContinuesWhenDiskIsFull(t *testing.T) { + originalConf := Conf + originalDB := DB + originalShared := TSDBShared + originalOpen := openTSDB + t.Cleanup(func() { + Conf = originalConf + DB = originalDB + TSDBShared = originalShared + openTSDB = originalOpen + }) + + Conf = &ConfigClass{Config: &model.Config{ + TSDB: model.TSDBConf{DataPath: "/full/tsdb"}, + }} + DB = nil + TSDBShared = nil + openTSDB = func(*tsdb.Config) (*tsdb.TSDB, error) { + return nil, fmt.Errorf("open failed: %w", tsdb.ErrDiskFull) + } + + require.NoError(t, InitTSDB()) + require.Nil(t, TSDBShared) +} + +func TestInitTSDBStillReturnsNonDiskErrors(t *testing.T) { + originalConf := Conf + originalShared := TSDBShared + originalOpen := openTSDB + t.Cleanup(func() { + Conf = originalConf + TSDBShared = originalShared + openTSDB = originalOpen + }) + + Conf = &ConfigClass{Config: &model.Config{ + TSDB: model.TSDBConf{DataPath: "/broken/tsdb"}, + }} + TSDBShared = nil + openTSDB = func(*tsdb.Config) (*tsdb.TSDB, error) { + return nil, fmt.Errorf("permission denied") + } + + require.ErrorContains(t, InitTSDB(), "permission denied") + require.Nil(t, TSDBShared) +}