diff --git a/base/stats.go b/base/stats.go index 72856f7333..676b4914cf 100644 --- a/base/stats.go +++ b/base/stats.go @@ -163,10 +163,23 @@ func init() { } // This String() is to satisfy the expvar.Var interface which is used to produce the expvar endpoint output. +// The dbStatsMapMutex is held only long enough to shallow-copy the per-db stats map; the (potentially +// slow) marshal runs outside the lock so stats serialization neither blocks, nor is blocked by, +// database registration/removal (NewDBStats/ClearDBStats). This keeps the stats logger flowing even +// while databases are being added or removed (CBG-5472). func (s *SgwStats) String() string { s.dbStatsMapMutex.Lock() - bytes, err := JSONMarshalCanonical(s) + statsCopy := &SgwStats{ + GlobalStats: s.GlobalStats, + DbStats: make(map[string]*DbStats, len(s.DbStats)), + ReplicatorStats: s.ReplicatorStats, + } + for name, dbStats := range s.DbStats { + statsCopy.DbStats[name] = dbStats + } s.dbStatsMapMutex.Unlock() + + bytes, err := JSONMarshalCanonical(statsCopy) if err != nil { ErrorfCtx(context.Background(), "Unable to Marshal SgwStats: %v", err) return "null" @@ -1346,8 +1359,9 @@ type QueryStat struct { } func (s *SgwStats) NewDBStats(name string, deltaSyncEnabled bool, importEnabled bool, viewsEnabled bool, metadataMigrationEnabled bool, queryNames []string, collections []string) (*DbStats, error) { - s.dbStatsMapMutex.Lock() - defer s.dbStatsMapMutex.Unlock() + // Build and register (with Prometheus) the db's stats without holding dbStatsMapMutex, so this + // doesn't block stats serialization (String) - the mutex is only taken to insert into the map + // below (CBG-5472). dbStats := &DbStats{ dbName: name, DbReplicatorStats: make(map[string]*DbReplicatorStats), @@ -1416,45 +1430,53 @@ func (s *SgwStats) NewDBStats(name string, deltaSyncEnabled bool, importEnabled return nil, err } + s.dbStatsMapMutex.Lock() s.DbStats[name] = dbStats + s.dbStatsMapMutex.Unlock() return dbStats, nil } func (s *SgwStats) ClearDBStats(name string) { + // Remove the db from the map under dbStatsMapMutex, then unregister its stats from Prometheus + // outside the lock. The unregister* calls only touch the Prometheus registry (not the DbStats + // maps), so serialization (String) - which marshals a shallow copy - never blocks behind this + // (CBG-5472). s.dbStatsMapMutex.Lock() - defer s.dbStatsMapMutex.Unlock() - - if _, ok := s.DbStats[name]; !ok { + dbStats, ok := s.DbStats[name] + if ok { + delete(s.DbStats, name) + } + s.dbStatsMapMutex.Unlock() + if !ok { return } - for scopeAndCollectionName := range s.DbStats[name].CollectionStats { - s.DbStats[name].unregisterCollectionStats(scopeAndCollectionName) + for scopeAndCollectionName := range dbStats.CollectionStats { + dbStats.unregisterCollectionStats(scopeAndCollectionName) } - s.DbStats[name].unregisterCacheStats() - s.DbStats[name].unregisterCBLReplicationPullStats() - s.DbStats[name].unregisterCBLReplicationPushStats() - for replName := range s.DbStats[name].DbReplicatorStats { - s.DbStats[name].unregisterReplicationStats(replName) + dbStats.unregisterCacheStats() + dbStats.unregisterCBLReplicationPullStats() + dbStats.unregisterCBLReplicationPushStats() + for replName := range dbStats.DbReplicatorStats { + dbStats.unregisterReplicationStats(replName) } - s.DbStats[name].unregisterDatabaseStats() - s.DbStats[name].unregisterSecurityStats() + dbStats.unregisterDatabaseStats() + dbStats.unregisterSecurityStats() - if s.DbStats[name].DeltaSyncStats != nil { - s.DbStats[name].unregisterDeltaSyncStats() + if dbStats.DeltaSyncStats != nil { + dbStats.unregisterDeltaSyncStats() } - if s.DbStats[name].SharedBucketImportStats != nil { - s.DbStats[name].unregisterSharedBucketImportStats() + if dbStats.SharedBucketImportStats != nil { + dbStats.unregisterSharedBucketImportStats() } - if s.DbStats[name].MigrationStats != nil { - s.DbStats[name].unregisterMigrationStats() + if dbStats.MigrationStats != nil { + dbStats.unregisterMigrationStats() } - s.DbStats[name].unregisterQueryStats() - delete(s.DbStats, name) + dbStats.unregisterQueryStats() } // Removes the per-database stats for this database by removing the database from the map diff --git a/base/stats_test.go b/base/stats_test.go index 8b05a7f9d8..49fddfdfcd 100644 --- a/base/stats_test.go +++ b/base/stats_test.go @@ -12,7 +12,9 @@ package base import ( "expvar" + "fmt" "math" + "sync" "testing" "github.com/couchbase/sync_gateway/testing/require" @@ -197,3 +199,51 @@ func TestSgwFloatStatMarshalNonFinite(t *testing.T) { }) } } + +// TestClearDBStatsRemovesFromSerialization is a characterization test for CBG-5472: a database's +// stats appear in the serialized output after NewDBStats and are gone after ClearDBStats. The +// CBG-5472 change moves Prometheus (un)registration out of the dbStatsMapMutex critical section, so +// this pins the externally-observable serialization behaviour that must be preserved. +func TestClearDBStatsRemovesFromSerialization(t *testing.T) { + stats := &SgwStats{DbStats: map[string]*DbStats{}} + const dbName = "TestClearDBStatsRemovesFromSerialization_db" + + _, err := stats.NewDBStats(dbName, false, false, false, false, nil, nil) + require.NoError(t, err) + require.Contains(t, stats.String(), dbName) + + stats.ClearDBStats(dbName) + require.NotContains(t, stats.String(), dbName) +} + +// TestStatsSerializationConcurrentWithDBRegistration verifies CBG-5472: serializing stats (String) +// runs concurrently with database registration/removal without racing or deadlocking. Most valuable +// under -race, which must stay clean after the dbStatsMapMutex critical sections are narrowed. +func TestStatsSerializationConcurrentWithDBRegistration(t *testing.T) { + stats := &SgwStats{DbStats: map[string]*DbStats{}} + + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 200; j++ { + _ = stats.String() + } + }() + } + for i := 0; i < 4; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + // Distinct db name per goroutine so concurrent registrations never collide in Prometheus. + name := fmt.Sprintf("TestStatsSerializationConcurrentWithDBRegistration_db_%d", id) + for j := 0; j < 50; j++ { + _, err := stats.NewDBStats(name, false, false, false, false, nil, nil) + assert.NoError(t, err) + stats.ClearDBStats(name) + } + }(i) + } + wg.Wait() +} diff --git a/rest/server_context.go b/rest/server_context.go index d42d402a89..985f20b94e 100644 --- a/rest/server_context.go +++ b/rest/server_context.go @@ -78,6 +78,12 @@ type ServerContext struct { _databases map[string]*db.DatabaseContext // _databases is a map of dbname to db.DatabaseContext _databasesLock sync.RWMutex // Lock for _databases and other db-specific maps above + // databasesSnapshot is a lock-free copy of the _databases values, read by the stats logger so + // stats logging never blocks on _databasesLock (which config updates can hold for a long time + // while waiting on index readiness - CBG-5472). Refreshed under _databasesLock whenever + // _databases changes, via _updateDatabasesSnapshot. No _ prefix: it is read without the lock. + databasesSnapshot atomic.Pointer[[]*db.DatabaseContext] + // serverCtx is cancelled by Close() to broadcast server shutdown to all background // goroutines that hold a reference to this ServerContext (e.g. handleDbOnline). serverCtx context.Context @@ -323,6 +329,7 @@ func (sc *ServerContext) Close(ctx context.Context) { _ = db.EventMgr.RaiseDBStateChangeEvent(ctx, db.Name, "offline", "Database context closed", &sc.Config.API.AdminInterface) } sc._databases = nil + sc._updateDatabasesSnapshot() sc.invalidDatabaseConfigTracking.dbNames = nil } @@ -1204,6 +1211,7 @@ func (sc *ServerContext) _getOrAddDatabaseFromConfig(ctx context.Context, config // Register it so HTTP handlers can find it: sc._databases[dbcontext.Name] = dbcontext + sc._updateDatabasesSnapshot() sc._dbConfigs[dbcontext.Name] = &RuntimeDatabaseConfig{DatabaseConfig: config} sc._dbRegistry[dbName] = struct{}{} for _, name := range fqCollections { @@ -1811,6 +1819,7 @@ func (sc *ServerContext) _unloadDatabase(ctx context.Context, dbName string) boo base.InfofCtx(ctx, base.KeyAll, "Closing db /%s (bucket %q)", base.MD(dbCtx.Name), base.MD(dbCtx.Bucket.GetName())) dbCtx.Close(ctx) delete(sc._databases, dbName) + sc._updateDatabasesSnapshot() return true } @@ -1927,17 +1936,29 @@ func (sc *ServerContext) logNetworkInterfaceStats(ctx context.Context) { } -// Updates stats that are more efficient to calculate at stats collection time -func (sc *ServerContext) updateCalculatedStats(ctx context.Context) { - sc._databasesLock.RLock() - defer sc._databasesLock.RUnlock() +// _updateDatabasesSnapshot refreshes the lock-free snapshot of database contexts read by the stats +// logger (see databasesSnapshot). The caller must hold sc._databasesLock (write). +func (sc *ServerContext) _updateDatabasesSnapshot() { + snapshot := make([]*db.DatabaseContext, 0, len(sc._databases)) for _, dbContext := range sc._databases { - dbState := atomic.LoadUint32(&dbContext.State) - if dbState == db.DBOnline { + snapshot = append(snapshot, dbContext) + } + sc.databasesSnapshot.Store(&snapshot) +} + +// Updates stats that are more efficient to calculate at stats collection time. +// Reads a lock-free snapshot of the databases so it never blocks on _databasesLock, which a config +// update can hold for a long time while waiting on index readiness (CBG-5472). +func (sc *ServerContext) updateCalculatedStats(ctx context.Context) { + snapshot := sc.databasesSnapshot.Load() + if snapshot == nil { + return + } + for _, dbContext := range *snapshot { + if atomic.LoadUint32(&dbContext.State) == db.DBOnline { dbContext.UpdateCalculatedStats(ctx) } } - } // initializeGoCBAgent Obtains a gocb agent from the current server connection. Requires the agent to be closed after use. diff --git a/rest/server_context_test.go b/rest/server_context_test.go index bb206edc75..ba769bbd6d 100644 --- a/rest/server_context_test.go +++ b/rest/server_context_test.go @@ -382,6 +382,83 @@ func TestStatsLoggerStopped(t *testing.T) { time.Sleep(time.Millisecond * 10) } +// TestStatsLoggerIndependentOfDatabaseLock verifies CBG-5472: the stats logger must not block on +// _databasesLock. A config update holds that write lock across a potentially very long (infinite) +// index-readiness wait; while it does, updateCalculatedStats must still be able to run so stats +// keep being logged. Here we simulate the long-held write lock and assert the stats collection +// completes promptly instead of blocking behind it. +func TestStatsLoggerIndependentOfDatabaseLock(t *testing.T) { + rt := NewRestTester(t, nil) + defer rt.Close() + + // Read the database set (takes _databasesLock) before we grab the write lock below. + require.Len(t, rt.ServerContext().AllDatabases(), 1) + + sc := rt.ServerContext() + ctx := base.TestCtx(t) + + // Simulate a config update holding the write lock for a long time (e.g. blocking on indexes). + sc._databasesLock.Lock() + defer sc._databasesLock.Unlock() + + done := make(chan struct{}) + go func() { + sc.updateCalculatedStats(ctx) + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("updateCalculatedStats blocked on _databasesLock held by a simulated config update (CBG-5472)") + } +} + +// TestDatabasesSnapshotTracksDatabases verifies the lock-free stats-logger snapshot (CBG-5472) stays +// in sync with _databases as databases are added and removed, so the stats logger neither misses a +// live database nor processes a removed one. +func TestDatabasesSnapshotTracksDatabases(t *testing.T) { + rt := NewRestTester(t, nil) + defer rt.Close() + + sc := rt.ServerContext() + dbName := rt.GetDatabase().Name + + snapshot := sc.databasesSnapshot.Load() + require.NotNil(t, snapshot) + require.Len(t, *snapshot, 1) + + require.True(t, sc.RemoveDatabase(base.TestCtx(t), dbName, "test cleanup")) + + snapshot = sc.databasesSnapshot.Load() + require.NotNil(t, snapshot) + require.Empty(t, *snapshot) +} + +// TestStatsLoggerConcurrentWithDatabaseRemoval exercises the CBG-5472 lock-free stats reader against +// a concurrent database removal/close. Because the reader no longer holds _databasesLock, it can +// observe (via the snapshot) a database that is being closed; this must remain memory-safe. Most +// valuable under -race. +func TestStatsLoggerConcurrentWithDatabaseRemoval(t *testing.T) { + rt := NewRestTester(t, nil) + defer rt.Close() + + sc := rt.ServerContext() + ctx := base.TestCtx(t) + dbName := rt.GetDatabase().Name + + readerDone := make(chan struct{}) + go func() { + defer close(readerDone) + for i := 0; i < 2000; i++ { + sc.updateCalculatedStats(ctx) + } + }() + + require.True(t, sc.RemoveDatabase(ctx, dbName, "concurrent stats test")) + <-readerDone +} + func TestObtainManagementEndpointsFromServerContext(t *testing.T) { if base.UnitTestUrlIsWalrus() { t.Skip("Test requires Couchbase Server")