fix tproxy sameness groups (#17468)

agent/consul/state/config_entry_intention.go

@@ -79,6 +79,122 @@ func (s *ServiceIntentionLegacyIDIndex) PrefixFromArgs(args ...interface{}) ([]b
 	return val, nil
 }
 
+type SamenessGroupMemberIndex struct {
+}
+
+// Compile-time assert that these interfaces hold to ensure that the
+// methods correctly exist across the oss/ent split.
+var _ memdb.Indexer = (*SamenessGroupMemberIndex)(nil)
+var _ memdb.MultiIndexer = (*SamenessGroupMemberIndex)(nil)
+
+func (s *SamenessGroupMemberIndex) FromObject(obj interface{}) (bool, [][]byte, error) {
+	entry, ok := obj.(structs.ConfigEntry)
+	if !ok {
+		return false, nil, fmt.Errorf("object is not a ConfigEntry")
+	}
+
+	sg, ok := entry.(*structs.SamenessGroupConfigEntry)
+	if !ok {
+		return false, nil, nil
+	}
+
+	vals := make([][]byte, 0)
+	for _, m := range sg.AllMembers() {
+		if m.Partition == "" {
+			continue
+		}
+
+		// add 1 for null separator after each string
+		buf := newIndexBuilder(len(m.Partition) + 1)
+		buf.String(m.Partition)
+		vals = append(vals, buf.Bytes())
+	}
+
+	if len(vals) == 0 {
+		return false, nil, nil
+	}
+
+	return true, vals, nil
+}
+
+func (s *SamenessGroupMemberIndex) FromArgs(args ...interface{}) ([]byte, error) {
+	if len(args) != 1 {
+		return nil, fmt.Errorf("must provide only a single argument")
+	}
+	arg, ok := args[0].(string)
+	if !ok {
+		return nil, fmt.Errorf("argument must be a string: %#v", args[0])
+	}
+	buf := newIndexBuilder(len(arg) + 1)
+	buf.String(arg)
+	// Add the null character as a terminator
+	return buf.Bytes(), nil
+}
+
+type ServiceIntentionSourceSamenessGroupIndex struct {
+}
+
+// Compile-time assert that these interfaces hold to ensure that the
+// methods correctly exist across the oss/ent split.
+var _ memdb.Indexer = (*ServiceIntentionSourceSamenessGroupIndex)(nil)
+var _ memdb.MultiIndexer = (*ServiceIntentionSourceSamenessGroupIndex)(nil)
+
+func (s *ServiceIntentionSourceSamenessGroupIndex) FromObject(obj interface{}) (bool, [][]byte, error) {
+	entry, ok := obj.(structs.ConfigEntry)
+	if !ok {
+		return false, nil, fmt.Errorf("object is not a ConfigEntry")
+	}
+
+	ixnEntry, ok := entry.(*structs.ServiceIntentionsConfigEntry)
+	if !ok {
+		return false, nil, nil
+	}
+
+	vals := make([][]byte, 0, len(ixnEntry.Sources))
+	for _, src := range ixnEntry.Sources {
+		sg := src.SamenessGroup
+		if sg == "" {
+			continue
+		}
+
+		sn := structs.ServiceName{
+			Name:           src.Name,
+			EnterpriseMeta: acl.NewEnterpriseMetaWithPartition(ixnEntry.PartitionOrDefault(), src.NamespaceOrDefault()),
+		}.String()
+
+		// add 2 for null separator after each string
+		buf := newIndexBuilder(len(sg) + len(sn) + 2)
+		buf.String(sg)
+		buf.String(sn)
+		vals = append(vals, buf.Bytes())
+	}
+
+	if len(vals) == 0 {
+		return false, nil, nil
+	}
+
+	return true, vals, nil
+}
+
+func (s *ServiceIntentionSourceSamenessGroupIndex) FromArgs(args ...interface{}) ([]byte, error) {
+	if len(args) != 1 {
+		return nil, fmt.Errorf("must provide only a single argument")
+	}
+	arg, ok := args[0].(structs.ServiceNameWithSamenessGroup)
+	if !ok {
+		return nil, fmt.Errorf("argument must be a structs.ServiceID: %#v", args[0])
+	}
+	// Intention queries cannot use a peered service as a source
+	sg := arg.SamenessGroup
+	sn := arg.ServiceName.String()
+	// add 2 for null separator after each string
+	buf := newIndexBuilder(len(sg) + len(sn) + 2)
+	buf.String(sg)
+	buf.String(sn)
+	// Add the null character as a terminator
+	return buf.Bytes(), nil
+}
+
 type ServiceIntentionSourceIndex struct {
 }
 
@@ -104,6 +220,10 @@ func (s *ServiceIntentionSourceIndex) FromObject(obj interface{}) (bool, [][]byt
 
 	vals := make([][]byte, 0, len(ixnEntry.Sources))
 	for _, src := range ixnEntry.Sources {
+		if src.SamenessGroup != "" {
+			continue
+		}
+
 		peer := src.Peer
 		if peer == "" {
 			peer = structs.LocalPeerKeyword
@@ -282,6 +402,11 @@ func readSourceIntentionsFromConfigEntriesTxn(
 		if err != nil {
 			return 0, nil, err
 		}
+
+		results, err = readSourceSamenessIntentionsFromConfigEntriesForServiceTxn(tx, ws, sn.Name, &sn.EnterpriseMeta, results, targetType)
+		if err != nil {
+			return 0, nil, err
+		}
 	}
 
 	// Sort the results by precedence
@@ -294,12 +419,11 @@ func readSourceIntentionsFromConfigEntriesForServiceTxn(
 	tx ReadTxn,
 	ws memdb.WatchSet,
 	serviceName string,
-	entMeta *acl.EnterpriseMeta,
+	sourceEntMeta *acl.EnterpriseMeta,
 	results structs.Intentions,
 	targetType structs.IntentionTargetType,
 ) (structs.Intentions, error) {
-	sn := structs.NewServiceName(serviceName, entMeta)
-
+	sn := structs.NewServiceName(serviceName, sourceEntMeta)
 	iter, err := tx.Get(tableConfigEntries, indexSource, sn)
 	if err != nil {
 		return nil, fmt.Errorf("failed config entry lookup: %s", err)
@@ -309,29 +433,21 @@ func readSourceIntentionsFromConfigEntriesForServiceTxn(
 	for v := iter.Next(); v != nil; v = iter.Next() {
 		entry := v.(*structs.ServiceIntentionsConfigEntry)
 		entMeta := entry.DestinationServiceName().EnterpriseMeta
-		// if we have a wildcard namespace or partition assume we are querying a service intention
-		// as destination intentions will never be queried as wildcard
-		kind := structs.GatewayServiceKindService
-		if entMeta.NamespaceOrDefault() != acl.WildcardName && entMeta.PartitionOrDefault() != acl.WildcardName {
-			kind, err = GatewayServiceKind(tx, entry.DestinationServiceName().Name, &entMeta)
-			if err != nil {
-				return nil, err
-			}
+
+		kind, err := serviceIntentionsToGatewayServiceKind(tx, entry.DestinationServiceName().Name, entMeta)
+		if err != nil {
+			return nil, err
 		}
+
 		for _, src := range entry.Sources {
 			if src.SourceServiceName() == sn {
-				switch targetType {
-				case structs.IntentionTargetService:
-					if kind == structs.GatewayServiceKindService || kind == structs.GatewayServiceKindUnknown {
-						results = append(results, entry.ToIntention(src))
-					}
-				case structs.IntentionTargetDestination:
-					// wildcard is needed here to be able to consider destinations in the wildcard intentions
-					if kind == structs.GatewayServiceKindDestination || entry.HasWildcardDestination() {
-						results = append(results, entry.ToIntention(src))
-					}
-				default:
-					return nil, fmt.Errorf("invalid target type")
+				canAdd, err := intentionMatches(targetType, kind, entry.HasWildcardDestination())
+				if err != nil {
+					return nil, err
+				}
+
+				if canAdd {
+					results = append(results, entry.ToIntention(src))
 				}
 			}
 		}
@@ -340,6 +456,37 @@ func readSourceIntentionsFromConfigEntriesForServiceTxn(
 	return results, nil
 }
 
+func serviceIntentionsToGatewayServiceKind(tx ReadTxn, serviceName string, entMeta acl.EnterpriseMeta) (structs.GatewayServiceKind, error) {
+	var err error
+	kind := structs.GatewayServiceKindService
+
+	// if we have a wildcard namespace or partition assume we are querying a service intention
+	// as destination intentions will never be queried as wildcard
+	if entMeta.NamespaceOrDefault() != acl.WildcardName && entMeta.PartitionOrDefault() != acl.WildcardName {
+		kind, err = GatewayServiceKind(tx, serviceName, &entMeta)
+		if err != nil {
+			return kind, err
+		}
+	}
+
+	return kind, nil
+}
+
+func intentionMatches(targetType structs.IntentionTargetType, kind structs.GatewayServiceKind, wildcardDestination bool) (bool, error) {
+	var canAdd bool
+	switch targetType {
+	case structs.IntentionTargetService:
+		canAdd = kind == structs.GatewayServiceKindService || kind == structs.GatewayServiceKindUnknown
+	case structs.IntentionTargetDestination:
+		// wildcard is needed here to be able to consider destinations in the wildcard intentions
+		canAdd = kind == structs.GatewayServiceKindDestination || wildcardDestination
+	default:
+		return false, fmt.Errorf("invalid target type")
+	}
+
+	return canAdd, nil
+}
+
 func readDestinationIntentionsFromConfigEntriesTxn(tx ReadTxn, ws memdb.WatchSet, serviceName string, entMeta *acl.EnterpriseMeta) (uint64, structs.Intentions, error) {
 	idx := maxIndexTxn(tx, tableConfigEntries)
 

agent/consul/state/config_entry_oss.go

@@ -76,3 +76,14 @@ func getExportedServicesMatchServiceNames(serviceName string, entMeta *acl.Enter
 		structs.NewServiceName(structs.WildcardSpecifier, entMeta),
 	}
 }
+
+func readSourceSamenessIntentionsFromConfigEntriesForServiceTxn(
+	tx ReadTxn,
+	ws memdb.WatchSet,
+	serviceName string,
+	sourceEntMeta *acl.EnterpriseMeta,
+	results structs.Intentions,
+	targetType structs.IntentionTargetType,
+) (structs.Intentions, error) {
+	return results, nil
+}

agent/consul/state/config_entry_schema.go

@@ -17,6 +17,8 @@ const (
 	indexLink                 = "link"
 	indexIntentionLegacyID    = "intention-legacy-id"
 	indexSource               = "intention-source"
+	indexSourceSamenessGroup  = "intention-source-sameness-group"
+	indexSamenessGroupMember  = "sameness-group-member"
 	indexSamenessGroupDefault = "sameness-group-default-for-failover"
 )
 
@@ -54,6 +56,18 @@ func configTableSchema() *memdb.TableSchema {
 				Unique:       false,
 				Indexer:      &ServiceIntentionSourceIndex{},
 			},
+			indexSourceSamenessGroup: {
+				Name:         indexSourceSamenessGroup,
+				AllowMissing: true,
+				Unique:       false,
+				Indexer:      &ServiceIntentionSourceSamenessGroupIndex{},
+			},
+			indexSamenessGroupMember: {
+				Name:         indexSamenessGroupMember,
+				AllowMissing: true,
+				Unique:       false,
+				Indexer:      &SamenessGroupMemberIndex{},
+			},
 			indexSamenessGroupDefault: {
 				Name:         indexSamenessGroupDefault,
 				AllowMissing: true,

agent/consul/state/intention.go

@@ -829,18 +829,37 @@ func (s *Store) IntentionMatch(ws memdb.WatchSet, args *structs.IntentionQueryMa
 	var out []structs.Intentions
 	for i, ixns := range ixnsList {
 		entry := args.Entries[i]
-		idx, simplifiedIxns, err := getSimplifiedIntentions(tx, ws, ixns, *entry.GetEnterpriseMeta())
+		idx, simplifiedIxns, err := getSimplifiedIntentions(tx, ws, ixns)
 		if err != nil {
 			return 0, nil, err
 		}
 		if idx > maxIdx {
 			maxIdx = idx
 		}
-		out = append(out, simplifiedIxns)
+
+		filteredIxns := filterIntentionsMatching(simplifiedIxns, args.Type, entry.GetEnterpriseMeta().PartitionOrDefault())
+
+		out = append(out, filteredIxns)
 	}
+
 	return maxIdx, out, nil
 }
 
+func filterIntentionsMatching(ixns structs.Intentions, matchType structs.IntentionMatchType, partition string) structs.Intentions {
+	var filteredIxns structs.Intentions
+	if matchType == structs.IntentionMatchSource {
+		for _, ixn := range ixns {
+			if partition == ixn.SourcePartitionOrDefault() {
+				filteredIxns = append(filteredIxns, ixn)
+			}
+		}
+	} else {
+		filteredIxns = ixns
+	}
+
+	return filteredIxns
+}
+
 func (s *Store) legacyIntentionMatchTxn(tx ReadTxn, ws memdb.WatchSet, args *structs.IntentionQueryMatch) (uint64, []structs.Intentions, error) {
 	// Get the table index.
 	idx := maxIndexTxn(tx, tableConnectIntentions)
@@ -909,15 +928,18 @@ func compatIntentionMatchOneTxn(
 		return 0, nil, err
 	}
 
-	idx, simplifiedIxns, err := getSimplifiedIntentions(tx, ws, ixns, *entry.GetEnterpriseMeta())
+	idx, simplifiedIxns, err := getSimplifiedIntentions(tx, ws, ixns)
 	if err != nil {
 		return 0, nil, err
 	}
+
 	if idx > maxIdx {
 		maxIdx = idx
 	}
 
-	return maxIdx, structs.SimplifiedIntentions(simplifiedIxns), nil
+	filteredIxns := filterIntentionsMatching(simplifiedIxns, matchType, entry.GetEnterpriseMeta().PartitionOrDefault())
+
+	return maxIdx, structs.SimplifiedIntentions(filteredIxns), nil
 }
 
 func legacyIntentionMatchOneTxn(

agent/consul/state/intention_oss.go

@@ -22,7 +22,6 @@ func getSimplifiedIntentions(
 	tx ReadTxn,
 	ws memdb.WatchSet,
 	ixns structs.Intentions,
-	entMeta acl.EnterpriseMeta,
 ) (uint64, structs.Intentions, error) {
 	return 0, ixns, nil
 }

agent/structs/intention_oss.go

@@ -78,3 +78,7 @@ func (ixn *Intention) FillPartitionAndNamespace(entMeta *acl.EnterpriseMeta, fil
 	ixn.SourcePartition = ""
 	ixn.DestinationPartition = ""
 }
+
+func (ixn *Intention) SourcePartitionOrDefault() string {
+	return "default"
+}

agent/structs/structs.go

@@ -2321,6 +2321,11 @@ func (psn PeeredServiceName) String() string {
 	return fmt.Sprintf("%v:%v", psn.ServiceName.String(), psn.Peer)
 }
 
+type ServiceNameWithSamenessGroup struct {
+	SamenessGroup string
+	ServiceName
+}
+
 type ServiceName struct {
 	Name               string
 	acl.EnterpriseMeta `mapstructure:",squash"`