MM-24310: Recycle DB connections properly (#14378)

Automatic Merge

app/admin.go

@@ -161,18 +161,14 @@ func (a *App) InvalidateAllCachesSkipSend() {
 }
 
 func (a *App) RecycleDatabaseConnection() {
-	oldStore := a.Srv().Store
+	mlog.Info("Attempting to recycle database connections.")
 
-	mlog.Warn("Attempting to recycle the database connection.")
-	a.Srv().Store = a.Srv().newStore()
-	a.Srv().Jobs.Store = a.Srv().Store
+	// This works by setting 10 seconds as the max conn lifetime for all DB connections.
+	// This allows in gradually closing connections as they expire. In future, we can think
+	// of exposing this as a param from the REST api.
+	a.Srv().Store.RecycleDBConnections(10 * time.Second)
 
-	if a.Srv().Store != oldStore {
-		time.Sleep(20 * time.Second)
-		oldStore.Close()
-	}
-
-	mlog.Warn("Finished recycling the database connection.")
+	mlog.Info("Finished recycling database connections.")
 }
 
 func (a *App) TestSiteURL(siteURL string) *model.AppError {

store/sqlstore/supplier.go

@@ -426,7 +426,7 @@ func (ss *SqlSupplier) DoesTableExist(tableName string) bool {
 
 	} else if ss.DriverName() == model.DATABASE_DRIVER_SQLITE {
 		count, err := ss.GetMaster().SelectInt(
-			`SELECT name FROM sqlite_master WHERE type='table' AND name=?`,
+			`SELECT count(name) FROM sqlite_master WHERE type='table' AND name=?`,
 			tableName,
 		)
 
@@ -1004,6 +1004,23 @@ func (ss *SqlSupplier) GetAllConns() []*gorp.DbMap {
 	return all
 }
 
+// RecycleDBConnections closes active connections by setting the max conn lifetime
+// to d, and then resets them back to their original duration.
+func (ss *SqlSupplier) RecycleDBConnections(d time.Duration) {
+	// Get old time.
+	originalDuration := time.Duration(*ss.settings.ConnMaxLifetimeMilliseconds) * time.Millisecond
+	// Set the max lifetimes for all connections.
+	for _, conn := range ss.GetAllConns() {
+		conn.Db.SetConnMaxLifetime(d)
+	}
+	// Wait for that period with an additional 2 seconds of scheduling delay.
+	time.Sleep(d + 2*time.Second)
+	// Reset max lifetime back to original value.
+	for _, conn := range ss.GetAllConns() {
+		conn.Db.SetConnMaxLifetime(originalDuration)
+	}
+}
+
 func (ss *SqlSupplier) Close() {
 	ss.master.Db.Close()
 	for _, replica := range ss.replicas {

store/sqlstore/supplier_test.go

@@ -5,7 +5,9 @@ package sqlstore_test
 
 import (
 	"regexp"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/mattermost/gorp"
 	_ "github.com/mattn/go-sqlite3"
@@ -151,6 +153,46 @@ func TestGetDbVersion(t *testing.T) {
 	}
 }
 
+func TestRecycleDBConns(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping recycle DBConns test")
+	}
+	testDrivers := []string{
+		model.DATABASE_DRIVER_POSTGRES,
+		model.DATABASE_DRIVER_MYSQL,
+		model.DATABASE_DRIVER_SQLITE,
+	}
+
+	for _, driver := range testDrivers {
+		t.Run(driver, func(t *testing.T) {
+			settings := makeSqlSettings(driver)
+			supplier := sqlstore.NewSqlSupplier(*settings, nil)
+
+			var wg sync.WaitGroup
+			tables := []string{"Posts", "Channels", "Users"}
+			for _, table := range tables {
+				wg.Add(1)
+				go func(table string) {
+					defer wg.Done()
+					query := `SELECT count(*) FROM ` + table
+					_, err := supplier.GetMaster().SelectInt(query)
+					assert.NoError(t, err)
+				}(table)
+			}
+			wg.Wait()
+
+			stats := supplier.GetMaster().Db.Stats()
+			assert.Equal(t, 0, int(stats.MaxLifetimeClosed), "unexpected number of connections closed due to maxlifetime")
+
+			supplier.RecycleDBConnections(2 * time.Second)
+			// We cannot reliably control exactly how many open connections are there. So we
+			// just do a basic check and confirm that atleast one has been closed.
+			stats = supplier.GetMaster().Db.Stats()
+			assert.Greater(t, int(stats.MaxLifetimeClosed), 0, "unexpected number of connections closed due to maxlifetime")
+		})
+	}
+}
+
 func TestGetAllConns(t *testing.T) {
 	t.Parallel()
 	testCases := []struct {

store/store.go

@@ -7,6 +7,7 @@ package store
 
 import (
 	"context"
+	"time"
 
 	"github.com/mattermost/mattermost-server/v5/model"
 )
@@ -53,6 +54,7 @@ type Store interface {
 	LockToMaster()
 	UnlockFromMaster()
 	DropAllTables()
+	RecycleDBConnections(d time.Duration)
 	GetCurrentSchemaVersion() string
 	GetDbVersion() (string, error)
 	TotalMasterDbConnections() int

store/storetest/mocks/Store.go

@@ -9,6 +9,8 @@ import (
 
 	store "github.com/mattermost/mattermost-server/v5/store"
 	mock "github.com/stretchr/testify/mock"
+
+	time "time"
 )
 
 // Store is an autogenerated mock type for the Store type
@@ -407,6 +409,11 @@ func (_m *Store) Reaction() store.ReactionStore {
 	return r0
 }
 
+// RecycleDBConnections provides a mock function with given fields: d
+func (_m *Store) RecycleDBConnections(d time.Duration) {
+	_m.Called(d)
+}
+
 // Role provides a mock function with given fields:
 func (_m *Store) Role() store.RoleStore {
 	ret := _m.Called()

store/storetest/store.go

@@ -5,6 +5,7 @@ package storetest
 
 import (
 	"context"
+	"time"
 
 	"github.com/mattermost/mattermost-server/v5/store"
 	"github.com/mattermost/mattermost-server/v5/store/storetest/mocks"
@@ -88,6 +89,7 @@ func (s *Store) LockToMaster()                         { /* do nothing */ }
 func (s *Store) UnlockFromMaster()                     { /* do nothing */ }
 func (s *Store) DropAllTables()                        { /* do nothing */ }
 func (s *Store) GetDbVersion() (string, error)         { return "", nil }
+func (s *Store) RecycleDBConnections(time.Duration)    {}
 func (s *Store) TotalMasterDbConnections() int         { return 1 }
 func (s *Store) TotalReadDbConnections() int           { return 1 }
 func (s *Store) TotalSearchDbConnections() int         { return 1 }