Skip to content

Commit de9149d

Browse files
bbrksclaude
andcommitted
CBG-5472: Make the stats logger independent of the database lock
The stats logger could stall, leaving gaps in sg_stats.log, whenever a database config update held a lock the stats path also needed. Two independent locks were involved: - _databasesLock: updateCalculatedStats took RLock, but a config update holds the write lock across an infinite index-readiness wait, blocking the reader for the whole (unbounded) window. - base.SgwStats.dbStatsMapMutex: String() held it for the entire marshal, and NewDBStats/ClearDBStats held it for all Prometheus (un)registration under _databasesLock. Decouple the reader from both: - ServerContext now keeps a lock-free atomic snapshot of the database contexts, refreshed under _databasesLock at each _databases mutation. updateCalculatedStats reads the snapshot, never the lock. - dbStatsMapMutex is now held only for the map operation: String() marshals a shallow copy, and NewDBStats/ClearDBStats do their Prometheus work outside the lock. The unregister* calls only touch Prometheus (not the DbStats maps), so this stays race-free. The stats ticker now keeps emitting (stale-but-present) while databases are added, removed or reloaded. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 7eb230c commit de9149d

4 files changed

Lines changed: 200 additions & 30 deletions

File tree

base/stats.go

Lines changed: 45 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -163,10 +163,23 @@ func init() {
163163
}
164164

165165
// This String() is to satisfy the expvar.Var interface which is used to produce the expvar endpoint output.
166+
// The dbStatsMapMutex is held only long enough to shallow-copy the per-db stats map; the (potentially
167+
// slow) marshal runs outside the lock so stats serialization neither blocks, nor is blocked by,
168+
// database registration/removal (NewDBStats/ClearDBStats). This keeps the stats logger flowing even
169+
// while databases are being added or removed (CBG-5472).
166170
func (s *SgwStats) String() string {
167171
s.dbStatsMapMutex.Lock()
168-
bytes, err := JSONMarshalCanonical(s)
172+
statsCopy := &SgwStats{
173+
GlobalStats: s.GlobalStats,
174+
DbStats: make(map[string]*DbStats, len(s.DbStats)),
175+
ReplicatorStats: s.ReplicatorStats,
176+
}
177+
for name, dbStats := range s.DbStats {
178+
statsCopy.DbStats[name] = dbStats
179+
}
169180
s.dbStatsMapMutex.Unlock()
181+
182+
bytes, err := JSONMarshalCanonical(statsCopy)
170183
if err != nil {
171184
ErrorfCtx(context.Background(), "Unable to Marshal SgwStats: %v", err)
172185
return "null"
@@ -1346,8 +1359,9 @@ type QueryStat struct {
13461359
}
13471360

13481361
func (s *SgwStats) NewDBStats(name string, deltaSyncEnabled bool, importEnabled bool, viewsEnabled bool, metadataMigrationEnabled bool, queryNames []string, collections []string) (*DbStats, error) {
1349-
s.dbStatsMapMutex.Lock()
1350-
defer s.dbStatsMapMutex.Unlock()
1362+
// Build and register (with Prometheus) the db's stats without holding dbStatsMapMutex, so this
1363+
// doesn't block stats serialization (String) - the mutex is only taken to insert into the map
1364+
// below (CBG-5472).
13511365
dbStats := &DbStats{
13521366
dbName: name,
13531367
DbReplicatorStats: make(map[string]*DbReplicatorStats),
@@ -1416,45 +1430,53 @@ func (s *SgwStats) NewDBStats(name string, deltaSyncEnabled bool, importEnabled
14161430
return nil, err
14171431
}
14181432

1433+
s.dbStatsMapMutex.Lock()
14191434
s.DbStats[name] = dbStats
1435+
s.dbStatsMapMutex.Unlock()
14201436
return dbStats, nil
14211437
}
14221438

14231439
func (s *SgwStats) ClearDBStats(name string) {
1440+
// Remove the db from the map under dbStatsMapMutex, then unregister its stats from Prometheus
1441+
// outside the lock. The unregister* calls only touch the Prometheus registry (not the DbStats
1442+
// maps), so serialization (String) - which marshals a shallow copy - never blocks behind this
1443+
// (CBG-5472).
14241444
s.dbStatsMapMutex.Lock()
1425-
defer s.dbStatsMapMutex.Unlock()
1426-
1427-
if _, ok := s.DbStats[name]; !ok {
1445+
dbStats, ok := s.DbStats[name]
1446+
if ok {
1447+
delete(s.DbStats, name)
1448+
}
1449+
s.dbStatsMapMutex.Unlock()
1450+
if !ok {
14281451
return
14291452
}
14301453

1431-
for scopeAndCollectionName := range s.DbStats[name].CollectionStats {
1432-
s.DbStats[name].unregisterCollectionStats(scopeAndCollectionName)
1454+
for scopeAndCollectionName := range dbStats.CollectionStats {
1455+
dbStats.unregisterCollectionStats(scopeAndCollectionName)
14331456
}
14341457

1435-
s.DbStats[name].unregisterCacheStats()
1436-
s.DbStats[name].unregisterCBLReplicationPullStats()
1437-
s.DbStats[name].unregisterCBLReplicationPushStats()
1438-
for replName := range s.DbStats[name].DbReplicatorStats {
1439-
s.DbStats[name].unregisterReplicationStats(replName)
1458+
dbStats.unregisterCacheStats()
1459+
dbStats.unregisterCBLReplicationPullStats()
1460+
dbStats.unregisterCBLReplicationPushStats()
1461+
for replName := range dbStats.DbReplicatorStats {
1462+
dbStats.unregisterReplicationStats(replName)
14401463
}
1441-
s.DbStats[name].unregisterDatabaseStats()
1442-
s.DbStats[name].unregisterSecurityStats()
1464+
dbStats.unregisterDatabaseStats()
1465+
dbStats.unregisterSecurityStats()
14431466

1444-
if s.DbStats[name].DeltaSyncStats != nil {
1445-
s.DbStats[name].unregisterDeltaSyncStats()
1467+
if dbStats.DeltaSyncStats != nil {
1468+
dbStats.unregisterDeltaSyncStats()
14461469
}
14471470

1448-
if s.DbStats[name].SharedBucketImportStats != nil {
1449-
s.DbStats[name].unregisterSharedBucketImportStats()
1471+
if dbStats.SharedBucketImportStats != nil {
1472+
dbStats.unregisterSharedBucketImportStats()
14501473
}
14511474

1452-
if s.DbStats[name].MigrationStats != nil {
1453-
s.DbStats[name].unregisterMigrationStats()
1475+
if dbStats.MigrationStats != nil {
1476+
dbStats.unregisterMigrationStats()
14541477
}
14551478

1456-
s.DbStats[name].unregisterQueryStats()
1457-
delete(s.DbStats, name)
1479+
dbStats.unregisterQueryStats()
14581480
}
14591481

14601482
// Removes the per-database stats for this database by removing the database from the map

base/stats_test.go

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@ package base
1212

1313
import (
1414
"expvar"
15+
"fmt"
1516
"math"
17+
"sync"
1618
"testing"
1719

1820
"github.com/couchbase/sync_gateway/testing/require"
@@ -197,3 +199,51 @@ func TestSgwFloatStatMarshalNonFinite(t *testing.T) {
197199
})
198200
}
199201
}
202+
203+
// TestClearDBStatsRemovesFromSerialization is a characterization test for CBG-5472: a database's
204+
// stats appear in the serialized output after NewDBStats and are gone after ClearDBStats. The
205+
// CBG-5472 change moves Prometheus (un)registration out of the dbStatsMapMutex critical section, so
206+
// this pins the externally-observable serialization behaviour that must be preserved.
207+
func TestClearDBStatsRemovesFromSerialization(t *testing.T) {
208+
stats := &SgwStats{DbStats: map[string]*DbStats{}}
209+
const dbName = "TestClearDBStatsRemovesFromSerialization_db"
210+
211+
_, err := stats.NewDBStats(dbName, false, false, false, false, nil, nil)
212+
require.NoError(t, err)
213+
require.Contains(t, stats.String(), dbName)
214+
215+
stats.ClearDBStats(dbName)
216+
require.NotContains(t, stats.String(), dbName)
217+
}
218+
219+
// TestStatsSerializationConcurrentWithDBRegistration verifies CBG-5472: serializing stats (String)
220+
// runs concurrently with database registration/removal without racing or deadlocking. Most valuable
221+
// under -race, which must stay clean after the dbStatsMapMutex critical sections are narrowed.
222+
func TestStatsSerializationConcurrentWithDBRegistration(t *testing.T) {
223+
stats := &SgwStats{DbStats: map[string]*DbStats{}}
224+
225+
var wg sync.WaitGroup
226+
for i := 0; i < 4; i++ {
227+
wg.Add(1)
228+
go func() {
229+
defer wg.Done()
230+
for j := 0; j < 200; j++ {
231+
_ = stats.String()
232+
}
233+
}()
234+
}
235+
for i := 0; i < 4; i++ {
236+
wg.Add(1)
237+
go func(id int) {
238+
defer wg.Done()
239+
// Distinct db name per goroutine so concurrent registrations never collide in Prometheus.
240+
name := fmt.Sprintf("TestStatsSerializationConcurrentWithDBRegistration_db_%d", id)
241+
for j := 0; j < 50; j++ {
242+
_, err := stats.NewDBStats(name, false, false, false, false, nil, nil)
243+
assert.NoError(t, err)
244+
stats.ClearDBStats(name)
245+
}
246+
}(i)
247+
}
248+
wg.Wait()
249+
}

rest/server_context.go

Lines changed: 28 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,12 @@ type ServerContext struct {
7878
_databases map[string]*db.DatabaseContext // _databases is a map of dbname to db.DatabaseContext
7979
_databasesLock sync.RWMutex // Lock for _databases and other db-specific maps above
8080

81+
// databasesSnapshot is a lock-free copy of the _databases values, read by the stats logger so
82+
// stats logging never blocks on _databasesLock (which config updates can hold for a long time
83+
// while waiting on index readiness - CBG-5472). Refreshed under _databasesLock whenever
84+
// _databases changes, via _updateDatabasesSnapshot. No _ prefix: it is read without the lock.
85+
databasesSnapshot atomic.Pointer[[]*db.DatabaseContext]
86+
8187
// serverCtx is cancelled by Close() to broadcast server shutdown to all background
8288
// goroutines that hold a reference to this ServerContext (e.g. handleDbOnline).
8389
serverCtx context.Context
@@ -323,6 +329,7 @@ func (sc *ServerContext) Close(ctx context.Context) {
323329
_ = db.EventMgr.RaiseDBStateChangeEvent(ctx, db.Name, "offline", "Database context closed", &sc.Config.API.AdminInterface)
324330
}
325331
sc._databases = nil
332+
sc._updateDatabasesSnapshot()
326333
sc.invalidDatabaseConfigTracking.dbNames = nil
327334
}
328335

@@ -1204,6 +1211,7 @@ func (sc *ServerContext) _getOrAddDatabaseFromConfig(ctx context.Context, config
12041211

12051212
// Register it so HTTP handlers can find it:
12061213
sc._databases[dbcontext.Name] = dbcontext
1214+
sc._updateDatabasesSnapshot()
12071215
sc._dbConfigs[dbcontext.Name] = &RuntimeDatabaseConfig{DatabaseConfig: config}
12081216
sc._dbRegistry[dbName] = struct{}{}
12091217
for _, name := range fqCollections {
@@ -1811,6 +1819,7 @@ func (sc *ServerContext) _unloadDatabase(ctx context.Context, dbName string) boo
18111819
base.InfofCtx(ctx, base.KeyAll, "Closing db /%s (bucket %q)", base.MD(dbCtx.Name), base.MD(dbCtx.Bucket.GetName()))
18121820
dbCtx.Close(ctx)
18131821
delete(sc._databases, dbName)
1822+
sc._updateDatabasesSnapshot()
18141823
return true
18151824
}
18161825

@@ -1927,17 +1936,29 @@ func (sc *ServerContext) logNetworkInterfaceStats(ctx context.Context) {
19271936

19281937
}
19291938

1930-
// Updates stats that are more efficient to calculate at stats collection time
1931-
func (sc *ServerContext) updateCalculatedStats(ctx context.Context) {
1932-
sc._databasesLock.RLock()
1933-
defer sc._databasesLock.RUnlock()
1939+
// _updateDatabasesSnapshot refreshes the lock-free snapshot of database contexts read by the stats
1940+
// logger (see databasesSnapshot). The caller must hold sc._databasesLock (write).
1941+
func (sc *ServerContext) _updateDatabasesSnapshot() {
1942+
snapshot := make([]*db.DatabaseContext, 0, len(sc._databases))
19341943
for _, dbContext := range sc._databases {
1935-
dbState := atomic.LoadUint32(&dbContext.State)
1936-
if dbState == db.DBOnline {
1944+
snapshot = append(snapshot, dbContext)
1945+
}
1946+
sc.databasesSnapshot.Store(&snapshot)
1947+
}
1948+
1949+
// Updates stats that are more efficient to calculate at stats collection time.
1950+
// Reads a lock-free snapshot of the databases so it never blocks on _databasesLock, which a config
1951+
// update can hold for a long time while waiting on index readiness (CBG-5472).
1952+
func (sc *ServerContext) updateCalculatedStats(ctx context.Context) {
1953+
snapshot := sc.databasesSnapshot.Load()
1954+
if snapshot == nil {
1955+
return
1956+
}
1957+
for _, dbContext := range *snapshot {
1958+
if atomic.LoadUint32(&dbContext.State) == db.DBOnline {
19371959
dbContext.UpdateCalculatedStats(ctx)
19381960
}
19391961
}
1940-
19411962
}
19421963

19431964
// initializeGoCBAgent Obtains a gocb agent from the current server connection. Requires the agent to be closed after use.

rest/server_context_test.go

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -382,6 +382,83 @@ func TestStatsLoggerStopped(t *testing.T) {
382382
time.Sleep(time.Millisecond * 10)
383383
}
384384

385+
// TestStatsLoggerIndependentOfDatabaseLock verifies CBG-5472: the stats logger must not block on
386+
// _databasesLock. A config update holds that write lock across a potentially very long (infinite)
387+
// index-readiness wait; while it does, updateCalculatedStats must still be able to run so stats
388+
// keep being logged. Here we simulate the long-held write lock and assert the stats collection
389+
// completes promptly instead of blocking behind it.
390+
func TestStatsLoggerIndependentOfDatabaseLock(t *testing.T) {
391+
rt := NewRestTester(t, nil)
392+
defer rt.Close()
393+
394+
// Read the database set (takes _databasesLock) before we grab the write lock below.
395+
require.Len(t, rt.ServerContext().AllDatabases(), 1)
396+
397+
sc := rt.ServerContext()
398+
ctx := base.TestCtx(t)
399+
400+
// Simulate a config update holding the write lock for a long time (e.g. blocking on indexes).
401+
sc._databasesLock.Lock()
402+
defer sc._databasesLock.Unlock()
403+
404+
done := make(chan struct{})
405+
go func() {
406+
sc.updateCalculatedStats(ctx)
407+
close(done)
408+
}()
409+
410+
select {
411+
case <-done:
412+
case <-time.After(10 * time.Second):
413+
t.Fatal("updateCalculatedStats blocked on _databasesLock held by a simulated config update (CBG-5472)")
414+
}
415+
}
416+
417+
// TestDatabasesSnapshotTracksDatabases verifies the lock-free stats-logger snapshot (CBG-5472) stays
418+
// in sync with _databases as databases are added and removed, so the stats logger neither misses a
419+
// live database nor processes a removed one.
420+
func TestDatabasesSnapshotTracksDatabases(t *testing.T) {
421+
rt := NewRestTester(t, nil)
422+
defer rt.Close()
423+
424+
sc := rt.ServerContext()
425+
dbName := rt.GetDatabase().Name
426+
427+
snapshot := sc.databasesSnapshot.Load()
428+
require.NotNil(t, snapshot)
429+
require.Len(t, *snapshot, 1)
430+
431+
require.True(t, sc.RemoveDatabase(base.TestCtx(t), dbName, "test cleanup"))
432+
433+
snapshot = sc.databasesSnapshot.Load()
434+
require.NotNil(t, snapshot)
435+
require.Empty(t, *snapshot)
436+
}
437+
438+
// TestStatsLoggerConcurrentWithDatabaseRemoval exercises the CBG-5472 lock-free stats reader against
439+
// a concurrent database removal/close. Because the reader no longer holds _databasesLock, it can
440+
// observe (via the snapshot) a database that is being closed; this must remain memory-safe. Most
441+
// valuable under -race.
442+
func TestStatsLoggerConcurrentWithDatabaseRemoval(t *testing.T) {
443+
rt := NewRestTester(t, nil)
444+
defer rt.Close()
445+
446+
sc := rt.ServerContext()
447+
ctx := base.TestCtx(t)
448+
dbName := rt.GetDatabase().Name
449+
450+
readerDone := make(chan struct{})
451+
go func() {
452+
defer close(readerDone)
453+
for i := 0; i < 2000; i++ {
454+
sc.updateCalculatedStats(ctx)
455+
}
456+
}()
457+
458+
require.True(t, sc.RemoveDatabase(ctx, dbName, "concurrent stats test"))
459+
<-readerDone
460+
}
461+
385462
func TestObtainManagementEndpointsFromServerContext(t *testing.T) {
386463
if base.UnitTestUrlIsWalrus() {
387464
t.Skip("Test requires Couchbase Server")

0 commit comments

Comments
 (0)