Implement config store for MySQL (cadence-workflow#5403)

avmi · Sep 20, 2023 · eb55629 · eb55629
1 parent 45ca74f
commit eb55629
Show file tree

Hide file tree

Showing 10 changed files with 87 additions and 37 deletions.
diff --git a/common/persistence/persistence-tests/configStorePersistenceTest.go b/common/persistence/persistence-tests/configStorePersistenceTest.go
@@ -70,10 +70,7 @@ func (s *ConfigStorePersistenceSuite) TestFetchFromEmptyTable() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
 	defer cancel()
 
-	s.DefaultTestCluster.TearDownTestDatabase()
-	s.DefaultTestCluster.SetupTestDatabase()
-
-	snapshot, err := s.FetchDynamicConfig(ctx)
+	snapshot, err := s.FetchDynamicConfig(ctx, 0)
 	s.Nil(snapshot)
 	s.NotNil(err)
 }
@@ -82,15 +79,12 @@ func (s *ConfigStorePersistenceSuite) TestUpdateSimpleSuccess() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
 	defer cancel()
 
-	s.DefaultTestCluster.TearDownTestDatabase()
-	s.DefaultTestCluster.SetupTestDatabase()
-
 	snapshot := generateRandomSnapshot(1)
-	err := s.UpdateDynamicConfig(ctx, snapshot)
+	err := s.UpdateDynamicConfig(ctx, snapshot, 1)
 	s.Nil(err)
 
-	retSnapshot, err := s.FetchDynamicConfig(ctx)
-	s.NotNil(snapshot)
+	retSnapshot, err := s.FetchDynamicConfig(ctx, 1)
+	s.NotNil(retSnapshot)
 	s.Nil(err)
 	s.Equal(snapshot.Version, retSnapshot.Version)
 	s.Equal(snapshot.Values.Entries[0].Name, retSnapshot.Values.Entries[0].Name)
@@ -100,14 +94,11 @@ func (s *ConfigStorePersistenceSuite) TestUpdateVersionCollisionFailure() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
 	defer cancel()
 
-	s.DefaultTestCluster.TearDownTestDatabase()
-	s.DefaultTestCluster.SetupTestDatabase()
-
 	snapshot := generateRandomSnapshot(1)
-	err := s.UpdateDynamicConfig(ctx, snapshot)
+	err := s.UpdateDynamicConfig(ctx, snapshot, 2)
 	s.Nil(err)
 
-	err = s.UpdateDynamicConfig(ctx, snapshot)
+	err = s.UpdateDynamicConfig(ctx, snapshot, 2)
 	var condErr *p.ConditionFailedError
 	s.True(errors.As(err, &condErr))
 }
@@ -116,32 +107,26 @@ func (s *ConfigStorePersistenceSuite) TestUpdateIncrementalVersionSuccess() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
 	defer cancel()
 
-	s.DefaultTestCluster.TearDownTestDatabase()
-	s.DefaultTestCluster.SetupTestDatabase()
-
 	snapshot2 := generateRandomSnapshot(2)
-	err := s.UpdateDynamicConfig(ctx, snapshot2)
+	err := s.UpdateDynamicConfig(ctx, snapshot2, 3)
 	s.Nil(err)
 	snapshot3 := generateRandomSnapshot(3)
-	err = s.UpdateDynamicConfig(ctx, snapshot3)
+	err = s.UpdateDynamicConfig(ctx, snapshot3, 3)
 	s.Nil(err)
 }
 
 func (s *ConfigStorePersistenceSuite) TestFetchLatestVersionSuccess() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
 	defer cancel()
 
-	s.DefaultTestCluster.TearDownTestDatabase()
-	s.DefaultTestCluster.SetupTestDatabase()
-
 	snapshot2 := generateRandomSnapshot(2)
-	err := s.UpdateDynamicConfig(ctx, snapshot2)
+	err := s.UpdateDynamicConfig(ctx, snapshot2, 4)
 	s.Nil(err)
 	snapshot3 := generateRandomSnapshot(3)
-	err = s.UpdateDynamicConfig(ctx, snapshot3)
+	err = s.UpdateDynamicConfig(ctx, snapshot3, 4)
 	s.Nil(err)
 
-	snapshot, err := s.FetchDynamicConfig(ctx)
+	snapshot, err := s.FetchDynamicConfig(ctx, 4)
 	s.NotNil(snapshot)
 	s.Nil(err)
 	s.Equal(int64(3), snapshot.Version)
@@ -174,8 +159,8 @@ func generateRandomSnapshot(version int64) *p.DynamicConfigSnapshot {
 	}
 }
 
-func (s *ConfigStorePersistenceSuite) FetchDynamicConfig(ctx context.Context) (*p.DynamicConfigSnapshot, error) {
-	response, err := s.ConfigStoreManager.FetchDynamicConfig(ctx, p.DynamicConfig)
+func (s *ConfigStorePersistenceSuite) FetchDynamicConfig(ctx context.Context, rowType int) (*p.DynamicConfigSnapshot, error) {
+	response, err := s.ConfigStoreManager.FetchDynamicConfig(ctx, p.ConfigType(rowType))
 	if err != nil {
 		return nil, err
 	}
@@ -185,6 +170,6 @@ func (s *ConfigStorePersistenceSuite) FetchDynamicConfig(ctx context.Context) (*
 	return response.Snapshot, nil
 }
 
-func (s *ConfigStorePersistenceSuite) UpdateDynamicConfig(ctx context.Context, snapshot *p.DynamicConfigSnapshot) error {
-	return s.ConfigStoreManager.UpdateDynamicConfig(ctx, &p.UpdateDynamicConfigRequest{Snapshot: snapshot}, p.DynamicConfig)
+func (s *ConfigStorePersistenceSuite) UpdateDynamicConfig(ctx context.Context, snapshot *p.DynamicConfigSnapshot, rowType int) error {
+	return s.ConfigStoreManager.UpdateDynamicConfig(ctx, &p.UpdateDynamicConfigRequest{Snapshot: snapshot}, p.ConfigType(rowType))
 }
diff --git a/common/persistence/sql/sqlConfigStore.go b/common/persistence/sql/sqlConfigStore.go
@@ -28,7 +28,6 @@ import (
 
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/persistence"
-	"github.com/uber/cadence/common/persistence/nosql/nosqlplugin"
 	"github.com/uber/cadence/common/persistence/serialization"
 	"github.com/uber/cadence/common/persistence/sql/sqlplugin"
 )
@@ -68,7 +67,7 @@ func (m *sqlConfigStore) FetchConfig(ctx context.Context, configType persistence
 func (m *sqlConfigStore) UpdateConfig(ctx context.Context, value *persistence.InternalConfigStoreEntry) error {
 	err := m.db.InsertConfig(ctx, value)
 	if err != nil {
-		if _, ok := err.(*nosqlplugin.ConditionFailure); ok {
+		if m.db.IsDupEntryError(err) {
 			return &persistence.ConditionFailedError{Msg: fmt.Sprintf("Version %v already exists. Condition Failed", value.Version)}
 		}
 		return convertCommonErrors(m.db, "UpdateConfig", "", err)

diff --git a/common/persistence/sql/sqlplugin/interfaces.go b/common/persistence/sql/sqlplugin/interfaces.go
@@ -569,6 +569,15 @@ type (
 		Data      []byte
 	}
 
+	// ClusterConfigRow represents a row in cluster_config table
+	ClusterConfigRow struct {
+		RowType      int
+		Version      int64
+		Timestamp    time.Time
+		Data         []byte
+		DataEncoding string
+	}
+
 	// tableCRUD defines the API for interacting with the database tables
 	tableCRUD interface {
 		InsertIntoDomain(ctx context.Context, rows *DomainRow) (sql.Result, error)

diff --git a/common/persistence/sql/sqlplugin/mysql/configstore.go b/common/persistence/sql/sqlplugin/mysql/configstore.go
@@ -24,15 +24,37 @@ package mysql
 
 import (
 	"context"
-	"fmt"
 
+	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/persistence"
+	"github.com/uber/cadence/common/persistence/sql/sqlplugin"
+)
+
+const (
+	_selectLatestConfigQuery = "SELECT row_type, version, timestamp, data, data_encoding FROM cluster_config WHERE row_type = ? ORDER BY version LIMIT 1;"
+
+	_insertConfigQuery = "INSERT INTO cluster_config (row_type, version, timestamp, data, data_encoding) VALUES(?, ?, ?, ?, ?)"
 )
 
 func (mdb *db) InsertConfig(ctx context.Context, row *persistence.InternalConfigStoreEntry) error {
-	return fmt.Errorf("not implemented")
+	_, err := mdb.driver.ExecContext(ctx, sqlplugin.DbDefaultShard, _insertConfigQuery, row.RowType, -1*row.Version, mdb.converter.ToMySQLDateTime(row.Timestamp), row.Values.Data, row.Values.Encoding)
+	return err
 }
 
 func (mdb *db) SelectLatestConfig(ctx context.Context, rowType int) (*persistence.InternalConfigStoreEntry, error) {
-	return nil, fmt.Errorf("not implemented")
+	var row sqlplugin.ClusterConfigRow
+	err := mdb.driver.GetContext(ctx, sqlplugin.DbDefaultShard, &row, _selectLatestConfigQuery, rowType)
+	if err != nil {
+		return nil, err
+	}
+	row.Version *= -1
+	return &persistence.InternalConfigStoreEntry{
+		RowType:   row.RowType,
+		Version:   row.Version,
+		Timestamp: mdb.converter.FromMySQLDateTime(row.Timestamp),
+		Values: &persistence.DataBlob{
+			Data:     row.Data,
+			Encoding: common.EncodingType(row.DataEncoding),
+		},
+	}, nil
 }
diff --git a/common/persistence/sql/sqlplugin/mysql/mysql_persistence_test.go b/common/persistence/sql/sqlplugin/mysql/mysql_persistence_test.go
@@ -92,3 +92,11 @@ func TestMySQLQueuePersistence(t *testing.T) {
 	s.TestBase.Setup()
 	suite.Run(t, s)
 }
+
+func TestMySQLConfigPersistence(t *testing.T) {
+	testflags.RequireMySQL(t)
+	s := new(pt.ConfigStorePersistenceSuite)
+	s.TestBase = pt.NewTestBaseWithSQL(GetTestClusterOption())
+	s.TestBase.Setup()
+	suite.Run(t, s)
+}
diff --git a/schema/mysql/v8/cadence/schema.sql b/schema/mysql/v8/cadence/schema.sql
@@ -265,3 +265,13 @@ CREATE TABLE queue_metadata (
   data MEDIUMBLOB NOT NULL,
   PRIMARY KEY(queue_type)
 );
+
+CREATE TABLE cluster_config (
+  row_type INT NOT NULL,
+  version BIGINT NOT NULL,
+  --
+  timestamp DATETIME(6) NOT NULL,
+  data           MEDIUMBLOB NOT NULL,
+  data_encoding  VARCHAR(16) NOT NULL,
+  PRIMARY KEY (row_type, version)
+);
diff --git a/schema/mysql/v8/cadence/versioned/v0.6/cluster_config.sql b/schema/mysql/v8/cadence/versioned/v0.6/cluster_config.sql
@@ -0,0 +1,9 @@
+CREATE TABLE cluster_config (
+  row_type INT NOT NULL,
+  version BIGINT NOT NULL,
+  --
+  timestamp DATETIME(6) NOT NULL,
+  data           MEDIUMBLOB NOT NULL,
+  data_encoding  VARCHAR(16) NOT NULL,
+  PRIMARY KEY (row_type, version)
+);
diff --git a/schema/mysql/v8/cadence/versioned/v0.6/manifest.json b/schema/mysql/v8/cadence/versioned/v0.6/manifest.json
@@ -0,0 +1,8 @@
+{
+  "CurrVersion": "0.6",
+  "MinCompatibleVersion": "0.6",
+  "Description": "create cluster config table",
+  "SchemaUpdateCqlFiles": [
+    "cluster_config.sql"
+  ]
+}
diff --git a/schema/mysql/version.go b/schema/mysql/version.go
@@ -23,7 +23,7 @@ package mysql
 // NOTE: whenever there is a new data base schema update, plz update the following versions
 
 // Version is the MySQL database release version
-const Version = "0.5"
+const Version = "0.6"
 
 // VisibilityVersion is the MySQL visibility database release version
 const VisibilityVersion = "0.7"
diff --git a/tools/common/schema/updatetask_test.go b/tools/common/schema/updatetask_test.go
@@ -120,7 +120,7 @@ func (s *UpdateTaskTestSuite) TestReadSchemaDirFromEmbeddings() {
 	s.NoError(err)
 	ans, err = readSchemaDir(fsys, "0.3", "")
 	s.NoError(err)
-	s.Equal([]string{"v0.4", "v0.5"}, ans)
+	s.Equal([]string{"v0.4", "v0.5", "v0.6"}, ans)
 
 	fsys, err = fs.Sub(mysql.SchemaFS, "v8/visibility/versioned")
 	s.NoError(err)