feat: [Dynamic config] Support for namespace filter and onboarding property (#7358)

<!-- Describe what has changed in this PR -->
**What changed?**
- Adding namespace as a filter for dynamic properties
- Model the onboarding mode as a dynamic property 


<!-- Tell your future self why have you made these changes -->
**Why?**
We need to onboard to the shard distributor incrementally using a
dynamic configuration

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
Unit tests, the dynamic config it is not integrated yet and used in the
code

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

---------

Signed-off-by: edigregorio <[email protected]>

Author: eleonoradgr

common/dynamicconfig/config.go

@@ -447,6 +447,22 @@ func (c *Collection) GetStringPropertyFilteredByTaskListInfo(key dynamicproperti
 	}
 }
 
+// GetStringPropertyFilteredByNamespace gets property with domain filter and asserts that it's a string
+func (c *Collection) GetStringPropertyFilteredByNamespace(key dynamicproperties.StringKey) dynamicproperties.StringPropertyFnWithNamespaceFilters {
+	return func(namespace string) string {
+		filters := c.toFilterMap(dynamicproperties.NamespaceFilter(namespace))
+		val, err := c.client.GetStringValue(
+			key,
+			filters,
+		)
+		if err != nil {
+			c.logError(key, filters, err)
+			return key.DefaultString()
+		}
+		return val
+	}
+}
+
 // GetBoolPropertyFilteredByDomain gets property with domain filter and asserts that it's a bool
 func (c *Collection) GetBoolPropertyFilteredByDomain(key dynamicproperties.BoolKey) dynamicproperties.BoolPropertyFnWithDomainFilter {
 	return func(domain string) bool {

common/dynamicconfig/config_test.go

@@ -120,6 +120,15 @@ func (s *configSuite) TestGetStringPropertyFnByTaskListInfo() {
 	s.Equal("round-robin", value(domain, taskList, taskType))
 }
 
+func (s *configSuite) TestGetStringPropertyFnWithNamespaceFilter() {
+	key := dynamicproperties.MigrationMode
+	namespace := "testService"
+	value := s.cln.GetStringPropertyFilteredByNamespace(key)
+	s.Equal(key.DefaultString(), value(namespace))
+	s.client.SetValue(key, "efg")
+	s.Equal("efg", value(namespace))
+}
+
 func (s *configSuite) TestGetStringPropertyFilteredByRatelimitKey() {
 	key := dynamicproperties.FrontendGlobalRatelimiterMode
 	ratelimitKey := "user:testDomain"

common/dynamicconfig/dynamicproperties/constants.go

@@ -2422,14 +2422,28 @@ const (
 	// Default value: "hash_ring"
 	MatchingShardDistributionMode
 
-	// LastStringKey must be the last one in this const group
-	LastStringKey
-
 	// SerializationEncoding is the encoding type for blobs
 	// KeyName: history.serializationEncoding
 	// Value type: String
 	// Default value: "thriftrw"
 	SerializationEncoding
+
+	// MigrationMode is the mode the at represent the state of the migration to rely on shard distributor for the sharding mechanism
+	//
+	// "invalid" invalid mode for the migration, not expected to be used
+	// "local_pass" the executor library is integrated but no external call to the SD happening
+	// "local_pass_shadow" heartbeat calls to the SD to update the sharding state in SD
+	// "distributed_pass" the local sharding mechanism is sent to SD, returned by SD and applied in the onboarded service
+	// "onboarded" the sharding logic in SD is used
+	//
+	// KeyName: shardDistributor.migrationMode
+	// Value type: String
+	// Default value: local_pass
+	// Allowed filters: namespace
+	MigrationMode
+
+	// LastStringKey must be the last one in this const group
+	LastStringKey
 )
 
 const (
@@ -4939,6 +4953,12 @@ var StringKeys = map[StringKey]DynamicString{
 		Description:  "SerializationEncoding is the encoding type for blobs",
 		DefaultValue: string(constants.EncodingTypeThriftRW),
 	},
+	MigrationMode: {
+		KeyName:      "shardDistributor.migrationMode",
+		Description:  "MigrationMode is the mode the at represent the state of the migration to rely on shard distributor for the sharding mechanism",
+		DefaultValue: "local_pass",
+		Filters:      []Filter{Namespace},
+	},
 }
 
 var DurationKeys = map[DurationKey]DynamicDuration{

common/dynamicconfig/dynamicproperties/constants_test.go

@@ -248,6 +248,13 @@ func (s *constantSuite) TestStringKey() {
 			Description:  "ReadVisibilityStoreName is key to identify which store to read visibility data from",
 			DefaultValue: "es",
 		},
+		"MigrationMode": {
+			Key:          MigrationMode,
+			KeyName:      "shardDistributor.migrationMode",
+			Filters:      []Filter{Namespace},
+			Description:  "MigrationMode is the mode the at represent the state of the migration to rely on shard distributor for the sharding mechanism",
+			DefaultValue: "local_pass",
+		},
 	}
 
 	for _, value := range testStringKeys {

common/dynamicconfig/dynamicproperties/definitions.go

@@ -86,6 +86,9 @@ type StringPropertyFnWithDomainFilter func(domain string) string
 // StringPropertyFnWithTaskListInfoFilters is a wrapper to get string property from dynamic config with domainID as filter
 type StringPropertyFnWithTaskListInfoFilters func(domain string, taskList string, taskType int) string
 
+// StringPropertyFnWithNamespaceFilters is a wrapper to get string property from dynamic config with namespace as filter
+type StringPropertyFnWithNamespaceFilters func(namespace string) string
+
 // BoolPropertyFnWithDomainFilter is a wrapper to get bool property from dynamic config with domain as filter
 type BoolPropertyFnWithDomainFilter func(domain string) bool
 
@@ -113,6 +116,9 @@ type ListPropertyFn func(opts ...FilterOption) []interface{}
 // StringPropertyWithRatelimitKeyFilter is a wrapper to get strings (currently global ratelimiter modes) per global ratelimit key
 type StringPropertyWithRatelimitKeyFilter func(globalRatelimitKey string) string
 
+// StringPropertyWithNamespaceFilter is a wrapper to get strings per namespace
+type StringPropertyWithNamespaceFilter func(namespace string) string
+
 func (f IntPropertyFn) AsFloat64(opts ...FilterOption) func() float64 {
 	return func() float64 { return float64(f(opts...)) }
 }

common/dynamicconfig/dynamicproperties/filter.go

@@ -57,6 +57,8 @@ func ParseFilter(filterName string) Filter {
 		return WorkflowType
 	case "ratelimitKey":
 		return RatelimitKey
+	case "namespace":
+		return Namespace
 	default:
 		return UnknownFilter
 	}
@@ -73,6 +75,7 @@ var filters = []string{
 	"workflowID",
 	"workflowType",
 	"ratelimitKey",
+	"namespace",
 }
 
 const (
@@ -95,6 +98,8 @@ const (
 	WorkflowType
 	// RatelimitKey is the global ratelimit key (not a local key name)
 	RatelimitKey
+	// Namespace is the entity of independent shard distribution mechanism
+	Namespace
 
 	// LastFilterTypeForTest must be the last one in this const group for testing purpose
 	LastFilterTypeForTest
@@ -166,6 +171,13 @@ func RatelimitKeyFilter(key string) FilterOption {
 	}
 }
 
+// NamespaceFilter filters by namespace
+func NamespaceFilter(namespace string) FilterOption {
+	return func(filterMap map[Filter]interface{}) {
+		filterMap[Namespace] = namespace
+	}
+}
+
 // ToGetDynamicConfigFilterRequest generates a GetDynamicConfigRequest object
 // by converting filters to DynamicConfigFilter objects and setting values
 func ToGetDynamicConfigFilterRequest(configName string, filters []FilterOption) *types.GetDynamicConfigRequest {