CBG-5472: Make the stats logger independent of the database lock

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

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

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.

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")