PLT-6355: Use separate Read Replicas for Search. (#6216)

api4/system_test.go

@@ -67,6 +67,9 @@ func TestGetConfig(t *testing.T) {
 	if !strings.Contains(strings.Join(cfg.SqlSettings.DataSourceReplicas, " "), model.FAKE_SETTING) && len(cfg.SqlSettings.DataSourceReplicas) != 0 {
 		t.Fatal("did not sanitize properly")
 	}
+	if !strings.Contains(strings.Join(cfg.SqlSettings.DataSourceSearchReplicas, " "), model.FAKE_SETTING) && len(cfg.SqlSettings.DataSourceSearchReplicas) != 0 {
+		t.Fatal("did not sanitize properly")
+	}
 }
 
 func TestReloadConfig(t *testing.T) {

config/config.json

@@ -73,6 +73,7 @@
         "DriverName": "mysql",
         "DataSource": "mmuser:mostest@tcp(dockerhost:3306)/mattermost_test?charset=utf8mb4,utf8&readTimeout=30s&writeTimeout=30s",
         "DataSourceReplicas": [],
+        "DataSourceSearchReplicas": [],
         "MaxIdleConns": 20,
         "MaxOpenConns": 300,
         "Trace": false,

model/config.go

@@ -184,13 +184,14 @@ type SSOSettings struct {
 }
 
 type SqlSettings struct {
-	DriverName         string
+	DriverName               string
-	DataSource         string
+	DataSource               string
-	DataSourceReplicas []string
+	DataSourceReplicas       []string
-	MaxIdleConns       int
+	DataSourceSearchReplicas []string
-	MaxOpenConns       int
+	MaxIdleConns             int
-	Trace              bool
+	MaxOpenConns             int
-	AtRestEncryptKey   string
+	Trace                    bool
+	AtRestEncryptKey         string
 }
 
 type LogSettings struct {
@@ -1446,6 +1447,10 @@ func (o *Config) Sanitize() {
 	for i := range o.SqlSettings.DataSourceReplicas {
 		o.SqlSettings.DataSourceReplicas[i] = FAKE_SETTING
 	}
+
+	for i := range o.SqlSettings.DataSourceSearchReplicas {
+		o.SqlSettings.DataSourceSearchReplicas[i] = FAKE_SETTING
+	}
 }
 
 func (o *Config) defaultWebrtcSettings() {

store/sql_post_store.go

@@ -1065,7 +1065,7 @@ func (s SqlPostStore) Search(teamId string, userId string, params *model.SearchP
 
 		list := model.NewPostList()
 
-		_, err := s.GetReplica().Select(&posts, searchQuery, queryParams)
+		_, err := s.GetSearchReplica().Select(&posts, searchQuery, queryParams)
 		if err != nil {
 			l4g.Warn(utils.T("store.sql_post.search.warn"), err.Error())
 			// Don't return the error to the caller as it is of no use to the user. Instead return an empty set of search results.

store/sql_store.go

@@ -66,33 +66,36 @@ const (
 )
 
 type SqlStore struct {
-	master        *gorp.DbMap
+	master         *gorp.DbMap
-	replicas      []*gorp.DbMap
+	replicas       []*gorp.DbMap
-	team          TeamStore
+	searchReplicas []*gorp.DbMap
-	channel       ChannelStore
+	team           TeamStore
-	post          PostStore
+	channel        ChannelStore
-	user          UserStore
+	post           PostStore
-	audit         AuditStore
+	user           UserStore
-	compliance    ComplianceStore
+	audit          AuditStore
-	session       SessionStore
+	compliance     ComplianceStore
-	oauth         OAuthStore
+	session        SessionStore
-	system        SystemStore
+	oauth          OAuthStore
-	webhook       WebhookStore
+	system         SystemStore
-	command       CommandStore
+	webhook        WebhookStore
-	preference    PreferenceStore
+	command        CommandStore
-	license       LicenseStore
+	preference     PreferenceStore
-	token         TokenStore
+	license        LicenseStore
-	emoji         EmojiStore
+	token          TokenStore
-	status        StatusStore
+	emoji          EmojiStore
-	fileInfo      FileInfoStore
+	status         StatusStore
-	reaction      ReactionStore
+	fileInfo       FileInfoStore
-	SchemaVersion string
+	reaction       ReactionStore
-	rrCounter     int64
+	SchemaVersion  string
+	rrCounter      int64
+	srCounter      int64
 }
 
 func initConnection() *SqlStore {
 	sqlStore := &SqlStore{
 		rrCounter: 0,
+		srCounter: 0,
 	}
 
 	sqlStore.master = setupConnection("master", utils.Cfg.SqlSettings.DriverName,
@@ -111,6 +114,17 @@ func initConnection() *SqlStore {
 		}
 	}
 
+	if len(utils.Cfg.SqlSettings.DataSourceSearchReplicas) == 0 {
+		sqlStore.searchReplicas = sqlStore.replicas
+	} else {
+		sqlStore.searchReplicas = make([]*gorp.DbMap, len(utils.Cfg.SqlSettings.DataSourceSearchReplicas))
+		for i, replica := range utils.Cfg.SqlSettings.DataSourceSearchReplicas {
+			sqlStore.searchReplicas[i] = setupConnection(fmt.Sprintf("search-replica-%v", i), utils.Cfg.SqlSettings.DriverName, replica,
+				utils.Cfg.SqlSettings.MaxIdleConns, utils.Cfg.SqlSettings.MaxOpenConns,
+				utils.Cfg.SqlSettings.Trace)
+		}
+	}
+
 	sqlStore.SchemaVersion = sqlStore.GetCurrentSchemaVersion()
 	return sqlStore
 }
@@ -231,6 +245,19 @@ func (ss *SqlStore) TotalReadDbConnections() int {
 	return count
 }
 
+func (ss *SqlStore) TotalSearchDbConnections() int {
+	if len(utils.Cfg.SqlSettings.DataSourceSearchReplicas) == 0 {
+		return 0
+	}
+
+	count := 0
+	for _, db := range ss.searchReplicas {
+		count = count + db.Db.Stats().OpenConnections
+	}
+
+	return count
+}
+
 func (ss *SqlStore) GetCurrentSchemaVersion() string {
 	version, _ := ss.GetMaster().SelectStr("SELECT Value FROM Systems WHERE Name='Version'")
 	return version
@@ -611,6 +638,11 @@ func (ss *SqlStore) GetMaster() *gorp.DbMap {
 	return ss.master
 }
 
+func (ss *SqlStore) GetSearchReplica() *gorp.DbMap {
+	rrNum := atomic.AddInt64(&ss.srCounter, 1) % int64(len(ss.searchReplicas))
+	return ss.searchReplicas[rrNum]
+}
+
 func (ss *SqlStore) GetReplica() *gorp.DbMap {
 	rrNum := atomic.AddInt64(&ss.rrCounter, 1) % int64(len(ss.replicas))
 	return ss.replicas[rrNum]

store/store.go

@@ -52,6 +52,7 @@ type Store interface {
 	DropAllTables()
 	TotalMasterDbConnections() int
 	TotalReadDbConnections() int
+	TotalSearchDbConnections() int
 }
 
 type TeamStore interface {

utils/config.go

@@ -562,4 +562,8 @@ func Desanitize(cfg *model.Config) {
 	for i := range cfg.SqlSettings.DataSourceReplicas {
 		cfg.SqlSettings.DataSourceReplicas[i] = Cfg.SqlSettings.DataSourceReplicas[i]
 	}
+
+	for i := range cfg.SqlSettings.DataSourceSearchReplicas {
+		cfg.SqlSettings.DataSourceSearchReplicas[i] = Cfg.SqlSettings.DataSourceSearchReplicas[i]
+	}
 }