Multi-namespace search attribute config format (#95)

Author: pglass

config/config.go

@@ -111,11 +111,26 @@ type (
 		MuxTransports              []MuxTransportConfig  `yaml:"mux"`
 		HealthCheck                *HealthCheckConfig    `yaml:"healthCheck"`
 		NamespaceNameTranslation   NameTranslationConfig `yaml:"namespaceNameTranslation"`
-		SearchAttributeTranslation NameTranslationConfig `yaml:"searchAttributeTranslation"`
+		SearchAttributeTranslation SATranslationConfig   `yaml:"searchAttributeTranslation"`
 		Metrics                    *MetricsConfig        `yaml:"metrics"`
 		ProfilingConfig            ProfilingConfig       `yaml:"profiling"`
 	}
 
+	SATranslationConfig struct {
+		NamespaceMappings []SANamespaceMapping `yaml:"namespaceMappings"`
+	}
+
+	SANamespaceMapping struct {
+		Name        string      `yaml:"name"`
+		NamespaceId string      `yaml:"namespaceId"`
+		Mappings    []SAMapping `yaml:"mappings"`
+	}
+
+	SAMapping struct {
+		LocalName  string `yaml:"localFieldName"`
+		RemoteName string `yaml:"remoteFieldName"`
+	}
+
 	ProfilingConfig struct {
 		PProfHTTPAddress string `yaml:"pprofAddress"`
 	}
@@ -320,3 +335,36 @@ func (c *ProfilingConfig) UnmarshalYAML(unmarshal func(interface{}) error) error
 	type plain ProfilingConfig
 	return unmarshal((*plain)(c))
 }
+
+func (s SATranslationConfig) IsEnabled() bool {
+	return len(s.NamespaceMappings) > 0
+}
+
+// ToMaps returns request and response mappings.
+func (s SATranslationConfig) ToMaps(inBound bool) (map[string]map[string]string, map[string]map[string]string) {
+	reqMap := make(map[string]map[string]string)
+	respMap := make(map[string]map[string]string)
+	for _, ns := range s.NamespaceMappings {
+		reqMap[ns.NamespaceId] = make(map[string]string, len(ns.Mappings))
+		respMap[ns.NamespaceId] = make(map[string]string, len(ns.Mappings))
+
+		if inBound {
+			// For inbound listener,
+			//   - incoming requests from remote server are modifed to match local server
+			//   - outgoing responses to local server are modified to match remote server
+			for _, tr := range ns.Mappings {
+				reqMap[ns.NamespaceId][tr.RemoteName] = tr.LocalName
+				respMap[ns.NamespaceId][tr.LocalName] = tr.RemoteName
+			}
+		} else {
+			// For outbound listener,
+			//   - incoming requests from local server are modifed to match remote server
+			//   - outgoing responses to remote server are modified to match local server
+			for _, tr := range ns.Mappings {
+				reqMap[ns.NamespaceId][tr.LocalName] = tr.RemoteName
+				respMap[ns.NamespaceId][tr.RemoteName] = tr.LocalName
+			}
+		}
+	}
+	return reqMap, respMap
+}

interceptor/search_attribute_translator.go

@@ -0,0 +1,63 @@
+package interceptor
+
+import (
+	"strings"
+
+	"go.temporal.io/server/common/api"
+)
+
+type (
+	saTranslator struct {
+		matchMethod func(string) bool
+		reqMap      map[string]stringMatcher
+		respMap     map[string]stringMatcher
+	}
+)
+
+func NewSearchAttributeTranslator(reqMap, respMap map[string]map[string]string) Translator {
+	return &saTranslator{
+		matchMethod: func(method string) bool {
+			// In workflowservice APIs, responses only contain the search attribute alias.
+			// We should never translate these responses to the search attribute's indexed field.
+			return !strings.HasPrefix(method, api.WorkflowServicePrefix)
+		},
+		reqMap:  createStringMatchers(reqMap),
+		respMap: createStringMatchers(respMap),
+	}
+}
+
+func (s *saTranslator) MatchMethod(m string) bool {
+	return s.matchMethod(m)
+}
+
+func (s *saTranslator) TranslateRequest(req any) (bool, error) {
+	return visitSearchAttributes(req, s.getNamespaceReqMatcher(""))
+}
+
+func (s *saTranslator) TranslateResponse(resp any) (bool, error) {
+	return visitSearchAttributes(resp, s.getNamespaceRespMatcher(""))
+}
+
+func (s *saTranslator) getNamespaceReqMatcher(namespaceId string) stringMatcher {
+	// Placeholder: Just return the first one (only support one namespace mapping)
+	for _, matcher := range s.reqMap {
+		return matcher
+	}
+	return createStringMatcher(nil)
+}
+
+func (s *saTranslator) getNamespaceRespMatcher(namespaceId string) stringMatcher {
+	// Placeholder: Just return the first one (only support one namespace mappping)
+	for _, matcher := range s.respMap {
+		return matcher
+	}
+	return createStringMatcher(nil)
+}
+
+func createStringMatchers(nsMappings map[string]map[string]string) map[string]stringMatcher {
+	result := make(map[string]stringMatcher, len(nsMappings))
+	for nsId, mapping := range nsMappings {
+		result[nsId] = createStringMatcher(mapping)
+	}
+	return result
+}

interceptor/translator.go

@@ -1,11 +1,5 @@
 package interceptor
 
-import (
-	"strings"
-
-	"go.temporal.io/server/common/api"
-)
-
 type (
 	Translator interface {
 		MatchMethod(string) bool
@@ -30,19 +24,6 @@ func NewNamespaceNameTranslator(reqMap, respMap map[string]string) Translator {
 	}
 }
 
-func NewSearchAttributeTranslator(reqMap, respMap map[string]string) Translator {
-	return &translatorImpl{
-		matchMethod: func(method string) bool {
-			// In workflowservice APIs, responses only contain the search attribute alias.
-			// We should never translate these responses to the search attribute's indexed field.
-			return !strings.HasPrefix(method, api.WorkflowServicePrefix)
-		},
-		matchReq:  createStringMatcher(reqMap),
-		matchResp: createStringMatcher(respMap),
-		visitor:   visitSearchAttributes,
-	}
-}
-
 func (n *translatorImpl) MatchMethod(m string) bool {
 	return n.matchMethod(m)
 }

proxy/proxy.go

@@ -61,7 +61,10 @@ func makeServerOptions(
 	}
 
 	if tln := proxyOpts.Config.SearchAttributeTranslation; tln.IsEnabled() {
-		logger.Info("search attribute translation enabled", tag.NewAnyTag("mappings", tln.Mappings))
+		logger.Info("search attribute translation enabled", tag.NewAnyTag("mappings", tln.NamespaceMappings))
+		if len(tln.NamespaceMappings) > 1 {
+			panic("multiple namespace search attribute mappings are not supported")
+		}
 		translators = append(translators,
 			interceptor.NewSearchAttributeTranslator(tln.ToMaps(proxyOpts.IsInbound)))
 	}