* MM-33818: Add replica lag metric We add two new metrics for monitoring replica lag: - Monitor absolute lag based on binlog distance/transaction queue length. - Monitor time taken for the replica to catch up. To achieve this, we add a config setting to run a user defined SQL query on the database. We need to specify a separate datasource field as part of the config because in some databases, querying the replica lag value requires elevated credentials which are not needed for usual running of the application, and can even be a security risk. Arguably, a peculiar part of the design is the requirement of the query output to be in a (node, value) format. But since from the application, the SQL query is a black box and the user can set any query they want, we cannot, in any way templatize this. And as an extra note, freno also does it in a similar way. The last bit is because we need to have a separate datasources, now we consume one extra connection rather than sharing it with the pool. This is an unfortunate result of the design, and while one extra connection doesn't make much of a difference in a single-tenant scenario. It does make so, in a multi-tenant scenario. But in a multi-tenant scenario, the expectation would already be to use a connection pool. So this is not a big concern. https://mattermost.atlassian.net/browse/MM-33818 ```release-note Two new gauge metrics were added: mattermost_db_replica_lag_abs and mattermost_db_replica_lag_time, both containing a label of "node", signifying which db host is the metric from. These metrics signify the replica lag in absolute terms and in the time dimension capturing the whole picture of replica lag. To use these metrics, a separate config section ReplicaLagSettings was added under SqlSettings. This is an array of maps which contain three keys: DataSource, QueryAbsoluteLag, and QueryTimeLag. Each map entry is for a single replica instance. DataSource contains the DB credentials to connect to the replica instance. QueryAbsoluteLag is a plain SQL query that must return a single row of which the first column must be the node value of the Prometheus metric, and the second column must be the value of the lag. QueryTimeLag is the same as above, but used to measure the time lag. As an example, for AWS Aurora instances, the QueryAbsoluteLag can be: select server_id, highest_lsn_rcvd-durable_lsn as bindiff from aurora_global_db_instance_status() where server_id=<> and QueryTimeLag can be: select server_id, visibility_lag_in_msec from aurora_global_db_instance_status() where server_id=<> For MySQL Group Replication, the absolute lag can be measured from the number of pending transactions in the applier queue: select member_id, count_transaction_remote_in_applier_queue FROM performance_schema.replication_group_member_stats where member_id=<> Overall, what query to choose is left to the administrator, and depending on the database and need, an appropriate query can be chosen. ``` * Trigger CI * Fix tests * address review comments * Remove t.Parallel It was spawning too many connections, and overloading the docker container. Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
620 строки
14 KiB
Go
620 строки
14 KiB
Go
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
package sqlstore
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"regexp"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/go-sql-driver/mysql"
|
|
"github.com/lib/pq"
|
|
"github.com/mattermost/gorp"
|
|
"github.com/pkg/errors"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/mattermost/mattermost-server/v5/einterfaces/mocks"
|
|
"github.com/mattermost/mattermost-server/v5/model"
|
|
"github.com/mattermost/mattermost-server/v5/store"
|
|
"github.com/mattermost/mattermost-server/v5/store/searchtest"
|
|
"github.com/mattermost/mattermost-server/v5/store/storetest"
|
|
)
|
|
|
|
type storeType struct {
|
|
Name string
|
|
SqlSettings *model.SqlSettings
|
|
SqlStore *SqlStore
|
|
Store store.Store
|
|
}
|
|
|
|
var storeTypes []*storeType
|
|
|
|
func newStoreType(name, driver string) *storeType {
|
|
return &storeType{
|
|
Name: name,
|
|
SqlSettings: storetest.MakeSqlSettings(driver, false),
|
|
}
|
|
}
|
|
|
|
func StoreTest(t *testing.T, f func(*testing.T, store.Store)) {
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
tearDownStores()
|
|
panic(err)
|
|
}
|
|
}()
|
|
for _, st := range storeTypes {
|
|
st := st
|
|
t.Run(st.Name, func(t *testing.T) {
|
|
if testing.Short() {
|
|
t.SkipNow()
|
|
}
|
|
f(t, st.Store)
|
|
})
|
|
}
|
|
}
|
|
|
|
func StoreTestWithSearchTestEngine(t *testing.T, f func(*testing.T, store.Store, *searchtest.SearchTestEngine)) {
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
tearDownStores()
|
|
panic(err)
|
|
}
|
|
}()
|
|
|
|
for _, st := range storeTypes {
|
|
st := st
|
|
searchTestEngine := &searchtest.SearchTestEngine{
|
|
Driver: *st.SqlSettings.DriverName,
|
|
}
|
|
|
|
t.Run(st.Name, func(t *testing.T) { f(t, st.Store, searchTestEngine) })
|
|
}
|
|
}
|
|
|
|
func StoreTestWithSqlStore(t *testing.T, f func(*testing.T, store.Store, storetest.SqlStore)) {
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
tearDownStores()
|
|
panic(err)
|
|
}
|
|
}()
|
|
for _, st := range storeTypes {
|
|
st := st
|
|
t.Run(st.Name, func(t *testing.T) {
|
|
if testing.Short() {
|
|
t.SkipNow()
|
|
}
|
|
f(t, st.Store, st.SqlStore)
|
|
})
|
|
}
|
|
}
|
|
|
|
func initStores() {
|
|
if testing.Short() {
|
|
return
|
|
}
|
|
// In CI, we already run the entire test suite for both mysql and postgres in parallel.
|
|
// So we just run the tests for the current database set.
|
|
if os.Getenv("IS_CI") == "true" {
|
|
switch os.Getenv("MM_SQLSETTINGS_DRIVERNAME") {
|
|
case "mysql":
|
|
storeTypes = append(storeTypes, newStoreType("MySQL", model.DATABASE_DRIVER_MYSQL))
|
|
case "postgres":
|
|
storeTypes = append(storeTypes, newStoreType("PostgreSQL", model.DATABASE_DRIVER_POSTGRES))
|
|
}
|
|
} else {
|
|
storeTypes = append(storeTypes, newStoreType("MySQL", model.DATABASE_DRIVER_MYSQL),
|
|
newStoreType("PostgreSQL", model.DATABASE_DRIVER_POSTGRES))
|
|
}
|
|
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
tearDownStores()
|
|
panic(err)
|
|
}
|
|
}()
|
|
var wg sync.WaitGroup
|
|
for _, st := range storeTypes {
|
|
st := st
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
st.SqlStore = New(*st.SqlSettings, nil)
|
|
st.Store = st.SqlStore
|
|
st.Store.DropAllTables()
|
|
st.Store.MarkSystemRanUnitTests()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
var tearDownStoresOnce sync.Once
|
|
|
|
func tearDownStores() {
|
|
if testing.Short() {
|
|
return
|
|
}
|
|
tearDownStoresOnce.Do(func() {
|
|
var wg sync.WaitGroup
|
|
wg.Add(len(storeTypes))
|
|
for _, st := range storeTypes {
|
|
st := st
|
|
go func() {
|
|
if st.Store != nil {
|
|
st.Store.Close()
|
|
}
|
|
if st.SqlSettings != nil {
|
|
storetest.CleanupSqlSettings(st.SqlSettings)
|
|
}
|
|
wg.Done()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
})
|
|
}
|
|
|
|
// This test was used to consistently reproduce the race
|
|
// before the fix in MM-28397.
|
|
// Keeping it here to help avoiding future regressions.
|
|
func TestStoreLicenseRace(t *testing.T) {
|
|
settings := makeSqlSettings(model.DATABASE_DRIVER_POSTGRES)
|
|
store := New(*settings, nil)
|
|
defer func() {
|
|
store.Close()
|
|
storetest.CleanupSqlSettings(settings)
|
|
}()
|
|
|
|
wg := sync.WaitGroup{}
|
|
wg.Add(3)
|
|
|
|
go func() {
|
|
store.UpdateLicense(&model.License{})
|
|
wg.Done()
|
|
}()
|
|
|
|
go func() {
|
|
store.GetReplica()
|
|
wg.Done()
|
|
}()
|
|
|
|
go func() {
|
|
store.GetSearchReplica()
|
|
wg.Done()
|
|
}()
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestGetReplica(t *testing.T) {
|
|
t.Parallel()
|
|
testCases := []struct {
|
|
Description string
|
|
DataSourceReplicaNum int
|
|
DataSourceSearchReplicaNum int
|
|
}{
|
|
{
|
|
"no replicas",
|
|
0,
|
|
0,
|
|
},
|
|
{
|
|
"one source replica",
|
|
1,
|
|
0,
|
|
},
|
|
{
|
|
"multiple source replicas",
|
|
3,
|
|
0,
|
|
},
|
|
{
|
|
"one source search replica",
|
|
0,
|
|
1,
|
|
},
|
|
{
|
|
"multiple source search replicas",
|
|
0,
|
|
3,
|
|
},
|
|
{
|
|
"one source replica, one source search replica",
|
|
1,
|
|
1,
|
|
},
|
|
{
|
|
"one source replica, multiple source search replicas",
|
|
1,
|
|
3,
|
|
},
|
|
{
|
|
"multiple source replica, one source search replica",
|
|
3,
|
|
1,
|
|
},
|
|
{
|
|
"multiple source replica, multiple source search replicas",
|
|
3,
|
|
3,
|
|
},
|
|
}
|
|
|
|
for _, testCase := range testCases {
|
|
testCase := testCase
|
|
t.Run(testCase.Description+" with license", func(t *testing.T) {
|
|
|
|
settings := makeSqlSettings(model.DATABASE_DRIVER_POSTGRES)
|
|
dataSourceReplicas := []string{}
|
|
dataSourceSearchReplicas := []string{}
|
|
for i := 0; i < testCase.DataSourceReplicaNum; i++ {
|
|
dataSourceReplicas = append(dataSourceReplicas, *settings.DataSource)
|
|
}
|
|
for i := 0; i < testCase.DataSourceSearchReplicaNum; i++ {
|
|
dataSourceSearchReplicas = append(dataSourceSearchReplicas, *settings.DataSource)
|
|
}
|
|
|
|
settings.DataSourceReplicas = dataSourceReplicas
|
|
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
|
|
store := New(*settings, nil)
|
|
defer func() {
|
|
store.Close()
|
|
storetest.CleanupSqlSettings(settings)
|
|
}()
|
|
|
|
store.UpdateLicense(&model.License{})
|
|
|
|
replicas := make(map[*gorp.DbMap]bool)
|
|
for i := 0; i < 5; i++ {
|
|
replicas[store.GetReplica()] = true
|
|
}
|
|
|
|
searchReplicas := make(map[*gorp.DbMap]bool)
|
|
for i := 0; i < 5; i++ {
|
|
searchReplicas[store.GetSearchReplica()] = true
|
|
}
|
|
|
|
if testCase.DataSourceReplicaNum > 0 {
|
|
// If replicas were defined, ensure none are the master.
|
|
assert.Len(t, replicas, testCase.DataSourceReplicaNum)
|
|
|
|
for replica := range replicas {
|
|
assert.NotSame(t, store.GetMaster(), replica)
|
|
}
|
|
|
|
} else if assert.Len(t, replicas, 1) {
|
|
// Otherwise ensure the replicas contains only the master.
|
|
for replica := range replicas {
|
|
assert.Same(t, store.GetMaster(), replica)
|
|
}
|
|
}
|
|
|
|
if testCase.DataSourceSearchReplicaNum > 0 {
|
|
// If search replicas were defined, ensure none are the master nor the replicas.
|
|
assert.Len(t, searchReplicas, testCase.DataSourceSearchReplicaNum)
|
|
|
|
for searchReplica := range searchReplicas {
|
|
assert.NotSame(t, store.GetMaster(), searchReplica)
|
|
for replica := range replicas {
|
|
assert.NotSame(t, searchReplica, replica)
|
|
}
|
|
}
|
|
} else if testCase.DataSourceReplicaNum > 0 {
|
|
assert.Equal(t, len(replicas), len(searchReplicas))
|
|
for k := range replicas {
|
|
assert.True(t, searchReplicas[k])
|
|
}
|
|
} else if testCase.DataSourceReplicaNum == 0 && assert.Len(t, searchReplicas, 1) {
|
|
// Otherwise ensure the search replicas contains the master.
|
|
for searchReplica := range searchReplicas {
|
|
assert.Same(t, store.GetMaster(), searchReplica)
|
|
}
|
|
}
|
|
})
|
|
|
|
t.Run(testCase.Description+" without license", func(t *testing.T) {
|
|
|
|
settings := makeSqlSettings(model.DATABASE_DRIVER_POSTGRES)
|
|
dataSourceReplicas := []string{}
|
|
dataSourceSearchReplicas := []string{}
|
|
for i := 0; i < testCase.DataSourceReplicaNum; i++ {
|
|
dataSourceReplicas = append(dataSourceReplicas, *settings.DataSource)
|
|
}
|
|
for i := 0; i < testCase.DataSourceSearchReplicaNum; i++ {
|
|
dataSourceSearchReplicas = append(dataSourceSearchReplicas, *settings.DataSource)
|
|
}
|
|
|
|
settings.DataSourceReplicas = dataSourceReplicas
|
|
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
|
|
store := New(*settings, nil)
|
|
defer func() {
|
|
store.Close()
|
|
storetest.CleanupSqlSettings(settings)
|
|
}()
|
|
|
|
replicas := make(map[*gorp.DbMap]bool)
|
|
for i := 0; i < 5; i++ {
|
|
replicas[store.GetReplica()] = true
|
|
}
|
|
|
|
searchReplicas := make(map[*gorp.DbMap]bool)
|
|
for i := 0; i < 5; i++ {
|
|
searchReplicas[store.GetSearchReplica()] = true
|
|
}
|
|
|
|
if testCase.DataSourceReplicaNum > 0 {
|
|
// If replicas were defined, ensure none are the master.
|
|
assert.Len(t, replicas, 1)
|
|
|
|
for replica := range replicas {
|
|
assert.Same(t, store.GetMaster(), replica)
|
|
}
|
|
|
|
} else if assert.Len(t, replicas, 1) {
|
|
// Otherwise ensure the replicas contains only the master.
|
|
for replica := range replicas {
|
|
assert.Same(t, store.GetMaster(), replica)
|
|
}
|
|
}
|
|
|
|
if testCase.DataSourceSearchReplicaNum > 0 {
|
|
// If search replicas were defined, ensure none are the master nor the replicas.
|
|
assert.Len(t, searchReplicas, 1)
|
|
|
|
for searchReplica := range searchReplicas {
|
|
assert.Same(t, store.GetMaster(), searchReplica)
|
|
}
|
|
|
|
} else if testCase.DataSourceReplicaNum > 0 {
|
|
assert.Equal(t, len(replicas), len(searchReplicas))
|
|
for k := range replicas {
|
|
assert.True(t, searchReplicas[k])
|
|
}
|
|
} else if assert.Len(t, searchReplicas, 1) {
|
|
// Otherwise ensure the search replicas contains the master.
|
|
for searchReplica := range searchReplicas {
|
|
assert.Same(t, store.GetMaster(), searchReplica)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGetDbVersion(t *testing.T) {
|
|
testDrivers := []string{
|
|
model.DATABASE_DRIVER_POSTGRES,
|
|
model.DATABASE_DRIVER_MYSQL,
|
|
}
|
|
|
|
for _, driver := range testDrivers {
|
|
t.Run("Should return db version for "+driver, func(t *testing.T) {
|
|
t.Parallel()
|
|
settings := makeSqlSettings(driver)
|
|
store := New(*settings, nil)
|
|
|
|
version, err := store.GetDbVersion(false)
|
|
require.NoError(t, err)
|
|
require.Regexp(t, regexp.MustCompile(`\d+\.\d+(\.\d+)?`), version)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestUpAndDownMigrations(t *testing.T) {
|
|
testDrivers := []string{
|
|
model.DATABASE_DRIVER_POSTGRES,
|
|
model.DATABASE_DRIVER_MYSQL,
|
|
}
|
|
|
|
for _, driver := range testDrivers {
|
|
t.Run("Should be reversible for "+driver, func(t *testing.T) {
|
|
t.Parallel()
|
|
settings := makeSqlSettings(driver)
|
|
store := New(*settings, nil)
|
|
defer store.Close()
|
|
|
|
err := store.migrate(migrationsDirectionDown)
|
|
assert.NoError(t, err, "downing migrations should not error")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGetAllConns(t *testing.T) {
|
|
t.Parallel()
|
|
testCases := []struct {
|
|
Description string
|
|
DataSourceReplicaNum int
|
|
DataSourceSearchReplicaNum int
|
|
ExpectedNumConnections int
|
|
}{
|
|
{
|
|
"no replicas",
|
|
0,
|
|
0,
|
|
1,
|
|
},
|
|
{
|
|
"one source replica",
|
|
1,
|
|
0,
|
|
2,
|
|
},
|
|
{
|
|
"multiple source replicas",
|
|
3,
|
|
0,
|
|
4,
|
|
},
|
|
{
|
|
"one source search replica",
|
|
0,
|
|
1,
|
|
1,
|
|
},
|
|
{
|
|
"multiple source search replicas",
|
|
0,
|
|
3,
|
|
1,
|
|
},
|
|
{
|
|
"one source replica, one source search replica",
|
|
1,
|
|
1,
|
|
2,
|
|
},
|
|
{
|
|
"one source replica, multiple source search replicas",
|
|
1,
|
|
3,
|
|
2,
|
|
},
|
|
{
|
|
"multiple source replica, one source search replica",
|
|
3,
|
|
1,
|
|
4,
|
|
},
|
|
{
|
|
"multiple source replica, multiple source search replicas",
|
|
3,
|
|
3,
|
|
4,
|
|
},
|
|
}
|
|
|
|
for _, testCase := range testCases {
|
|
testCase := testCase
|
|
t.Run(testCase.Description, func(t *testing.T) {
|
|
t.Parallel()
|
|
settings := makeSqlSettings(model.DATABASE_DRIVER_POSTGRES)
|
|
dataSourceReplicas := []string{}
|
|
dataSourceSearchReplicas := []string{}
|
|
for i := 0; i < testCase.DataSourceReplicaNum; i++ {
|
|
dataSourceReplicas = append(dataSourceReplicas, *settings.DataSource)
|
|
}
|
|
for i := 0; i < testCase.DataSourceSearchReplicaNum; i++ {
|
|
dataSourceSearchReplicas = append(dataSourceSearchReplicas, *settings.DataSource)
|
|
}
|
|
|
|
settings.DataSourceReplicas = dataSourceReplicas
|
|
settings.DataSourceSearchReplicas = dataSourceSearchReplicas
|
|
store := New(*settings, nil)
|
|
defer func() {
|
|
store.Close()
|
|
storetest.CleanupSqlSettings(settings)
|
|
}()
|
|
|
|
assert.Len(t, store.GetAllConns(), testCase.ExpectedNumConnections)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestIsDuplicate(t *testing.T) {
|
|
testErrors := map[error]bool{
|
|
&pq.Error{Code: "42P06"}: false,
|
|
&pq.Error{Code: PGDupTableErrorCode}: true,
|
|
&mysql.MySQLError{Number: uint16(1000)}: false,
|
|
&mysql.MySQLError{Number: MySQLDupTableErrorCode}: true,
|
|
errors.New("Random error"): false,
|
|
}
|
|
|
|
for err, expected := range testErrors {
|
|
t.Run(fmt.Sprintf("Should return %t for %s", expected, err.Error()), func(t *testing.T) {
|
|
t.Parallel()
|
|
assert.Equal(t, expected, IsDuplicate(err))
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestVersionString(t *testing.T) {
|
|
versions := []struct {
|
|
input int
|
|
output string
|
|
}{
|
|
{
|
|
input: 100000,
|
|
output: "10.0",
|
|
},
|
|
{
|
|
input: 90603,
|
|
output: "9.603",
|
|
},
|
|
{
|
|
input: 120005,
|
|
output: "12.5",
|
|
},
|
|
}
|
|
|
|
for _, v := range versions {
|
|
out := VersionString(v.input)
|
|
assert.Equal(t, v.output, out)
|
|
}
|
|
}
|
|
|
|
func TestReplicaLagQuery(t *testing.T) {
|
|
testDrivers := []string{
|
|
model.DATABASE_DRIVER_POSTGRES,
|
|
model.DATABASE_DRIVER_MYSQL,
|
|
}
|
|
|
|
for _, driver := range testDrivers {
|
|
settings := makeSqlSettings(driver)
|
|
var query string
|
|
var tableName string
|
|
// Just any random query which returns a row in (string, int) format.
|
|
switch driver {
|
|
case model.DATABASE_DRIVER_POSTGRES:
|
|
query = `SELECT relname, count(relname) FROM pg_class WHERE relname='posts' GROUP BY relname`
|
|
tableName = "posts"
|
|
case model.DATABASE_DRIVER_MYSQL:
|
|
query = `SELECT table_name, count(table_name) FROM information_schema.tables WHERE table_name='Posts' and table_schema=Database() GROUP BY table_name`
|
|
tableName = "Posts"
|
|
}
|
|
|
|
settings.ReplicaLagSettings = []*model.ReplicaLagSettings{{
|
|
DataSource: model.NewString(*settings.DataSource),
|
|
QueryAbsoluteLag: model.NewString(query),
|
|
QueryTimeLag: model.NewString(query),
|
|
}}
|
|
|
|
mockMetrics := &mocks.MetricsInterface{}
|
|
defer mockMetrics.AssertExpectations(t)
|
|
mockMetrics.On("SetReplicaLagAbsolute", tableName, float64(1))
|
|
mockMetrics.On("SetReplicaLagTime", tableName, float64(1))
|
|
|
|
store := &SqlStore{
|
|
rrCounter: 0,
|
|
srCounter: 0,
|
|
settings: settings,
|
|
metrics: mockMetrics,
|
|
}
|
|
|
|
store.initConnection()
|
|
store.stores.post = newSqlPostStore(store, mockMetrics)
|
|
err := store.GetMaster().CreateTablesIfNotExists()
|
|
require.NoError(t, err)
|
|
|
|
defer store.Close()
|
|
|
|
err = store.ReplicaLagAbs()
|
|
require.NoError(t, err)
|
|
err = store.ReplicaLagTime()
|
|
require.NoError(t, err)
|
|
}
|
|
}
|
|
|
|
func makeSqlSettings(driver string) *model.SqlSettings {
|
|
switch driver {
|
|
case model.DATABASE_DRIVER_POSTGRES:
|
|
return storetest.MakeSqlSettings(driver, false)
|
|
case model.DATABASE_DRIVER_MYSQL:
|
|
return storetest.MakeSqlSettings(driver, false)
|
|
}
|
|
|
|
return nil
|
|
}
|