Skip to content

Commit

Permalink
Add unit tests for persistence workflow utils part 1
Browse files Browse the repository at this point in the history
  • Loading branch information
@taylanisikdemir
taylanisikdemir committed Dec 13, 2023
1 parent 0c066ec commit 4664711
Show file tree
Hide file tree
Showing 19 changed files with 1,768 additions and 776 deletions.
7 changes: 0 additions & 7 deletions ...osql/nosqlplugin/cassandra/configStore.go → ...l/nosqlplugin/cassandra/cluster_config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,13 +29,6 @@ import (
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin"
)

const (
// version is the clustering key(DESC order) so this query will always return the record with largest version
templateSelectLatestConfig = `SELECT row_type, version, timestamp, values, encoding FROM cluster_config WHERE row_type = ? LIMIT 1;`

templateInsertConfig = `INSERT INTO cluster_config (row_type, version, timestamp, values, encoding) VALUES (?, ?, ?, ?, ?) IF NOT EXISTS;`
)

func (db *cdb) InsertConfig(ctx context.Context, row *persistence.InternalConfigStoreEntry) error {
query := db.session.Query(templateInsertConfig, row.RowType, row.Version, row.Timestamp, row.Values.Data, row.Values.Encoding).WithContext(ctx)
applied, err := query.MapScanCAS(make(map[string]interface{}))
Expand Down
33 changes: 33 additions & 0 deletions common/persistence/nosql/nosqlplugin/cassandra/cluster_config_cql.go
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
// Copyright (c) 2021 Uber Technologies, Inc.
// Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package cassandra

const (
// version is the clustering key(DESC order) so this query will always return the record with largest version
templateSelectLatestConfig = `SELECT row_type, version, timestamp, values, encoding FROM cluster_config ` +
`WHERE row_type = ? ` +
`LIMIT 1;`

templateInsertConfig = `INSERT INTO cluster_config (row_type, version, timestamp, values, encoding) ` +
`VALUES (?, ?, ?, ?, ?) ` +
`IF NOT EXISTS;`
)
147 changes: 16 additions & 131 deletions common/persistence/nosql/nosqlplugin/cassandra/domain.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import (

"github.com/uber/cadence/common"
"github.com/uber/cadence/common/log/tag"
p "github.com/uber/cadence/common/persistence"
"github.com/uber/cadence/common/persistence"
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin"
"github.com/uber/cadence/common/persistence/nosql/nosqlplugin/cassandra/gocql"
"github.com/uber/cadence/common/types"
Expand All @@ -39,121 +39,6 @@ const (
emptyFailoverEndTime = int64(0)
)

const (
templateDomainInfoType = `{` +
`id: ?, ` +
`name: ?, ` +
`status: ?, ` +
`description: ?, ` +
`owner_email: ?, ` +
`data: ? ` +
`}`

templateDomainConfigType = `{` +
`retention: ?, ` +
`emit_metric: ?, ` +
`archival_bucket: ?, ` +
`archival_status: ?,` +
`history_archival_status: ?, ` +
`history_archival_uri: ?, ` +
`visibility_archival_status: ?, ` +
`visibility_archival_uri: ?, ` +
`bad_binaries: ?,` +
`bad_binaries_encoding: ?,` +
`isolation_groups: ?,` +
`isolation_groups_encoding: ?` +
`}`

templateDomainReplicationConfigType = `{` +
`active_cluster_name: ?, ` +
`clusters: ? ` +
`}`

templateCreateDomainQuery = `INSERT INTO domains (` +
`id, domain) ` +
`VALUES(?, {name: ?}) IF NOT EXISTS`

templateGetDomainQuery = `SELECT domain.name ` +
`FROM domains ` +
`WHERE id = ?`

templateDeleteDomainQuery = `DELETE FROM domains ` +
`WHERE id = ?`

templateCreateDomainByNameQueryWithinBatchV2 = `INSERT INTO domains_by_name_v2 (` +
`domains_partition, name, domain, config, replication_config, is_global_domain, config_version, failover_version, failover_notification_version, previous_failover_version, failover_end_time, last_updated_time, notification_version) ` +
`VALUES(?, ?, ` + templateDomainInfoType + `, ` + templateDomainConfigType + `, ` + templateDomainReplicationConfigType + `, ?, ?, ?, ?, ?, ?, ?, ?) IF NOT EXISTS`

templateGetDomainByNameQueryV2 = `SELECT domain.id, domain.name, domain.status, domain.description, ` +
`domain.owner_email, domain.data, config.retention, config.emit_metric, ` +
`config.archival_bucket, config.archival_status, ` +
`config.history_archival_status, config.history_archival_uri, ` +
`config.visibility_archival_status, config.visibility_archival_uri, ` +
`config.bad_binaries, config.bad_binaries_encoding, ` +
`replication_config.active_cluster_name, replication_config.clusters, ` +
`config.isolation_groups,` +
`config.isolation_groups_encoding,` +
`is_global_domain, ` +
`config_version, ` +
`failover_version, ` +
`failover_notification_version, ` +
`previous_failover_version, ` +
`failover_end_time, ` +
`last_updated_time, ` +
`notification_version ` +
`FROM domains_by_name_v2 ` +
`WHERE domains_partition = ? ` +
`and name = ?`

templateUpdateDomainByNameQueryWithinBatchV2 = `UPDATE domains_by_name_v2 ` +
`SET domain = ` + templateDomainInfoType + `, ` +
`config = ` + templateDomainConfigType + `, ` +
`replication_config = ` + templateDomainReplicationConfigType + `, ` +
`config_version = ? ,` +
`failover_version = ? ,` +
`failover_notification_version = ? , ` +
`previous_failover_version = ? , ` +
`failover_end_time = ?,` +
`last_updated_time = ?,` +
`notification_version = ? ` +
`WHERE domains_partition = ? ` +
`and name = ?`

templateGetMetadataQueryV2 = `SELECT notification_version ` +
`FROM domains_by_name_v2 ` +
`WHERE domains_partition = ? ` +
`and name = ? `

templateUpdateMetadataQueryWithinBatchV2 = `UPDATE domains_by_name_v2 ` +
`SET notification_version = ? ` +
`WHERE domains_partition = ? ` +
`and name = ? ` +
`IF notification_version = ? `

templateDeleteDomainByNameQueryV2 = `DELETE FROM domains_by_name_v2 ` +
`WHERE domains_partition = ? ` +
`and name = ?`

templateListDomainQueryV2 = `SELECT name, domain.id, domain.name, domain.status, domain.description, ` +
`domain.owner_email, domain.data, config.retention, config.emit_metric, ` +
`config.archival_bucket, config.archival_status, ` +
`config.history_archival_status, config.history_archival_uri, ` +
`config.visibility_archival_status, config.visibility_archival_uri, ` +
`config.bad_binaries, config.bad_binaries_encoding, ` +
`config.isolation_groups, config.isolation_groups_encoding, ` +
`replication_config.active_cluster_name, replication_config.clusters, ` +
`is_global_domain, ` +
`config_version, ` +
`failover_version, ` +
`failover_notification_version, ` +
`previous_failover_version, ` +
`failover_end_time, ` +
`last_updated_time, ` +
`notification_version ` +
`FROM domains_by_name_v2 ` +
`WHERE domains_partition = ? `
)

// Insert a new record to domain
// return types.DomainAlreadyExistsError error if failed or already exists
// Must return ConditionFailure error if other condition doesn't match
Expand Down Expand Up @@ -204,11 +89,11 @@ func (db *cdb) InsertDomain(
isolationGroupData,
isolationGroupEncoding,
row.ReplicationConfig.ActiveClusterName,
p.SerializeClusterConfigs(row.ReplicationConfig.Clusters),
persistence.SerializeClusterConfigs(row.ReplicationConfig.Clusters),
row.IsGlobalDomain,
row.ConfigVersion,
row.FailoverVersion,
p.InitialFailoverNotificationVersion,
persistence.InitialFailoverNotificationVersion,
common.InitialPreviousFailoverVersion,
failoverEndTime,
row.LastUpdatedTime.UnixNano(),
Expand Down Expand Up @@ -307,7 +192,7 @@ func (db *cdb) UpdateDomain(
isolationGroupData,
isolationGroupEncoding,
row.ReplicationConfig.ActiveClusterName,
p.SerializeClusterConfigs(row.ReplicationConfig.Clusters),
persistence.SerializeClusterConfigs(row.ReplicationConfig.Clusters),
row.ConfigVersion,
row.FailoverVersion,
row.FailoverNotificationVersion,
Expand Down Expand Up @@ -359,9 +244,9 @@ func (db *cdb) SelectDomain(
}
}

info := &p.DomainInfo{}
info := &persistence.DomainInfo{}
config := &nosqlplugin.NoSQLInternalDomainConfig{}
replicationConfig := &p.DomainReplicationConfig{}
replicationConfig := &persistence.DomainReplicationConfig{}

// because of encoding/types, we can't directly read from config struct
var badBinariesData []byte
Expand Down Expand Up @@ -416,10 +301,10 @@ func (db *cdb) SelectDomain(
return nil, err
}

config.IsolationGroups = p.NewDataBlob(isolationGroupData, common.EncodingType(isolationGroupEncoding))
config.BadBinaries = p.NewDataBlob(badBinariesData, common.EncodingType(badBinariesDataEncoding))
config.IsolationGroups = persistence.NewDataBlob(isolationGroupData, common.EncodingType(isolationGroupEncoding))
config.BadBinaries = persistence.NewDataBlob(badBinariesData, common.EncodingType(badBinariesDataEncoding))
config.Retention = common.DaysToDuration(retentionDays)
replicationConfig.Clusters = p.DeserializeClusterConfigs(replicationClusters)
replicationConfig.Clusters = persistence.DeserializeClusterConfigs(replicationClusters)

dr := &nosqlplugin.DomainRow{
Info: info,
Expand Down Expand Up @@ -456,9 +341,9 @@ func (db *cdb) SelectAllDomains(

var name string
domain := &nosqlplugin.DomainRow{
Info: &p.DomainInfo{},
Info: &persistence.DomainInfo{},
Config: &nosqlplugin.NoSQLInternalDomainConfig{},
ReplicationConfig: &p.DomainReplicationConfig{},
ReplicationConfig: &persistence.DomainReplicationConfig{},
}
var replicationClusters []map[string]interface{}
var badBinariesData []byte
Expand Down Expand Up @@ -502,10 +387,10 @@ func (db *cdb) SelectAllDomains(
) {
if name != domainMetadataRecordName {
// do not include the metadata record
domain.Config.BadBinaries = p.NewDataBlob(badBinariesData, common.EncodingType(badBinariesDataEncoding))
domain.ReplicationConfig.Clusters = p.DeserializeClusterConfigs(replicationClusters)
domain.Config.BadBinaries = persistence.NewDataBlob(badBinariesData, common.EncodingType(badBinariesDataEncoding))
domain.ReplicationConfig.Clusters = persistence.DeserializeClusterConfigs(replicationClusters)
domain.Config.Retention = common.DaysToDuration(retentionDays)
domain.Config.IsolationGroups = p.NewDataBlob(isolationGroups, common.EncodingType(isolationGroupsEncoding))
domain.Config.IsolationGroups = persistence.NewDataBlob(isolationGroups, common.EncodingType(isolationGroupsEncoding))
domain.LastUpdatedTime = time.Unix(0, lastUpdateTime)
if failoverEndTime > emptyFailoverEndTime {
domain.FailoverEndTime = common.TimePtr(time.Unix(0, failoverEndTime))
Expand All @@ -519,9 +404,9 @@ func (db *cdb) SelectAllDomains(
lastUpdateTime = 0
retentionDays = 0
domain = &nosqlplugin.DomainRow{
Info: &p.DomainInfo{},
Info: &persistence.DomainInfo{},
Config: &nosqlplugin.NoSQLInternalDomainConfig{},
ReplicationConfig: &p.DomainReplicationConfig{},
ReplicationConfig: &persistence.DomainReplicationConfig{},
}
}

Expand Down
Loading

0 comments on commit 4664711

Please sign in to comment.