Physical sharding for NoSQL-based persistence #5187
Conversation
common/config/persistence.go
Outdated
| if _, found := connections[shardRange.Shard]; !found { | ||
| return fmt.Errorf("ShardedNosql config: Unknown history shard name: %v", shardRange.Shard) | ||
| } | ||
| if shardRange.Start != currentShardID+1 { |
There was a problem hiding this comment.
I prefer to make start inclusive but end exclusive, because that's a pretty common behavior across different areas.
There was a problem hiding this comment.
Yeah, I also debated the two options when modeling this part of the config. In a usual scenario, I agree that the start-inclusive & end-exclusive pattern works the best. However, when I was thinking about more arbitrary shard distributions (i.e. creating a mapping for a single shard) it felt like using inclusive start & end was making the config a little more readable.
Thinking about it again though, there are other ways to cleanly handle single shard pinning if it ever becomes a need. So I'll change this to the standard range pattern.
| ) | ||
|
|
||
| type ( | ||
| // Factory vends datastore implementations backed by cassandra | ||
| Factory struct { | ||
| sync.RWMutex | ||
| cfg config.Cassandra | ||
| cfg config.ShardedNoSQL |
There was a problem hiding this comment.
is this strictly a feature superset?
There was a problem hiding this comment.
Cassandra is same as NoSQL and ShardedNoSQL is strictly a superset of it, so a NoSQL configuration can be parsed into a ShardedNoSQL config object.
| return &shard, nil | ||
| } | ||
|
|
||
| s, err := sn.connectToShard(cfg.NoSQLPlugin, shardName) |
There was a problem hiding this comment.
if we don't have any already (and I may have missed them) it'd be great to get some metrics around the number of connections occurring
Ignore this, I think I was somehow wildly overstating the number of shards in my head for some reason.
| } | ||
| tlShards := sp.config.ShardingPolicy.TaskListHashing.ShardOrder | ||
| tlShardCount := len(tlShards) | ||
| hash := farm.Hash32([]byte(domainID+"_"+taskListName)) % uint32(tlShardCount) |
There was a problem hiding this comment.
having just made a silly mistake with this myself, just double-cehcking, are there any conditions under which tlShardCount could be zero? https://go.dev/play/p/Pv73y1KpbOZ
There was a problem hiding this comment.
It should be impossible with the if !sp.hasShardedTasklist at the top of the func.
| // ShardingPolicy contains configuration for physical DB sharding | ||
| ShardingPolicy struct { | ||
| // HistoryShardMapping defines the ranges of history shards stored by each DB shard. Ranges listed here *MUST* | ||
| // be continuous and non-overlapping, such that the first range in the list starts from Shard 0, each following |
There was a problem hiding this comment.
Nonblocking question: I could be lazy to look for this, but i didn't see it. Do we have any runtime checks to ensure this is true?
| @@ -198,6 +198,8 @@ type ( | |||
| SQL *SQL `yaml:"sql"` | |||
| // NoSQL contains the config for a NoSQL based datastore | |||
| NoSQL *NoSQL `yaml:"nosql"` | |||
There was a problem hiding this comment.
Are we deprecating NoSQL field?
Add a comment to announce the deprecation?
There was a problem hiding this comment.
We should not deprecate it right away. Most users outside Uber have very small Cadence deployments and the ShardedNoSQL config will just be unnecessarily more complex for them to maintain.
Good idea but I'll leave that out as a separate task. |
There is already a partitioning implementation for SQL and I think it has various weaknesses and needs to be refactored so that it can support resharding and does better job at colocation of related records across executions and history tables. Unifying the two sharding approaches by refactoring the SQL implementation is a good next step, and reusing some of the pieces in this PR might make sense at that time. I'll keep shardingPolicy in NoSQL implementation for now until it's needed outside the package. |
What changed? Added a sharding layer to the NoSQL persistence stack so that Cadence can use multiple Cassandra clusters at once in a physically sharded manner. Cadence is a heavily storage-bounded system, so the limits for the load per Cadence cluster is strictly limited by the underlying storage system. Given the massive adoption of Cadence at Uber, this scale limitation forces us to create more Cadence clusters than we want to operate. This capability will let us have one or two orders of magnitude larger Cadence clusters than we have today. Note that this feature only enables bootstrapping a brand-new cluster with multiple databases behind it. Resharding is designed but not implemented yet. Why? So that a Cadence cluster can be bootstrapped with multiple Cassandra clusters powering it. How did you test it? Added unit tests. Ran samples and tested bench tests in a staging environment. Potential risks Since this change significantly changes the low-level persistence logic, it can cause data loss if something goes terribly wrong. Release notes The change is backward compatible. Existing Cadence cluster configurations can be updated, if desired, to use the sharded NoSQL config format. However, they must continue having a single shard since Cadence still doesn't have the ability to reshard data. Documentation Changes There is a sample config file included in this PR that shows how to make use of the feature in a new cluster. Co-authored-by: emrahs <emrah@uber.com>
What changed?
Added a sharding layer to the NoSQL persistence stack so that Cadence can use multiple Cassandra clusters at once in a physically sharded manner.
Cadence is a heavily storage-bounded system, so the limits for the load per Cadence cluster is strictly limited by the underlying storage system. Given the massive adoption of Cadence at Uber, this scale limitation forces us to create more Cadence clusters than we want to operate. This capability will let us have one or two orders of magnitude larger Cadence clusters than we have today.
Note that this feature only enables bootstrapping a brand-new cluster with multiple databases behind it. Resharding is designed but not implemented yet.
Why?
So that a Cadence cluster can be bootstrapped with multiple Cassandra clusters powering it.
How did you test it?
Added unit tests. Ran samples and tested bench tests in a staging environment.
Potential risks
Since this change significantly changes the low-level persistence logic, it can cause data loss if something goes terribly wrong.
Release notes
The change is backward compatible. Existing Cadence cluster configurations can be updated, if desired, to use the sharded NoSQL config format. However, they must continue having a single shard since Cadence still doesn't have the ability to reshard data.
Documentation Changes
There is a sample config file included in this PR that shows how to make use of the feature in a new cluster.