From c52b1c9373abe044b85ff15c99692c7b2c9227b1 Mon Sep 17 00:00:00 2001 From: JH Date: Sun, 23 Aug 2026 18:03:29 -0700 Subject: [PATCH 1/3] multi es --- config/config.go | 55 ++- config/config_test.go | 235 +++++++++++ interceptor/reflection.go | 155 +++++++- interceptor/reflection_test.go | 72 ++++ interceptor/search_attribute_translator.go | 57 ++- .../search_attribute_translator_test.go | 367 +++++++++++++++++- interceptor/translation_interceptor.go | 5 +- interceptor/translation_interceptor_test.go | 36 +- interceptor/translator.go | 7 +- metrics/prometheus_defs.go | 12 + proxy/cluster_connection.go | 7 +- proxy/cluster_connection_test.go | 64 +++ 12 files changed, 1028 insertions(+), 44 deletions(-) diff --git a/config/config.go b/config/config.go index f23d0a3a..7ea16fdc 100644 --- a/config/config.go +++ b/config/config.go @@ -2,6 +2,7 @@ package config import ( "bytes" + "fmt" "maps" "os" @@ -33,6 +34,11 @@ const ( HTTP HealthCheckProtocol = "http" ) +// LegacyWildcardNamespaceID is the empty namespaceId used by older configs that predate +// per-namespace search attribute translation. When it is the sole mapping it is applied to +// every namespace, preserving the behaviour those configs shipped with. +const LegacyWildcardNamespaceID = "" + type ( ConfigProvider interface { GetS2SProxyConfig() S2SProxyConfig @@ -242,6 +248,36 @@ func (s *SATranslationConfig) IsEnabled() bool { return len(s.NamespaceMappings) > 0 } +// Validate reports the first reason the configured namespace mappings cannot be applied per +// namespace: search attributes are translated by namespaceId, so every mapping needs an id that +// is present and unique. A single mapping is allowed to omit the namespaceId, in which case it is +// applied to every namespace, see LegacyWildcardNamespaceID. +func (s *SATranslationConfig) Validate() error { + namesByNamespaceId := make(map[string]string, len(s.NamespaceMappings)) + seenNames := make(map[string]struct{}, len(s.NamespaceMappings)) + for _, m := range s.NamespaceMappings { + // The missing id is reported before the duplicate id so that a config with several + // mappings and no ids gets the actionable message rather than `duplicate namespaceId ""`. + if m.NamespaceId == LegacyWildcardNamespaceID && len(s.NamespaceMappings) > 1 { + return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] has an empty namespaceId; namespaceId is required when more than one namespace is configured", m.Name) + } + if existing, found := namesByNamespaceId[m.NamespaceId]; found { + return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] and namespaceMappings[name=%q] have duplicate namespaceId %q", existing, m.Name, m.NamespaceId) + } + namesByNamespaceId[m.NamespaceId] = m.Name + + // The name is informational, so only reject duplicates of a name that was actually set. + if m.Name == "" { + continue + } + if _, found := seenNames[m.Name]; found { + return fmt.Errorf("searchAttributeTranslation: namespaceMappings has duplicate name %q", m.Name) + } + seenNames[m.Name] = struct{}{} + } + return nil +} + // 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) @@ -324,18 +360,28 @@ func (s SearchAttributeTranslation) FlattenMaps() map[string]map[string]string { return raw } +// HasLegacyWildcard reports whether the translation was built from a config that omitted the +// namespaceId, in which case its mappings apply to every namespace. See LegacyWildcardNamespaceID. +func (s SearchAttributeTranslation) HasLegacyWildcard() bool { + _, found := s.inner[LegacyWildcardNamespaceID] + return found +} + // AsLocalToRemoteSATranslation converts the flat list of namespace + local/remote pairs into a map of BiMaps, with local->remote // as the direction returned. The remote->local mapping can be accessed with saTranslator[namespaceId].Inverse() func (s *SATranslationConfig) AsLocalToRemoteSATranslation() (SearchAttributeTranslation, error) { if s.cachedBiMap.inner != nil { return s.cachedBiMap, nil } + // This is the only path that builds the translation, so it is where the config is checked. + if err := s.Validate(); err != nil { + return SearchAttributeTranslation{}, err + } saTranslation := SearchAttributeTranslation{ inner: make(map[string]collect.StaticBiMap[string, string], len(s.NamespaceMappings)), } for _, mapping := range s.NamespaceMappings { - var err error - saTranslation.inner[mapping.NamespaceId], err = collect.NewStaticBiMap(func(yield func(string, string) bool) { + nsBiMap, err := collect.NewStaticBiMap(func(yield func(string, string) bool) { for _, attrPair := range mapping.Mappings { if !yield(attrPair.LocalName, attrPair.RemoteName) { return @@ -343,8 +389,11 @@ func (s *SATranslationConfig) AsLocalToRemoteSATranslation() (SearchAttributeTra } }, len(mapping.Mappings)) if err != nil { - return SearchAttributeTranslation{}, err + return SearchAttributeTranslation{}, fmt.Errorf( + "searchAttributeTranslation: namespaceMappings[name=%q namespaceId=%q]: %w", + mapping.Name, mapping.NamespaceId, err) } + saTranslation.inner[mapping.NamespaceId] = nsBiMap } s.cachedBiMap = saTranslation return saTranslation, nil diff --git a/config/config_test.go b/config/config_test.go index f42ce17d..92e5d528 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -322,3 +322,238 @@ func TestExampleChart(t *testing.T) { require.Equal(t, ConnectionType("mux-client"), cc.Remote.ConnectionType) require.Equal(t, "s2s-proxy-sample.example.tmprl.cloud:8233", cc.Remote.MuxAddressInfo.ConnectionString) } + +func TestSATranslationConfigValidate(t *testing.T) { + cases := []struct { + name string + cfg SATranslationConfig + // wantValidateErr lists substrings the Validate error must contain. Empty means the + // config is valid. + wantValidateErr []string + // wantValidateErrExcludes lists substrings the Validate error must not contain. + wantValidateErrExcludes []string + // wantTranslationErr lists substrings the AsLocalToRemoteSATranslation error must + // contain. Leave nil when the translation fails for the same reason as Validate. + wantTranslationErr []string + // verify runs against the built translation when it is expected to succeed. + verify func(t *testing.T, saTranslation SearchAttributeTranslation) + }{ + { + name: "no namespace mappings", + cfg: SATranslationConfig{}, + verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { + require.Equal(t, 0, saTranslation.LenNamespaces()) + require.False(t, saTranslation.HasLegacyWildcard()) + }, + }, + { + // Two namespaces sharing an id used to silently overwrite each other, leaving one + // namespace translated with the other namespace's mappings. + name: "duplicate namespaceId", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "namespace1", + NamespaceId: "namespace-id-1", + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + { + Name: "namespace2", + NamespaceId: "namespace-id-1", + Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}}, + }, + }, + }, + wantValidateErr: []string{ + `namespaceMappings[name="namespace1"]`, + `namespaceMappings[name="namespace2"]`, + `duplicate namespaceId "namespace-id-1"`, + }, + }, + { + name: "empty namespaceId alongside another namespace", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "legacyNamespace", + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + { + Name: "namespace2", + NamespaceId: "namespace-id-2", + Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}}, + }, + }, + }, + wantValidateErr: []string{ + `namespaceMappings[name="legacyNamespace"]`, + "namespaceId is required when more than one namespace is configured", + }, + }, + { + // The missing id is the actionable problem, so it is reported ahead of the duplicate + // id the empty values also form. + name: "every mapping missing its namespaceId", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "namespace1", + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + { + Name: "namespace2", + Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}}, + }, + }, + }, + wantValidateErr: []string{ + `namespaceMappings[name="namespace1"]`, + "namespaceId is required", + }, + wantValidateErrExcludes: []string{"duplicate namespaceId"}, + }, + { + // Configs written before per-namespace translation omit the namespaceId entirely. + // A single such mapping keeps working and is applied to every namespace. + name: "single mapping with empty namespaceId", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + }, + }, + verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { + require.Equal(t, 1, saTranslation.LenNamespaces()) + require.True(t, saTranslation.HasLegacyWildcard()) + require.Equal(t, "remoteOne", saTranslation.Get(LegacyWildcardNamespaceID, "localOne")) + require.Equal(t, "localOne", saTranslation.Inverse().Get(LegacyWildcardNamespaceID, "remoteOne")) + }, + }, + { + // The shape deployed today: the namespace is named but the namespaceId is blank. + name: "named mapping with empty namespaceId", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "migration-namespace", + Mappings: []SAMapping{ + {LocalName: "CustomKeywordField", RemoteName: "Keyword01"}, + {LocalName: "CustomStringField", RemoteName: "Text01"}, + }, + }, + }, + }, + verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { + require.Equal(t, 1, saTranslation.LenNamespaces()) + require.True(t, saTranslation.HasLegacyWildcard()) + require.Equal(t, 2, saTranslation.Len(LegacyWildcardNamespaceID)) + require.Equal(t, "Keyword01", saTranslation.Get(LegacyWildcardNamespaceID, "CustomKeywordField")) + require.Equal(t, "Text01", saTranslation.Get(LegacyWildcardNamespaceID, "CustomStringField")) + }, + }, + { + // The namespace is well formed, the mappings inside it are not: the bimap rejects + // them and the error must say which namespace it came from. + name: "duplicate localFieldName within one namespace", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "namespace1", + NamespaceId: "namespace-id-1", + Mappings: []SAMapping{ + {LocalName: "localOne", RemoteName: "remoteOne"}, + {LocalName: "localOne", RemoteName: "remoteTwo"}, + }, + }, + }, + }, + wantTranslationErr: []string{ + `namespaceMappings[name="namespace1" namespaceId="namespace-id-1"]`, + }, + }, + { + name: "duplicate name across namespaces", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "namespace1", + NamespaceId: "namespace-id-1", + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + { + Name: "namespace1", + NamespaceId: "namespace-id-2", + Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}}, + }, + }, + }, + wantValidateErr: []string{`duplicate name "namespace1"`}, + }, + { + name: "distinct namespaceIds translate independently", + cfg: SATranslationConfig{ + NamespaceMappings: []SANamespaceMapping{ + { + Name: "namespace1", + NamespaceId: "namespace-id-1", + Mappings: []SAMapping{{LocalName: "localOne", RemoteName: "remoteOne"}}, + }, + { + Name: "namespace2", + NamespaceId: "namespace-id-2", + Mappings: []SAMapping{{LocalName: "localTwo", RemoteName: "remoteTwo"}}, + }, + }, + }, + verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { + require.Equal(t, 2, saTranslation.LenNamespaces()) + require.False(t, saTranslation.HasLegacyWildcard()) + require.Equal(t, "remoteOne", saTranslation.Get("namespace-id-1", "localOne")) + require.Equal(t, "remoteTwo", saTranslation.Get("namespace-id-2", "localTwo")) + // Each namespace only knows its own attributes. + require.Equal(t, "", saTranslation.Get("namespace-id-1", "localTwo")) + require.Equal(t, "", saTranslation.Get("namespace-id-2", "localOne")) + require.Equal(t, NewTuple("", false), NewTuple(saTranslation.GetExists(LegacyWildcardNamespaceID, "localOne"))) + require.Equal(t, "localOne", saTranslation.Inverse().Get("namespace-id-1", "remoteOne")) + require.Equal(t, "localTwo", saTranslation.Inverse().Get("namespace-id-2", "remoteTwo")) + }, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + cfg := c.cfg + err := cfg.Validate() + if len(c.wantValidateErr) == 0 { + require.NoError(t, err) + } else { + require.Error(t, err) + for _, want := range c.wantValidateErr { + require.Contains(t, err.Error(), want) + } + for _, unwanted := range c.wantValidateErrExcludes { + require.NotContains(t, err.Error(), unwanted) + } + } + + // Validate runs inside AsLocalToRemoteSATranslation, so an invalid config fails + // there for the same reason unless the case says otherwise. + wantTranslationErr := c.wantTranslationErr + if wantTranslationErr == nil { + wantTranslationErr = c.wantValidateErr + } + saTranslation, err := cfg.AsLocalToRemoteSATranslation() + if len(wantTranslationErr) == 0 { + require.NoError(t, err) + c.verify(t, saTranslation) + return + } + require.Error(t, err) + for _, want := range wantTranslationErr { + require.Contains(t, err.Error(), want) + } + require.Equal(t, 0, saTranslation.LenNamespaces()) + }) + } +} diff --git a/interceptor/reflection.go b/interceptor/reflection.go index 9f836fa4..d2a35ce4 100644 --- a/interceptor/reflection.go +++ b/interceptor/reflection.go @@ -10,6 +10,8 @@ import ( "go.temporal.io/api/history/v1" "go.temporal.io/api/namespace/v1" "go.temporal.io/api/workflowservice/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + replicationspb "go.temporal.io/server/api/replication/v1" "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/persistence/serialization" @@ -128,6 +130,22 @@ var ( } ) +const namespaceIDFieldName = "NamespaceId" + +// namespaceIDOwners maps struct types that OWN a namespace to their NamespaceId field index. +// Deliberately an allowlist, not a name match: other types have a NamespaceId naming a +// DIFFERENT namespace -- notably history.StartChildWorkflowExecutionInitiatedEventAttributes, +// whose NamespaceId is the *child's* and which also carries SearchAttributes. Matching on the +// field name would translate a parent's history event with the child's mapping. +// +// Built once at init and never mutated afterwards, so reads need no lock. +var namespaceIDOwners = mustBuildNamespaceIDOwners( + reflect.TypeFor[replicationspb.HistoryTaskAttributes](), + reflect.TypeFor[replicationspb.BackfillHistoryTaskAttributes](), + reflect.TypeFor[replicationspb.SyncVersionedTransitionTaskAttributes](), + reflect.TypeFor[persistencespb.WorkflowExecutionInfo](), +) + // stringMatcher returns 2 values: // 1. new name. If there is no change, new name equals to input name // 2. whether or not the input name matches the defined rule(s). @@ -137,6 +155,23 @@ type stringMatcher func(name string) (string, bool) // It returns whether anything was matched and any error it encountered. type visitor func(logger log.Logger, obj any, match stringMatcher) (bool, error) +// saMatcherResolver returns the search attribute matcher configured for a namespace id. +// The second return value is false when that namespace has no mapping, in which case the +// search attributes at hand must be left untouched. +type saMatcherResolver func(namespaceID string) (stringMatcher, bool) + +// constMatcherResolver adapts a single matcher to saMatcherResolver, for callers that +// intentionally apply one mapping to every namespace. +func constMatcherResolver(match stringMatcher) saMatcherResolver { + return func(string) (stringMatcher, bool) { return match, true } +} + +// blobVisitor visits the history events deserialized from a data blob. +// It returns whether anything was matched and any error it encountered. +// Callers close over whatever matching state they need, since a data blob +// starts a fresh traversal that cannot see the enclosing message. +type blobVisitor func(events []*history.HistoryEvent) (bool, error) + // visitNamespace uses reflection to recursively visit all fields // in the given object. When it finds namespace string fields, it invokes // the provided match function. @@ -180,7 +215,9 @@ func visitNamespace(logger log.Logger, obj any, match stringMatcher) (bool, erro } return visit.Skip, nil } else if dataBlobFieldNames[fieldType.Name] { - changed, err := visitDataBlobs(logger, vwp, match, visitNamespace) + changed, err := visitDataBlobs(logger, vwp, func(events []*history.HistoryEvent) (bool, error) { + return visitNamespace(logger, events, match) + }) matched = matched || changed if err != nil { return visit.Stop, err @@ -208,9 +245,12 @@ func visitNamespace(logger log.Logger, obj any, match stringMatcher) (bool, erro } // visitSearchAttributes uses reflection to recursively visit all fields -// in the given object. When it finds namespace string fields, it invokes -// the provided match function. -func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (bool, error) { +// in the given object. When it finds search attribute fields, it resolves the namespace +// that owns them and applies the matcher configured for that namespace, if any. +// +// boundNamespaceID is the namespace to fall back on when the parent chain reaches no +// namespace owner. See resolveNamespaceID. +func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver, boundNamespaceID string) (bool, error) { var matched bool // The visitor function can return Skip, Stop, or Continue to control recursion. @@ -225,12 +265,28 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo return action, nil } if dataBlobFieldNames[fieldType.Name] { - changed, err := visitDataBlobs(logger, vwp, match, visitSearchAttributes) + // Resolve once, here at the boundary: the parent chain is still intact, whereas the + // events inside the blob are visited in a fresh traversal that cannot see the + // enclosing namespace owner. Descend even when the namespace is unresolved, since + // visitDataBlobs also repairs invalid UTF-8 independently of any translation. + nsID := resolveNamespaceID(vwp, boundNamespaceID) + changed, err := visitDataBlobs(logger, vwp, func(events []*history.HistoryEvent) (bool, error) { + return visitSearchAttributes(logger, events, resolve, nsID) + }) matched = matched || changed if err != nil { return visit.Stop, err } } else if searchAttributeFieldNames[fieldType.Name] { + nsID := resolveNamespaceID(vwp, boundNamespaceID) + match, ok := resolve(nsID) + if !ok { + logSkippedSearchAttributes(logger, obj, nsID) + + // Leave these search attributes untouched. + return visit.Continue, nil + } + // This could be *common.SearchAttributes, or it could be map[string]*common.Payload (indexed fields) var changed bool switch attrs := vwp.Interface().(type) { @@ -244,7 +300,14 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo } } default: - return visit.Stop, fmt.Errorf("unhandled search attribute type: %T", attrs) + // Reachable: Add- and RemoveSearchAttributesRequest name their fields + // SearchAttributes too, but they hold a map[string]enums.IndexedValueType and a + // []string. Skip them instead of aborting translation of the whole message. + logger.Warn("unhandled search attribute type", + tag.NewStringTag("type", fmt.Sprintf("%T", attrs))) + metrics.SearchAttrTranslationSkipped.WithLabelValues( + metrics.SkipReasonUnsupportedType, metrics.SanitizedTypeName(obj)).Inc() + return visit.Continue, nil } matched = matched || changed @@ -257,6 +320,26 @@ func visitSearchAttributes(logger log.Logger, obj any, match stringMatcher) (boo return matched, err } +// logSkippedSearchAttributes reports search attributes left untranslated because no matcher +// resolved for their namespace. +func logSkippedSearchAttributes(logger log.Logger, obj any, nsID string) { + msgType := metrics.SanitizedTypeName(obj) + if nsID == "" { + // No enclosing namespace owner and nothing seeded a fallback. That is a gap in + // namespaceIDOwners or an unpaired response, not a configuration decision. + logger.Warn("could not resolve namespace for search attributes", + tag.NewStringTag("type", msgType)) + metrics.SearchAttrTranslationSkipped.WithLabelValues( + metrics.SkipReasonUnresolvedNamespace, msgType).Inc() + return + } + + // A namespace with no configured mapping is the expected steady state for every namespace + // that is not being migrated, so it is not counted. + logger.Debug("no search attribute mapping configured for namespace", + tag.NewStringTag("namespace-id", nsID), tag.NewStringTag("type", msgType)) +} + func translateIndexedFields(fields map[string]*common.Payload, match stringMatcher) (map[string]*common.Payload, bool) { if fields == nil { return fields, false @@ -287,10 +370,56 @@ func getParentFieldType(vwp visit.ValueWithParent) (result reflect.StructField, return fieldType, action } -func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMatcher, visitor visitor) (bool, error) { +// mustBuildNamespaceIDOwners resolves each owner's NamespaceId to a field index up front, so +// that resolveNamespaceID never calls reflect.Type.FieldByName on the hot path: that linear +// scans the 20+ fields of a generated proto struct on every hop. The owner set is fixed at +// compile time, so an absent or retyped field is a programmer error (or an upstream proto +// rename) and panics here rather than silently disabling translation. +func mustBuildNamespaceIDOwners(ownerTypes ...reflect.Type) map[reflect.Type]int { + owners := make(map[reflect.Type]int, len(ownerTypes)) + for _, ownerType := range ownerTypes { + field, ok := ownerType.FieldByName(namespaceIDFieldName) + if !ok { + panic(fmt.Sprintf("namespace owner %v has no %s field", ownerType, namespaceIDFieldName)) + } + if len(field.Index) != 1 || field.Type.Kind() != reflect.String { + panic(fmt.Sprintf("namespace owner %v has a %s field that is not a direct string: %v", + ownerType, namespaceIDFieldName, field.Type)) + } + owners[ownerType] = field.Index[0] + } + return owners +} + +// resolveNamespaceID walks UP from vwp to the nearest enclosing namespace owner and returns its +// NamespaceId. Walking only upward keeps the result independent of traversal order, which is +// unspecified: visit.ValuesUnsafe pops the front of its worklist but swaps in the last element, +// so it is neither breadth- nor depth-first. Tracking the most recent NamespaceId seen while +// descending would be non-deterministic. +// +// fallback carries context across a boundary the parent chain cannot cross: a data blob, whose +// events are visited in a fresh traversal, or a unary response whose paired request holds the +// only namespace id. +func resolveNamespaceID(vwp visit.ValueWithParent, fallback string) string { + for p := vwp.Parent; p != nil; p = p.Parent { + if p.Kind() != reflect.Struct { + continue + } + fieldIdx, ok := namespaceIDOwners[p.Type()] + if !ok { + continue + } + if nsID := p.Field(fieldIdx).String(); nsID != "" { + return nsID + } + } + return fallback +} + +func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, bv blobVisitor) (bool, error) { switch evt := vwp.Interface().(type) { case []*common.DataBlob: - newEvts, matched, changed, err := translateDataBlobs(logger, match, visitor, evt...) + newEvts, matched, changed, err := translateDataBlobs(logger, bv, evt...) if err != nil { return matched, err } @@ -301,7 +430,7 @@ func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMa } return matched, nil case *common.DataBlob: - newEvt, matched, changed, err := translateOneDataBlob(logger, match, visitor, evt) + newEvt, matched, changed, err := translateOneDataBlob(logger, bv, evt) if err != nil { return matched, err } @@ -316,9 +445,9 @@ func visitDataBlobs(logger log.Logger, vwp visit.ValueWithParent, match stringMa } } -func translateDataBlobs(logger log.Logger, match stringMatcher, visitor visitor, blobs ...*common.DataBlob) (result []*common.DataBlob, anyMatched, anyChanged bool, retErr error) { +func translateDataBlobs(logger log.Logger, bv blobVisitor, blobs ...*common.DataBlob) (result []*common.DataBlob, anyMatched, anyChanged bool, retErr error) { for i, blob := range blobs { - newBlob, matched, changed, err := translateOneDataBlob(logger, match, visitor, blob) + newBlob, matched, changed, err := translateOneDataBlob(logger, bv, blob) anyChanged = anyChanged || changed anyMatched = anyMatched || matched if err != nil { @@ -329,7 +458,7 @@ func translateDataBlobs(logger log.Logger, match stringMatcher, visitor visitor, return blobs, anyMatched, anyChanged, nil } -func translateOneDataBlob(logger log.Logger, match stringMatcher, visitor visitor, blob *common.DataBlob) (result *common.DataBlob, matched, changed bool, retErr error) { +func translateOneDataBlob(logger log.Logger, bv blobVisitor, blob *common.DataBlob) (result *common.DataBlob, matched, changed bool, retErr error) { if blob == nil || len(blob.Data) == 0 { return blob, matched, changed, nil } @@ -356,7 +485,7 @@ func translateOneDataBlob(logger log.Logger, match stringMatcher, visitor visito } } - m, err := visitor(logger, events, match) + m, err := bv(events) matched = matched || m if err != nil { return blob, matched, changed, err diff --git a/interceptor/reflection_test.go b/interceptor/reflection_test.go index d9bce11a..66917be8 100644 --- a/interceptor/reflection_test.go +++ b/interceptor/reflection_test.go @@ -1,11 +1,43 @@ package interceptor import ( + "reflect" "testing" + "github.com/stretchr/testify/require" + "go.temporal.io/api/history/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + replicationspb "go.temporal.io/server/api/replication/v1" "go.temporal.io/server/common/log" ) +// TestNamespaceIDOwnersAreValid guards the allowlist against an upstream proto rename and, +// more importantly, against anyone replacing it with a NamespaceId field-name match. +func TestNamespaceIDOwnersAreValid(t *testing.T) { + expectedOwners := []reflect.Type{ + reflect.TypeFor[replicationspb.HistoryTaskAttributes](), + reflect.TypeFor[replicationspb.BackfillHistoryTaskAttributes](), + reflect.TypeFor[replicationspb.SyncVersionedTransitionTaskAttributes](), + reflect.TypeFor[persistencespb.WorkflowExecutionInfo](), + } + require.Len(t, namespaceIDOwners, len(expectedOwners)) + + for _, ownerType := range expectedOwners { + fieldIdx, ok := namespaceIDOwners[ownerType] + require.True(t, ok, "%v is missing from namespaceIDOwners", ownerType) + + field := ownerType.Field(fieldIdx) + require.Equal(t, namespaceIDFieldName, field.Name, "owner %v", ownerType) + require.Equal(t, reflect.String, field.Type.Kind(), "owner %v", ownerType) + } + + // StartChildWorkflowExecutionInitiatedEventAttributes carries the *child's* NamespaceId + // alongside its own SearchAttributes. Treating it as an owner would translate a parent's + // history event with the child's mapping. + require.NotContains(t, namespaceIDOwners, + reflect.TypeFor[history.StartChildWorkflowExecutionInitiatedEventAttributes]()) +} + func BenchmarkVisitNamespace(b *testing.B) { variants := []struct { testName string @@ -43,3 +75,43 @@ func BenchmarkVisitNamespace(b *testing.B) { }) } } + +func BenchmarkVisitSearchAttributes(b *testing.B) { + variants := []struct { + testName string + inputSAName string + mapping map[string]string + }{ + { + testName: "name changed", + inputSAName: "orig", + mapping: map[string]string{"orig": "orig.cloud"}, + }, + { + testName: "name unchanged", + inputSAName: "orig", + mapping: map[string]string{"other": "other.cloud"}, + }, + } + // Includes the deeply nested SyncVersionedTransitionTaskAttributes case, where the namespace + // owner sits several hops above the search attributes. + cases := generateSearchAttributeObjs() + + logger := log.NewTestLogger() + for _, c := range cases { + b.Run(c.objName, func(b *testing.B) { + for _, variant := range variants { + resolve := constMatcherResolver(createStringMatcher(variant.mapping)) + b.Run(variant.testName, func(b *testing.B) { + for i := 0; i < b.N; i++ { + b.StopTimer() + input := c.makeType(variant.inputSAName) + + b.StartTimer() + _, _ = visitSearchAttributes(logger, input, resolve, "") + } + }) + } + }) + } +} diff --git a/interceptor/search_attribute_translator.go b/interceptor/search_attribute_translator.go index cae88ae7..c246c30d 100644 --- a/interceptor/search_attribute_translator.go +++ b/interceptor/search_attribute_translator.go @@ -3,6 +3,7 @@ package interceptor import ( "strings" + "go.temporal.io/server/api/adminservice/v1" "go.temporal.io/server/common/api" "go.temporal.io/server/common/log" @@ -13,8 +14,8 @@ type ( saTranslator struct { logger log.Logger matchMethod func(string) bool - reqMap map[string]stringMatcher - respMap map[string]stringMatcher + resolveReq saMatcherResolver + resolveResp saMatcherResolver } ) @@ -26,8 +27,10 @@ func NewSearchAttributeTranslator(logger log.Logger, reqMap, respMap map[string] // 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), + // The resolvers are built once here and hold no per-message state, so a single + // translator serves every concurrent stream. + resolveReq: newSAMatcherResolver(reqMap), + resolveResp: newSAMatcherResolver(respMap), } } @@ -40,27 +43,43 @@ func (s *saTranslator) MatchMethod(m string) bool { } func (s *saTranslator) TranslateRequest(req any) (bool, error) { - return visitSearchAttributes(s.logger, req, s.getNamespaceReqMatcher("")) + return visitSearchAttributes(s.logger, req, s.resolveReq, "") } -func (s *saTranslator) TranslateResponse(resp any) (bool, error) { - return visitSearchAttributes(s.logger, 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 +// TranslateResponse translates the search attributes in resp. Some admin service responses +// carry history blobs but no namespace field of their own, so the paired request is used to +// seed the namespace. req is nil for streams, which have no request to pair with. +// +// This relies on NamespaceId surviving TranslateRequest: the namespace name translator rewrites +// Namespace, never NamespaceId. Adding namespace id translation later would silently break it. +func (s *saTranslator) TranslateResponse(req, resp any) (bool, error) { + var boundNamespaceID string + switch r := req.(type) { + case *adminservice.GetWorkflowExecutionRawHistoryV2Request: + boundNamespaceID = r.NamespaceId + case *adminservice.GetWorkflowExecutionRawHistoryRequest: + boundNamespaceID = r.NamespaceId } - return createStringMatcher(nil) + return visitSearchAttributes(s.logger, resp, s.resolveResp, boundNamespaceID) } -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 +// newSAMatcherResolver builds a resolver over per-namespace search attribute mappings. +func newSAMatcherResolver(nsMappings map[string]map[string]string) saMatcherResolver { + matchers := createStringMatchers(nsMappings) + + // Legacy configs express a single mapping keyed by an empty namespace id, meaning "apply to + // every namespace". Mirrors config.LegacyWildcardNamespaceID. + wildcard, hasWildcard := matchers[""] + + return func(nsID string) (stringMatcher, bool) { + if match, ok := matchers[nsID]; ok { + return match, true + } + if hasWildcard { + return wildcard, true + } + return nil, false } - return createStringMatcher(nil) } func createStringMatchers(nsMappings map[string]map[string]string) map[string]stringMatcher { diff --git a/interceptor/search_attribute_translator_test.go b/interceptor/search_attribute_translator_test.go index f33b7a83..0b6b2c2f 100644 --- a/interceptor/search_attribute_translator_test.go +++ b/interceptor/search_attribute_translator_test.go @@ -1,6 +1,7 @@ package interceptor import ( + "sort" "testing" "github.com/stretchr/testify/require" @@ -10,6 +11,7 @@ import ( "go.temporal.io/server/api/adminservice/v1" "go.temporal.io/server/api/persistence/v1" replicationspb "go.temporal.io/server/api/replication/v1" + "go.temporal.io/server/common/log" "go.temporal.io/server/common/persistence/serialization" ) @@ -45,7 +47,11 @@ type ( ) func TestTranslateSearchAttribute(t *testing.T) { - testTranslateObj(t, visitSearchAttributes, generateSearchAttributeObjs(), require.EqualExportedValues) + // These cases exercise a single namespace, so one matcher applies everywhere. + adapter := func(l log.Logger, obj any, m stringMatcher) (bool, error) { + return visitSearchAttributes(l, obj, constMatcherResolver(m), "") + } + testTranslateObj(t, adapter, generateSearchAttributeObjs(), require.EqualExportedValues) } func generateSearchAttributeObjs() []objCase { @@ -233,3 +239,362 @@ func makeTestIndexedFieldMap(name string) map[string]*common.Payload { }, } } + +const ( + testNsA = "ns-a" + testNsB = "ns-b" + testNsC = "ns-c" + + testSAName = "TestSA" + keywordA = "Keyword01" + keywordB = "Keyword02" +) + +// testSAMappings maps the same search attribute to a different indexed field per namespace, +// so a translation applied with the wrong namespace's mapping is visible in assertions. +func testSAMappings() map[string]map[string]string { + return map[string]map[string]string{ + testNsA: {testSAName: keywordA}, + testNsB: {testSAName: keywordB}, + } +} + +func newTestSATranslator(t *testing.T, nsMappings map[string]map[string]string) Translator { + t.Helper() + return NewSearchAttributeTranslator(log.NewTestLogger(), nsMappings, nsMappings) +} + +// makeMultiNamespaceFrame builds one replication frame carrying four tasks spanning three +// namespaces, which is what the wire actually looks like: resolution has to happen per subtree. +// +// Every blob and IndexedFields map is built fresh. visit.Values keeps a set of pointers it has +// already seen and skips repeats, so a hoisted, shared blob would leave the second subtree +// untranslated and could still read as a pass. +func makeMultiNamespaceFrame(saName string) *adminservice.StreamWorkflowReplicationMessagesResponse { + return &adminservice.StreamWorkflowReplicationMessagesResponse{ + Attributes: &adminservice.StreamWorkflowReplicationMessagesResponse_Messages{ + Messages: &replicationspb.WorkflowReplicationMessages{ + ReplicationTasks: []*replicationspb.ReplicationTask{ + { + Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{ + HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{ + NamespaceId: testNsA, + WorkflowId: "wf-a", + Events: makeHistoryEventsBlobWithSearchAttribute(saName), + NewRunEvents: makeHistoryEventsBlobWithSearchAttribute(saName), + }, + }, + }, + { + Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{ + HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{ + NamespaceId: testNsB, + WorkflowId: "wf-b", + Events: makeHistoryEventsBlobWithSearchAttribute(saName), + }, + }, + }, + { + // WorkflowExecutionInfo owns its namespace directly: no blob involved. + Attributes: &replicationspb.ReplicationTask_SyncWorkflowStateTaskAttributes{ + SyncWorkflowStateTaskAttributes: &replicationspb.SyncWorkflowStateTaskAttributes{ + WorkflowState: &persistence.WorkflowMutableState{ + ExecutionInfo: &persistence.WorkflowExecutionInfo{ + NamespaceId: testNsA, + WorkflowId: "wf-a-state", + SearchAttributes: makeTestIndexedFieldMap(saName), + // Memo is the same type as SearchAttributes. It must not be rewritten. + Memo: makeTestIndexedFieldMap(saName), + }, + }, + }, + }, + }, + { + Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{ + HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{ + NamespaceId: testNsC, + WorkflowId: "wf-c", + Events: makeHistoryEventsBlobWithSearchAttribute(saName), + }, + }, + }, + }, + }, + }, + } +} + +// makeHistoryTaskFrame wraps the given events in a blob owned by nsID. +func makeHistoryTaskFrame(nsID string, events ...*history.HistoryEvent) *adminservice.StreamWorkflowReplicationMessagesResponse { + blob, err := serialization.NewSerializer().SerializeEvents(events) + if err != nil { + panic(err) + } + return &adminservice.StreamWorkflowReplicationMessagesResponse{ + Attributes: &adminservice.StreamWorkflowReplicationMessagesResponse_Messages{ + Messages: &replicationspb.WorkflowReplicationMessages{ + ReplicationTasks: []*replicationspb.ReplicationTask{ + { + Attributes: &replicationspb.ReplicationTask_HistoryTaskAttributes{ + HistoryTaskAttributes: &replicationspb.HistoryTaskAttributes{ + NamespaceId: nsID, + Events: blob, + }, + }, + }, + }, + }, + }, + } +} + +func firstTaskEvents(resp *adminservice.StreamWorkflowReplicationMessagesResponse) *common.DataBlob { + return resp.GetMessages().GetReplicationTasks()[0].GetHistoryTaskAttributes().GetEvents() +} + +// blobSAKeys returns every search attribute key in the blob, sorted. Duplicates are kept so +// that a partial translation (one event rewritten, another missed) fails the assertion. +func blobSAKeys(t *testing.T, blob *common.DataBlob) []string { + t.Helper() + events, err := serialization.NewSerializer().DeserializeEvents(blob) + require.NoError(t, err) + + var keys []string + for _, evt := range events { + keys = append(keys, mapKeys(evt.GetWorkflowExecutionStartedEventAttributes().GetSearchAttributes().GetIndexedFields())...) + keys = append(keys, mapKeys(evt.GetStartChildWorkflowExecutionInitiatedEventAttributes().GetSearchAttributes().GetIndexedFields())...) + } + sort.Strings(keys) + return keys +} + +func mapKeys(fields map[string]*common.Payload) []string { + keys := make([]string, 0, len(fields)) + for key := range fields { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + +func TestTranslateSearchAttributePerNamespace(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) + + // A single pass can pass by luck: visit.ValuesUnsafe pops the front of its worklist but + // swaps in the last element, so the order in which the four tasks are reached is + // unspecified. Rebuild the frame every iteration so no subtree is ever seen pre-translated. + for i := 0; i < 25; i++ { + frame := makeMultiNamespaceFrame(testSAName) + + changed, err := tr.TranslateResponse(nil, frame) + require.NoError(t, err) + require.True(t, changed) + + tasks := frame.GetMessages().GetReplicationTasks() + require.Len(t, tasks, 4) + + nsATask := tasks[0].GetHistoryTaskAttributes() + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetEvents())) + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetNewRunEvents())) + + require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, tasks[1].GetHistoryTaskAttributes().GetEvents()), + "ns-b must get its own mapping, not ns-a's") + + execInfo := tasks[2].GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo() + require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes())) + require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetMemo()), "Memo must not be rewritten") + + require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, tasks[3].GetHistoryTaskAttributes().GetEvents()), + "ns-c has no mapping and must be left untouched") + } +} + +func TestTranslateSearchAttributeInBlobUsesEnclosingNamespace(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) + + // The event names a different namespace as its parent workflow's. The enclosing + // HistoryTaskAttributes is what owns the namespace, so ns-a's mapping must win. + frame := makeHistoryTaskFrame(testNsA, &history.HistoryEvent{ + EventId: 1, + EventType: enums.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, + Attributes: &history.HistoryEvent_WorkflowExecutionStartedEventAttributes{ + WorkflowExecutionStartedEventAttributes: &history.WorkflowExecutionStartedEventAttributes{ + ParentWorkflowNamespaceId: testNsB, + SearchAttributes: &common.SearchAttributes{ + IndexedFields: makeTestIndexedFieldMap(testSAName), + }, + }, + }, + }) + + changed, err := tr.TranslateResponse(nil, frame) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, []string{keywordA}, blobSAKeys(t, firstTaskEvents(frame))) +} + +func TestTranslateSearchAttributeIgnoresChildNamespaceId(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) + + // StartChildWorkflowExecutionInitiatedEventAttributes has both a NamespaceId (the child's) + // and SearchAttributes (the parent's). Resolving by nearest struct with a NamespaceId field + // would translate this event with ns-b's mapping. This test fails if the owner allowlist is + // ever replaced with a field-name match. + frame := makeHistoryTaskFrame(testNsA, &history.HistoryEvent{ + EventId: 1, + EventType: enums.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED, + Attributes: &history.HistoryEvent_StartChildWorkflowExecutionInitiatedEventAttributes{ + StartChildWorkflowExecutionInitiatedEventAttributes: &history.StartChildWorkflowExecutionInitiatedEventAttributes{ + Namespace: "child-ns", + NamespaceId: testNsB, + SearchAttributes: &common.SearchAttributes{ + IndexedFields: makeTestIndexedFieldMap(testSAName), + }, + }, + }, + }) + + changed, err := tr.TranslateResponse(nil, frame) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, []string{keywordA}, blobSAKeys(t, firstTaskEvents(frame))) +} + +func TestTranslateSearchAttributeIgnoresParentNamespaceId(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) + + // WorkflowExecutionInfo carries both NamespaceId and ParentNamespaceId. Only the former + // owns the search attributes. + execInfo := &persistence.WorkflowExecutionInfo{ + NamespaceId: testNsA, + ParentNamespaceId: testNsB, + SearchAttributes: makeTestIndexedFieldMap(testSAName), + } + + changed, err := tr.TranslateResponse(nil, execInfo) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes())) +} + +func TestTranslateSearchAttributeRawHistoryUsesPairedRequest(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) + + // These responses carry history blobs but no namespace field of their own. Without the + // paired request there is nothing to resolve, and the response must be left alone rather + // than erroring or being translated with an arbitrary namespace's mapping. + newV2Resp := func() *adminservice.GetWorkflowExecutionRawHistoryV2Response { + return &adminservice.GetWorkflowExecutionRawHistoryV2Response{ + HistoryBatches: []*common.DataBlob{makeHistoryEventsBlobWithSearchAttribute(testSAName)}, + } + } + newResp := func() *adminservice.GetWorkflowExecutionRawHistoryResponse { + return &adminservice.GetWorkflowExecutionRawHistoryResponse{ + HistoryBatches: []*common.DataBlob{makeHistoryEventsBlobWithSearchAttribute(testSAName)}, + } + } + + t.Run("V2 unpaired", func(t *testing.T) { + resp := newV2Resp() + changed, err := tr.TranslateResponse(nil, resp) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, resp.HistoryBatches[0])) + }) + + t.Run("V2 paired", func(t *testing.T) { + resp := newV2Resp() + req := &adminservice.GetWorkflowExecutionRawHistoryV2Request{NamespaceId: testNsB} + changed, err := tr.TranslateResponse(req, resp) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, resp.HistoryBatches[0])) + }) + + t.Run("unpaired", func(t *testing.T) { + resp := newResp() + changed, err := tr.TranslateResponse(nil, resp) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, resp.HistoryBatches[0])) + }) + + t.Run("paired", func(t *testing.T) { + resp := newResp() + req := &adminservice.GetWorkflowExecutionRawHistoryRequest{NamespaceId: testNsB} + changed, err := tr.TranslateResponse(req, resp) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, []string{keywordB, keywordB}, blobSAKeys(t, resp.HistoryBatches[0])) + }) +} + +func TestTranslateSearchAttributeLegacyWildcard(t *testing.T) { + // A sole mapping keyed by an empty namespace id means "every namespace". Deployed configs + // rely on this, so it must keep behaving exactly as it did before per-namespace support. + tr := newTestSATranslator(t, map[string]map[string]string{"": {testSAName: keywordA}}) + + frame := makeMultiNamespaceFrame(testSAName) + changed, err := tr.TranslateResponse(nil, frame) + require.NoError(t, err) + require.True(t, changed) + + tasks := frame.GetMessages().GetReplicationTasks() + require.Len(t, tasks, 4) + + nsATask := tasks[0].GetHistoryTaskAttributes() + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetEvents())) + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetNewRunEvents())) + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, tasks[1].GetHistoryTaskAttributes().GetEvents())) + require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, tasks[3].GetHistoryTaskAttributes().GetEvents()), + "the wildcard mapping applies even to namespaces with no entry of their own") + + execInfo := tasks[2].GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo() + require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes())) + require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetMemo()), "Memo must not be rewritten") +} + +func TestTranslateSearchAttributeUnsupportedFieldTypes(t *testing.T) { + // Add- and RemoveSearchAttributesRequest name their fields SearchAttributes too, but hold a + // map[string]enums.IndexedValueType and a []string. They must be skipped, not treated as an + // error that aborts translation of the message. + // + // The two configs reach that outcome down different paths, so both are needed: + // - namespace keyed: neither request type is enclosed by a namespace owner, so the + // namespace resolves to "", no matcher resolves, and the field is skipped before the + // type switch runs. The unsupported-type branch is unreachable in this configuration. + // - legacy wildcard: the wildcard matcher resolves for every namespace, so the type + // switch does run and its unsupported-type branch is what prevents the error. This is + // the configuration deployed proxies use, which is what makes that branch load-bearing. + configs := map[string]map[string]map[string]string{ + "namespace keyed": testSAMappings(), + "legacy wildcard": {"": {testSAName: keywordA}}, + } + + for name, nsMappings := range configs { + t.Run(name, func(t *testing.T) { + tr := newTestSATranslator(t, nsMappings) + + addReq := &adminservice.AddSearchAttributesRequest{ + SearchAttributes: map[string]enums.IndexedValueType{ + testSAName: enums.INDEXED_VALUE_TYPE_KEYWORD, + }, + } + changed, err := tr.TranslateRequest(addReq) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, map[string]enums.IndexedValueType{ + testSAName: enums.INDEXED_VALUE_TYPE_KEYWORD, + }, addReq.SearchAttributes) + + removeReq := &adminservice.RemoveSearchAttributesRequest{ + SearchAttributes: []string{testSAName}, + } + changed, err = tr.TranslateRequest(removeReq) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, []string{testSAName}, removeReq.SearchAttributes) + }) + } +} diff --git a/interceptor/translation_interceptor.go b/interceptor/translation_interceptor.go index f06b8c4f..f59b85bc 100644 --- a/interceptor/translation_interceptor.go +++ b/interceptor/translation_interceptor.go @@ -61,7 +61,7 @@ func (i *TranslationInterceptor) Intercept( for _, tr := range i.translators { if tr.MatchMethod(info.FullMethod) { start := time.Now() - changed, trErr := tr.TranslateResponse(resp) + changed, trErr := tr.TranslateResponse(req, resp) logTranslateResult(tr, i.logger, changed, trErr, methodName+"Response", resp, time.Since(start)) } } @@ -106,7 +106,8 @@ func (w *streamTranslator) RecvMsg(m any) error { func (w *streamTranslator) SendMsg(m any) error { for _, tr := range w.translators { start := time.Now() - changed, trErr := tr.TranslateResponse(m) + // Streams have no request to pair with this message. + changed, trErr := tr.TranslateResponse(nil, m) logTranslateResult(tr, w.logger, changed, trErr, "SendMsg", m, time.Since(start)) } return w.ServerStream.SendMsg(m) diff --git a/interceptor/translation_interceptor_test.go b/interceptor/translation_interceptor_test.go index e67b4787..fae7e679 100644 --- a/interceptor/translation_interceptor_test.go +++ b/interceptor/translation_interceptor_test.go @@ -20,13 +20,16 @@ type spyTranslator struct { matchCalls int translateReqCalls int translateRespCalls int + // pairedReqs records the request passed alongside each response, in call order. + pairedReqs []any } func (s *spyTranslator) Kind() string { return "spy" } func (s *spyTranslator) MatchMethod(string) bool { s.matchCalls++; return true } func (s *spyTranslator) TranslateRequest(any) (bool, error) { s.translateReqCalls++; return false, nil } -func (s *spyTranslator) TranslateResponse(any) (bool, error) { +func (s *spyTranslator) TranslateResponse(req, _ any) (bool, error) { s.translateRespCalls++ + s.pairedReqs = append(s.pairedReqs, req) return false, nil } @@ -73,12 +76,41 @@ func TestTranslationInterceptor(t *testing.T) { if tc.incomingHeaders != nil { ctx = metadata.NewIncomingContext(ctx, metadata.New(tc.incomingHeaders)) } - _, err := ti.Intercept(ctx, &workflowservice.DescribeWorkflowExecutionRequest{}, info, handler) + req := &workflowservice.DescribeWorkflowExecutionRequest{} + _, err := ti.Intercept(ctx, req, info, handler) require.NoError(t, err) require.Equal(t, tc.expectedMatchCalls, spy.matchCalls, "MatchMethod call count") require.Equal(t, tc.expectedReqCalls, spy.translateReqCalls, "TranslateRequest call count") require.Equal(t, tc.expectedRespCalls, spy.translateRespCalls, "TranslateResponse call count") + if tc.expectedRespCalls > 0 { + require.Equal(t, []any{req}, spy.pairedReqs, + "unary responses must be translated alongside their request") + } }) } } + +// fakeServerStream captures the messages a streamTranslator forwards. +type fakeServerStream struct { + grpc.ServerStream + sent []any +} + +func (f *fakeServerStream) SendMsg(m any) error { + f.sent = append(f.sent, m) + return nil +} + +func TestStreamTranslatorSendMsgHasNoPairedRequest(t *testing.T) { + spy := &spyTranslator{} + inner := &fakeServerStream{} + st := newStreamTranslator(inner, log.NewTestLogger(), []Translator{spy}) + + msg := &workflowservice.DescribeWorkflowExecutionResponse{} + require.NoError(t, st.SendMsg(msg)) + + require.Equal(t, 1, spy.translateRespCalls) + require.Equal(t, []any{nil}, spy.pairedReqs, "stream messages have no request to pair with") + require.Equal(t, []any{msg}, inner.sent) +} diff --git a/interceptor/translator.go b/interceptor/translator.go index eb6f8704..5d432ab2 100644 --- a/interceptor/translator.go +++ b/interceptor/translator.go @@ -10,7 +10,10 @@ type ( Translator interface { MatchMethod(string) bool TranslateRequest(any) (bool, error) - TranslateResponse(any) (bool, error) + // TranslateResponse receives the request paired with resp, so that a translator can + // take context from it that the response itself does not carry. req is nil for + // streams, whose messages have no request to pair with. + TranslateResponse(req, resp any) (bool, error) Kind() string } @@ -47,7 +50,7 @@ func (n *translatorImpl) TranslateRequest(req any) (bool, error) { return n.visitor(n.logger, req, n.matchReq) } -func (n *translatorImpl) TranslateResponse(resp any) (bool, error) { +func (n *translatorImpl) TranslateResponse(_, resp any) (bool, error) { return n.visitor(n.logger, resp, n.matchResp) } diff --git a/metrics/prometheus_defs.go b/metrics/prometheus_defs.go index fbb26e04..97abe1d0 100644 --- a/metrics/prometheus_defs.go +++ b/metrics/prometheus_defs.go @@ -75,10 +75,21 @@ var ( TranslationErrors = DefaultCounterVec("translation_error", "Count of message translation errors", translationLabels...) TranslationLatency = DefaultHistogramVec("translation_latency", "Latency of message translations", translationLabels...) + // SearchAttrTranslationSkipped counts search attribute sites the translator deliberately + // left alone. It is not emitted for the expected steady state where a namespace resolves + // but has no configured mapping, which is true of every namespace not being migrated. + // The namespace id is logged rather than labelled, since its cardinality is unbounded. + SearchAttrTranslationSkipped = DefaultCounterVec("search_attribute_translation_skipped", + "Count of search attribute sites left untranslated", "reason", "message_type") + UTF8RepairTranslationKind = "utf8repair" NamespaceTranslationKind = "namespace" SearchAttrTranslationKind = "search-attribute" HistoryBlobMessageType = "HistoryEventBlob" + + // Reasons for SearchAttrTranslationSkipped. + SkipReasonUnresolvedNamespace = "unresolved_namespace" + SkipReasonUnsupportedType = "unsupported_type" ) // GetGRPCClientMetrics helps the GRPC client metrics objects feel more like the server one @@ -139,4 +150,5 @@ func init() { prometheus.MustRegister(TranslationCount) prometheus.MustRegister(TranslationErrors) prometheus.MustRegister(TranslationLatency) + prometheus.MustRegister(SearchAttrTranslationSkipped) } diff --git a/proxy/cluster_connection.go b/proxy/cluster_connection.go index f93a2da6..6a578496 100644 --- a/proxy/cluster_connection.go +++ b/proxy/cluster_connection.go @@ -388,8 +388,11 @@ func makeServerOptions(c serverConfiguration, tlsConfig encryption.TLSConfig) ([ if c.saTranslations.LenNamespaces() > 0 { c.loggers.Get(LogClusterConnection).Info("search attribute translation enabled", tag.NewAnyTag("mappings", c.saTranslations)) - if c.saTranslations.LenNamespaces() > 1 { - panic("multiple namespace search attribute mappings are not supported") + if c.saTranslations.HasLegacyWildcard() { + // Validated as the sole mapping (see config.SATranslationConfig.Validate), so it is applied + // to every namespace. Set namespaceId to translate more than one namespace concurrently. + c.loggers.Get(LogClusterConnection).Warn("DEPRECATED: searchAttributeTranslation has a " + + "namespaceMappings entry with an empty namespaceId; its mappings are applied to every namespace") } translators = append(translators, interceptor.NewSearchAttributeTranslator(c.loggers.Get(LogInterceptor), c.saTranslations.FlattenMaps(), c.saTranslations.Inverse().FlattenMaps())) diff --git a/proxy/cluster_connection_test.go b/proxy/cluster_connection_test.go index fcc52484..d286826a 100644 --- a/proxy/cluster_connection_test.go +++ b/proxy/cluster_connection_test.go @@ -15,7 +15,9 @@ import ( "go.temporal.io/server/common/log/tag" "google.golang.org/grpc" + "github.com/temporalio/s2s-proxy/collect" "github.com/temporalio/s2s-proxy/config" + "github.com/temporalio/s2s-proxy/encryption" "github.com/temporalio/s2s-proxy/endtoendtest/testservices" "github.com/temporalio/s2s-proxy/logging" "github.com/temporalio/s2s-proxy/metrics" @@ -361,3 +363,65 @@ func TestTCPListenerIsReleasedWhenNeverStarted(t *testing.T) { } runtime.KeepAlive(cc) } + +// TestMakeServerOptionsMultiNamespaceSATranslation guards the removal of the former +// panic("multiple namespace search attribute mappings are not supported"), which fired on the +// startup path and so crash-looped the whole proxy -- including namespaces that used no custom +// search attributes. +func TestMakeServerOptionsMultiNamespaceSATranslation(t *testing.T) { + saMappings := func(pairs ...config.SANamespaceMapping) config.SearchAttributeTranslation { + t.Helper() + translation, err := (&config.SATranslationConfig{NamespaceMappings: pairs}).AsLocalToRemoteSATranslation() + require.NoError(t, err) + return translation + } + + for _, tc := range []struct { + name string + saTranslator config.SearchAttributeTranslation + }{ + { + name: "two namespaces", + saTranslator: saMappings( + config.SANamespaceMapping{ + Name: "ns-a", + NamespaceId: "11111111-1111-1111-1111-111111111111", + Mappings: []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}}, + }, + config.SANamespaceMapping{ + Name: "ns-b", + NamespaceId: "22222222-2222-2222-2222-222222222222", + Mappings: []config.SAMapping{{LocalName: "Keyword02", RemoteName: "TestSA"}}, + }, + ), + }, + { + // The shape shipped by configs that predate per-namespace translation. + name: "legacy wildcard, empty namespaceId", + saTranslator: saMappings(config.SANamespaceMapping{ + Name: "ns-a", + Mappings: []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}}, + }), + }, + } { + t.Run(tc.name, func(t *testing.T) { + // nsTranslations is an interface, so it must be non-nil: makeServerOptions calls + // Len() on it unconditionally. + emptyNsTranslations, err := collect.NewStaticBiMap(func(func(string, string) bool) {}, 0) + require.NoError(t, err) + + cfg := serverConfiguration{ + directionLabel: "outbound", + nsTranslations: emptyNsTranslations, + saTranslations: tc.saTranslator, + loggers: logging.NewLoggerProvider(log.NewTestLogger(), config.NewMockConfigProvider(config.S2SProxyConfig{})), + } + + require.NotPanics(t, func() { + opts, err := makeServerOptions(cfg, encryption.TLSConfig{}) + require.NoError(t, err) + require.NotEmpty(t, opts) + }) + }) + } +} From 50348e610f26f2575d569ae0b137926c9885968e Mon Sep 17 00:00:00 2001 From: JH Date: Mon, 24 Aug 2026 13:51:53 -0700 Subject: [PATCH 2/3] Trim self-evident comments and align namespace fallback naming Drop comments that restate the code, keeping those that record facts the code cannot show: why namespace owners are an allowlist rather than a field name match, why resolution walks only upward, why a data blob is descended into even when the namespace is unresolved, and why one skip reason is counted and the other is not. Rename visitSearchAttributes' boundNamespaceID to fallbackNamespaceID so it matches resolveNamespaceID's fallback parameter, which is the same value. Move constMatcherResolver to reflection_test.go; it has no production caller and exists so the pre-existing table cases can apply one matcher to every namespace. Co-Authored-By: Claude Opus 5 --- interceptor/reflection.go | 35 +++++++--------------- interceptor/reflection_test.go | 6 ++++ interceptor/search_attribute_translator.go | 9 +++--- metrics/prometheus_defs.go | 1 - 4 files changed, 21 insertions(+), 30 deletions(-) diff --git a/interceptor/reflection.go b/interceptor/reflection.go index d2a35ce4..35d619a9 100644 --- a/interceptor/reflection.go +++ b/interceptor/reflection.go @@ -156,20 +156,12 @@ type stringMatcher func(name string) (string, bool) type visitor func(logger log.Logger, obj any, match stringMatcher) (bool, error) // saMatcherResolver returns the search attribute matcher configured for a namespace id. -// The second return value is false when that namespace has no mapping, in which case the -// search attributes at hand must be left untouched. +// false means the namespace has no mapping and its search attributes must be left untouched. type saMatcherResolver func(namespaceID string) (stringMatcher, bool) -// constMatcherResolver adapts a single matcher to saMatcherResolver, for callers that -// intentionally apply one mapping to every namespace. -func constMatcherResolver(match stringMatcher) saMatcherResolver { - return func(string) (stringMatcher, bool) { return match, true } -} - -// blobVisitor visits the history events deserialized from a data blob. -// It returns whether anything was matched and any error it encountered. -// Callers close over whatever matching state they need, since a data blob -// starts a fresh traversal that cannot see the enclosing message. +// blobVisitor visits the history events deserialized from a data blob. Callers close over +// whatever matching state they need, since a data blob starts a fresh traversal that cannot +// see the enclosing message. type blobVisitor func(events []*history.HistoryEvent) (bool, error) // visitNamespace uses reflection to recursively visit all fields @@ -244,13 +236,12 @@ func visitNamespace(logger log.Logger, obj any, match stringMatcher) (bool, erro return matched, err } -// visitSearchAttributes uses reflection to recursively visit all fields -// in the given object. When it finds search attribute fields, it resolves the namespace -// that owns them and applies the matcher configured for that namespace, if any. +// visitSearchAttributes translates the search attributes in obj using the mapping configured +// for whichever namespace owns them. // -// boundNamespaceID is the namespace to fall back on when the parent chain reaches no -// namespace owner. See resolveNamespaceID. -func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver, boundNamespaceID string) (bool, error) { +// fallbackNamespaceID is used when the parent chain reaches no namespace owner. See +// resolveNamespaceID. +func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver, fallbackNamespaceID string) (bool, error) { var matched bool // The visitor function can return Skip, Stop, or Continue to control recursion. @@ -269,7 +260,7 @@ func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver // events inside the blob are visited in a fresh traversal that cannot see the // enclosing namespace owner. Descend even when the namespace is unresolved, since // visitDataBlobs also repairs invalid UTF-8 independently of any translation. - nsID := resolveNamespaceID(vwp, boundNamespaceID) + nsID := resolveNamespaceID(vwp, fallbackNamespaceID) changed, err := visitDataBlobs(logger, vwp, func(events []*history.HistoryEvent) (bool, error) { return visitSearchAttributes(logger, events, resolve, nsID) }) @@ -278,12 +269,10 @@ func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver return visit.Stop, err } } else if searchAttributeFieldNames[fieldType.Name] { - nsID := resolveNamespaceID(vwp, boundNamespaceID) + nsID := resolveNamespaceID(vwp, fallbackNamespaceID) match, ok := resolve(nsID) if !ok { logSkippedSearchAttributes(logger, obj, nsID) - - // Leave these search attributes untouched. return visit.Continue, nil } @@ -320,8 +309,6 @@ func visitSearchAttributes(logger log.Logger, obj any, resolve saMatcherResolver return matched, err } -// logSkippedSearchAttributes reports search attributes left untranslated because no matcher -// resolved for their namespace. func logSkippedSearchAttributes(logger log.Logger, obj any, nsID string) { msgType := metrics.SanitizedTypeName(obj) if nsID == "" { diff --git a/interceptor/reflection_test.go b/interceptor/reflection_test.go index 66917be8..ed5f0e3a 100644 --- a/interceptor/reflection_test.go +++ b/interceptor/reflection_test.go @@ -115,3 +115,9 @@ func BenchmarkVisitSearchAttributes(b *testing.B) { }) } } + +// constMatcherResolver applies one matcher to every namespace, for tests that are not +// exercising per-namespace resolution. +func constMatcherResolver(match stringMatcher) saMatcherResolver { + return func(string) (stringMatcher, bool) { return match, true } +} diff --git a/interceptor/search_attribute_translator.go b/interceptor/search_attribute_translator.go index c246c30d..a1ac0440 100644 --- a/interceptor/search_attribute_translator.go +++ b/interceptor/search_attribute_translator.go @@ -53,17 +53,16 @@ func (s *saTranslator) TranslateRequest(req any) (bool, error) { // This relies on NamespaceId surviving TranslateRequest: the namespace name translator rewrites // Namespace, never NamespaceId. Adding namespace id translation later would silently break it. func (s *saTranslator) TranslateResponse(req, resp any) (bool, error) { - var boundNamespaceID string + var fallbackNamespaceID string switch r := req.(type) { case *adminservice.GetWorkflowExecutionRawHistoryV2Request: - boundNamespaceID = r.NamespaceId + fallbackNamespaceID = r.NamespaceId case *adminservice.GetWorkflowExecutionRawHistoryRequest: - boundNamespaceID = r.NamespaceId + fallbackNamespaceID = r.NamespaceId } - return visitSearchAttributes(s.logger, resp, s.resolveResp, boundNamespaceID) + return visitSearchAttributes(s.logger, resp, s.resolveResp, fallbackNamespaceID) } -// newSAMatcherResolver builds a resolver over per-namespace search attribute mappings. func newSAMatcherResolver(nsMappings map[string]map[string]string) saMatcherResolver { matchers := createStringMatchers(nsMappings) diff --git a/metrics/prometheus_defs.go b/metrics/prometheus_defs.go index 97abe1d0..57779a33 100644 --- a/metrics/prometheus_defs.go +++ b/metrics/prometheus_defs.go @@ -87,7 +87,6 @@ var ( SearchAttrTranslationKind = "search-attribute" HistoryBlobMessageType = "HistoryEventBlob" - // Reasons for SearchAttrTranslationSkipped. SkipReasonUnresolvedNamespace = "unresolved_namespace" SkipReasonUnsupportedType = "unsupported_type" ) From 57a95542724faa6d32dc2af9c385d9d672583a67 Mon Sep 17 00:00:00 2001 From: JH Date: Mon, 24 Aug 2026 14:35:26 -0700 Subject: [PATCH 3/3] Require namespaceId; drop the empty-id wildcard An earlier revision let a single mapping omit the namespaceId and applied it to every namespace, matching what configs deployed before per-namespace translation happen to do today. That keeps a config shape alive that nobody should write, and a blank id that silently applies to everything is the same ambiguity this change exists to remove. Validate now rejects any mapping without a namespaceId, naming the entry, so a config that cannot be applied per namespace fails at startup rather than translating an arbitrary namespace. Removes LegacyWildcardNamespaceID, HasLegacyWildcard and the resolver fallback. Consequence: the migration tooling still emits an empty namespaceId, so it has to emit real ids before this ships, and cells configured with a blank id need their configmap corrected in the same change. With no wildcard, visitSearchAttributes' unsupported-type branch is no longer reachable from Add/RemoveSearchAttributesRequest: neither is enclosed by a namespace owner, so the namespace resolves to empty, nothing matches, and the field is skipped before the type switch. Kept as a defensive path and its test collapsed to the one case that still exercises anything. Co-Authored-By: Claude Opus 5 --- config/config.go | 23 ++----- config/config_test.go | 32 ++++------ interceptor/search_attribute_translator.go | 13 +--- .../search_attribute_translator_test.go | 60 +++++++++---------- proxy/cluster_connection.go | 6 -- proxy/cluster_connection_test.go | 8 +-- 6 files changed, 50 insertions(+), 92 deletions(-) diff --git a/config/config.go b/config/config.go index 7ea16fdc..16902e85 100644 --- a/config/config.go +++ b/config/config.go @@ -34,11 +34,6 @@ const ( HTTP HealthCheckProtocol = "http" ) -// LegacyWildcardNamespaceID is the empty namespaceId used by older configs that predate -// per-namespace search attribute translation. When it is the sole mapping it is applied to -// every namespace, preserving the behaviour those configs shipped with. -const LegacyWildcardNamespaceID = "" - type ( ConfigProvider interface { GetS2SProxyConfig() S2SProxyConfig @@ -250,16 +245,15 @@ func (s *SATranslationConfig) IsEnabled() bool { // Validate reports the first reason the configured namespace mappings cannot be applied per // namespace: search attributes are translated by namespaceId, so every mapping needs an id that -// is present and unique. A single mapping is allowed to omit the namespaceId, in which case it is -// applied to every namespace, see LegacyWildcardNamespaceID. +// is present and unique. func (s *SATranslationConfig) Validate() error { namesByNamespaceId := make(map[string]string, len(s.NamespaceMappings)) seenNames := make(map[string]struct{}, len(s.NamespaceMappings)) for _, m := range s.NamespaceMappings { - // The missing id is reported before the duplicate id so that a config with several - // mappings and no ids gets the actionable message rather than `duplicate namespaceId ""`. - if m.NamespaceId == LegacyWildcardNamespaceID && len(s.NamespaceMappings) > 1 { - return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] has an empty namespaceId; namespaceId is required when more than one namespace is configured", m.Name) + // Reported before the duplicate check so that a config with several mappings and no ids + // gets the actionable message rather than `duplicate namespaceId ""`. + if m.NamespaceId == "" { + return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] has no namespaceId; search attributes are translated per namespace so namespaceId is required", m.Name) } if existing, found := namesByNamespaceId[m.NamespaceId]; found { return fmt.Errorf("searchAttributeTranslation: namespaceMappings[name=%q] and namespaceMappings[name=%q] have duplicate namespaceId %q", existing, m.Name, m.NamespaceId) @@ -360,13 +354,6 @@ func (s SearchAttributeTranslation) FlattenMaps() map[string]map[string]string { return raw } -// HasLegacyWildcard reports whether the translation was built from a config that omitted the -// namespaceId, in which case its mappings apply to every namespace. See LegacyWildcardNamespaceID. -func (s SearchAttributeTranslation) HasLegacyWildcard() bool { - _, found := s.inner[LegacyWildcardNamespaceID] - return found -} - // AsLocalToRemoteSATranslation converts the flat list of namespace + local/remote pairs into a map of BiMaps, with local->remote // as the direction returned. The remote->local mapping can be accessed with saTranslator[namespaceId].Inverse() func (s *SATranslationConfig) AsLocalToRemoteSATranslation() (SearchAttributeTranslation, error) { diff --git a/config/config_test.go b/config/config_test.go index 92e5d528..af22b06e 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -343,7 +343,6 @@ func TestSATranslationConfigValidate(t *testing.T) { cfg: SATranslationConfig{}, verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { require.Equal(t, 0, saTranslation.LenNamespaces()) - require.False(t, saTranslation.HasLegacyWildcard()) }, }, { @@ -387,7 +386,7 @@ func TestSATranslationConfigValidate(t *testing.T) { }, wantValidateErr: []string{ `namespaceMappings[name="legacyNamespace"]`, - "namespaceId is required when more than one namespace is configured", + "has no namespaceId", }, }, { @@ -408,13 +407,13 @@ func TestSATranslationConfigValidate(t *testing.T) { }, wantValidateErr: []string{ `namespaceMappings[name="namespace1"]`, - "namespaceId is required", + "has no namespaceId", }, wantValidateErrExcludes: []string{"duplicate namespaceId"}, }, { - // Configs written before per-namespace translation omit the namespaceId entirely. - // A single such mapping keeps working and is applied to every namespace. + // namespaceId is required even for one namespace: there is no mapping that applies + // to every namespace, so an omitted id has nothing to match against. name: "single mapping with empty namespaceId", cfg: SATranslationConfig{ NamespaceMappings: []SANamespaceMapping{ @@ -423,15 +422,12 @@ func TestSATranslationConfigValidate(t *testing.T) { }, }, }, - verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { - require.Equal(t, 1, saTranslation.LenNamespaces()) - require.True(t, saTranslation.HasLegacyWildcard()) - require.Equal(t, "remoteOne", saTranslation.Get(LegacyWildcardNamespaceID, "localOne")) - require.Equal(t, "localOne", saTranslation.Inverse().Get(LegacyWildcardNamespaceID, "remoteOne")) - }, + wantValidateErr: []string{"has no namespaceId", "namespaceId is required"}, }, { - // The shape deployed today: the namespace is named but the namespaceId is blank. + // The shape the migration tooling emits today: the namespace is named but the + // namespaceId is blank. It must fail at startup naming the entry, rather than + // translating some arbitrary namespace. name: "named mapping with empty namespaceId", cfg: SATranslationConfig{ NamespaceMappings: []SANamespaceMapping{ @@ -444,12 +440,9 @@ func TestSATranslationConfigValidate(t *testing.T) { }, }, }, - verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { - require.Equal(t, 1, saTranslation.LenNamespaces()) - require.True(t, saTranslation.HasLegacyWildcard()) - require.Equal(t, 2, saTranslation.Len(LegacyWildcardNamespaceID)) - require.Equal(t, "Keyword01", saTranslation.Get(LegacyWildcardNamespaceID, "CustomKeywordField")) - require.Equal(t, "Text01", saTranslation.Get(LegacyWildcardNamespaceID, "CustomStringField")) + wantValidateErr: []string{ + `namespaceMappings[name="migration-namespace"]`, + "has no namespaceId", }, }, { @@ -508,13 +501,12 @@ func TestSATranslationConfigValidate(t *testing.T) { }, verify: func(t *testing.T, saTranslation SearchAttributeTranslation) { require.Equal(t, 2, saTranslation.LenNamespaces()) - require.False(t, saTranslation.HasLegacyWildcard()) require.Equal(t, "remoteOne", saTranslation.Get("namespace-id-1", "localOne")) require.Equal(t, "remoteTwo", saTranslation.Get("namespace-id-2", "localTwo")) // Each namespace only knows its own attributes. require.Equal(t, "", saTranslation.Get("namespace-id-1", "localTwo")) require.Equal(t, "", saTranslation.Get("namespace-id-2", "localOne")) - require.Equal(t, NewTuple("", false), NewTuple(saTranslation.GetExists(LegacyWildcardNamespaceID, "localOne"))) + require.Equal(t, NewTuple("", false), NewTuple(saTranslation.GetExists("", "localOne"))) require.Equal(t, "localOne", saTranslation.Inverse().Get("namespace-id-1", "remoteOne")) require.Equal(t, "localTwo", saTranslation.Inverse().Get("namespace-id-2", "remoteTwo")) }, diff --git a/interceptor/search_attribute_translator.go b/interceptor/search_attribute_translator.go index a1ac0440..df6e661a 100644 --- a/interceptor/search_attribute_translator.go +++ b/interceptor/search_attribute_translator.go @@ -66,18 +66,9 @@ func (s *saTranslator) TranslateResponse(req, resp any) (bool, error) { func newSAMatcherResolver(nsMappings map[string]map[string]string) saMatcherResolver { matchers := createStringMatchers(nsMappings) - // Legacy configs express a single mapping keyed by an empty namespace id, meaning "apply to - // every namespace". Mirrors config.LegacyWildcardNamespaceID. - wildcard, hasWildcard := matchers[""] - return func(nsID string) (stringMatcher, bool) { - if match, ok := matchers[nsID]; ok { - return match, true - } - if hasWildcard { - return wildcard, true - } - return nil, false + match, ok := matchers[nsID] + return match, ok } } diff --git a/interceptor/search_attribute_translator_test.go b/interceptor/search_attribute_translator_test.go index 0b6b2c2f..c9ec083f 100644 --- a/interceptor/search_attribute_translator_test.go +++ b/interceptor/search_attribute_translator_test.go @@ -530,51 +530,45 @@ func TestTranslateSearchAttributeRawHistoryUsesPairedRequest(t *testing.T) { }) } -func TestTranslateSearchAttributeLegacyWildcard(t *testing.T) { - // A sole mapping keyed by an empty namespace id means "every namespace". Deployed configs - // rely on this, so it must keep behaving exactly as it did before per-namespace support. +func TestTranslateSearchAttributeEmptyNamespaceIdMatchesNothing(t *testing.T) { + // There is no mapping that applies to every namespace: an entry keyed by an empty namespace + // id matches only an unresolved namespace, which no well formed replication task produces. + // config.SATranslationConfig.Validate rejects such a config before it gets this far; this is + // the second line of defence, and it fails if a wildcard fallback is ever reintroduced. tr := newTestSATranslator(t, map[string]map[string]string{"": {testSAName: keywordA}}) frame := makeMultiNamespaceFrame(testSAName) changed, err := tr.TranslateResponse(nil, frame) require.NoError(t, err) - require.True(t, changed) + require.False(t, changed) tasks := frame.GetMessages().GetReplicationTasks() require.Len(t, tasks, 4) - - nsATask := tasks[0].GetHistoryTaskAttributes() - require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetEvents())) - require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, nsATask.GetNewRunEvents())) - require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, tasks[1].GetHistoryTaskAttributes().GetEvents())) - require.Equal(t, []string{keywordA, keywordA}, blobSAKeys(t, tasks[3].GetHistoryTaskAttributes().GetEvents()), - "the wildcard mapping applies even to namespaces with no entry of their own") - - execInfo := tasks[2].GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo() - require.Equal(t, []string{keywordA}, mapKeys(execInfo.GetSearchAttributes())) - require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetMemo()), "Memo must not be rewritten") + for i, task := range tasks { + if hta := task.GetHistoryTaskAttributes(); hta != nil { + require.Equal(t, []string{testSAName, testSAName}, blobSAKeys(t, hta.GetEvents()), + "task %d must be untouched", i) + continue + } + execInfo := task.GetSyncWorkflowStateTaskAttributes().GetWorkflowState().GetExecutionInfo() + require.Equal(t, []string{testSAName}, mapKeys(execInfo.GetSearchAttributes()), + "task %d must be untouched", i) + } } - func TestTranslateSearchAttributeUnsupportedFieldTypes(t *testing.T) { // Add- and RemoveSearchAttributesRequest name their fields SearchAttributes too, but hold a - // map[string]enums.IndexedValueType and a []string. They must be skipped, not treated as an - // error that aborts translation of the message. + // map[string]enums.IndexedValueType and a []string, so neither must be treated as an error + // that aborts translation of the whole message. // - // The two configs reach that outcome down different paths, so both are needed: - // - namespace keyed: neither request type is enclosed by a namespace owner, so the - // namespace resolves to "", no matcher resolves, and the field is skipped before the - // type switch runs. The unsupported-type branch is unreachable in this configuration. - // - legacy wildcard: the wildcard matcher resolves for every namespace, so the type - // switch does run and its unsupported-type branch is what prevents the error. This is - // the configuration deployed proxies use, which is what makes that branch load-bearing. - configs := map[string]map[string]map[string]string{ - "namespace keyed": testSAMappings(), - "legacy wildcard": {"": {testSAName: keywordA}}, - } - - for name, nsMappings := range configs { - t.Run(name, func(t *testing.T) { - tr := newTestSATranslator(t, nsMappings) + // Neither request type is enclosed by a namespace owner, so the namespace resolves to "", + // no matcher resolves, and the field is skipped before the type switch runs. That makes + // visitSearchAttributes' unsupported-type branch defensive rather than load-bearing: it + // only matters if a future message carries an odd SearchAttributes type *inside* a + // namespace owner. Skipping is still the right outcome there, which is why it warns and + // continues instead of returning visit.Stop. + { + t.Run("namespace keyed", func(t *testing.T) { + tr := newTestSATranslator(t, testSAMappings()) addReq := &adminservice.AddSearchAttributesRequest{ SearchAttributes: map[string]enums.IndexedValueType{ diff --git a/proxy/cluster_connection.go b/proxy/cluster_connection.go index 6a578496..04b858c7 100644 --- a/proxy/cluster_connection.go +++ b/proxy/cluster_connection.go @@ -388,12 +388,6 @@ func makeServerOptions(c serverConfiguration, tlsConfig encryption.TLSConfig) ([ if c.saTranslations.LenNamespaces() > 0 { c.loggers.Get(LogClusterConnection).Info("search attribute translation enabled", tag.NewAnyTag("mappings", c.saTranslations)) - if c.saTranslations.HasLegacyWildcard() { - // Validated as the sole mapping (see config.SATranslationConfig.Validate), so it is applied - // to every namespace. Set namespaceId to translate more than one namespace concurrently. - c.loggers.Get(LogClusterConnection).Warn("DEPRECATED: searchAttributeTranslation has a " + - "namespaceMappings entry with an empty namespaceId; its mappings are applied to every namespace") - } translators = append(translators, interceptor.NewSearchAttributeTranslator(c.loggers.Get(LogInterceptor), c.saTranslations.FlattenMaps(), c.saTranslations.Inverse().FlattenMaps())) } diff --git a/proxy/cluster_connection_test.go b/proxy/cluster_connection_test.go index d286826a..3ad8dd7a 100644 --- a/proxy/cluster_connection_test.go +++ b/proxy/cluster_connection_test.go @@ -396,11 +396,11 @@ func TestMakeServerOptionsMultiNamespaceSATranslation(t *testing.T) { ), }, { - // The shape shipped by configs that predate per-namespace translation. - name: "legacy wildcard, empty namespaceId", + name: "one namespace", saTranslator: saMappings(config.SANamespaceMapping{ - Name: "ns-a", - Mappings: []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}}, + Name: "ns-a", + NamespaceId: "11111111-1111-1111-1111-111111111111", + Mappings: []config.SAMapping{{LocalName: "Keyword01", RemoteName: "TestSA"}}, }), }, } {